| """Centralized admin authentication dependency. |
| |
| Admin access is granted when: |
| 1. The user's role is "admin", OR |
| 2. The user's email appears in the ADMIN_EMAILS environment variable |
| |
| Non-admin users receive a standard 403 error. |
| """ |
| from __future__ import annotations |
|
|
| from fastapi import Depends, HTTPException, status |
|
|
| from app.core.auth import require_user |
| from app.core.config import get_settings |
| from app.models.user import User |
|
|
|
|
| def _is_admin(user: User) -> bool: |
| """Check if a user has admin privileges.""" |
| if user.role == "admin": |
| return True |
|
|
| settings = get_settings() |
| admin_emails_raw = settings.admin_emails |
| if admin_emails_raw: |
| admin_emails = { |
| e.strip().lower() |
| for e in admin_emails_raw.split(",") |
| if e.strip() |
| } |
| if user.email.lower() in admin_emails: |
| return True |
|
|
| return False |
|
|
|
|
| def require_admin(user: User = Depends(require_user)) -> User: |
| """FastAPI dependency that requires the current user to be an admin. |
| |
| Returns the user if admin, raises 403 otherwise. |
| """ |
| if not _is_admin(user): |
| raise HTTPException( |
| status_code=status.HTTP_403_FORBIDDEN, |
| detail={ |
| "code": "FORBIDDEN", |
| "message": "Admin access required.", |
| }, |
| ) |
| return user |
|
|