gitscope
/
← marzneshin / app/routes/admin.py / blame
Files Commits 446 Branches 1 Tags 43 Markers 4 README License
https://dev.kronoz.win/r/marzneshin.git
app/routes/admin.py · v0.3.0
Source
1 d4ec019 Dawsh 2y ago from typing import Optional, Annotated
2 a47eb4c Dawsh 2y ago
3 a47eb4c Dawsh 2y ago import sqlalchemy
4 f7af2b3 Dawsh 2y ago from fastapi import APIRouter
5 f7af2b3 Dawsh 2y ago from fastapi import Depends, HTTPException, status
6 f7af2b3 Dawsh 2y ago from fastapi.security import OAuth2PasswordRequestForm
7 d4ec019 Dawsh 2y ago from fastapi_pagination import Page
8 d4ec019 Dawsh 2y ago from fastapi_pagination.ext.sqlalchemy import paginate
9 f7af2b3 Dawsh 2y ago
10 c13a1aa Dawsh 2y ago from app.db import Session, crud
11 d4ec019 Dawsh 2y ago from app.db.models import Admin as DBAdmin
12 f7af2b3 Dawsh 2y ago from app.dependencies import AdminDep, SudoAdminDep, DBDep
13 a47eb4c Dawsh 2y ago from app.models.admin import Admin, AdminCreate, AdminInDB, AdminModify, Token
14 a47eb4c Dawsh 2y ago from app.utils.jwt import create_admin_token
15 f7af2b3 Dawsh 2y ago
16 5bb2712 Mardin 2y ago router = APIRouter(tags=["Admin"], prefix="/admins")
17 a47eb4c Dawsh 2y ago
18 a47eb4c Dawsh 2y ago
19 a47eb4c Dawsh 2y ago def authenticate_admin(db: Session, username: str, password: str) -> Optional[Admin]:
20 a47eb4c Dawsh 2y ago dbadmin = crud.get_admin(db, username)
21 a47eb4c Dawsh 2y ago if not dbadmin:
22 a47eb4c Dawsh 2y ago return None
23 a47eb4c Dawsh 2y ago
24 01cdf2e Dawsh 2y ago return (
25 01cdf2e Dawsh 2y ago dbadmin if AdminInDB.model_validate(dbadmin).verify_password(password) else None
26 01cdf2e Dawsh 2y ago )
27 a47eb4c Dawsh 2y ago
28 a47eb4c Dawsh 2y ago
29 d4ec019 Dawsh 2y ago @router.get("", response_model=Page[Admin])
30 01cdf2e Dawsh 2y ago def get_admins(db: DBDep, admin: SudoAdminDep, username: str = None):
31 d4ec019 Dawsh 2y ago query = db.query(DBAdmin)
32 d4ec019 Dawsh 2y ago if username:
33 01cdf2e Dawsh 2y ago query = query.filter(DBAdmin.username.ilike(f"%{username}%"))
34 d4ec019 Dawsh 2y ago return paginate(db, query)
35 d4ec019 Dawsh 2y ago
36 d4ec019 Dawsh 2y ago
37 d4ec019 Dawsh 2y ago @router.post("", response_model=Admin)
38 01cdf2e Dawsh 2y ago def create_admin(new_admin: AdminCreate, db: DBDep, admin: SudoAdminDep):
39 d4ec019 Dawsh 2y ago try:
40 d4ec019 Dawsh 2y ago dbadmin = crud.create_admin(db, new_admin)
41 d4ec019 Dawsh 2y ago except sqlalchemy.exc.IntegrityError:
42 d4ec019 Dawsh 2y ago db.rollback()
43 d4ec019 Dawsh 2y ago raise HTTPException(status_code=409, detail="Admin already exists")
44 d4ec019 Dawsh 2y ago
45 d4ec019 Dawsh 2y ago return dbadmin
46 d4ec019 Dawsh 2y ago
47 d4ec019 Dawsh 2y ago
48 d4ec019 Dawsh 2y ago @router.get("/current", response_model=Admin)
49 d4ec019 Dawsh 2y ago def get_current_admin(admin: AdminDep):
50 d4ec019 Dawsh 2y ago return admin
51 d4ec019 Dawsh 2y ago
52 d4ec019 Dawsh 2y ago
53 2de9d72 Dawsh 2y ago @router.post("/token", response_model=Token)
54 01cdf2e Dawsh 2y ago def admin_token(form_data: Annotated[OAuth2PasswordRequestForm, Depends()], db: DBDep):
55 a47eb4c Dawsh 2y ago if dbadmin := authenticate_admin(db, form_data.username, form_data.password):
56 01cdf2e Dawsh 2y ago return Token(
57 01cdf2e Dawsh 2y ago access_token=create_admin_token(form_data.username, is_sudo=dbadmin.is_sudo)
58 01cdf2e Dawsh 2y ago )
59 a47eb4c Dawsh 2y ago
60 a47eb4c Dawsh 2y ago raise HTTPException(
61 a47eb4c Dawsh 2y ago status_code=status.HTTP_401_UNAUTHORIZED,
62 a47eb4c Dawsh 2y ago detail="Incorrect username or password",
63 a47eb4c Dawsh 2y ago headers={"WWW-Authenticate": "Bearer"},
64 a47eb4c Dawsh 2y ago )
65 a47eb4c Dawsh 2y ago
66 a47eb4c Dawsh 2y ago
67 2de9d72 Dawsh 2y ago @router.put("/{username}", response_model=Admin)
68 01cdf2e Dawsh 2y ago def modify_admin(
69 01cdf2e Dawsh 2y ago username: str, modified_admin: AdminModify, db: DBDep, admin: AdminDep
70 01cdf2e Dawsh 2y ago ):
71 a47eb4c Dawsh 2y ago if not (admin.is_sudo or admin.username == username):
72 a47eb4c Dawsh 2y ago raise HTTPException(status_code=403, detail="You're not allowed")
73 a47eb4c Dawsh 2y ago
74 a47eb4c Dawsh 2y ago # If a non-sudoer admin is making itself a sudoer
75 a47eb4c Dawsh 2y ago if (admin.username == username) and (modified_admin.is_sudo and not admin.is_sudo):
76 a47eb4c Dawsh 2y ago raise HTTPException(status_code=403, detail="You can't make yourself sudoer!")
77 a47eb4c Dawsh 2y ago
78 a47eb4c Dawsh 2y ago dbadmin = crud.get_admin(db, username)
79 a47eb4c Dawsh 2y ago if not dbadmin:
80 a47eb4c Dawsh 2y ago raise HTTPException(status_code=404, detail="Admin not found")
81 a47eb4c Dawsh 2y ago
82 a47eb4c Dawsh 2y ago # If a sudoer admin wants to edit another sudoer
83 a47eb4c Dawsh 2y ago if (username != admin.username) and dbadmin.is_sudo:
84 a47eb4c Dawsh 2y ago raise HTTPException(
85 a47eb4c Dawsh 2y ago status_code=403,
86 c13a1aa Dawsh 2y ago detail="You're not allowed to edit another sudoers account. Use marzban-cli instead.",
87 a47eb4c Dawsh 2y ago )
88 a47eb4c Dawsh 2y ago
89 a47eb4c Dawsh 2y ago dbadmin = crud.update_admin(db, dbadmin, modified_admin)
90 a47eb4c Dawsh 2y ago return dbadmin
91 a47eb4c Dawsh 2y ago
92 a47eb4c Dawsh 2y ago
93 2de9d72 Dawsh 2y ago @router.delete("/{username}")
94 01cdf2e Dawsh 2y ago def remove_admin(username: str, db: DBDep, admin: SudoAdminDep):
95 a47eb4c Dawsh 2y ago dbadmin = crud.get_admin(db, username)
96 a47eb4c Dawsh 2y ago if not dbadmin:
97 a47eb4c Dawsh 2y ago raise HTTPException(status_code=404, detail="Admin not found")
98 a47eb4c Dawsh 2y ago
99 a47eb4c Dawsh 2y ago if dbadmin.is_sudo:
100 a47eb4c Dawsh 2y ago raise HTTPException(
101 a47eb4c Dawsh 2y ago status_code=403,
102 c13a1aa Dawsh 2y ago detail="You're not allowed to delete sudoers accounts. Use marzban-cli instead.",
103 a47eb4c Dawsh 2y ago )
104 a47eb4c Dawsh 2y ago
105 c13a1aa Dawsh 2y ago crud.remove_admin(db, dbadmin)
106 a47eb4c Dawsh 2y ago return {}