"""Clients for the executor's owners (NK-WP-0056-T13): approval-engine (carrier) and audit-core (sink). Stdlib HTTP only. Bearer tokens are read from files at call time (projected or ESO-delivered), never logged. Errors carry status codes only, never response bodies. """ from __future__ import annotations import base64 import json import pathlib import ssl import urllib.error import urllib.parse import urllib.request class ClientError(Exception): pass class _NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, *args, **kwargs): # credentials never follow a redirect return None def _opener(ca_file: str | None): context = ssl.create_default_context(cafile=ca_file) if ca_file else ssl.create_default_context() return urllib.request.build_opener(urllib.request.ProxyHandler({}), _NoRedirect(), urllib.request.HTTPSHandler(context=context)).open class KeyCapeClientCredentials: """client_credentials exchange at KeyCape (src/internal/server/oidc/token.go: client_secret_basic, every requested scope must be allowed for the client). A fresh token per call and scope; nothing is cached or logged, as secrets-engine's approval consumer does. `secret` is a callable so the value stays in its owner's frame. `host` sends the public Host (and X-Forwarded-*) when the token URL is the in-cluster service, the way the synthetic journey reaches KeyCape (sso-mfa/k8s/synthetic-journey/journey.py host_map).""" def __init__(self, token_url: str, client_id: str, secret, *, host: str | None = None, ca_file: str | None = None, timeout: float = 15): parsed = urllib.parse.urlsplit(token_url) if parsed.scheme not in ("https", "http") or not parsed.netloc or parsed.query or parsed.fragment: raise ClientError("token url") if not callable(secret): raise ClientError("secret provider") self.token_url, self.client_id, self.secret, self.host = token_url, client_id, secret, host self.open = _opener(ca_file) self.timeout = timeout def token(self, scope: str) -> str: if not scope or " " in scope: raise ClientError("one scope per token") basic = base64.b64encode((urllib.parse.quote(self.client_id, safe="") + ":" + urllib.parse.quote(self.secret(), safe="")).encode()).decode() headers = {"Authorization": "Basic " + basic, "Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"} if self.host: headers.update({"Host": self.host, "X-Forwarded-Host": self.host, "X-Forwarded-Proto": "https"}) body = urllib.parse.urlencode({"grant_type": "client_credentials", "scope": scope}).encode() request = urllib.request.Request(self.token_url, method="POST", headers=headers, data=body) try: with self.open(request, timeout=self.timeout) as response: data = json.loads(response.read(1 << 16)) except urllib.error.HTTPError as error: error.close() raise ClientError(f"token {error.code}") from None token = data.get("access_token") if (not isinstance(token, str) or not token or len(token) > 16384 or any(c.isspace() for c in token) or str(data.get("token_type", "")).lower() != "bearer" or not isinstance(data.get("expires_in"), int) or not 0 < data["expires_in"] <= 900): raise ClientError("token response") return token class _Http: def __init__(self, base: str, token_file: str | None = None, ca_file: str | None = None, timeout: float = 15, token_provider=None): parsed = urllib.parse.urlsplit(base) if parsed.scheme not in ("https", "http") or not parsed.netloc or parsed.query or parsed.fragment: raise ClientError("base url") if (token_file is None) == (token_provider is None): raise ClientError("exactly one of token_file and token_provider") self.base = base.rstrip("/") self.token_file = token_file self.token_provider = token_provider self.open = _opener(ca_file) self.timeout = timeout def bearer(self, scope: str | None) -> str: if self.token_provider is not None: return self.token_provider(scope) return pathlib.Path(self.token_file).read_text().strip() def call(self, method: str, path: str, body: dict | None = None, headers: dict | None = None, scope: str | None = None) -> tuple[int, dict]: head = {"Authorization": "Bearer " + self.bearer(scope), "Accept": "application/json", **(headers or {})} data = None if body is not None: head["Content-Type"] = "application/json" data = json.dumps(body).encode() request = urllib.request.Request(self.base + path, method=method, headers=head, data=data) try: with self.open(request, timeout=self.timeout) as response: raw = response.read(1 << 20) return response.status, (json.loads(raw) if raw else {}) except urllib.error.HTTPError as error: error.close() raise ClientError(f"{method} {error.code}") from None class ApprovalEngine: """POST /v1/approvals, GET /v1/approvals/{id} and POST /v1/approvals/{id}/consume (approval-engine api.py). Each call asks for exactly the scope it needs (approval:create, approval:read, approval:consume).""" def __init__(self, base: str, token_file: str | None = None, ca_file: str | None = None, token_provider=None): self.http = _Http(base, token_file, ca_file, token_provider=token_provider) def create(self, binding: dict, validity: dict, required_count: int) -> dict: """File the approval request as the actor itself. human_control is always on and pdp_path off (pilot).""" status, data = self.http.call("POST", "/v1/approvals", { "binding": binding, "validity": validity, "required_count": required_count, "human_control": True, "pdp_path": False}, scope="approval:create") if status != 201: raise ClientError(f"create {status}") return data @staticmethod def _path(approval_id: str) -> str: if not approval_id or "/" in approval_id or len(approval_id) > 128: raise ClientError("approval id") return "/v1/approvals/" + urllib.parse.quote(approval_id, safe="") def get(self, approval_id: str) -> dict: return self.http.call("GET", self._path(approval_id), scope="approval:read")[1] def consume(self, approval_id: str, request_digest: str, decision_id: str) -> dict: return self.http.call("POST", self._path(approval_id) + "/consume", {"request_digest": request_digest, "decision_id": decision_id}, scope="approval:consume")[1] class AuditCore: """POST /v1/events with Idempotency-Key = event id (audit-core docs/event-envelope.md).""" def __init__(self, base: str, token_file: str, ca_file: str | None = None): self.http = _Http(base, token_file, ca_file) def emit(self, event: dict) -> None: """audit-core answers 202 for a first accept and 200 for an idempotent duplicate (AUDIT-WP-0018-T02 tests). 409 means the run id was reused with a different body: a defect, never retried.""" try: status, _ = self.http.call("POST", "/v1/events", event, {"Idempotency-Key": event["id"]}) except ClientError as error: if str(error).endswith(" 409"): raise ClientError("audit conflict 409: run id reused with a different body") from None raise if status not in (200, 201, 202): raise ClientError(f"audit {status}")