secrets-engine/src/secrets_engine/service_auth.py

"""KeyCape service-JWT exchange shared by two separately validated profiles.

The OpenBao login profile and approval consumer profile have distinct clients,
audiences, tenants and scopes. Provider selection never falls back to another
identity. Token parsing is a claim preflight, not signature verification: the
receiving OpenBao or approval-engine service verifies RS256 before accepting it.
"""
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
    max_future_iat_seconds: int = field(default=60, init=False)

    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 resource-server use."""
    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("level") != "aal1"
        or assurance.get("methods") != ["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 + config.max_future_iat_seconds:
        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(65537)
                if len(raw) > 65536:
                    raise BackendError("KeyCape token response exceeds size bound")
        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)