From 2bd2d19a98988eb2c1229c197efdffb0ec764621 Mon Sep 17 00:00:00 2001 From: tegwick Date: Wed, 2 Sep 2026 00:52:04 +0200 Subject: [PATCH] Implement approval engine production readiness Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a05e2e-805b-7042-a750-71f473bceea2 --- Containerfile | 19 ++ Makefile | 8 +- README.md | 13 +- WORK-RECORDS.md | 12 +- approval_engine/__init__.py | 4 + approval_engine/api.py | 80 +++++++-- approval_engine/audit.py | 100 +++++++++++ approval_engine/auth.py | 169 ++++++++++++++++++ approval_engine/cli.py | 150 ++++++++++++++-- approval_engine/errors.py | 10 ++ approval_engine/pep.py | 114 ++++++++++++ approval_engine/store.py | 162 ++++++++++++++++- deploy/README.md | 26 +++ deploy/approval-engine.yaml | 118 ++++++++++++ deploy/networkpolicies.yaml | 60 +++++++ docs/caller-authentication.md | 34 ++++ docs/emission-cadence.md | 5 +- docs/outbox-contract.md | 15 +- docs/pep-integration.md | 20 +++ docs/storage-operations.md | 31 ++++ pyproject.toml | 2 + tests/conftest.py | 25 ++- tests/test_api.py | 18 +- tests/test_audit.py | 82 +++++++++ tests/test_auth.py | 167 +++++++++++++++++ tests/test_cli.py | 22 +++ tests/test_outbox.py | 16 ++ tests/test_pep.py | 105 +++++++++++ tests/test_storage.py | 55 ++++++ ...duction-readiness-and-consumer-adoption.md | 90 +++++++++- 30 files changed, 1679 insertions(+), 53 deletions(-) create mode 100644 Containerfile create mode 100644 approval_engine/audit.py create mode 100644 approval_engine/auth.py create mode 100644 approval_engine/pep.py create mode 100644 deploy/README.md create mode 100644 deploy/approval-engine.yaml create mode 100644 deploy/networkpolicies.yaml create mode 100644 docs/caller-authentication.md create mode 100644 docs/pep-integration.md create mode 100644 docs/storage-operations.md create mode 100644 tests/test_audit.py create mode 100644 tests/test_auth.py create mode 100644 tests/test_cli.py create mode 100644 tests/test_pep.py create mode 100644 tests/test_storage.py diff --git a/Containerfile b/Containerfile new file mode 100644 index 0000000..fa05b1f --- /dev/null +++ b/Containerfile @@ -0,0 +1,19 @@ +FROM python:3.12-slim@sha256:d764629ce0ddd8c71fd371e9901efb324a95789d2315a47db7e4d27e78f1b0e9 + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + PATH=/opt/venv/bin:$PATH + +RUN python -m venv /opt/venv \ + && addgroup --system --gid 10001 approval \ + && adduser --system --uid 10001 --ingroup approval --no-create-home approval + +WORKDIR /app +COPY pyproject.toml README.md ./ +COPY approval_engine ./approval_engine +RUN pip install --no-cache-dir '.[serve]' + +USER 10001:10001 +EXPOSE 8080 +ENTRYPOINT ["approval-engine"] +CMD ["serve", "--help"] diff --git a/Makefile b/Makefile index a608802..aa61807 100644 --- a/Makefile +++ b/Makefile @@ -4,8 +4,14 @@ SHELL := /usr/bin/env bash test: ## Run unit tests python3 -m pytest -q +image-build: ## Build the production image locally + docker build -f Containerfile -t approval-engine:local . + +deploy-dry-run: ## Validate Kubernetes manifests without applying them + kubectl apply --dry-run=client -f deploy/approval-engine.yaml -f deploy/networkpolicies.yaml + help: ## Show this help @awk 'BEGIN {FS = ":.*##"; printf "\nUsage:\n make \033[36m\033[0m\n"} \ /^[a-zA-Z_-]+:.*?##/ { printf " \033[36m%-24s\033[0m %s\n", $$1, $$2 }' $(MAKEFILE_LIST) -.PHONY: test help +.PHONY: test image-build deploy-dry-run help diff --git a/README.md b/README.md index 9c0484e..4f2d8ae 100644 --- a/README.md +++ b/README.md @@ -27,9 +27,20 @@ layer model. ```bash make test -python3 -m approval_engine.cli serve --db approvals.sqlite +approval-engine migrate --db approvals.sqlite +approval-engine verify --db approvals.sqlite +approval-engine serve --db approvals.sqlite \ + --jwt-issuer https://auth.netkingdom.local \ + --jwt-audience approval-engine \ + --jwks-url http://key-cape.sso.svc.cluster.local:8080/jwks ``` `POST /v1/approvals/{id}/consume` implements `GH-DEC-2026-003`: the PEP atomically spends the approval before the protected side effect. Same-digest retries are idempotent; a different digest conflicts. + +Production operations, caller scopes, audit delivery, and PEP sequencing are +documented in `docs/storage-operations.md`, `docs/caller-authentication.md`, +`docs/outbox-contract.md`, and `docs/pep-integration.md`. The checked-in +StatefulSet deliberately refuses production startup without a migrated +persistent store, KeyCape JWT verification, and authenticated audit delivery. diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 2b368aa..723d195 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -9,7 +9,7 @@ | Kind | ID | Status | Lane | Source | | --- | --- | --- | --- | --- | | workplan | APPROVAL-WP-0001 | finished | — | workplans/APPROVAL-WP-0001-v07-alignment-and-engine-spine.md | -| workplan | APPROVAL-WP-0002 | proposed | — | workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md | +| workplan | APPROVAL-WP-0002 | active | — | workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md | | task | APPROVAL-WP-0001-T01 | done | — | workplans/APPROVAL-WP-0001-v07-alignment-and-engine-spine.md | | task | APPROVAL-WP-0001-T02 | done | — | workplans/APPROVAL-WP-0001-v07-alignment-and-engine-spine.md | | task | APPROVAL-WP-0001-T03 | done | — | workplans/APPROVAL-WP-0001-v07-alignment-and-engine-spine.md | @@ -19,9 +19,9 @@ | task | APPROVAL-WP-0001-T07 | done | — | workplans/APPROVAL-WP-0001-v07-alignment-and-engine-spine.md | | task | APPROVAL-WP-0001-T08 | done | — | workplans/APPROVAL-WP-0001-v07-alignment-and-engine-spine.md | | task | APPROVAL-WP-0001-T09 | done | — | workplans/APPROVAL-WP-0001-v07-alignment-and-engine-spine.md | -| task | APPROVAL-WP-0002-T01 | todo | — | workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md | -| task | APPROVAL-WP-0002-T02 | todo | — | workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md | -| task | APPROVAL-WP-0002-T03 | todo | — | workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md | -| task | APPROVAL-WP-0002-T04 | todo | — | workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md | -| task | APPROVAL-WP-0002-T05 | todo | — | workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md | +| task | APPROVAL-WP-0002-T01 | progress | — | workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md | +| task | APPROVAL-WP-0002-T02 | done | — | workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md | +| task | APPROVAL-WP-0002-T03 | wait | — | workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md | +| task | APPROVAL-WP-0002-T04 | wait | — | workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md | +| task | APPROVAL-WP-0002-T05 | wait | — | workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md | | intake | APPROVAL-IN-0001 | open | blue | intakes/intakes.md | diff --git a/approval_engine/__init__.py b/approval_engine/__init__.py index 12dfc66..27ac5f5 100644 --- a/approval_engine/__init__.py +++ b/approval_engine/__init__.py @@ -5,8 +5,10 @@ from .errors import ( ApprovalError, Conflict, DuplicateApprover, + Forbidden, NotFound, StoreUnavailable, + Unauthenticated, Unprocessable, ) from .store import Engine @@ -16,8 +18,10 @@ __all__ = [ "ApprovalError", "Conflict", "DuplicateApprover", + "Forbidden", "NotFound", "StoreUnavailable", + "Unauthenticated", "Unprocessable", "binding_digest", "canonical_binding", diff --git a/approval_engine/api.py b/approval_engine/api.py index afcff8f..ee2ab28 100644 --- a/approval_engine/api.py +++ b/approval_engine/api.py @@ -5,17 +5,25 @@ from __future__ import annotations import json from typing import Any, Callable -from .errors import ApprovalError +from .auth import Authenticator, DenyAllAuthenticator, Identity +from .errors import ApprovalError, Forbidden from .store import Engine FORBIDDEN_DECISION_KEYS = frozenset({"effect", "decision", "allow", "deny"}) def _read_json(environ: dict[str, Any]) -> dict[str, Any]: - length = int(environ.get("CONTENT_LENGTH") or 0) + try: + length = int(environ.get("CONTENT_LENGTH") or 0) + except (TypeError, ValueError) as exc: + raise ApprovalError("invalid content length") from exc + if length < 0 or length > 256 * 1024: + raise ApprovalError("request body is too large") if length == 0: return {} raw = environ["wsgi.input"].read(length) + if len(raw) != length: + raise ApprovalError("truncated request body") if not raw: return {} try: @@ -40,8 +48,24 @@ def _assert_not_decision(payload: Any) -> None: class App: - def __init__(self, engine: Engine) -> None: + def __init__( + self, + engine: Engine, + authenticator: Authenticator | None = None, + *, + require_persistent: bool = False, + ) -> None: self.engine = engine + self.authenticator = authenticator or DenyAllAuthenticator() + self.require_persistent = require_persistent + + def identity(self, environ: dict[str, Any], scope: str) -> Identity: + identity = self.authenticator.authenticate( + environ.get("HTTP_AUTHORIZATION") + ).require(scope) + if identity.tenant != self.engine.tenant: + raise Forbidden("caller tenant does not match this approval store") + return identity def __call__(self, environ: dict[str, Any], start_response: Callable) -> list[bytes]: method = environ.get("REQUEST_METHOD", "GET").upper() @@ -64,19 +88,38 @@ class App: return [payload] def dispatch(self, method: str, path: str, environ: dict[str, Any]) -> tuple[int, dict[str, Any]]: + if path in ("/healthz", "/v1/healthz") and method == "GET": + return 200, {"status": "ok"} if path in ("/readyz", "/v1/readyz") and method == "GET": + storage = self.engine.storage_status() + ready = storage["schema_current"] and ( + storage["persistent"] or not self.require_persistent + ) self.engine.outbox_stats() - return 200, {"status": "ok", "store": "ok"} + return (200 if ready else 503), { + "status": "ok" if ready else "unavailable", + "store": "ok" if ready else "not-production-ready", + } + if path == "/v1/storage/status" and method == "GET": + self.identity(environ, "approval:observe") + return 200, self.engine.storage_status(integrity=True) if path == "/v1/cadence" and method == "GET": + self.identity(environ, "approval:observe") return 200, self.engine.transition_counts() | {"form": "heartbeat-or-reconciliation"} if path == "/v1/outbox/stats" and method == "GET": + self.identity(environ, "approval:observe") return 200, self.engine.outbox_stats() if path == "/v1/heartbeat" and method == "POST": + self.identity(environ, "approval:emit") return 200, self.engine.emit_heartbeat() if path == "/v1/approvals" and method == "POST": + identity = self.identity(environ, "approval:create") data = _read_json(environ) + binding = data.get("binding") or {} + if binding.get("actor") != identity.subject: + raise Forbidden("binding.actor must match the authenticated subject") obj = self.engine.create( - data.get("binding") or {}, + binding, data.get("validity") or {}, int(data.get("required_count") or 1), pdp_digest=data.get("pdp_digest"), @@ -88,24 +131,32 @@ class App: approval_id = parts[2] rest = parts[3:] if not rest and method == "GET": + self.identity(environ, "approval:read") return 200, self.engine.get(approval_id).as_dict() if rest == ["claim"] and method == "GET": + self.identity(environ, "approval:read") return 200, self.engine.claim(approval_id) if rest == ["entries"] and method == "POST": - data = _read_json(environ) + identity = self.identity(environ, "approval:approve") + _read_json(environ) obj = self.engine.add_entry( approval_id, - data.get("subject_id") or "", - assurance=data.get("assurance"), - evidence_ref=data.get("evidence_ref"), + identity.subject, + assurance=json.dumps(identity.assurance, sort_keys=True), + evidence_ref=identity.evidence_ref, ) return 200, obj.as_dict() if rest == ["revoke"] and method == "POST": + self.identity(environ, "approval:revoke") return 200, self.engine.revoke(approval_id).as_dict() if rest == ["supersede"] and method == "POST": + self.identity(environ, "approval:supersede") data = _read_json(environ) return 200, self.engine.supersede(approval_id, data.get("successor_id")) if rest == ["consume"] and method == "POST": + identity = self.identity(environ, "approval:consume") + if identity.principal_type not in {"service", "agent"}: + raise Forbidden("consume requires a service or agent principal") data = _read_json(environ) return 200, self.engine.consume( approval_id, @@ -117,7 +168,14 @@ class App: return 404, {"error": "not_found", "message": path} -def call(app: App, method: str, path: str, body: dict[str, Any] | None = None) -> tuple[int, dict[str, Any]]: +def call( + app: App, + method: str, + path: str, + body: dict[str, Any] | None = None, + *, + authorization: str | None = "Bearer test-token", +) -> tuple[int, dict[str, Any]]: """In-process WSGI helper for tests.""" raw = json.dumps(body or {}).encode("utf-8") if body is not None else b"" environ = { @@ -127,6 +185,8 @@ def call(app: App, method: str, path: str, body: dict[str, Any] | None = None) - "CONTENT_LENGTH": str(len(raw)) if body is not None else "0", "QUERY_STRING": "", } + if authorization is not None: + environ["HTTP_AUTHORIZATION"] = authorization status_headers: list[tuple[str, list]] = [] def start_response(status: str, headers: list[tuple[str, str]]) -> None: diff --git a/approval_engine/audit.py b/approval_engine/audit.py new file mode 100644 index 0000000..26053a1 --- /dev/null +++ b/approval_engine/audit.py @@ -0,0 +1,100 @@ +"""Asynchronous delivery of the transactional outbox to audit-core.""" + +from __future__ import annotations + +import json +import threading +from pathlib import Path +from typing import Any, Callable +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + +from .store import Engine + + +class AuditDeliveryError(RuntimeError): + """Bounded delivery failure; details deliberately exclude response bodies.""" + + +def audit_envelope(payload: dict[str, Any]) -> dict[str, Any]: + details = dict(payload.get("details") or {}) + return { + "id": payload["event_id"], + "type": payload["action"], + "source": payload["source"], + "subject": payload["resource"], + "tenant": payload["tenant"], + "correlation_id": details.get("approval_id") or payload["event_id"], + "occurred_at": payload["observed_at"], + "data": { + "schema_version": payload["schema_version"], + "scope": payload["scope"], + "actor": payload.get("actor"), + "resource": payload["resource"], + "outcome": payload["outcome"], + "reason": payload.get("reason"), + "details": details, + }, + } + + +class AuditCoreSink: + def __init__( + self, + base_url: str, + token_file: str | Path, + *, + timeout_seconds: float = 5, + opener: Callable[..., Any] = urlopen, + ) -> None: + self.url = base_url.rstrip("/") + "/v1/events" + self.token_file = Path(token_file) + self.timeout_seconds = timeout_seconds + self.opener = opener + + def __call__(self, payload: dict[str, Any]) -> None: + token = self.token_file.read_text(encoding="utf-8").strip() + if not token or any(ch.isspace() for ch in token): + raise AuditDeliveryError("audit credential is unavailable") + body = json.dumps(audit_envelope(payload), sort_keys=True).encode("utf-8") + request = Request( + self.url, + data=body, + method="POST", + headers={ + "Authorization": f"Bearer {token}", + "Content-Type": "application/json", + "Idempotency-Key": payload["event_id"], + }, + ) + try: + response = self.opener(request, timeout=self.timeout_seconds) + status = int(response.getcode()) + response.close() + except (HTTPError, URLError, OSError) as exc: + raise AuditDeliveryError(type(exc).__name__) from exc + if status not in {200, 202}: + raise AuditDeliveryError(f"audit ingest returned status {status}") + + +class OutboxWorker: + def __init__( + self, + engine: Engine, + sink: Callable[[dict[str, Any]], None], + *, + heartbeat_interval_seconds: int = 300, + ) -> None: + self.engine = engine + self.sink = sink + self.heartbeat_interval_seconds = heartbeat_interval_seconds + + def run_once(self) -> dict[str, int]: + if self.engine.heartbeat_due(self.heartbeat_interval_seconds): + self.engine.emit_heartbeat() + return self.engine.drain(self.sink) + + def run_forever(self, stop: threading.Event, poll_seconds: float = 5) -> None: + while not stop.is_set(): + self.run_once() + stop.wait(poll_seconds) diff --git a/approval_engine/auth.py b/approval_engine/auth.py new file mode 100644 index 0000000..a6ee5cc --- /dev/null +++ b/approval_engine/auth.py @@ -0,0 +1,169 @@ +"""Verified caller identity for approval-engine's HTTP boundary. + +The engine enforces scopes issued by the identity owner; it does not decide who +ought to hold them. Production uses KeyCape's RS256/JWKS contract. Tests use an +explicit static verifier so no unverified claim can accidentally become a +production fallback. +""" + +from __future__ import annotations + +import hashlib +from dataclasses import dataclass +from typing import Any, Mapping, Protocol + +import jwt + +from .errors import Forbidden, Unauthenticated + + +@dataclass(frozen=True) +class Identity: + subject: str + issuer: str + audiences: tuple[str, ...] + principal_type: str + tenant: str + roles: frozenset[str] + scopes: frozenset[str] + assurance: dict[str, Any] + evidence_ref: str + + def require(self, scope: str) -> "Identity": + if scope not in self.scopes: + raise Forbidden(f"caller lacks required scope {scope!r}") + return self + + +class Authenticator(Protocol): + def authenticate(self, authorization: str | None) -> Identity: ... + + +def _bearer(authorization: str | None) -> str: + value = str(authorization or "") + if not value.startswith("Bearer "): + raise Unauthenticated("bearer token is required") + token = value[7:].strip() + if not token or any(ch.isspace() for ch in token): + raise Unauthenticated("bearer token is invalid") + return token + + +def _texts(value: object, name: str, *, allow_empty: bool = False) -> tuple[str, ...]: + if isinstance(value, str): + items = tuple(item for item in value.split() if item) + elif isinstance(value, (list, tuple)): + items = tuple(item for item in value if isinstance(item, str) and item) + if len(items) != len(value): + raise Unauthenticated(f"token claim {name!r} is invalid") + else: + raise Unauthenticated(f"token claim {name!r} is invalid") + if not items and not allow_empty: + raise Unauthenticated(f"token claim {name!r} is empty") + return items + + +def identity_from_claims(claims: Mapping[str, Any], token: str) -> Identity: + try: + subject = str(claims["sub"]) + issuer = str(claims["iss"]) + principal_type = str(claims["principal_type"]) + tenant = str(claims["tenant"]) + assurance = claims["assurance"] + except (KeyError, TypeError) as exc: + raise Unauthenticated("token is missing required identity claims") from exc + if not subject or not issuer or principal_type not in {"human", "service", "agent"}: + raise Unauthenticated("token identity claims are invalid") + if not tenant or not isinstance(assurance, dict): + raise Unauthenticated("token tenant or assurance claim is invalid") + audiences = _texts(claims.get("aud"), "aud") + roles = frozenset(_texts(claims.get("roles"), "roles", allow_empty=True)) + scopes = frozenset(_texts(claims.get("scope", claims.get("scp")), "scope")) + fingerprint = hashlib.sha256(token.encode("utf-8")).hexdigest() + return Identity( + subject=subject, + issuer=issuer, + audiences=audiences, + principal_type=principal_type, + tenant=tenant, + roles=roles, + scopes=scopes, + assurance=dict(assurance), + evidence_ref=f"jwt-sha256:{fingerprint}", + ) + + +class JWTAuthenticator: + """Verify KeyCape JWT signature, issuer, audience, time, and profile claims.""" + + def __init__( + self, + *, + issuer: str, + audience: str, + jwks_url: str, + leeway_seconds: int = 30, + timeout_seconds: int = 3, + jwks_client: Any | None = None, + ) -> None: + if not issuer or not audience or not jwks_url: + raise ValueError("issuer, audience, and jwks_url are required") + self.issuer = issuer + self.audience = audience + self.leeway_seconds = leeway_seconds + self.jwks_client = jwks_client or jwt.PyJWKClient( + jwks_url, + cache_keys=True, + cache_jwk_set=True, + lifespan=300, + timeout=timeout_seconds, + ) + + def authenticate(self, authorization: str | None) -> Identity: + token = _bearer(authorization) + try: + signing_key = self.jwks_client.get_signing_key_from_jwt(token) + claims = jwt.decode( + token, + signing_key.key, + algorithms=["RS256"], + issuer=self.issuer, + audience=self.audience, + leeway=self.leeway_seconds, + options={ + "require": [ + "iss", + "sub", + "aud", + "exp", + "iat", + "tenant", + "principal_type", + "roles", + "scope", + "assurance", + ] + }, + ) + except Exception as exc: + raise Unauthenticated("token verification failed") from exc + return identity_from_claims(claims, token) + + +class StaticTokenAuthenticator: + """Explicit test/development verifier; never selected implicitly.""" + + def __init__(self, identities: Mapping[str, Identity]) -> None: + self.identities = dict(identities) + + def authenticate(self, authorization: str | None) -> Identity: + token = _bearer(authorization) + identity = self.identities.get(token) + if identity is None: + raise Unauthenticated("token verification failed") + return identity + + +class DenyAllAuthenticator: + def authenticate(self, authorization: str | None) -> Identity: + raise Unauthenticated("caller authentication is not configured") diff --git a/approval_engine/cli.py b/approval_engine/cli.py index 34acfdf..1d44e14 100644 --- a/approval_engine/cli.py +++ b/approval_engine/cli.py @@ -1,26 +1,154 @@ from __future__ import annotations import argparse +import json +import threading +from pathlib import Path from wsgiref.simple_server import make_server from .api import App +from .audit import AuditCoreSink, OutboxWorker +from .auth import Identity, JWTAuthenticator, StaticTokenAuthenticator from .store import Engine +ALL_SCOPES = frozenset( + { + "approval:observe", + "approval:emit", + "approval:create", + "approval:read", + "approval:approve", + "approval:revoke", + "approval:supersede", + "approval:consume", + } +) + + +def _common_db(sub: argparse.ArgumentParser) -> None: + sub.add_argument("--db", default="approvals.sqlite") + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="approval-engine") + sub = parser.add_subparsers(dest="cmd", required=True) + serve = sub.add_parser("serve", help="serve the authenticated HTTP API") + _common_db(serve) + serve.add_argument("--host", default="127.0.0.1") + serve.add_argument("--port", type=int, default=8787) + serve.add_argument("--production", action="store_true") + serve.add_argument("--tenant", default="platform") + serve.add_argument("--jwt-issuer") + serve.add_argument("--jwt-audience") + serve.add_argument("--jwks-url") + serve.add_argument("--dev-token-file") + serve.add_argument("--audit-url") + serve.add_argument("--audit-token-file") + serve.add_argument("--outbox-poll-seconds", type=float, default=5) + serve.add_argument("--heartbeat-seconds", type=int, default=86400) + migrate = sub.add_parser("migrate", help="apply repeatable schema migrations") + _common_db(migrate) + verify = sub.add_parser("verify", help="verify schema and SQLite integrity") + _common_db(verify) + backup = sub.add_parser("backup", help="create and verify an online backup") + _common_db(backup) + backup.add_argument("--output", required=True) + return parser + + +def _authenticator(args: argparse.Namespace, parser: argparse.ArgumentParser): + jwt_values = (args.jwt_issuer, args.jwt_audience, args.jwks_url) + if all(jwt_values): + return JWTAuthenticator( + issuer=args.jwt_issuer, + audience=args.jwt_audience, + jwks_url=args.jwks_url, + ) + if any(jwt_values): + parser.error("--jwt-issuer, --jwt-audience, and --jwks-url are a set") + if args.production: + parser.error("production requires KeyCape JWT verifier configuration") + if not args.dev_token_file: + parser.error("configure JWT verification or explicit --dev-token-file") + token = Path(args.dev_token_file).read_text(encoding="utf-8").strip() + if not token: + parser.error("development token file is empty") + identity = Identity( + subject="development-agent", + issuer="local-development", + audiences=("approval-engine",), + principal_type="agent", + tenant="development", + roles=frozenset({"developer"}), + scopes=ALL_SCOPES, + assurance={"method": "explicit-development-token"}, + evidence_ref="local-development-token", + ) + return StaticTokenAuthenticator({token: identity}) + + +def _serve(args: argparse.Namespace, parser: argparse.ArgumentParser) -> int: + if args.production and args.db == ":memory:": + parser.error("production requires a persistent database") + if bool(args.audit_url) != bool(args.audit_token_file): + parser.error("--audit-url and --audit-token-file are a set") + if args.production and not args.audit_url: + parser.error("production requires authenticated audit delivery") + authenticator = _authenticator(args, parser) + engine = Engine(args.db, auto_migrate=not args.production, tenant=args.tenant) + storage = engine.storage_status(integrity=True) + if args.production and (not storage["ok"] or not storage["persistent"]): + parser.error("production store is not ready; run migrate and verify") + app = App(engine, authenticator, require_persistent=args.production) + stop = threading.Event() + if args.audit_url: + sink = AuditCoreSink(args.audit_url, args.audit_token_file) + worker = OutboxWorker( + engine, + sink, + heartbeat_interval_seconds=args.heartbeat_seconds, + ) + threading.Thread( + target=worker.run_forever, + args=(stop, args.outbox_poll_seconds), + daemon=True, + name="approval-outbox", + ).start() + print(f"approval-engine listening on {args.host}:{args.port} db={args.db}") + try: + if args.production: + from waitress import serve + + serve(app, host=args.host, port=args.port, threads=4) + else: + make_server(args.host, args.port, app).serve_forever() + finally: + stop.set() + engine.close() + return 0 + def main(argv: list[str] | None = None) -> int: - parser = argparse.ArgumentParser(prog="approval-engine") - sub = parser.add_subparsers(dest="cmd", required=True) - serve = sub.add_parser("serve", help="serve the introspection API") - serve.add_argument("--db", default="approvals.sqlite") - serve.add_argument("--host", default="127.0.0.1") - serve.add_argument("--port", type=int, default=8787) + parser = _parser() args = parser.parse_args(argv) - if args.cmd == "serve": - engine = Engine(args.db) - httpd = make_server(args.host, args.port, App(engine)) - print(f"approval-engine on http://{args.host}:{args.port} db={args.db}") - httpd.serve_forever() + if args.cmd == "migrate": + engine = Engine(args.db, auto_migrate=True) + print(json.dumps(engine.storage_status(integrity=True), sort_keys=True)) + engine.close() return 0 + if args.cmd == "verify": + engine = Engine(args.db, auto_migrate=False) + status = engine.storage_status(integrity=True) + print(json.dumps(status, sort_keys=True)) + engine.close() + return 0 if status["ok"] else 1 + if args.cmd == "backup": + engine = Engine(args.db, auto_migrate=False) + print(json.dumps(engine.backup(args.output), sort_keys=True)) + engine.close() + return 0 + if args.cmd == "serve": + return _serve(args, parser) return 2 diff --git a/approval_engine/errors.py b/approval_engine/errors.py index 453e607..ebc5839 100644 --- a/approval_engine/errors.py +++ b/approval_engine/errors.py @@ -3,6 +3,16 @@ class ApprovalError(Exception): reason_code = "error" +class Unauthenticated(ApprovalError): + http_status = 401 + reason_code = "unauthenticated" + + +class Forbidden(ApprovalError): + http_status = 403 + reason_code = "forbidden" + + class NotFound(ApprovalError): http_status = 404 reason_code = "not_found" diff --git a/approval_engine/pep.py b/approval_engine/pep.py new file mode 100644 index 0000000..269476a --- /dev/null +++ b/approval_engine/pep.py @@ -0,0 +1,114 @@ +"""Fail-closed PEP client and sequencing harness for GH-DEC-2026-003.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Callable +from urllib.error import HTTPError, URLError +from urllib.parse import quote +from urllib.request import Request, urlopen + + +class ApprovalProtocolError(RuntimeError): + pass + + +class ApprovalHTTPClient: + def __init__( + self, + base_url: str, + token_file: str | Path, + *, + timeout_seconds: float = 3, + opener: Callable[..., Any] = urlopen, + ) -> None: + self.base_url = base_url.rstrip("/") + self.token_file = Path(token_file) + self.timeout_seconds = timeout_seconds + self.opener = opener + + def _request( + self, method: str, path: str, body: dict[str, Any] | None = None + ) -> dict[str, Any]: + token = self.token_file.read_text(encoding="utf-8").strip() + if not token or any(ch.isspace() for ch in token): + raise ApprovalProtocolError("approval credential is unavailable") + encoded = None if body is None else json.dumps(body).encode("utf-8") + request = Request( + self.base_url + path, + data=encoded, + method=method, + headers={ + "Authorization": f"Bearer {token}", + "Content-Type": "application/json", + }, + ) + try: + response = self.opener(request, timeout=self.timeout_seconds) + status = int(response.getcode()) + raw = response.read(256 * 1024 + 1) + response.close() + except (HTTPError, URLError, OSError) as exc: + raise ApprovalProtocolError(type(exc).__name__) from exc + if status != 200 or len(raw) > 256 * 1024: + raise ApprovalProtocolError(f"approval endpoint returned status {status}") + try: + result = json.loads(raw) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise ApprovalProtocolError("approval endpoint returned invalid JSON") from exc + if not isinstance(result, dict): + raise ApprovalProtocolError("approval endpoint returned invalid payload") + return result + + def claim(self, approval_id: str) -> dict[str, Any]: + encoded_id = quote(approval_id, safe="") + return self._request("GET", f"/v1/approvals/{encoded_id}/claim") + + def consume( + self, approval_id: str, request_digest: str, decision_id: str + ) -> dict[str, Any]: + encoded_id = quote(approval_id, safe="") + return self._request( + "POST", + f"/v1/approvals/{encoded_id}/consume", + {"request_digest": request_digest, "decision_id": decision_id}, + ) + + +class ProtectedActionHarness: + """Sequence a supplied PDP decision and dry-run/protected callback. + + The decision callback owns authorization. This class only enforces that a + fresh approval claim precedes it and a successful CAS consume precedes the + side effect. + """ + + def __init__(self, client: ApprovalHTTPClient) -> None: + self.client = client + + def execute( + self, + approval_id: str, + request_digest: str, + decide: Callable[[dict[str, Any]], dict[str, Any]], + side_effect: Callable[[], Any], + ) -> Any: + claim = self.client.claim(approval_id) + if claim.get("valid_now") is not True or claim.get("consumed") is not False: + raise ApprovalProtocolError("approval claim is not valid for use") + decision = decide(claim) + if decision.get("effect") != "ALLOW": + raise ApprovalProtocolError("authorization decision did not allow") + decision_id = decision.get("decision_id") + if not isinstance(decision_id, str) or not decision_id: + raise ApprovalProtocolError("authorization decision lacks decision_id") + if decision.get("request_digest") != request_digest: + raise ApprovalProtocolError("authorization decision digest does not match") + consumed = self.client.consume(approval_id, request_digest, decision_id) + if ( + consumed.get("status") != "consumed" + or consumed.get("request_digest") != request_digest + ): + raise ApprovalProtocolError("approval consumption was not confirmed") + return side_effect() diff --git a/approval_engine/store.py b/approval_engine/store.py index 8beaaa5..bf2c63d 100644 --- a/approval_engine/store.py +++ b/approval_engine/store.py @@ -8,8 +8,10 @@ request digest before the protected side effect. from __future__ import annotations import json +import os import sqlite3 import threading +import tempfile from dataclasses import dataclass from datetime import datetime, timedelta, timezone from pathlib import Path @@ -33,6 +35,7 @@ AUDIT_SCHEMA = "audit-core.event.v1alpha1" SOURCE = "approval-engine" SCOPE = "netkingdom-approvals" EVENT_CLASSES = ("issuance", "use", "supersession", "revocation", "heartbeat") +LATEST_SCHEMA_VERSION = 2 SCHEMA = """ CREATE TABLE IF NOT EXISTS approvals ( @@ -70,7 +73,10 @@ CREATE TABLE IF NOT EXISTS outbox ( approval_id TEXT, payload_json TEXT NOT NULL, created_at TEXT NOT NULL, - drained_at TEXT + drained_at TEXT, + attempts INTEGER NOT NULL DEFAULT 0, + last_attempt_at TEXT, + last_error TEXT ); """ @@ -157,12 +163,14 @@ class Engine: freshness_ttl: int = DEFAULT_FRESHNESS_TTL, fail_outbox: bool = False, tenant: str = "platform", + auto_migrate: bool = True, ) -> None: self.path = str(path) self.clock = clock self.freshness_ttl = freshness_ttl self.fail_outbox = fail_outbox self.tenant = tenant + self.auto_migrate = auto_migrate self._local = threading.local() self._init_schema() @@ -188,13 +196,40 @@ class Engine: def _init_schema(self) -> None: conn = self._conn() try: + current = int(conn.execute("PRAGMA user_version").fetchone()[0]) + if not self.auto_migrate: + tables = { + row[0] + for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ).fetchall() + } + if current != LATEST_SCHEMA_VERSION: + raise StoreUnavailable( + f"database schema version {current} is not {LATEST_SCHEMA_VERSION}; " + "run approval-engine migrate" + ) + if not {"approvals", "entries", "outbox"} <= tables: + raise StoreUnavailable("database schema is incomplete") + return conn.executescript(SCHEMA) - columns = { + approval_columns = { row["name"] for row in conn.execute("PRAGMA table_info(approvals)").fetchall() } for name in ("consumed_digest", "consumed_decision_id", "consumed_at"): - if name not in columns: + if name not in approval_columns: conn.execute(f"ALTER TABLE approvals ADD COLUMN {name} TEXT") + outbox_columns = { + row["name"] for row in conn.execute("PRAGMA table_info(outbox)").fetchall() + } + if "attempts" not in outbox_columns: + conn.execute( + "ALTER TABLE outbox ADD COLUMN attempts INTEGER NOT NULL DEFAULT 0" + ) + for name in ("last_attempt_at", "last_error"): + if name not in outbox_columns: + conn.execute(f"ALTER TABLE outbox ADD COLUMN {name} TEXT") + conn.execute(f"PRAGMA user_version={LATEST_SCHEMA_VERSION}") conn.commit() except sqlite3.Error as exc: raise StoreUnavailable(str(exc)) from exc @@ -208,6 +243,82 @@ class Engine: def now(self) -> datetime: return utc_now(self.clock) + def storage_status(self, *, integrity: bool = False) -> dict[str, Any]: + conn = self._conn() + try: + version = int(conn.execute("PRAGMA user_version").fetchone()[0]) + result: dict[str, Any] = { + "schema_version": version, + "expected_schema_version": LATEST_SCHEMA_VERSION, + "schema_current": version == LATEST_SCHEMA_VERSION, + "persistent": self.path != ":memory:", + } + tables = { + row[0] + for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ).fetchall() + } + result["required_tables"] = { + name: name in tables for name in ("approvals", "entries", "outbox") + } + if integrity: + check = [ + row[0] for row in conn.execute("PRAGMA integrity_check").fetchall() + ] + foreign_keys = conn.execute("PRAGMA foreign_key_check").fetchall() + result["integrity"] = check + result["foreign_key_violations"] = len(foreign_keys) + result["ok"] = ( + check == ["ok"] + and not foreign_keys + and result["schema_current"] + and all(result["required_tables"].values()) + ) + return result + except sqlite3.Error as exc: + raise StoreUnavailable(str(exc)) from exc + + def backup(self, output: str | Path) -> dict[str, Any]: + if self.path == ":memory:": + raise Unprocessable("an in-memory database cannot be backed up") + target = Path(output) + if target.exists(): + raise Conflict(f"backup target already exists: {target}") + if not target.parent.exists(): + raise Unprocessable(f"backup parent does not exist: {target.parent}") + handle = tempfile.NamedTemporaryFile( + prefix=target.name + ".", + suffix=".tmp", + dir=target.parent, + delete=False, + ) + temporary = Path(handle.name) + handle.close() + os.chmod(temporary, 0o600) + destination: sqlite3.Connection | None = None + try: + destination = sqlite3.connect(temporary) + self._conn().backup(destination) + destination.close() + destination = None + check = sqlite3.connect(temporary) + try: + integrity = [ + row[0] for row in check.execute("PRAGMA integrity_check") + ] + finally: + check.close() + if integrity != ["ok"]: + raise StoreUnavailable("backup integrity verification failed") + os.replace(temporary, target) + except Exception: + if destination is not None: + destination.close() + temporary.unlink(missing_ok=True) + raise + return {"path": str(target), "mode": "0600", "integrity": "ok"} + def create( self, binding: dict[str, Any], @@ -728,7 +839,8 @@ class Engine: now = self.now() try: rows = conn.execute( - "SELECT event_id, class, created_at, drained_at FROM outbox" + "SELECT event_id, class, created_at, drained_at, attempts, last_error " + "FROM outbox" ).fetchall() except sqlite3.Error as exc: raise StoreUnavailable(str(exc)) from exc @@ -738,14 +850,32 @@ class Engine: "total": len(rows), "pending": len(pending), "max_age_seconds": max(ages) if ages else 0, + "attempts": sum(int(row["attempts"]) for row in rows), + "failed_pending": sum( + 1 for row in pending if row["last_error"] is not None + ), "counts": self.transition_counts(), } + def heartbeat_due(self, interval_seconds: int) -> bool: + if interval_seconds < 1: + raise Unprocessable("heartbeat interval must be positive") + try: + row = self._conn().execute( + "SELECT MAX(created_at) AS created_at FROM outbox WHERE class='heartbeat'" + ).fetchone() + except sqlite3.Error as exc: + raise StoreUnavailable(str(exc)) from exc + if row["created_at"] is None: + return True + return (self.now() - parse_iso(row["created_at"])).total_seconds() >= interval_seconds + def undrained(self) -> list[dict[str, Any]]: conn = self._conn() try: rows = conn.execute( - "SELECT event_id, class, approval_id, payload_json, created_at " + "SELECT event_id, class, approval_id, payload_json, created_at, " + "attempts, last_attempt_at, last_error " "FROM outbox WHERE drained_at IS NULL ORDER BY created_at" ).fetchall() except sqlite3.Error as exc: @@ -757,6 +887,9 @@ class Engine: "approval_id": r["approval_id"], "payload": json.loads(r["payload_json"]), "created_at": r["created_at"], + "attempts": r["attempts"], + "last_attempt_at": r["last_attempt_at"], + "last_error": r["last_error"], } for r in rows ] @@ -770,15 +903,28 @@ class Engine: failed = 0 conn = self._conn() for item in self.undrained(): + attempted_at = iso(self.now()) try: sink(item["payload"]) - except Exception: + except Exception as exc: failed += 1 + try: + conn.execute( + "UPDATE outbox SET attempts=attempts+1, last_attempt_at=?, " + "last_error=? WHERE event_id=? AND drained_at IS NULL", + (attempted_at, type(exc).__name__, item["event_id"]), + ) + conn.commit() + except sqlite3.Error as db_exc: + conn.rollback() + raise StoreUnavailable(str(db_exc)) from db_exc continue try: conn.execute( - "UPDATE outbox SET drained_at=? WHERE event_id=? AND drained_at IS NULL", - (iso(self.now()), item["event_id"]), + "UPDATE outbox SET drained_at=?, attempts=attempts+1, " + "last_attempt_at=?, last_error=NULL " + "WHERE event_id=? AND drained_at IS NULL", + (iso(self.now()), attempted_at, item["event_id"]), ) conn.commit() delivered += 1 diff --git a/deploy/README.md b/deploy/README.md new file mode 100644 index 0000000..4fe75e8 --- /dev/null +++ b/deploy/README.md @@ -0,0 +1,26 @@ +# Deployment gates + +The checked-in StatefulSet is a reviewed release input, not evidence of a live +deployment. Replace both image placeholders with the same immutable digest. +SQLite is intentionally limited to one replica, `ReadWriteOnce` storage, and an +`OnDelete` update: never start two writers against a copied database. + +Before applying: + +1. Register the exact `approval-engine` audience, caller scopes, and service + clients in KeyCape. Confirm the in-cluster JWKS endpoint and configured + issuer match the manifest. +2. Register `approval-engine` as an audit-core sender, add matching audit-core + ingress, and provision `approval-engine-audit` through the credential owner. + The Secret must contain key `audit-token`; never commit its value. +3. Build and scan the image, replace `REPLACE_WITH_RELEASE_DIGEST`, then run + `make deploy-dry-run`. +4. Take a verified backup. Roll out by deleting the sole pod, then prove + readiness, restart persistence, outbox drain, heartbeat, and restore. +5. Enable a caller namespace only by applying label + `railiance.io/approval-engine-client=true`; JWT scope checks remain the inner + boundary. + +The audit-core receiver-side registration, cadence findings, and accepted-count +reconciliation are tracked by `AUDIT-WP-0009-T04/T06/T09` and cannot be closed +from this repository. diff --git a/deploy/approval-engine.yaml b/deploy/approval-engine.yaml new file mode 100644 index 0000000..586164f --- /dev/null +++ b/deploy/approval-engine.yaml @@ -0,0 +1,118 @@ +apiVersion: v1 +kind: Namespace +metadata: + name: approval-engine + labels: + kubernetes.io/metadata.name: approval-engine + railiance.io/workload-class: platform +--- +apiVersion: v1 +kind: Service +metadata: + name: approval-engine + namespace: approval-engine +spec: + selector: + app.kubernetes.io/name: approval-engine + ports: + - {name: http, port: 8080, targetPort: http, protocol: TCP} +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: approval-engine + namespace: approval-engine +spec: + serviceName: approval-engine + replicas: 1 + podManagementPolicy: OrderedReady + updateStrategy: {type: OnDelete} + selector: + matchLabels: + app.kubernetes.io/name: approval-engine + template: + metadata: + labels: + app.kubernetes.io/name: approval-engine + spec: + securityContext: + runAsNonRoot: true + runAsUser: 10001 + fsGroup: 10001 + seccompProfile: {type: RuntimeDefault} + initContainers: + - name: migrate + image: forgejo.coulomb.social/coulomb/approval-engine@sha256:REPLACE_WITH_RELEASE_DIGEST + args: ["migrate", "--db", "/data/approvals.sqlite"] + securityContext: + allowPrivilegeEscalation: false + capabilities: {drop: ["ALL"]} + readOnlyRootFilesystem: true + volumeMounts: + - {name: data, mountPath: /data} + - {name: tmp, mountPath: /tmp} + containers: + - name: approval-engine + image: forgejo.coulomb.social/coulomb/approval-engine@sha256:REPLACE_WITH_RELEASE_DIGEST + args: + - serve + - --production + - --db + - /data/approvals.sqlite + - --host + - 0.0.0.0 + - --port + - "8080" + - --tenant + - platform + - --jwt-issuer + - https://auth.netkingdom.local + - --jwt-audience + - approval-engine + - --jwks-url + - http://key-cape.sso.svc.cluster.local:8080/jwks + - --audit-url + - http://audit-core.audit-core.svc.cluster.local:8080 + - --audit-token-file + - /var/run/secrets/approval-engine/audit-token + ports: + - {name: http, containerPort: 8080} + resources: + requests: {cpu: 50m, memory: 64Mi} + limits: {cpu: 500m, memory: 256Mi} + securityContext: + allowPrivilegeEscalation: false + capabilities: {drop: ["ALL"]} + readOnlyRootFilesystem: true + startupProbe: + httpGet: {path: /healthz, port: http} + periodSeconds: 3 + failureThreshold: 20 + readinessProbe: + httpGet: {path: /readyz, port: http} + periodSeconds: 10 + timeoutSeconds: 2 + livenessProbe: + httpGet: {path: /healthz, port: http} + periodSeconds: 20 + timeoutSeconds: 2 + volumeMounts: + - {name: data, mountPath: /data} + - {name: tmp, mountPath: /tmp} + - name: audit-token + mountPath: /var/run/secrets/approval-engine + readOnly: true + volumes: + - name: tmp + emptyDir: {} + - name: audit-token + secret: + secretName: approval-engine-audit + defaultMode: 0440 + volumeClaimTemplates: + - metadata: + name: data + spec: + accessModes: ["ReadWriteOnce"] + resources: + requests: {storage: 1Gi} diff --git a/deploy/networkpolicies.yaml b/deploy/networkpolicies.yaml new file mode 100644 index 0000000..6301345 --- /dev/null +++ b/deploy/networkpolicies.yaml @@ -0,0 +1,60 @@ +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: approval-engine-default-deny + namespace: approval-engine +spec: + podSelector: {} + policyTypes: [Ingress, Egress] +--- +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: approval-engine-callers + namespace: approval-engine +spec: + podSelector: + matchLabels: + app.kubernetes.io/name: approval-engine + policyTypes: [Ingress] + ingress: + - from: + - namespaceSelector: + matchLabels: + kubernetes.io/metadata.name: secrets-engine + - namespaceSelector: + matchLabels: + railiance.io/approval-engine-client: "true" + ports: + - {protocol: TCP, port: 8080} +--- +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: approval-engine-egress + namespace: approval-engine +spec: + podSelector: + matchLabels: + app.kubernetes.io/name: approval-engine + policyTypes: [Egress] + egress: + - to: + - namespaceSelector: + matchLabels: + kubernetes.io/metadata.name: sso + ports: + - {protocol: TCP, port: 8080} + - to: + - namespaceSelector: + matchLabels: + kubernetes.io/metadata.name: audit-core + ports: + - {protocol: TCP, port: 8080} + - to: + - namespaceSelector: + matchLabels: + kubernetes.io/metadata.name: kube-system + ports: + - {protocol: UDP, port: 53} + - {protocol: TCP, port: 53} diff --git a/docs/caller-authentication.md b/docs/caller-authentication.md new file mode 100644 index 0000000..240c482 --- /dev/null +++ b/docs/caller-authentication.md @@ -0,0 +1,34 @@ +# Caller authentication + +Production accepts only RS256 JWTs verified against KeyCape JWKS with the exact +configured issuer and `approval-engine` audience. `exp`, `iat`, `sub`, +`principal_type`, `tenant`, `roles`, `scope`, and `assurance` are mandatory. +Missing or unverifiable credentials fail closed. The development static-token +mode is explicit, file-backed, and refused with `--production`. +The verified `tenant` must exactly match the service's configured store tenant; +cross-tenant reads and mutations are rejected before object lookup. + +| Route | Required scope | +|---|---| +| create approval | `approval:create` | +| get approval or claim | `approval:read` | +| add approval entry | `approval:approve` | +| revoke | `approval:revoke` | +| supersede | `approval:supersede` | +| consume | `approval:consume` and service/agent principal | +| cadence, outbox, storage | `approval:observe` | +| explicit heartbeat | `approval:emit` | + +Create additionally requires `binding.actor == sub`. Approval-entry subject, +assurance, and evidence reference are derived from the verified JWT, never the +request body. KeyCape owns client registration and scope grants; approval-engine +only verifies and enforces them. Requested registrations are: + +- audience/resource server `approval-engine` with the scopes above; +- the secrets-engine PEP service client with `approval:read` and + `approval:consume`; +- separately governed lifecycle/operator clients with only their needed + mutation or observation scopes. + +Client credentials belong in OpenBao/operator custody and must not be placed in +manifests, logs, State Hub, or this repository. diff --git a/docs/emission-cadence.md b/docs/emission-cadence.md index c59abbb..50482eb 100644 --- a/docs/emission-cadence.md +++ b/docs/emission-cadence.md @@ -14,9 +14,8 @@ surface; this declaration is what it reads. Machine-readable copy: ### Heartbeat -A signed positive claim: *nothing to report*, together with per-class -transition counts since the previous heartbeat (or since process start on -the first). The claim can itself go missing, which is the point — silence +An authenticated positive claim: *nothing to report*, together with cumulative +per-class committed transition counts. The claim can itself go missing, which is the point — silence becomes a missing positive rather than a quiet month. | Field | Value | diff --git a/docs/outbox-contract.md b/docs/outbox-contract.md index e27016d..83582ef 100644 --- a/docs/outbox-contract.md +++ b/docs/outbox-contract.md @@ -41,9 +41,12 @@ emitted class. The validity window is already on the object. | `event_id` | UUID | Stable. Drain retries reuse it. | | `class` | enum above | | | `approval_id` | UUID or null | Null only for heartbeat. | -| `payload` | object | The `audit-core.event.v1alpha1` record, ready to POST. | +| `payload` | object | The source record adapted to audit-core's HTTP envelope at drain time. | | `created_at` | RFC 3339 UTC | | | `drained_at` | RFC 3339 UTC or null | Set after a successful audit-core ack. | +| `attempts` | integer | Delivery attempts, including the successful attempt. | +| `last_attempt_at` | RFC 3339 UTC or null | Most recent delivery attempt. | +| `last_error` | class name or null | Bounded failure category; never exception text. | ## Payload (audit-core v1alpha1) @@ -76,9 +79,13 @@ approval-validity query; this payload does not invite one. ## Drain 1. Select undrained rows, oldest first. -2. POST each payload to `audit-core`. -3. On success, set `drained_at`. -4. On failure, leave the row; retry later. **Do not** roll back the object +2. Adapt it to audit-core's `{id,type,source,subject,tenant,correlation_id, + occurred_at,data}` envelope, using `event_id` as both `id` and the + `Idempotency-Key` header. +3. POST with the mounted sender credential, reread on every attempt. +4. On `202 accepted` or `200 duplicate`, set `drained_at`. +5. On failure, leave the row and record only a bounded failure class; retry + later. **Do not** roll back the object mutation — it already committed with the row. An `audit-core` outage therefore cannot block a revocation. This engine's diff --git a/docs/pep-integration.md b/docs/pep-integration.md new file mode 100644 index 0000000..2ef3d7c --- /dev/null +++ b/docs/pep-integration.md @@ -0,0 +1,20 @@ +# PEP integration sequence + +`approval_engine.pep` implements the fail-closed ordering from +`GH-DEC-2026-003` without becoming a PDP: + +1. fetch a fresh approval claim; +2. pass that claim to the consumer's authorization decision function; +3. require ALLOW, decision id, and the exact canonical request digest; +4. CAS-consume the approval; +5. only after confirmed consumption invoke the protected callback. + +Claim or consume unavailability, invalid/consumed claims, DENY, digest mismatch, +and consume conflicts all prevent the callback. A same-digest retry receives +the engine's idempotent success. If the callback fails after consume, the +approval stays spent; there is no unconsume. + +The module rereads the mounted bearer-token file on each HTTP request. Its unit +harness uses a dry-run callback and demonstrates the ordering, but live closure +requires the secrets-engine-owned handler to prove that no OpenBao request is +made in every failure case. diff --git a/docs/storage-operations.md b/docs/storage-operations.md new file mode 100644 index 0000000..e8d42e5 --- /dev/null +++ b/docs/storage-operations.md @@ -0,0 +1,31 @@ +# Storage operations + +Production uses one SQLite writer on persistent `ReadWriteOnce` storage. +Production serve disables automatic migration and refuses schema drift or an +in-memory database. + +```bash +approval-engine migrate --db /data/approvals.sqlite +approval-engine verify --db /data/approvals.sqlite +approval-engine backup --db /data/approvals.sqlite --output /backup/approval.sqlite +``` + +Migration is repeatable and sets an explicit `PRAGMA user_version`. Backup uses +SQLite's online backup API, verifies `PRAGMA integrity_check`, writes mode 0600, +and refuses to overwrite a target. + +Restore is a stopped-single-writer operation: + +1. Stop the StatefulSet and confirm no approval-engine or migration process has + the PVC open. +2. Preserve the failed database and its `-wal`/`-shm` companions for analysis. +3. Copy the verified backup to a new database path with owner 10001 and mode + 0600. Do not merge a backup with old WAL files. +4. Run `verify`, then `migrate` if the release schema is newer, then `verify` + again. +5. Start exactly one replica and prove approval/entry/outbox counts, readiness, + claim retrieval, and idempotent audit drain before reopening callers. + +Approval mutation and outbox insertion share `BEGIN IMMEDIATE` and one commit; +a failed outbox insert rolls the mutation back. Delivery occurs afterward and +does not roll back a committed mutation. diff --git a/pyproject.toml b/pyproject.toml index 80eb72a..ce365d8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,9 +4,11 @@ version = "0.1.0" description = "PIP for the approval object: durable, authenticated, consumable, atomically supersedable." readme = "README.md" requires-python = ">=3.11" +dependencies = ["PyJWT[crypto]>=2.7,<3"] [project.optional-dependencies] dev = ["pytest"] +serve = ["waitress>=3,<4"] [project.scripts] approval-engine = "approval_engine.cli:main" diff --git a/tests/conftest.py b/tests/conftest.py index fe8518f..473fad6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -3,6 +3,7 @@ from datetime import datetime, timezone import pytest from approval_engine.api import App +from approval_engine.auth import Identity, StaticTokenAuthenticator from approval_engine.store import Engine FROZEN = datetime(2026, 8, 29, 12, 0, tzinfo=timezone.utc) @@ -22,7 +23,29 @@ def engine(now): @pytest.fixture def app(engine): - return App(engine) + identity = Identity( + subject="agt-secrets-engine", + issuer="https://keycape.example", + audiences=("approval-engine",), + principal_type="service", + tenant="platform", + roles=frozenset({"secrets-engine"}), + scopes=frozenset( + { + "approval:create", + "approval:read", + "approval:approve", + "approval:revoke", + "approval:supersede", + "approval:consume", + "approval:observe", + "approval:emit", + } + ), + assurance={"level": "aal1", "methods": ["test"], "source": "test"}, + evidence_ref="test-identity:test-token", + ) + return App(engine, StaticTokenAuthenticator({"test-token": identity})) def binding(**overrides): diff --git a/tests/test_api.py b/tests/test_api.py index 5f1c21b..8304e00 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -127,13 +127,29 @@ def test_claim_after_revoke(app): def test_store_unavailable_is_503(): from approval_engine.api import App + from approval_engine.auth import Identity, StaticTokenAuthenticator from approval_engine.errors import StoreUnavailable from approval_engine.store import Engine class Dead(Engine): + def storage_status(self): + return {"schema_current": True, "persistent": True} + def outbox_stats(self): raise StoreUnavailable("down") - status, body = call(App(Dead.__new__(Dead)), "GET", "/v1/readyz") + identity = Identity( + subject="test", + issuer="test", + audiences=("approval-engine",), + principal_type="service", + tenant="test", + roles=frozenset(), + scopes=frozenset(), + assurance={}, + evidence_ref="test", + ) + app = App(Dead.__new__(Dead), StaticTokenAuthenticator({"test-token": identity})) + status, body = call(app, "GET", "/v1/readyz") assert status == 503 assert body["error"] == "store_unavailable" diff --git a/tests/test_audit.py b/tests/test_audit.py new file mode 100644 index 0000000..cb34aed --- /dev/null +++ b/tests/test_audit.py @@ -0,0 +1,82 @@ +import json +from datetime import timedelta + +import pytest + +from approval_engine.audit import AuditCoreSink, AuditDeliveryError, OutboxWorker +from approval_engine.store import Engine +from tests.conftest import FROZEN, approve + + +class Response: + def __init__(self, status): + self.status = status + + def getcode(self): + return self.status + + def close(self): + pass + + +def test_audit_sender_adapts_envelope_and_rereads_token(tmp_path): + token = tmp_path / "token" + token.write_text("first") + seen = [] + + def open_request(request, timeout): + seen.append((request, timeout)) + return Response(202) + + engine = Engine(":memory:", clock=lambda: FROZEN) + approve(engine) + sink = AuditCoreSink("http://audit-core:8080", token, opener=open_request) + sink(engine.undrained()[0]["payload"]) + token.write_text("second") + sink(engine.undrained()[0]["payload"]) + first, second = (item[0] for item in seen) + assert first.get_header("Authorization") == "Bearer first" + assert second.get_header("Authorization") == "Bearer second" + assert first.get_header("Idempotency-key") == engine.undrained()[0]["event_id"] + body = json.loads(first.data) + assert body["id"] == body["correlation_id"] or body["correlation_id"] + assert body["type"] == "approval.issuance" + assert body["source"] == "approval-engine" + engine.close() + + +def test_nonaccepted_audit_status_remains_pending(tmp_path): + token = tmp_path / "token" + token.write_text("token") + engine = Engine(":memory:", clock=lambda: FROZEN) + approve(engine) + sink = AuditCoreSink( + "http://audit-core:8080", token, opener=lambda *_args, **_kwargs: Response(503) + ) + result = engine.drain(sink) + assert result == {"delivered": 0, "failed": 1} + assert engine.undrained()[0]["last_error"] == "AuditDeliveryError" + engine.close() + + +def test_worker_emits_due_heartbeat_and_drains(): + now = [FROZEN] + engine = Engine(":memory:", clock=lambda: now[0]) + delivered = [] + worker = OutboxWorker(engine, delivered.append, heartbeat_interval_seconds=300) + first = worker.run_once() + assert first["delivered"] == 1 + assert delivered[0]["action"] == "approval.heartbeat" + now[0] += timedelta(seconds=301) + second = worker.run_once() + assert second["delivered"] == 1 + assert len(delivered) == 2 + engine.close() + + +def test_sender_rejects_empty_token(tmp_path): + token = tmp_path / "token" + token.write_text("") + sink = AuditCoreSink("http://audit-core", token) + with pytest.raises(AuditDeliveryError, match="credential"): + sink({}) diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 0000000..a8cdf2d --- /dev/null +++ b/tests/test_auth.py @@ -0,0 +1,167 @@ +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +import jwt +import pytest +from cryptography.hazmat.primitives.asymmetric import rsa + +from approval_engine.api import call +from approval_engine.auth import Identity, JWTAuthenticator, StaticTokenAuthenticator +from approval_engine.errors import Unauthenticated +from tests.conftest import binding, validity + + +class _Key: + def __init__(self, key): + self.key = key + + +class _JWKS: + def __init__(self, key): + self.key = key + + def get_signing_key_from_jwt(self, _token): + return _Key(self.key) + + +def _jwt(private_key, **overrides): + now = datetime.now(timezone.utc) + claims = { + "iss": "https://keycape.example", + "sub": "service:secrets-engine", + "aud": "approval-engine", + "iat": int(now.timestamp()), + "exp": int((now + timedelta(minutes=5)).timestamp()), + "tenant": "tenant:coulomb", + "principal_type": "service", + "roles": ["secrets-engine"], + "scope": "approval:read approval:consume", + "assurance": {"level": "aal1", "methods": ["client_secret"], "source": "key-cape"}, + } + claims.update(overrides) + return jwt.encode(claims, private_key, algorithm="RS256", headers={"kid": "test"}) + + +def test_jwt_authenticator_verifies_signature_issuer_audience_and_claims(): + private = rsa.generate_private_key(public_exponent=65537, key_size=2048) + auth = JWTAuthenticator( + issuer="https://keycape.example", + audience="approval-engine", + jwks_url="https://keycape.example/jwks", + jwks_client=_JWKS(private.public_key()), + ) + identity = auth.authenticate("Bearer " + _jwt(private)) + assert identity.subject == "service:secrets-engine" + assert identity.principal_type == "service" + assert identity.scopes == {"approval:read", "approval:consume"} + assert identity.evidence_ref.startswith("jwt-sha256:") + + +@pytest.mark.parametrize( + "claims", + [ + {"iss": "https://wrong.example"}, + {"aud": "somewhere-else"}, + {"exp": 1}, + ], +) +def test_jwt_authenticator_fails_closed(claims): + private = rsa.generate_private_key(public_exponent=65537, key_size=2048) + auth = JWTAuthenticator( + issuer="https://keycape.example", + audience="approval-engine", + jwks_url="https://keycape.example/jwks", + jwks_client=_JWKS(private.public_key()), + ) + with pytest.raises(Unauthenticated): + auth.authenticate("Bearer " + _jwt(private, **claims)) + + +def test_api_requires_scope_and_binds_create_actor(engine): + identity = Identity( + subject="service:creator", + issuer="test", + audiences=("approval-engine",), + principal_type="service", + tenant="platform", + roles=frozenset(), + scopes=frozenset({"approval:create"}), + assurance={"level": "aal1"}, + evidence_ref="test", + ) + from approval_engine.api import App + + app = App(engine, StaticTokenAuthenticator({"creator": identity})) + status, body = call( + app, + "POST", + "/v1/approvals", + {"binding": binding(), "validity": validity()}, + authorization="Bearer creator", + ) + assert status == 403 + assert body["error"] == "forbidden" + + exact = binding(actor="service:creator") + status, created = call( + app, + "POST", + "/v1/approvals", + {"binding": exact, "validity": validity()}, + authorization="Bearer creator", + ) + assert status == 201 + status, body = call( + app, "GET", f"/v1/approvals/{created['id']}/claim", authorization="Bearer creator" + ) + assert status == 403 + assert body["error"] == "forbidden" + + +def test_approval_entry_uses_verified_identity_not_body(app): + _, created = call( + app, + "POST", + "/v1/approvals", + {"binding": binding(), "validity": validity()}, + ) + _, approved = call( + app, + "POST", + f"/v1/approvals/{created['id']}/entries", + { + "subject_id": "user:spoofed", + "assurance": "spoofed", + "evidence_ref": "spoofed", + }, + ) + assert approved["entries"][0]["subject_id"] == "agt-secrets-engine" + assert approved["entries"][0]["evidence_ref"] == "test-identity:test-token" + assert "spoofed" not in approved["entries"][0]["assurance"] + + +def test_missing_token_is_unauthenticated(app): + status, body = call(app, "GET", "/v1/cadence", authorization=None) + assert status == 401 + assert body["error"] == "unauthenticated" + + +def test_wrong_tenant_is_forbidden(engine): + identity = Identity( + subject="agt-secrets-engine", + issuer="test", + audiences=("approval-engine",), + principal_type="service", + tenant="another-tenant", + roles=frozenset(), + scopes=frozenset({"approval:observe"}), + assurance={}, + evidence_ref="test", + ) + from approval_engine.api import App + + app = App(engine, StaticTokenAuthenticator({"wrong": identity})) + status, body = call(app, "GET", "/v1/cadence", authorization="Bearer wrong") + assert status == 403 + assert body["error"] == "forbidden" diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..63d197a --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,22 @@ +import json + +import pytest + +from approval_engine.cli import main + + +def test_migrate_verify_and_backup_commands(tmp_path, capsys): + database = tmp_path / "approval.sqlite" + backup = tmp_path / "approval.backup.sqlite" + assert main(["migrate", "--db", str(database)]) == 0 + migrated = json.loads(capsys.readouterr().out) + assert migrated["ok"] is True + assert main(["verify", "--db", str(database)]) == 0 + assert json.loads(capsys.readouterr().out)["schema_current"] is True + assert main(["backup", "--db", str(database), "--output", str(backup)]) == 0 + assert json.loads(capsys.readouterr().out)["integrity"] == "ok" + + +def test_production_refuses_memory_store_before_serving(): + with pytest.raises(SystemExit): + main(["serve", "--production", "--db", ":memory:"]) diff --git a/tests/test_outbox.py b/tests/test_outbox.py index cc7148c..de78849 100644 --- a/tests/test_outbox.py +++ b/tests/test_outbox.py @@ -82,3 +82,19 @@ def test_failed_outbox_rolls_back_consume(engine): pass assert engine.get(obj.id).status == "approved" assert all(item["class"] != "use" for item in engine.undrained()) + + +def test_drain_failure_records_bounded_attempt_state(engine): + approve(engine) + + class SensitiveFailure(Exception): + pass + + result = engine.drain(lambda _payload: (_ for _ in ()).throw(SensitiveFailure("secret"))) + assert result["failed"] == 1 + pending = engine.undrained() + assert pending[0]["attempts"] == 1 + assert pending[0]["last_error"] == "SensitiveFailure" + stats = engine.outbox_stats() + assert stats["failed_pending"] == 1 + assert stats["attempts"] == 1 diff --git a/tests/test_pep.py b/tests/test_pep.py new file mode 100644 index 0000000..6c09816 --- /dev/null +++ b/tests/test_pep.py @@ -0,0 +1,105 @@ +import json + +import pytest + +from approval_engine.pep import ( + ApprovalHTTPClient, + ApprovalProtocolError, + ProtectedActionHarness, +) + + +class Client: + def __init__(self, claim=None, consume=None, failure=None): + self.claim_result = claim or {"valid_now": True, "consumed": False} + self.consume_result = consume + self.failure = failure + self.calls = [] + + def claim(self, approval_id): + self.calls.append("claim") + if self.failure == "claim": + raise ApprovalProtocolError("down") + return self.claim_result + + def consume(self, approval_id, digest, decision_id): + self.calls.append("consume") + if self.failure == "consume": + raise ApprovalProtocolError("conflict") + return self.consume_result or {"status": "consumed", "request_digest": digest} + + +DIGEST = "sha256:" + "ab" * 32 + + +class Response: + def __init__(self, body): + self.body = json.dumps(body).encode() + + def getcode(self): + return 200 + + def read(self, _size): + return self.body + + def close(self): + pass + + +def test_http_client_rereads_mounted_token(tmp_path): + token = tmp_path / "token" + token.write_text("first") + seen = [] + + def opener(request, timeout): + seen.append((request.get_header("Authorization"), timeout)) + return Response({"valid_now": True, "consumed": False}) + + client = ApprovalHTTPClient("http://approval-engine:8080", token, opener=opener) + client.claim("approval-1") + token.write_text("second") + client.claim("approval-1") + assert [item[0] for item in seen] == ["Bearer first", "Bearer second"] + + +def allow(_claim): + return {"effect": "ALLOW", "decision_id": "decision:1", "request_digest": DIGEST} + + +def test_side_effect_occurs_only_after_claim_decision_and_consume(): + client = Client() + order = client.calls + result = ProtectedActionHarness(client).execute( + "approval:1", + DIGEST, + lambda claim: (order.append("decision"), allow(claim))[1], + lambda: (order.append("side-effect"), "dry-run-only")[1], + ) + assert result == "dry-run-only" + assert order == ["claim", "decision", "consume", "side-effect"] + + +@pytest.mark.parametrize("failure", ["claim", "consume"]) +def test_unavailable_or_conflicting_engine_prevents_side_effect(failure): + client = Client(failure=failure) + effects = [] + with pytest.raises(ApprovalProtocolError): + ProtectedActionHarness(client).execute( + "approval:1", DIGEST, allow, lambda: effects.append("called") + ) + assert effects == [] + + +def test_deny_or_digest_mismatch_prevents_consume_and_side_effect(): + for decision in ( + {"effect": "DENY", "decision_id": "decision:1", "request_digest": DIGEST}, + {"effect": "ALLOW", "decision_id": "decision:1", "request_digest": "sha256:" + "cd" * 32}, + ): + client = Client() + effects = [] + with pytest.raises(ApprovalProtocolError): + ProtectedActionHarness(client).execute( + "approval:1", DIGEST, lambda _claim, value=decision: value, lambda: effects.append("called") + ) + assert client.calls == ["claim"] + assert effects == [] diff --git a/tests/test_storage.py b/tests/test_storage.py new file mode 100644 index 0000000..e9e9169 --- /dev/null +++ b/tests/test_storage.py @@ -0,0 +1,55 @@ +import os +import sqlite3 +from pathlib import Path + +import pytest + +from approval_engine.errors import Conflict, StoreUnavailable +from approval_engine.store import Engine, LATEST_SCHEMA_VERSION +from tests.conftest import FROZEN, approve + + +def test_production_open_refuses_unmigrated_database(tmp_path): + path = tmp_path / "approval.sqlite" + sqlite3.connect(path).close() + with pytest.raises(StoreUnavailable, match="run approval-engine migrate"): + Engine(path, clock=lambda: FROZEN, auto_migrate=False) + + +def test_migrate_then_open_without_auto_migrate(tmp_path): + path = tmp_path / "approval.sqlite" + migrated = Engine(path, clock=lambda: FROZEN) + migrated.close() + production = Engine(path, clock=lambda: FROZEN, auto_migrate=False) + status = production.storage_status(integrity=True) + assert status["schema_version"] == LATEST_SCHEMA_VERSION + assert status["schema_current"] is True + assert status["persistent"] is True + assert status["ok"] is True + production.close() + + +def test_online_backup_is_mode_0600_and_restorable(tmp_path): + source = tmp_path / "approval.sqlite" + backup = tmp_path / "approval.backup.sqlite" + engine = Engine(source, clock=lambda: FROZEN) + obj = approve(engine) + result = engine.backup(backup) + assert result["integrity"] == "ok" + assert os.stat(backup).st_mode & 0o777 == 0o600 + restored = Engine(backup, clock=lambda: FROZEN, auto_migrate=False) + assert restored.get(obj.id).id == obj.id + assert restored.storage_status(integrity=True)["ok"] is True + restored.close() + engine.close() + + +def test_backup_refuses_overwrite(tmp_path): + source = tmp_path / "approval.sqlite" + target = tmp_path / "existing.sqlite" + target.write_text("do not overwrite") + engine = Engine(source, clock=lambda: FROZEN) + with pytest.raises(Conflict, match="already exists"): + engine.backup(target) + assert target.read_text() == "do not overwrite" + engine.close() diff --git a/workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md b/workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md index fb03c90..ec8af8c 100644 --- a/workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md +++ b/workplans/APPROVAL-WP-0002-production-readiness-and-consumer-adoption.md @@ -4,11 +4,20 @@ type: workplan title: "Production readiness and consumer adoption" domain: infotech repo: approval-engine -status: proposed +status: active owner: codex topic_slug: netkingdom created: "2026-09-01" -updated: "2026-09-01" +updated: "2026-09-02" +reviewed_at: "2026-09-01" +reviewed_against_commit: "ebce5abb276c01ab29ce2526f3b8abb332dc9e90" +reviewed_note: >- + Reviewed against approval-engine's finished spine, KeyCape's RS256/JWKS and + service-token contract, access-engine caller-auth/binding surface, + audit-core's deployed authenticated ingestion plus open AUDIT-WP-0009 + approval-source/cadence/reconciliation tasks, and secrets-engine's waiting + exact-action/decision-consumption tasks. Repo-owned implementation can + proceed; live T03-T05 closure remains evidence-gated on those owners. origin: residual origin_ref: APPROVAL-WP-0001 state_hub_workstream_id: "4fa25ad5-f5d0-5592-aa59-085f8ee3edaf" @@ -23,11 +32,21 @@ the residual production scope deliberately excluded from APPROVAL-WP-0001. The workplan is proposed pending review against the deployment estate and the current key-cape, access-engine, audit-core, and secrets-engine contracts. +Review completed 2026-09-01. The plan is active. Production mode will verify +KeyCape JWT signatures and exact issuer/audience/scopes; local caller-supplied +identity never becomes authenticated evidence. SQLite remains the first +production store only as a single-replica StatefulSet with explicit migration, +backup, integrity, and restore gates. audit-core delivery can be implemented +against its existing authenticated idempotent ingest, while heartbeat findings, +count reconciliation, and sender registration remain external gates in +`AUDIT-WP-0009` T04/T06/T09. The first live PEP proof remains jointly gated on +secrets-engine T04/T02 and deployment of this service. + ## Authenticate lifecycle mutations and approver evidence ```task id: APPROVAL-WP-0002-T01 -status: todo +status: progress priority: high state_hub_task_id: "dc4523f5-e0af-5734-a0ef-07aa7f2b2a27" ``` @@ -37,11 +56,24 @@ authenticated identities. An API-supplied `subject_id`, `actor`, or `decision_id` is provenance only until independently authenticated. Keep authorization decisions in access-engine and approval doctrine in gate-house. +Acceptance: production startup requires a signature-valid KeyCape JWT verifier; +every non-health API route requires an explicit scope; approval entry identity +and assurance come only from verified claims; create binds `binding.actor` to +the authenticated subject; consume is restricted to service/agent principals; +missing, expired, wrong-issuer, wrong-audience, wrong-scope, or unverifiable +tokens fail closed without mutation. + +Repository implementation complete 2026-09-02: RS256/JWKS verification, +issuer/audience/time/profile validation, exact scopes, store-tenant isolation, +verified approver evidence, and a deny-all default are covered by tests. Remains +`progress` until KeyCape owns and proves the production audience/client/scope +registrations. + ## Harden durable storage and migrations ```task id: APPROVAL-WP-0002-T02 -status: todo +status: done priority: high state_hub_task_id: "2bef94ca-9482-5eff-9bd9-39adef292a78" ``` @@ -50,11 +82,21 @@ Define the production persistence, backup/restore, migration, concurrency, and recovery posture. Prove schema upgrades preserve existing approvals and that crash recovery cannot separate mutations from outbox evidence. +Acceptance: schema version is explicit; production serve refuses an unmigrated +or in-memory store; migration is a separate repeatable command; backup uses +SQLite's online backup API and integrity verification; restore is documented as +a stopped-single-writer operation; tests cover legacy upgrade, backup/restore, +and outbox atomicity after restart. + +Completed 2026-09-02: schema v2 migration, production no-auto-migrate gate, +integrity/status surface, online mode-0600 backup, stopped-writer restore runbook, +retry-attempt state, and migration/backup/atomicity tests are in place. + ## Package and deploy the service ```task id: APPROVAL-WP-0002-T03 -status: todo +status: wait priority: high state_hub_task_id: "f0aa2e6d-19e6-5b43-886c-efa4e3de5f22" ``` @@ -63,11 +105,23 @@ Add the governed image/deployment surface, health and readiness behavior, resource bounds, and fail-closed caller configuration. A local WSGI development server is not production evidence. +Acceptance: a digest-pin-ready image and single-writer StatefulSet manifest +exist with non-root/read-only-root controls, PVC, migration init container, +resource bounds, probes, and default-deny network policy. Live completion also +requires an immutable image digest, KeyCape registrations, audit sender +credential, successful rollout, and restart/restore evidence. + +Repository implementation complete 2026-09-02: the digest-pin-ready non-root +image builds and runs; the single-writer StatefulSet, PVC, migration init, +read-only root, resources, probes, and default-deny policies pass client dry-run. +Waiting on release digest, KeyCape/audit registrations and credentials, rollout, +restart, and restore evidence. + ## Wire outbox delivery and reconciliation ```task id: APPROVAL-WP-0002-T04 -status: todo +status: wait priority: high state_hub_task_id: "777a5cf5-c1db-5e07-8de6-ab109db5ccdc" ``` @@ -76,11 +130,22 @@ Deliver the local outbox asynchronously to audit-core, preserve event-id deduplication, publish lag/depth signals, emit the declared heartbeat, and prove the Gate House reconciliation contract against accepted event counts. +Acceptance: the sender adapts the local event to audit-core's authenticated +HTTP ingest, reuses event id as idempotency key, rereads a mounted token file, +marks drained only on accepted/duplicate, retains retryable failures, exposes +attempt/lag state, and emits the declared heartbeat. Live reconciliation waits +on `AUDIT-WP-0009-T04/T06/T09`; do not invent that receiver surface here. + +Repository implementation complete 2026-09-02: audit-core envelope adaptation, +event-id idempotency, mounted-token reread, accepted/duplicate handling, +retry-attempt/lag metrics, and periodic heartbeats are tested. Waiting on the +audit-core sender registration/ingress and receiver-owned reconciliation work. + ## Prove one live PEP consumption path ```task id: APPROVAL-WP-0002-T05 -status: todo +status: wait priority: high state_hub_task_id: "3196fb77-3df0-5358-ae12-2d07bac12041" ``` @@ -89,3 +154,14 @@ Integrate one protected-system consumer under `GH-DEC-2026-003`: claim before decision, CAS consume after ALLOW and before side effect, same-digest retry, different-digest conflict, spent-on-failure behavior, and no protected action when approval-engine is unavailable. + +Acceptance: a repeatable harness proves the PEP sequence against the real HTTP +surface without performing a protected action; live closure requires a +secrets-engine-owned handler and evidence that no OpenBao call occurs on every +failure case. This repo may supply the protocol client and fixture, but may not +claim the consumer's side effect. + +Repository implementation complete 2026-09-02: the HTTP PEP client and +fail-closed sequencing harness prove claim-before-decision and CAS-consume-before +callback, including unavailable, DENY, digest mismatch, and conflict paths. +Waiting on the secrets-engine-owned handler and live no-OpenBao-on-failure proof.