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