Spaces:
Sleeping
Sleeping
File size: 5,033 Bytes
62516b8 | 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 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 | from __future__ import annotations
import hmac
from collections.abc import Callable, Iterable
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from uuid import uuid4
from typing import Any, Annotated
import jwt
from fastapi import Header, HTTPException, status
from jwt import InvalidTokenError
from src.auth.revocation import is_token_revoked
from src.config import settings
def _unauthorized(detail: str) -> HTTPException:
return HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=detail,
headers={"WWW-Authenticate": "Bearer"},
)
def _extract_bearer_token(authorization: str | None) -> str:
if not authorization:
raise _unauthorized("Missing Authorization header.")
scheme, _, token = authorization.partition(" ")
if not hmac.compare_digest(scheme.lower(), "bearer") or not token.strip():
raise _unauthorized("Invalid Authorization header.")
return token.strip()
def _audience_matches(expected: str, claim: Any) -> bool:
if isinstance(claim, str):
return hmac.compare_digest(claim, expected)
if isinstance(claim, list):
return any(isinstance(item, str) and hmac.compare_digest(item, expected) for item in claim)
return False
def _extract_scope_set(payload: dict[str, Any]) -> set[str]:
raw_scope = payload.get("scope")
if isinstance(raw_scope, str):
return {part for part in raw_scope.split(" ") if part}
return set()
def _normalize_scopes(scopes: Iterable[str]) -> tuple[str, ...]:
seen: set[str] = set()
normalized: list[str] = []
for scope in scopes:
value = scope.strip()
if not value or value in seen:
continue
seen.add(value)
normalized.append(value)
return tuple(normalized)
def _ensure_client_subject(payload: dict[str, Any]) -> None:
subject = payload.get("sub")
if not isinstance(subject, str):
raise _unauthorized("Invalid token subject.")
if not subject.startswith("client:") or len(subject) <= len("client:"):
raise _unauthorized("Invalid token subject.")
@dataclass(frozen=True)
class IssuedAccessToken:
access_token: str
token_type: str
expires_in: int
scope: str
jti: str
def decode_and_verify_jwt(token: str) -> dict[str, Any]:
try:
payload = jwt.decode(
token,
settings.jwt_secret,
algorithms=["HS256"],
options={
"require": ["iss", "aud", "sub", "exp", "iat"],
"verify_signature": True,
"verify_exp": True,
"verify_iat": True,
"verify_iss": False,
"verify_aud": False,
},
leeway=settings.jwt_clock_skew_seconds,
)
except InvalidTokenError as exc:
raise _unauthorized("Invalid or expired token.") from exc
issuer = payload.get("iss")
if not isinstance(issuer, str) or not hmac.compare_digest(issuer, settings.jwt_issuer):
raise _unauthorized("Invalid token issuer.")
if not _audience_matches(settings.jwt_audience, payload.get("aud")):
raise _unauthorized("Invalid token audience.")
return payload
def issue_client_access_token(client_id: str, scopes: Iterable[str]) -> IssuedAccessToken:
now = datetime.now(UTC)
expires_in = max(1, settings.oauth_token_ttl_seconds)
normalized_scopes = _normalize_scopes(scopes)
jti = uuid4().hex
payload = {
"iss": settings.jwt_issuer,
"aud": settings.jwt_audience,
"sub": f"client:{client_id}",
"iat": int(now.timestamp()),
"exp": int((now + timedelta(seconds=expires_in)).timestamp()),
"scope": " ".join(normalized_scopes),
"jti": jti,
}
token = jwt.encode(payload, settings.jwt_secret, algorithm="HS256")
return IssuedAccessToken(
access_token=token,
token_type="Bearer",
expires_in=expires_in,
scope=payload["scope"],
jti=jti,
)
def require_jwt(required_scopes: list[str] | None = None) -> Callable[..., dict[str, Any]]:
async def dependency(
authorization: Annotated[str | None, Header(alias="Authorization")] = None,
) -> dict[str, Any]:
if not settings.jwt_enabled:
return {}
token = _extract_bearer_token(authorization)
payload = decode_and_verify_jwt(token)
_ensure_client_subject(payload)
jti = payload.get("jti")
if isinstance(jti, str) and jti and await is_token_revoked(jti):
raise _unauthorized("Token has been revoked.")
needed = set(required_scopes or [])
if not needed:
return payload
provided_scopes = _extract_scope_set(payload)
if not needed.issubset(provided_scopes):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Insufficient scope.",
)
return payload
return dependency
|