feat: exchange scoped approval service tokens per request

Assistant: codex
Assistant-Model: gpt-5.6-luna
Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc

src/secrets_engine/approval_auth.py

@@ -0,0 +1,113 @@
+"""The approval consumer's own KeyCape identity, distinct from OpenBao login.
+
+Access tokens stay in memory for one request. Claim preflight is not signature
+verification or an authorization decision; approval-engine verifies the JWT.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+from urllib.parse import urlsplit
+from urllib.request import HTTPRedirectHandler, build_opener
+
+from secrets_engine.errors import BackendError, DecisionError
+from secrets_engine.openbao import read_strict_token_file
+from secrets_engine.service_auth import KeyCapeServiceAuthConfig, KeyCapeServiceAuthProvider
+
+
+class _NoRedirect(HTTPRedirectHandler):
+    def redirect_request(self, req, fp, code, msg, headers, newurl):
+        # urllib otherwise forwards Basic/Bearer headers on a redirected GET.
+        return None
+
+
+def credential_urlopen(request, *, timeout):
+    return build_opener(_NoRedirect()).open(request, timeout=timeout)
+
+
+@dataclass(frozen=True)
+class KeyCapeApprovalAuthConfig(KeyCapeServiceAuthConfig):
+    client_id: str = "secrets-engine-approval"
+    audience: str = "approval-engine"
+    tenant: str = "tenant:platform"
+    scope: str = "approval:read"
+    max_future_iat_seconds: int = field(default=30, init=False)
+
+    def __post_init__(self) -> None:
+        # Deliberately do not relax the separate OpenBao profile's validator.
+        if (
+            self.client_id != "secrets-engine-approval"
+            or self.audience != "approval-engine"
+            or self.subject != "service:secrets-engine"
+            or self.tenant != "tenant:platform"
+            or self.required_role != "secrets-engine"
+            or self.scope not in {"approval:read", "approval:consume"}
+            or self.max_future_iat_seconds != 30
+        ):
+            raise BackendError("KeyCape approval identity does not match accepted contract")
+        try:
+            issuer = urlsplit(self.issuer)
+            valid = (
+                issuer.scheme == "https" and bool(issuer.hostname)
+                and issuer.username is None and issuer.password is None
+                and not issuer.query and not issuer.fragment
+                and issuer.path in {"", "/"}
+                and self.token_url == self.issuer.rstrip("/") + "/token"
+                and issuer.port != 0
+            )
+        except ValueError:
+            valid = False
+        if not valid:
+            raise BackendError("KeyCape approval exchange requires the issuer's HTTPS /token")
+        if self.timeout_seconds <= 0:
+            raise BackendError("KeyCape timeout must be positive")
+
+
+def approval_auth_configured(cfg: Any) -> bool:
+    token_file = getattr(cfg, "approval_token_file", None)
+    secret_file = getattr(cfg, "approval_client_secret_file", None)
+    if token_file and secret_file:
+        raise DecisionError("approval auth requires one provider; token and client-secret files conflict")
+    return bool(token_file or secret_file)
+
+
+def require_approval_address(base_url: str) -> None:
+    """The first CLI consumer uses TLS or the owner's literal loopback tunnel.
+
+    A loopback address alone does not establish the tunnel's target identity;
+    the owner must bind kubectl port-forward to the admitted cluster and pod.
+    """
+    try:
+        url = urlsplit(base_url)
+        host = (url.hostname or "").rstrip(".").lower()
+        valid = (
+            bool(host) and url.username is None and url.password is None
+            and not url.query and not url.fragment and url.path in {"", "/"}
+            and url.port != 0
+            and not host.endswith((".svc", ".svc.cluster.local", ".cluster.local"))
+            and (url.scheme == "https" or (
+                url.scheme == "http" and host in {"127.0.0.1", "::1"}
+            ))
+        )
+    except ValueError:
+        valid = False
+    if not valid:
+        raise DecisionError("approval service auth requires HTTPS or an owner-bound literal loopback tunnel")
+
+
+def approval_token(cfg: Any, *, scope: str) -> str:
+    """Select one explicit provider. Never fall back after exchange failure."""
+    if not approval_auth_configured(cfg):
+        raise DecisionError("approval credential is unconfigured")
+    secret_file = getattr(cfg, "approval_client_secret_file", None)
+    if secret_file:
+        require_approval_address(str(getattr(cfg, "approval_url", "") or ""))
+        config = KeyCapeApprovalAuthConfig(
+            token_url=str(getattr(cfg, "keycape_token_url", "") or ""),
+            issuer=str(getattr(cfg, "keycape_issuer", "") or ""),
+            client_secret_file=Path(secret_file),
+            scope=scope,
+        )
+        return KeyCapeServiceAuthProvider(config, transport=credential_urlopen).exchange().token
+    return read_strict_token_file(Path(cfg.approval_token_file), purpose="approval credential")