tegwick
ba9d266b7e
All checks were successful
CLI entry point and approval-engine/audit-core clients; events carry occurred_at (audit-core envelope); hash-locked image with pinned lldap stage builds locally; image workflow and executor tests in CI; pilot example pins the executed files. rapp reshaping still open. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 994874@bnt-lap001 Assistant-Session: 6017c416-d619-47eb-bd1a-f670a02d7d87
79 lines
3.3 KiB
Python
79 lines
3.3 KiB
Python
"""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 json
|
|
import pathlib
|
|
import ssl
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
|
|
class ClientError(Exception):
|
|
pass
|
|
|
|
|
|
class _Http:
|
|
def __init__(self, base: str, token_file: str, ca_file: str | None = None, timeout: float = 15):
|
|
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")
|
|
self.base = base.rstrip("/")
|
|
self.token_file = token_file
|
|
context = ssl.create_default_context(cafile=ca_file) if ca_file else ssl.create_default_context()
|
|
self.open = urllib.request.build_opener(urllib.request.ProxyHandler({}),
|
|
urllib.request.HTTPSHandler(context=context)).open
|
|
self.timeout = timeout
|
|
|
|
def call(self, method: str, path: str, body: dict | None = None, headers: dict | None = None) -> tuple[int, dict]:
|
|
head = {"Authorization": "Bearer " + pathlib.Path(self.token_file).read_text().strip(),
|
|
"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:
|
|
"""GET /v1/approvals/{id} and POST /v1/approvals/{id}/consume (approval-engine api.py)."""
|
|
|
|
def __init__(self, base: str, token_file: str, ca_file: str | None = None):
|
|
self.http = _Http(base, token_file, ca_file)
|
|
|
|
@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))[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})[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:
|
|
status, _ = self.http.call("POST", "/v1/events", event, {"Idempotency-Key": event["id"]})
|
|
if status not in (200, 201):
|
|
raise ClientError(f"audit {status}")
|