1from datetime import datetime
  2from typing import Annotated
  3
  4from fastapi import Depends, HTTPException, status
  5from sqlalchemy.orm import Session
  6
  7from app.db import crud, User, GetDB
  8from app.db.models import Service
  9from app.models.admin import Admin, oauth2_scheme
 10from app.utils.auth import get_admin_payload
 11
 12
 13def get_db():
 14    with GetDB() as db:
 15        yield db
 16
 17
 18def get_admin(
 19    db: Annotated[Session, Depends(get_db)],
 20    token: Annotated[str, Depends(oauth2_scheme)],
 21):
 22    payload = get_admin_payload(token)
 23    if not payload:
 24        return
 25
 26    dbadmin = crud.get_admin(db, payload["username"])
 27    if not dbadmin:
 28        return
 29
 30    if dbadmin.password_reset_at:
 31        created_at = payload.get("created_at")
 32        if not created_at or dbadmin.password_reset_at > created_at:
 33            return
 34
 35    if not dbadmin.is_sudo and not dbadmin.enabled:
 36        return
 37
 38    return Admin.model_validate(dbadmin)
 39
 40
 41def get_current_admin(admin: Annotated[Admin, Depends(get_admin)]):
 42    if not admin:
 43        raise HTTPException(
 44            status_code=status.HTTP_401_UNAUTHORIZED,
 45            detail="Could not validate credentials",
 46            headers={"WWW-Authenticate": "Bearer"},
 47        )
 48
 49    return admin
 50
 51
 52def sudo_admin(admin: Annotated[Admin, Depends(get_current_admin)]):
 53    if not admin.is_sudo:
 54        raise HTTPException(
 55            status_code=status.HTTP_403_FORBIDDEN,
 56            detail="Access Denied",
 57        )
 58    return admin
 59
 60
 61def get_subscription_user(
 62    username: str, key: str, db: Annotated[Session, Depends(get_db)]
 63):
 64    try:
 65        int(key, 16)
 66    except ValueError:
 67        raise HTTPException(status_code=404)
 68
 69    db_user = crud.get_user(db, username)
 70    if db_user and db_user.key == key:
 71        return db_user
 72    else:
 73        raise HTTPException(status_code=404)
 74
 75
 76def get_user(
 77    username: str,
 78    admin: Annotated[Admin, Depends(get_current_admin)],
 79    db: Annotated[Session, Depends(get_db)],
 80):
 81    db_user = crud.get_user(db, username)
 82    if not (
 83        admin.is_sudo or (db_user and db_user.admin.username == admin.username)
 84    ):
 85        raise HTTPException(status_code=403, detail="You're not allowed")
 86
 87    if not db_user:
 88        raise HTTPException(status_code=404, detail="User not found")
 89
 90    return db_user
 91
 92
 93def user_modification_access(
 94    admin: Annotated[Admin, Depends(get_current_admin)],
 95):
 96    if not admin.is_sudo and not admin.modify_users_access:
 97        raise HTTPException(status_code=403, detail="You're not allowed")
 98
 99
100def parse_start_date(start: str | None = None):
101    if not start:
102        return datetime.fromtimestamp(
103            datetime.utcnow().timestamp() - 30 * 24 * 3600
104        )
105    else:
106        return datetime.fromisoformat(start)
107
108
109def parse_end_date(end: str | None = None):
110    if not end:
111        return datetime.utcnow()
112    else:
113        return datetime.fromisoformat(end)
114
115
116def get_service(id: int, db: Annotated[Session, Depends(get_db)]):
117    service = crud.get_service(db, id)
118    if not service:
119        raise HTTPException(status_code=404, detail="Service not found")
120    return service
121
122
123SubUserDep = Annotated[User, Depends(get_subscription_user)]
124UserDep = Annotated[User, Depends(get_user)]
125AdminDep = Annotated[Admin, Depends(get_current_admin)]
126SudoAdminDep = Annotated[Admin, Depends(sudo_admin)]
127DBDep = Annotated[Session, Depends(get_db)]
128StartDateDep = Annotated[datetime, Depends(parse_start_date)]
129EndDateDep = Annotated[datetime, Depends(parse_end_date)]
130ModifyUsersAccess = Annotated[None, Depends(user_modification_access)]
131ServiceDep = Annotated[Service, Depends(get_service)]