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-alpha4
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 a47eb4c Dawsh 2y ago from config import SUDOERS
16 f7af2b3 Dawsh 2y ago
17 5bb2712 Mardin 2y ago router = APIRouter(tags=["Admin"], prefix="/admins")
18 a47eb4c Dawsh 2y ago
19 a47eb4c Dawsh 2y ago
20 a47eb4c Dawsh 2y ago def authenticate_env_sudo(username: str, password: str) -> bool:
21 a47eb4c Dawsh 2y ago try:
22 a47eb4c Dawsh 2y ago return password == SUDOERS[username]
23 a47eb4c Dawsh 2y ago except KeyError:
24 a47eb4c Dawsh 2y ago return False
25 a47eb4c Dawsh 2y ago
26 a47eb4c Dawsh 2y ago
27 a47eb4c Dawsh 2y ago def authenticate_admin(db: Session, username: str, password: str) -> Optional[Admin]:
28 a47eb4c Dawsh 2y ago dbadmin = crud.get_admin(db, username)
29 a47eb4c Dawsh 2y ago if not dbadmin:
30 a47eb4c Dawsh 2y ago return None
31 a47eb4c Dawsh 2y ago
32 01cdf2e Dawsh 2y ago return (
33 01cdf2e Dawsh 2y ago dbadmin if AdminInDB.model_validate(dbadmin).verify_password(password) else None
34 01cdf2e Dawsh 2y ago )
35 a47eb4c Dawsh 2y ago
36 a47eb4c Dawsh 2y ago
37 d4ec019 Dawsh 2y ago @router.get("", response_model=Page[Admin])
38 01cdf2e Dawsh 2y ago def get_admins(db: DBDep, admin: SudoAdminDep, username: str = None):
39 d4ec019 Dawsh 2y ago query = db.query(DBAdmin)
40 d4ec019 Dawsh 2y ago if username:
41 01cdf2e Dawsh 2y ago query = query.filter(DBAdmin.username.ilike(f"%{username}%"))
42 d4ec019 Dawsh 2y ago return paginate(db, query)
43 d4ec019 Dawsh 2y ago
44 d4ec019 Dawsh 2y ago
45 d4ec019 Dawsh 2y ago @router.post("", response_model=Admin)
46 01cdf2e Dawsh 2y ago def create_admin(new_admin: AdminCreate, db: DBDep, admin: SudoAdminDep):
47 d4ec019 Dawsh 2y ago try:
48 d4ec019 Dawsh 2y ago dbadmin = crud.create_admin(db, new_admin)
49 d4ec019 Dawsh 2y ago except sqlalchemy.exc.IntegrityError:
50 d4ec019 Dawsh 2y ago db.rollback()
51 d4ec019 Dawsh 2y ago raise HTTPException(status_code=409, detail="Admin already exists")
52 d4ec019 Dawsh 2y ago
53 d4ec019 Dawsh 2y ago return dbadmin
54 d4ec019 Dawsh 2y ago
55 d4ec019 Dawsh 2y ago
56 d4ec019 Dawsh 2y ago @router.get("/current", response_model=Admin)
57 d4ec019 Dawsh 2y ago def get_current_admin(admin: AdminDep):
58 d4ec019 Dawsh 2y ago return admin
59 d4ec019 Dawsh 2y ago
60 d4ec019 Dawsh 2y ago
61 2de9d72 Dawsh 2y ago @router.post("/token", response_model=Token)
62 01cdf2e Dawsh 2y ago def admin_token(form_data: Annotated[OAuth2PasswordRequestForm, Depends()], db: DBDep):
63 a47eb4c Dawsh 2y ago if authenticate_env_sudo(form_data.username, form_data.password):
64 a47eb4c Dawsh 2y ago return Token(access_token=create_admin_token(form_data.username, is_sudo=True))
65 a47eb4c Dawsh 2y ago
66 a47eb4c Dawsh 2y ago if dbadmin := authenticate_admin(db, form_data.username, form_data.password):
67 01cdf2e Dawsh 2y ago return Token(
68 01cdf2e Dawsh 2y ago access_token=create_admin_token(form_data.username, is_sudo=dbadmin.is_sudo)
69 01cdf2e Dawsh 2y ago )
70 a47eb4c Dawsh 2y ago
71 a47eb4c Dawsh 2y ago raise HTTPException(
72 a47eb4c Dawsh 2y ago status_code=status.HTTP_401_UNAUTHORIZED,
73 a47eb4c Dawsh 2y ago detail="Incorrect username or password",
74 a47eb4c Dawsh 2y ago headers={"WWW-Authenticate": "Bearer"},
75 a47eb4c Dawsh 2y ago )
76 a47eb4c Dawsh 2y ago
77 a47eb4c Dawsh 2y ago
78 2de9d72 Dawsh 2y ago @router.put("/{username}", response_model=Admin)
79 01cdf2e Dawsh 2y ago def modify_admin(
80 01cdf2e Dawsh 2y ago username: str, modified_admin: AdminModify, db: DBDep, admin: AdminDep
81 01cdf2e Dawsh 2y ago ):
82 a47eb4c Dawsh 2y ago if not (admin.is_sudo or admin.username == username):
83 a47eb4c Dawsh 2y ago raise HTTPException(status_code=403, detail="You're not allowed")
84 a47eb4c Dawsh 2y ago
85 a47eb4c Dawsh 2y ago # If a non-sudoer admin is making itself a sudoer
86 a47eb4c Dawsh 2y ago if (admin.username == username) and (modified_admin.is_sudo and not admin.is_sudo):
87 a47eb4c Dawsh 2y ago raise HTTPException(status_code=403, detail="You can't make yourself sudoer!")
88 a47eb4c Dawsh 2y ago
89 a47eb4c Dawsh 2y ago dbadmin = crud.get_admin(db, username)
90 a47eb4c Dawsh 2y ago if not dbadmin:
91 a47eb4c Dawsh 2y ago raise HTTPException(status_code=404, detail="Admin not found")
92 a47eb4c Dawsh 2y ago
93 a47eb4c Dawsh 2y ago # If a sudoer admin wants to edit another sudoer
94 a47eb4c Dawsh 2y ago if (username != admin.username) and dbadmin.is_sudo:
95 a47eb4c Dawsh 2y ago raise HTTPException(
96 a47eb4c Dawsh 2y ago status_code=403,
97 c13a1aa Dawsh 2y ago detail="You're not allowed to edit another sudoers account. Use marzban-cli instead.",
98 a47eb4c Dawsh 2y ago )
99 a47eb4c Dawsh 2y ago
100 a47eb4c Dawsh 2y ago dbadmin = crud.update_admin(db, dbadmin, modified_admin)
101 a47eb4c Dawsh 2y ago return dbadmin
102 a47eb4c Dawsh 2y ago
103 a47eb4c Dawsh 2y ago
104 2de9d72 Dawsh 2y ago @router.delete("/{username}")
105 01cdf2e Dawsh 2y ago def remove_admin(username: str, db: DBDep, admin: SudoAdminDep):
106 a47eb4c Dawsh 2y ago dbadmin = crud.get_admin(db, username)
107 a47eb4c Dawsh 2y ago if not dbadmin:
108 a47eb4c Dawsh 2y ago raise HTTPException(status_code=404, detail="Admin not found")
109 a47eb4c Dawsh 2y ago
110 a47eb4c Dawsh 2y ago if dbadmin.is_sudo:
111 a47eb4c Dawsh 2y ago raise HTTPException(
112 a47eb4c Dawsh 2y ago status_code=403,
113 c13a1aa Dawsh 2y ago detail="You're not allowed to delete sudoers accounts. Use marzban-cli instead.",
114 a47eb4c Dawsh 2y ago )
115 a47eb4c Dawsh 2y ago
116 c13a1aa Dawsh 2y ago crud.remove_admin(db, dbadmin)
117 a47eb4c Dawsh 2y ago return {}