Spaces:
Running
Running
File size: 2,927 Bytes
f1fa34c | 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 | """
JWT access tokens and the authenticated-user dependency.
This module is the **only** place a request's identity may come from. Routers
take `user_id: str = Depends(current_user)`; no endpoint accepts a caller-supplied
user id, so identity cannot be spoofed by editing a request body.
Kept in `api/` because it is a pure HTTP concern — `app/` stays reusable by the
Streamlit application, which has its own session handling.
"""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from api.settings import get_api_settings
_ALGORITHM = "HS256"
_TOKEN_TYPE = "access"
# auto_error=False so we can raise our own 401 with a WWW-Authenticate header
# instead of FastAPI's bare 403 for a missing Authorization header.
_bearer = HTTPBearer(auto_error=False)
def _unauthorized(detail: str) -> HTTPException:
return HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=detail,
headers={"WWW-Authenticate": "Bearer"},
)
def issue_access_token(username: str) -> tuple[str, int]:
"""
Mint a short-lived access token.
Returns:
(token, expires_in_seconds) — the client uses the second value to
refresh proactively rather than waiting for a 401.
"""
cfg = get_api_settings()
ttl = timedelta(minutes=cfg.access_token_ttl_minutes)
now = datetime.now(timezone.utc)
payload = {
"sub": username,
"type": _TOKEN_TYPE,
"iat": int(now.timestamp()),
"exp": int((now + ttl).timestamp()),
}
token = jwt.encode(payload, cfg.jwt_secret, algorithm=_ALGORITHM)
return token, int(ttl.total_seconds())
def decode_access_token(token: str) -> str:
"""Verify an access token and return its subject. Raises 401 when invalid."""
cfg = get_api_settings()
try:
payload = jwt.decode(token, cfg.jwt_secret, algorithms=[_ALGORITHM])
except jwt.ExpiredSignatureError:
raise _unauthorized("Access token expired") from None
except jwt.PyJWTError:
raise _unauthorized("Invalid access token") from None
# Reject a refresh token presented where an access token is required.
if payload.get("type") != _TOKEN_TYPE:
raise _unauthorized("Wrong token type")
subject = payload.get("sub")
if not subject:
raise _unauthorized("Malformed token")
return str(subject)
def current_user(
credentials: HTTPAuthorizationCredentials | None = Depends(_bearer),
) -> str:
"""
FastAPI dependency resolving the authenticated user id.
This is the single source of identity for every protected route.
"""
if credentials is None or not credentials.credentials:
raise _unauthorized("Not authenticated")
return decode_access_token(credentials.credentials)
|