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