Accept Tessera (Keycloak-compatible) OIDC tokens as API bearer

Adds an additive bearer-verification path: verify RS256 access tokens against
Tessera's JWKS (iss/aud/exp), map sub/preferred_username/email + roles
(realm_access.roles, resource_access.<audience>.roles) to the app's identity.
Existing auth (API keys / app JWTs / sessions) is unchanged. Issuer + audience
are env-configurable. Validated end-to-end against the local sim.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
h z
2026-06-02 15:11:31 +01:00
parent 1a5a3ed1b1
commit 0bdc432215
3 changed files with 266 additions and 5 deletions

View File

@@ -1,5 +1,6 @@
"""Shared auth dependencies."""
import hashlib
import logging
from datetime import datetime, timedelta
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, APIKeyHeader
@@ -12,6 +13,8 @@ from app.core.config import get_db, settings
from app.models import models
from app.models.apikey import APIKey
logger = logging.getLogger("harborforge.deps")
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token", auto_error=False)
apikey_header = APIKeyHeader(name="X-API-Key", auto_error=False)
@@ -86,6 +89,10 @@ async def get_current_user_or_apikey(
"""Authenticate via JWT token (Authorization: Bearer <jwt>) OR API key
(X-API-Key: <key>, OR — as a convenience for CLI clients that only know
Bearer — Authorization: Bearer <api-key>; falls back when JWT decode fails).
Bearer tokens are tried in order: local HS256 JWT → external Tessera
(OIDC) RS256 access token → API key. The Tessera path is purely additive
and never affects local-JWT/API-key callers.
"""
# Native X-API-Key header
if api_key:
@@ -93,15 +100,31 @@ async def get_current_user_or_apikey(
if user:
return user
# Bearer header — try JWT first, then API key on decode failure
# Bearer header — local JWT first, then Tessera, then API key.
if token:
try:
return await get_current_user(token=token, db=db)
except HTTPException:
user = _lookup_api_key(db, token)
if user:
return user
raise
pass
# External Tessera (OIDC) RS256 access token.
try:
from app.api.tessera import authenticate_tessera
return authenticate_tessera(db, token)
except HTTPException:
pass
except Exception: # JWKS fetch / unexpected verifier error → don't 500
logger.warning("Tessera token verification error", exc_info=True)
# Bearer-carried API key (CLI convenience).
user = _lookup_api_key(db, token)
if user:
return user
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
raise HTTPException(status_code=401, detail="Not authenticated")