Fusion
54 lines
1.7 KiB
Python
54 lines
1.7 KiB
Python
from typing import Any
|
|
|
|
import jwt
|
|
|
|
from app.config import settings
|
|
|
|
|
|
def decode_token(token: str) -> dict[str, Any]:
|
|
"""Decode a JWT token.
|
|
|
|
When authentik_issuer_url is configured, validates the token
|
|
against the OIDC discovery document JWKS.
|
|
Otherwise, decodes without verification (local development only).
|
|
"""
|
|
if settings.authentik_issuer_url:
|
|
import httpx
|
|
|
|
issuer = settings.authentik_issuer_url.rstrip("/")
|
|
discovery_url = f"{issuer}/.well-known/openid-configuration"
|
|
with httpx.Client() as client:
|
|
resp = client.get(discovery_url)
|
|
resp.raise_for_status()
|
|
discovery = resp.json()
|
|
jwks_uri = discovery["jwks_uri"]
|
|
|
|
jwks_resp = client.get(jwks_uri)
|
|
jwks_resp.raise_for_status()
|
|
jwks = jwks_resp.json()
|
|
|
|
signing_key = jwt.algorithms.RSAAlgorithm.from_jwk(
|
|
_find_matching_key(jwks, token)
|
|
)
|
|
|
|
return jwt.decode(
|
|
token,
|
|
signing_key, # type: ignore[arg-type]
|
|
algorithms=["RS256"],
|
|
audience=settings.authentik_client_id,
|
|
issuer=settings.authentik_issuer_url,
|
|
)
|
|
|
|
return jwt.decode(token, options={"verify_signature": False})
|
|
|
|
|
|
def _find_matching_key(jwks: dict[str, Any], token: str) -> dict[str, Any]:
|
|
"""Find the key in JWKS that matches the token's kid header."""
|
|
unverified_header = jwt.get_unverified_header(token)
|
|
kid = unverified_header.get("kid")
|
|
for key in jwks.get("keys", []):
|
|
key_dict: dict[str, Any] = key
|
|
if key_dict.get("kid") == kid:
|
|
return key_dict
|
|
raise RuntimeError(f"No matching JWKS key found for kid={kid}")
|