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