27 lines
1.2 KiB
Python
27 lines
1.2 KiB
Python
from fastapi import Depends, HTTPException, status
|
|
|
|
from app.core.config import get_settings
|
|
from app.schemas.auth import AuthentikPrincipal
|
|
from app.security.authentik_jwt import require_authenticated_principal
|
|
|
|
|
|
def require_admin_principal(
|
|
principal: AuthentikPrincipal = Depends(require_authenticated_principal),
|
|
) -> AuthentikPrincipal:
|
|
settings = get_settings()
|
|
allowed_emails = {email.lower() for email in settings.admin_allowlist_emails}
|
|
allowed_subs = set(settings.admin_allowlist_subs)
|
|
required_groups = {group.lower() for group in settings.admin_required_groups}
|
|
|
|
if not allowed_emails and not allowed_subs and not required_groups:
|
|
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="admin_policy_not_configured")
|
|
|
|
email_ok = bool(principal.email and principal.email.lower() in allowed_emails)
|
|
sub_ok = principal.sub in allowed_subs
|
|
principal_groups = {group.lower() for group in principal.groups}
|
|
group_ok = bool(required_groups.intersection(principal_groups))
|
|
|
|
if not (email_ok or sub_ok or group_ok):
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="admin_forbidden")
|
|
return principal
|