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