"""Explicit KeyCape service-JWT provider for future OpenBao JWT login. This module implements the accepted KeyCape consumer contract without wiring it into OpenBao yet. The platform owner still needs to publish the exact OpenBao JWT auth mount and role. Keeping provider selection separate prevents an auth failure from falling back to bootstrap, operator, or AppRole credentials. JWT parsing here is a claim preflight, not signature verification. OpenBao must verify the RS256 signature against the configured issuer before issuing a token. """ from __future__ import annotations import base64 import binascii import json from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable from urllib.error import HTTPError, URLError from urllib.parse import urlencode from urllib.request import Request, urlopen from secrets_engine.errors import BackendError from secrets_engine.openbao import read_strict_token_file CLIENT_ID = "secrets-engine-openbao" SUBJECT = "service:secrets-engine" PRINCIPAL_TYPE = "service" TENANT = "tenant:coulomb" ROLE = "secrets-engine" SCOPE = "openbao:login" MAX_TOKEN_SECONDS = 15 * 60 RENEW_WINDOW_SECONDS = 3 * 60 @dataclass(frozen=True) class KeyCapeServiceAuthConfig: token_url: str issuer: str client_secret_file: Path client_id: str = CLIENT_ID subject: str = SUBJECT audience: str = CLIENT_ID tenant: str = TENANT required_role: str = ROLE scope: str = SCOPE timeout_seconds: float = 10.0 def __post_init__(self) -> None: if not self.token_url.startswith("https://"): raise BackendError("KeyCape token URL must use HTTPS") if not self.issuer.startswith("https://"): raise BackendError("KeyCape issuer must use HTTPS") if self.client_id != CLIENT_ID or self.audience != CLIENT_ID: raise BackendError("KeyCape service client/audience must match accepted contract") if ( self.subject != SUBJECT or self.tenant != TENANT or self.required_role != ROLE or self.scope != SCOPE ): raise BackendError("KeyCape service identity claims must match accepted contract") if self.timeout_seconds <= 0: raise BackendError("KeyCape timeout must be positive") @dataclass(frozen=True) class ServiceJWT: token: str = field(repr=False) issued_at: int expires_at: int claims: dict[str, Any] = field(repr=False) def needs_renewal(self, now: datetime | None = None) -> bool: current = int((now or datetime.now(timezone.utc)).timestamp()) return self.expires_at - current <= RENEW_WINDOW_SECONDS def _decode_segment(value: str, label: str) -> dict[str, Any]: try: padded = value + "=" * (-len(value) % 4) decoded = base64.urlsafe_b64decode(padded.encode("ascii")) parsed = json.loads(decoded) except (UnicodeEncodeError, binascii.Error, json.JSONDecodeError) as e: raise BackendError(f"KeyCape JWT has invalid {label}") from e if not isinstance(parsed, dict): raise BackendError(f"KeyCape JWT {label} must be an object") return parsed def preflight_service_jwt( token: str, config: KeyCapeServiceAuthConfig, *, now: datetime | None = None, ) -> ServiceJWT: """Validate non-cryptographic JWT shape/claims before OpenBao login.""" parts = token.split(".") if len(parts) != 3 or not all(parts): raise BackendError("KeyCape access token is not a compact JWT") header = _decode_segment(parts[0], "header") claims = _decode_segment(parts[1], "payload") if header.get("alg") != "RS256": raise BackendError("KeyCape JWT algorithm is not RS256") exact_claims = { "iss": config.issuer, "sub": config.subject, "aud": config.audience, "principal_type": PRINCIPAL_TYPE, "tenant": config.tenant, "groups": [], } for name, expected in exact_claims.items(): if claims.get(name) != expected: raise BackendError(f"KeyCape JWT claim '{name}' does not match contract") if claims.get("roles") != [config.required_role]: raise BackendError("KeyCape JWT roles do not match contract") if claims.get("scope") != config.scope: raise BackendError("KeyCape JWT scope does not match contract") assurance = claims.get("assurance") if not isinstance(assurance, dict) or ( assurance.get("aal") != "AAL1" or assurance.get("method") != "client_secret" or assurance.get("mfa") is not False or assurance.get("source") != "key-cape" ): raise BackendError("KeyCape JWT assurance does not match contract") issued_at = claims.get("iat") expires_at = claims.get("exp") if ( not isinstance(issued_at, int) or isinstance(issued_at, bool) or not isinstance(expires_at, int) or isinstance(expires_at, bool) ): raise BackendError("KeyCape JWT iat/exp must be integer timestamps") current = int((now or datetime.now(timezone.utc)).timestamp()) if issued_at > current + 60: raise BackendError("KeyCape JWT issue time is in the future") if expires_at <= current: raise BackendError("KeyCape JWT has expired") if expires_at <= issued_at or expires_at - issued_at > MAX_TOKEN_SECONDS: raise BackendError("KeyCape JWT lifetime exceeds accepted 15-minute bound") return ServiceJWT( token=token, issued_at=issued_at, expires_at=expires_at, claims=dict(claims), ) Transport = Callable[..., Any] @dataclass(frozen=True) class KeyCapeServiceAuthProvider: config: KeyCapeServiceAuthConfig transport: Transport = field(default=urlopen, repr=False, compare=False) def exchange(self, *, now: datetime | None = None) -> ServiceJWT: """Perform one client-credentials exchange; never retry or fall back.""" secret = read_strict_token_file( self.config.client_secret_file, purpose="KeyCape client secret", ) basic = base64.b64encode( f"{self.config.client_id}:{secret}".encode("utf-8") ).decode("ascii") body = urlencode( {"grant_type": "client_credentials", "scope": self.config.scope} ).encode("ascii") request = Request( self.config.token_url, data=body, method="POST", headers={ "Authorization": f"Basic {basic}", "Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json", }, ) try: response = self.transport(request, timeout=self.config.timeout_seconds) with response: status = getattr(response, "status", 200) raw = response.read() except HTTPError as e: raise BackendError(f"KeyCape token exchange failed with HTTP {e.code}") from e except (URLError, TimeoutError, OSError) as e: raise BackendError("KeyCape token exchange failed") from e if status != 200: raise BackendError(f"KeyCape token exchange failed with HTTP {status}") try: payload = json.loads(raw) except (UnicodeDecodeError, json.JSONDecodeError) as e: raise BackendError("KeyCape token response is not valid JSON") from e if not isinstance(payload, dict): raise BackendError("KeyCape token response must be an object") if payload.get("id_token") or payload.get("refresh_token"): raise BackendError("KeyCape service exchange returned a forbidden extra token") token = payload.get("access_token") if not isinstance(token, str) or not token: raise BackendError("KeyCape token response has no access token") if str(payload.get("token_type", "")).lower() != "bearer": raise BackendError("KeyCape token response type is not Bearer") expires_in = payload.get("expires_in") if ( not isinstance(expires_in, int) or isinstance(expires_in, bool) or not 0 < expires_in <= MAX_TOKEN_SECONDS ): raise BackendError("KeyCape token response lifetime is outside contract") return preflight_service_jwt(token, self.config, now=now)