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