76 lines
2.8 KiB
Python
76 lines
2.8 KiB
Python
from collections.abc import Callable, Generator
|
|
from typing import Annotated
|
|
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.config import Settings, get_settings
|
|
from app.core.security import decode_admin_access_token, decode_user_access_token
|
|
from app.db.session import get_db as _get_db
|
|
from app.models.entities import AdminUser
|
|
from app.models.enums import AdminRole
|
|
from app.models.user import User
|
|
|
|
_bearer = HTTPBearer(auto_error=False)
|
|
|
|
|
|
def get_db() -> Generator[Session, None, None]:
|
|
yield from _get_db()
|
|
|
|
|
|
SettingsDep = Annotated[Settings, Depends(get_settings)]
|
|
DbSession = Annotated[Session, Depends(get_db)]
|
|
|
|
|
|
def get_current_user(
|
|
credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(_bearer)],
|
|
db: DbSession,
|
|
settings: SettingsDep,
|
|
) -> User:
|
|
if credentials is None or credentials.scheme.lower() != "bearer":
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
|
|
user_id = decode_user_access_token(credentials.credentials, settings)
|
|
if user_id is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired token")
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
if not user or not user.is_active:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
|
|
return user
|
|
|
|
|
|
CurrentUser = Annotated[User, Depends(get_current_user)]
|
|
|
|
|
|
def get_current_admin(
|
|
credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(_bearer)],
|
|
db: DbSession,
|
|
settings: SettingsDep,
|
|
) -> AdminUser:
|
|
if credentials is None or credentials.scheme.lower() != "bearer":
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
|
|
admin_id = decode_admin_access_token(credentials.credentials, settings)
|
|
if admin_id is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired token")
|
|
admin = db.query(AdminUser).filter(AdminUser.id == admin_id).first()
|
|
if not admin:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Admin not found")
|
|
return admin
|
|
|
|
|
|
CurrentAdmin = Annotated[AdminUser, Depends(get_current_admin)]
|
|
|
|
|
|
def require_admin_roles(*roles: AdminRole) -> Callable:
|
|
allowed = set(roles)
|
|
|
|
def _checker(admin: CurrentAdmin) -> AdminUser:
|
|
if admin.role not in allowed:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Insufficient permissions")
|
|
return admin
|
|
|
|
return _checker
|
|
|
|
|
|
EditorAdmin = Annotated[AdminUser, Depends(require_admin_roles(AdminRole.editor, AdminRole.super_admin))]
|
|
SuperAdmin = Annotated[AdminUser, Depends(require_admin_roles(AdminRole.super_admin))]
|