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