87 lines
2.4 KiB
Python
87 lines
2.4 KiB
Python
# import logging
|
|
from uuid import UUID
|
|
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
from jose import JWTError, jwt
|
|
|
|
from app.core.config import settings
|
|
|
|
oauth2_optional = OAuth2PasswordBearer(
|
|
tokenUrl="keycloak", auto_error=False
|
|
)
|
|
|
|
# logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _decode_keycloak_token(token: str) -> dict:
|
|
if not settings.KEYCLOAK_PUBLIC_KEY:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="Keycloak public key is not configured",
|
|
)
|
|
|
|
key = settings.KEYCLOAK_PUBLIC_KEY.replace("\\n", "\n")
|
|
|
|
return jwt.decode(
|
|
token,
|
|
key,
|
|
algorithms=[settings.KEYCLOAK_ALGORITHM],
|
|
audience=settings.KEYCLOAK_AUDIENCE or None,
|
|
)
|
|
|
|
|
|
async def get_current_keycloak_payload(
|
|
token: str | None = Depends(oauth2_optional),
|
|
) -> dict:
|
|
if not token:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Not authenticated",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
try:
|
|
return _decode_keycloak_token(token)
|
|
except JWTError as exc:
|
|
# logger.warning("Keycloak token validation failed: %s", exc)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid token",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
) from exc
|
|
|
|
|
|
async def get_current_keycloak_sub(
|
|
payload: dict = Depends(get_current_keycloak_payload),
|
|
) -> UUID:
|
|
sub = payload.get("sub")
|
|
if not sub:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Missing subject claim",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
try:
|
|
return UUID(sub)
|
|
except ValueError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid subject claim",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
) from exc
|
|
|
|
|
|
async def get_current_keycloak_username(
|
|
payload: dict = Depends(get_current_keycloak_payload),
|
|
) -> str:
|
|
username = payload.get("preferred_username") or payload.get("username")
|
|
if not username:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Missing username claim",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
return str(username)
|