34 lines
900 B
Python
34 lines
900 B
Python
from fastapi import Header, HTTPException, status
|
|
|
|
from app.core.config import settings
|
|
from app.domain.roles import UserRole
|
|
|
|
|
|
def resolve_role(api_key: str) -> str | None:
|
|
if api_key in settings.admin_keys:
|
|
return UserRole.ADMIN.value
|
|
if api_key in settings.operator_keys:
|
|
return UserRole.OPERATOR.value
|
|
if api_key in settings.viewer_keys:
|
|
return UserRole.VIEWER.value
|
|
return None
|
|
|
|
|
|
async def require_api_key(
|
|
x_api_key: str | None = Header(default=None, alias=settings.auth_header_name),
|
|
) -> str:
|
|
if not x_api_key:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Missing API key",
|
|
)
|
|
|
|
role = resolve_role(x_api_key)
|
|
if role is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Invalid API key",
|
|
)
|
|
|
|
return role
|