Spaces:
Sleeping
Sleeping
File size: 2,777 Bytes
db4ba8d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | """
TradeFlow AI — FastAPI Auth Dependencies (T-009)
Provides reusable dependency functions for role-based access control.
All protected endpoints must use one of these dependencies.
"""
from __future__ import annotations
from typing import Annotated, Any
from fastapi import Depends, HTTPException, Security, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from .keycloak import extract_roles, extract_user_id, verify_keycloak_token
bearer_scheme = HTTPBearer(auto_error=True)
async def get_current_token_payload(
credentials: Annotated[HTTPAuthorizationCredentials, Security(bearer_scheme)],
) -> dict[str, Any]:
"""Verify the Bearer JWT and return its decoded payload."""
return await verify_keycloak_token(credentials.credentials)
async def get_current_user_id(
payload: Annotated[dict[str, Any], Depends(get_current_token_payload)],
) -> str:
"""Returns the authenticated user's Keycloak sub (UUID)."""
return extract_user_id(payload)
async def get_current_roles(
payload: Annotated[dict[str, Any], Depends(get_current_token_payload)],
) -> list[str]:
"""Returns the list of Keycloak realm roles for the current user."""
return extract_roles(payload)
def require_roles(*allowed_roles: str):
"""
Dependency factory that enforces role-based access.
Usage:
@router.post("/submit")
async def submit(
_: None = Depends(require_roles("operator", "admin"))
):
"""
async def _check_roles(
roles: Annotated[list[str], Depends(get_current_roles)],
) -> None:
if not any(role in roles for role in allowed_roles):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Required roles: {list(allowed_roles)}",
)
return Depends(_check_roles)
# Convenience singletons for common role checks
RequireOperator = require_roles("operator", "admin")
RequireAdmin = require_roles("admin")
RequireSME = require_roles("sme", "operator", "admin")
RequireSupervisor = require_roles("supervisor", "admin")
class CurrentUser:
"""Dependency class bundling user_id + roles in one inject."""
def __init__(self, user_id: str, roles: list[str]) -> None:
self.user_id = user_id
self.roles = roles
def has_role(self, *roles: str) -> bool:
return any(r in self.roles for r in roles)
def is_admin(self) -> bool:
return "admin" in self.roles
async def get_current_user(
user_id: Annotated[str, Depends(get_current_user_id)],
roles: Annotated[list[str], Depends(get_current_roles)],
) -> CurrentUser:
"""Returns a CurrentUser object with id and roles."""
return CurrentUser(user_id=user_id, roles=roles)
|