from __future__ import annotations import base64 import json import time import urllib.parse from collections.abc import Sequence from dataclasses import dataclass from typing import Any from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import padding, rsa from fastapi import Depends, HTTPException, status from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer PRINCIPAL_TYPES = {"human", "service", "agent"} ASSURANCE_LEVELS = {"aal0", "aal1", "aal2", "aal3", "break_glass"} DEFAULT_SKEW_SECONDS = 60 class IamProfileError(ValueError): """Raised when a bearer token is not a valid NetKingdom IAM Profile token.""" @dataclass(frozen=True, slots=True) class IamProfileClaims: issuer: str subject: str audience: tuple[str, ...] tenant: str principal_type: str groups: tuple[str, ...] roles: tuple[str, ...] scopes: tuple[str, ...] assurance: dict[str, Any] raw: dict[str, Any] @dataclass(frozen=True, slots=True) class IamProfileVerifier: issuer: str audience: str jwks: dict[str, Any] environment: str = "production" skew_seconds: int = DEFAULT_SKEW_SECONDS def verify_token(self, token: str) -> IamProfileClaims: header, payload = _decode_jwt(token) _verify_signature(header, token, self.jwks) self._verify_issuer(payload) self._verify_audience(payload) self._verify_lifetime(payload) return _claims_from_payload(payload) def _verify_issuer(self, payload: dict[str, Any]) -> None: issuer = payload.get("iss") if ( not isinstance(issuer, str) or _normalize_issuer(issuer) != _normalize_issuer(self.issuer) ): raise IamProfileError("token issuer does not match configured issuer") if self.environment == "production" and _is_local_issuer(issuer): raise IamProfileError("production mode must reject local-development issuers") def _verify_audience(self, payload: dict[str, Any]) -> None: if self.audience not in _audiences(payload.get("aud")): raise IamProfileError("token audience does not include this service") def _verify_lifetime(self, payload: dict[str, Any]) -> None: now = int(time.time()) skew = self.skew_seconds exp = payload.get("exp") iat = payload.get("iat") nbf = payload.get("nbf") if not isinstance(exp, int) or exp <= now - skew: raise IamProfileError("token is expired or missing exp") if not isinstance(iat, int) or iat > now + skew: raise IamProfileError("token iat is missing or in the future") if nbf is not None and (not isinstance(nbf, int) or nbf > now + skew): raise IamProfileError("token nbf is invalid") _bearer = HTTPBearer(auto_error=False) _bearer_dependency = Depends(_bearer) def iam_profile_dependency(verifier: IamProfileVerifier): async def dependency( credentials: HTTPAuthorizationCredentials | None = _bearer_dependency, ) -> IamProfileClaims: if credentials is None or credentials.scheme.lower() != "bearer": raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="missing bearer token", headers={"WWW-Authenticate": "Bearer"}, ) try: return verifier.verify_token(credentials.credentials) except IamProfileError as exc: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=str(exc), ) from exc return dependency def _b64url_decode(value: str) -> bytes: padding_len = (4 - len(value) % 4) % 4 return base64.urlsafe_b64decode(value + ("=" * padding_len)) def _decode_jwt(token: str) -> tuple[dict[str, Any], dict[str, Any]]: parts = token.split(".") if len(parts) != 3: raise IamProfileError("JWT must have three compact-serialization parts") try: header = json.loads(_b64url_decode(parts[0])) payload = json.loads(_b64url_decode(parts[1])) except (ValueError, json.JSONDecodeError) as exc: raise IamProfileError(f"JWT could not be decoded: {exc}") from exc if not isinstance(header, dict) or not isinstance(payload, dict): raise IamProfileError("JWT header and payload must be JSON objects") return header, payload def _jwk_to_rsa_public_key(jwk: dict[str, Any]): n = int.from_bytes(_b64url_decode(str(jwk["n"])), "big") e = int.from_bytes(_b64url_decode(str(jwk["e"])), "big") return rsa.RSAPublicNumbers(e, n).public_key() def _verify_signature(header: dict[str, Any], token: str, jwks: dict[str, Any]) -> None: if header.get("alg") != "RS256": raise IamProfileError("JWT alg must be RS256") kid = header.get("kid") keys = jwks.get("keys") if isinstance(jwks.get("keys"), list) else [] matching = [key for key in keys if isinstance(key, dict) and key.get("kid") == kid] if not matching: raise IamProfileError("JWT kid was not found in JWKS") parts = token.split(".") signing_input = f"{parts[0]}.{parts[1]}".encode("ascii") signature = _b64url_decode(parts[2]) try: public_key = _jwk_to_rsa_public_key(matching[0]) public_key.verify(signature, signing_input, padding.PKCS1v15(), hashes.SHA256()) except (InvalidSignature, ValueError, KeyError) as exc: raise IamProfileError("JWT signature verification failed") from exc def _normalize_issuer(value: str) -> str: return value.rstrip("/") def _is_local_issuer(issuer: str) -> bool: if issuer == "local-identity": return True parsed = urllib.parse.urlparse(issuer) host = (parsed.hostname or "").lower() if parsed.scheme == "http": return True return host in {"localhost", "127.0.0.1", "::1"} or host.endswith(".local") def _audiences(value: Any) -> tuple[str, ...]: if isinstance(value, str): return (value,) if isinstance(value, Sequence) and not isinstance(value, bytes): return tuple(str(item) for item in value) return () def _scopes(payload: dict[str, Any]) -> tuple[str, ...]: values: list[str] = [] for name in ("scope", "scp"): raw = payload.get(name) if isinstance(raw, str): values.extend(part for part in raw.split() if part) elif isinstance(raw, list): values.extend(str(part) for part in raw) return tuple(sorted(set(values))) def _string_list(payload: dict[str, Any], name: str) -> tuple[str, ...]: raw = payload.get(name) if not isinstance(raw, list): raise IamProfileError(f"{name} must be a list") return tuple(str(item) for item in raw) def _claims_from_payload(payload: dict[str, Any]) -> IamProfileClaims: required = {"iss", "sub", "aud", "exp", "iat", "tenant", "principal_type", "assurance"} missing = sorted(required - set(payload)) if missing: raise IamProfileError(f"token is missing required IAM Profile claims: {', '.join(missing)}") subject = payload.get("sub") issuer = payload.get("iss") tenant = payload.get("tenant") principal_type = payload.get("principal_type") assurance = payload.get("assurance") if not isinstance(subject, str) or not subject: raise IamProfileError("sub must be a non-empty string") if not isinstance(issuer, str): raise IamProfileError("iss must be a string") if not isinstance(tenant, str) or not tenant.startswith("tenant:"): raise IamProfileError("tenant must be a string like tenant:platform") if principal_type not in PRINCIPAL_TYPES: raise IamProfileError("principal_type must be human, service, or agent") if not isinstance(assurance, dict): raise IamProfileError("assurance must be an object") roles = _string_list(payload, "roles") groups = _string_list(payload, "groups") scopes = _scopes(payload) if not scopes: raise IamProfileError("scope or scp must be present") _validate_assurance(assurance) return IamProfileClaims( issuer=issuer, subject=subject, audience=_audiences(payload.get("aud")), tenant=tenant, principal_type=str(principal_type), groups=groups, roles=roles, scopes=scopes, assurance=assurance, raw=payload, ) def _validate_assurance(assurance: dict[str, Any]) -> None: level = assurance.get("level") methods = assurance.get("methods") mfa = assurance.get("mfa") source = assurance.get("source") if level not in ASSURANCE_LEVELS: raise IamProfileError("assurance.level is unsupported") if not isinstance(methods, list) or not all(isinstance(method, str) for method in methods): raise IamProfileError("assurance.methods must be a list of strings") if not isinstance(mfa, bool): raise IamProfileError("assurance.mfa must be boolean") if not isinstance(source, str) or not source: raise IamProfileError("assurance.source must be a non-empty string")