Spaces:
Sleeping
Sleeping
File size: 4,416 Bytes
67be617 | 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 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | """JWT Bearer authentication with lightweight ABAC checks.
The API uses symmetric HS256 JWTs. Tokens carry standard claims plus an
``attrs`` object used for attribute-based access control, e.g.:
{"sub": "alice", "exp": 1700000000,
"attrs": {"role": "admin", "scopes": ["harvest:read", "harvest:write"]}}
Configuration via environment:
JWT_SECRET required HMAC secret for signing / verifying tokens
JWT_ALGORITHM optional, defaults to HS256
JWT_ISSUER optional issuer claim to validate
JWT_AUDIENCE optional audience claim to validate
"""
from __future__ import annotations
import os
from dataclasses import dataclass
import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
_bearer_scheme = HTTPBearer(auto_error=False)
@dataclass(frozen=True)
class Principal:
"""Authenticated caller and their attributes."""
subject: str
attrs: dict
def has_scope(self, scope: str) -> bool:
scopes = self.attrs.get("scopes") or []
return scope in scopes
def has_role(self, role: str) -> bool:
return self.attrs.get("role") == role
def _get_secret() -> str:
secret = os.environ.get("JWT_SECRET")
if not secret:
# Fail closed — never accept tokens without a configured secret.
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="JWT_SECRET is not configured",
)
return secret
def _decode(token: str) -> dict:
algorithm = os.environ.get("JWT_ALGORITHM", "HS256")
options = {"require": ["exp"]}
kwargs: dict = {"algorithms": [algorithm], "options": options}
audience = os.environ.get("JWT_AUDIENCE")
if audience:
kwargs["audience"] = audience
issuer = os.environ.get("JWT_ISSUER")
if issuer:
kwargs["issuer"] = issuer
try:
return jwt.decode(token, _get_secret(), **kwargs)
except jwt.ExpiredSignatureError as exc:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token has expired",
headers={"WWW-Authenticate": "Bearer"},
) from exc
except jwt.InvalidTokenError as exc:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication token",
headers={"WWW-Authenticate": "Bearer"},
) from exc
def get_current_principal(
credentials: HTTPAuthorizationCredentials | None = Depends(_bearer_scheme),
) -> Principal:
"""FastAPI dependency that resolves the caller from a Bearer token."""
if credentials is None or credentials.scheme.lower() != "bearer":
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Missing Bearer credentials",
headers={"WWW-Authenticate": "Bearer"},
)
payload = _decode(credentials.credentials)
subject = payload.get("sub") or payload.get("subject")
if not subject:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token is missing 'sub' claim",
headers={"WWW-Authenticate": "Bearer"},
)
attrs = payload.get("attrs") or {}
if not isinstance(attrs, dict):
attrs = {}
# Fall back to top-level scope/role if not nested under attrs.
if "scopes" not in attrs and "scope" in payload:
scope = payload["scope"]
attrs["scopes"] = scope.split() if isinstance(scope, str) else list(scope)
if "role" not in attrs and "role" in payload:
attrs["role"] = payload["role"]
return Principal(subject=str(subject), attrs=attrs)
def require_scopes(*required: str):
"""Return a dependency enforcing that the caller has all given scopes."""
required_set = tuple(required)
def _checker(principal: Principal = Depends(get_current_principal)) -> Principal:
missing: list[str] = [s for s in required_set if not principal.has_scope(s)]
if missing:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Missing required scopes: {', '.join(required_set)}",
)
return principal
return _checker
|