5aa35b1d04
restrict ops endpoints to admin-only access block operator and viewer keys from admin maintenance routes cover destructive pricing cleanup in smoke execution, not only preview extend orchestration without regressing existing smoke stages
43 lines
1.1 KiB
Python
43 lines
1.1 KiB
Python
from fastapi import Depends, 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
|
|
|
|
|
|
async def require_admin_api_key(role: str = Depends(require_api_key)) -> str:
|
|
if role != UserRole.ADMIN.value:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Admin role required",
|
|
)
|
|
return role
|