tegwick
2b0d04e8e1
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
502 lines
30 KiB
Python
502 lines
30 KiB
Python
#!/usr/bin/env python3
|
|
"""Disposable KeyCape image -> actual Secrets Engine client -> Approval Engine.
|
|
|
|
Synthetic keys/credentials only. No OpenBao, cluster mutation, or model call.
|
|
The actual Flex Auth evaluator runs locally with its published policy/registry.
|
|
This is component conformance evidence, not production authorization evidence.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import copy
|
|
import hashlib
|
|
import http.server
|
|
import io
|
|
import ipaddress
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import secrets
|
|
import shutil
|
|
import socket
|
|
import ssl
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
from contextlib import contextmanager, redirect_stdout
|
|
from datetime import datetime, timedelta, timezone
|
|
from urllib.error import HTTPError
|
|
from urllib.parse import urlencode
|
|
from urllib.request import Request, urlopen
|
|
from unittest.mock import patch
|
|
from wsgiref.simple_server import WSGIRequestHandler, make_server
|
|
|
|
from cryptography import x509
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
from cryptography.x509.oid import NameOID
|
|
import yaml
|
|
|
|
KEYCAPE_IMAGE = "forgejo.coulomb.social/coulomb/key-cape@sha256:7ff54c54e63ee172ae9e6e7fd2da96e427352f712343d74626ee6fe0f6f82611"
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path[:0] = [str(ROOT / "src"), str(ROOT)]
|
|
|
|
|
|
def run(args):
|
|
result = subprocess.run(args, capture_output=True, timeout=45)
|
|
if result.returncode:
|
|
raise RuntimeError("contained command failed")
|
|
return result.stdout.decode().strip()
|
|
|
|
|
|
def commit(path):
|
|
return run(["git", "-C", str(path), "rev-parse", "HEAD"])
|
|
|
|
|
|
def private_write(path, value):
|
|
with path.open("x") as out:
|
|
os.chmod(path, 0o600)
|
|
out.write(value)
|
|
|
|
|
|
def exercise(args):
|
|
sys.path.insert(0, str(args.approval_engine_source))
|
|
from approval_engine.api import App
|
|
from approval_engine.auth import JWTAuthenticator
|
|
from approval_engine.store import Engine
|
|
from secrets_engine import cli
|
|
from secrets_engine.approval_auth import approval_token, credential_urlopen
|
|
from secrets_engine.approval_consume import ConsumeBinding, _expected_request, consume_approval
|
|
from secrets_engine.authorization import request_digest, validate_decision_envelope
|
|
from secrets_engine.catalog import validate_entry
|
|
from secrets_engine.config import Config
|
|
from secrets_engine.errors import BackendError, DecisionError
|
|
from tests.authorization_stub import AuthorizationStub
|
|
from tests.test_catalog import VALID
|
|
|
|
receipt = {
|
|
"schema_version": 1, "target": "disposable local processes; synthetic credentials",
|
|
"started_at": datetime.now(timezone.utc).isoformat(),
|
|
"keycape_image": KEYCAPE_IMAGE,
|
|
"approval_engine_commit": commit(args.approval_engine_source),
|
|
"flex_auth_commit": commit(args.flex_auth_source),
|
|
"keycape_contract_commit": commit(args.keycape_source),
|
|
"consumer_source_sha256": {
|
|
name: hashlib.sha256((ROOT / "src/secrets_engine" / name).read_bytes()).hexdigest()
|
|
for name in ("approval_auth.py", "approval_consume.py", "approval_claim.py", "authorization.py", "config.py", "service_auth.py", "cli.py", "catalog.py", "exec_owner.py", "exec_delivery.py", "routing.py")
|
|
},
|
|
"limitations": ["standalone Flex Auth source, not deployed pin", "local Approval Engine source, not deployed image",
|
|
"no live custody or client-side read grant", "no OpenBao effect or model execution"],
|
|
"checks": {},
|
|
}
|
|
with tempfile.TemporaryDirectory(prefix="approval-identity-private-") as temporary:
|
|
root = Path(temporary)
|
|
source = args.flex_auth_source.resolve()
|
|
run(["go", "build", "-C", str(source), "-o", str(root / "flex-auth"), "./cmd/flex-auth"])
|
|
producer_files = source / "examples/secrets-engine"
|
|
receipt["flex_auth_binary_sha256"] = hashlib.sha256((root / "flex-auth").read_bytes()).hexdigest()
|
|
receipt["producer_input_sha256"] = {
|
|
name: hashlib.sha256((producer_files / name).read_bytes()).hexdigest()
|
|
for name in ("policy_package.md", "registry_snapshot.json")
|
|
}
|
|
|
|
class ProducerPDP(AuthorizationStub):
|
|
"""Loopback transport only; every decision comes from the Go evaluator."""
|
|
last_decision = None
|
|
def decision(self, request):
|
|
path = root / "check-request.json"
|
|
path.write_text(json.dumps(request))
|
|
path.chmod(0o600)
|
|
self.last_decision = json.loads(run([str(root / "flex-auth"), "check",
|
|
"-policy", str(producer_files / "policy_package.md"),
|
|
"-registry", str(producer_files / "registry_snapshot.json"), "-request", str(path)]))
|
|
return self.last_decision
|
|
|
|
def bind_request(self, request):
|
|
# At issue, record the evaluator's claim-free digest, even if
|
|
# the destructive action denies until approval is supplied.
|
|
return self.decision(request)["binding"]["request_digest"]
|
|
|
|
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
private_write(root / "key.pem", key.private_bytes(serialization.Encoding.PEM,
|
|
serialization.PrivateFormat.PKCS8, serialization.NoEncryption()).decode())
|
|
now = datetime.now(timezone.utc)
|
|
name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "local approval identity exercise")])
|
|
cert = (x509.CertificateBuilder().subject_name(name).issuer_name(name).public_key(key.public_key())
|
|
.serial_number(x509.random_serial_number()).not_valid_before(now - timedelta(minutes=1))
|
|
.not_valid_after(now + timedelta(hours=1))
|
|
.add_extension(x509.SubjectAlternativeName([x509.IPAddress(ipaddress.ip_address("127.0.0.1"))]), False)
|
|
.add_extension(x509.BasicConstraints(ca=True, path_length=None), True)
|
|
.sign(key, hashes.SHA256()))
|
|
(root / "cert.pem").write_bytes(cert.public_bytes(serialization.Encoding.PEM))
|
|
with socket.socket() as sock:
|
|
sock.bind(("127.0.0.1", 0))
|
|
keycape_port = sock.getsockname()[1]
|
|
|
|
class Proxy(http.server.BaseHTTPRequestHandler):
|
|
def log_message(self, *_args):
|
|
pass
|
|
def proxy(self):
|
|
body = self.rfile.read(int(self.headers.get("Content-Length", 0))) if self.command == "POST" else None
|
|
req = Request(f"http://127.0.0.1:{keycape_port}" + self.path, data=body,
|
|
headers={k: self.headers[k] for k in ("Content-Type", "Authorization") if k in self.headers})
|
|
try:
|
|
response = urlopen(req, timeout=10)
|
|
except HTTPError as error:
|
|
response = error
|
|
with response:
|
|
self.send_response(response.code)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(response.read())
|
|
do_GET = proxy
|
|
do_POST = proxy
|
|
|
|
server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Proxy)
|
|
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
|
context.load_cert_chain(root / "cert.pem", root / "key.pem")
|
|
server.socket = context.wrap_socket(server.socket, server_side=True)
|
|
issuer = f"https://127.0.0.1:{server.server_port}"
|
|
proxy_thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
proxy_thread.start()
|
|
config = yaml.safe_load((args.keycape_source / "config/dev-config.yaml").read_text())
|
|
clients = yaml.safe_load((args.keycape_source / "config/service-clients.example.yaml").read_text())["clients"]
|
|
clients = [c for c in clients if c["clientId"] in ("secrets-engine-approval", "approval-engine-operator")]
|
|
assert len(clients) == 2
|
|
config.update(issuer=issuer, clients=clients)
|
|
config["authelia"]["issuer"] = "https://synthetic-upstream.invalid"
|
|
private_write(root / "config.yaml", yaml.safe_dump(config, sort_keys=False))
|
|
values = {c["clientId"]: secrets.token_urlsafe(48) for c in clients}
|
|
private_write(root / "env", "KEYCAPE_CONFIG=/etc/keycape/config.yaml\n" + "".join(
|
|
c["secretRef"].removeprefix("env:") + "=" + values[c["clientId"]] + "\n" for c in clients))
|
|
private_write(root / "approval.secret", values["secrets-engine-approval"])
|
|
private_write(root / "pdp.token", "synthetic-pdp-token")
|
|
container = "approval-identity-exercise-" + secrets.token_hex(5)
|
|
created = False
|
|
engine = None
|
|
api_server = None
|
|
pdp = None
|
|
try:
|
|
run(["docker", "run", "-d", "--rm", "--name", container, "--user", str(os.getuid()),
|
|
"--read-only", "--cap-drop=ALL", "--security-opt=no-new-privileges",
|
|
"--publish", f"127.0.0.1:{keycape_port}:8080", "--env-file", str(root / "env"),
|
|
"--mount", "type=bind,source=" + temporary + ",target=/etc/keycape,readonly", KEYCAPE_IMAGE])
|
|
created = True
|
|
with patch.dict(os.environ, {"SSL_CERT_FILE": str(root / "cert.pem"), "SECRETS_ENGINE_UNSAFE_DEMO": ""}):
|
|
for _ in range(40):
|
|
try:
|
|
with credential_urlopen(issuer + "/jwks", timeout=1) as response:
|
|
if response.status == 200:
|
|
break
|
|
except Exception:
|
|
time.sleep(0.25)
|
|
else:
|
|
raise RuntimeError("disposable KeyCape did not start")
|
|
engine = Engine(root / "approval.sqlite")
|
|
app = App(engine, JWTAuthenticator(issuer=issuer, audience="approval-engine", jwks_url=issuer + "/jwks"))
|
|
class Quiet(WSGIRequestHandler):
|
|
def log_message(self, *_args):
|
|
pass
|
|
api_server = make_server("127.0.0.1", 0, app, handler_class=Quiet)
|
|
threading.Thread(target=api_server.serve_forever, daemon=True).start()
|
|
api = f"http://127.0.0.1:{api_server.server_port}"
|
|
|
|
def operator_token(scope):
|
|
credential = base64.b64encode(("approval-engine-operator:" + values["approval-engine-operator"]).encode()).decode()
|
|
req = Request(issuer + "/token", data=urlencode({"grant_type": "client_credentials", "scope": scope}).encode(),
|
|
headers={"Authorization": "Basic " + credential})
|
|
with credential_urlopen(req, timeout=3) as response:
|
|
return json.load(response)["access_token"]
|
|
|
|
def operator_request(path, payload, scope):
|
|
req = Request(api + path, data=json.dumps(payload).encode(),
|
|
headers={"Content-Type": "application/json", "Authorization": "Bearer " + operator_token(scope)})
|
|
with credential_urlopen(req, timeout=3) as response:
|
|
return json.load(response)
|
|
|
|
raw = copy.deepcopy(VALID)
|
|
raw["stage"] = "prod"
|
|
raw["approval"] = {"model": "bootstrap-only", "authorization_id": "synthetic-approval", "purpose": "disposable identity proof"}
|
|
entry = validate_entry(raw)
|
|
pdp = ProducerPDP(approval_id="unused", package="secrets-engine.catalog-lane.lifecycle", version="v2").start()
|
|
cfg = Config(catalog_dir=root, policy_dir=root, evidence_dir=root / "evidence", hub_url="", bao_addr="", topic_id="",
|
|
approval_url=api, approval_client_secret_file=root / "approval.secret",
|
|
keycape_issuer=issuer, keycape_token_url=issuer + "/token",
|
|
authorization_subject_id="secrets-engine", authorization_subject_type="service",
|
|
authorization_policy_package="secrets-engine.catalog-lane.lifecycle", authorization_policy_version="v2",
|
|
pdp_url=pdp.url, pdp_token_file=root / "pdp.token")
|
|
request = _expected_request(cfg, entry, "apply")
|
|
digest = pdp.bind_request(request)
|
|
obj = operator_request("/v1/approvals", {
|
|
"id": "synthetic-approval",
|
|
"binding": {"action": "secrets.apply", "target": {"id": entry.id, "stage": "prod"},
|
|
"actor": "service:approval-engine-operator", "principal": "synthetic-operator", "purpose": "disposable identity proof"},
|
|
"validity": {"not_before": (now - timedelta(minutes=1)).isoformat(), "expires_at": (now + timedelta(minutes=10)).isoformat()},
|
|
"pdp_digest": digest, "pdp_path": True}, "approval:create")
|
|
operator_request(f"/v1/approvals/{obj['id']}/entries", {}, "approval:approve")
|
|
receipt["checks"]["operator_issued_and_approved_via_verified_jwt"] = True
|
|
try:
|
|
cli._require_lane_approval(cfg, entry, "destroy")
|
|
except DecisionError:
|
|
assert not engine.claim(obj["id"])["consumed"]
|
|
else:
|
|
raise AssertionError("wrong action accepted")
|
|
receipt["checks"]["wrong_action_refused_before_consume"] = True
|
|
cli._require_lane_approval(cfg, entry, "apply")
|
|
assert engine.claim(obj["id"])["consumed"]
|
|
assert pdp.calls == ["check", "check"]
|
|
accepted = pdp.last_decision
|
|
assert accepted["binding"]["approval_binding_digest"] == digest
|
|
assert accepted["binding"]["request_digest"] != digest
|
|
assert accepted["binding"]["submitted_request_digest"] != accepted["binding"]["request_digest"]
|
|
assert accepted["binding"]["context"]["approval"]["approval_id"] == obj["id"]
|
|
receipt["checks"]["producer_origin_join_with_carried_claim"] = True
|
|
receipt["checks"]["actual_consumer_claim_check_consume"] = True
|
|
binding = ConsumeBinding(obj["id"], accepted["binding"]["request_digest"], accepted["id"])
|
|
retry = consume_approval(base_url=api, binding=binding, token_provider=lambda: approval_token(cfg, scope="approval:consume"))
|
|
assert retry.idempotent
|
|
receipt["checks"]["same_digest_retry_idempotent"] = True
|
|
try:
|
|
consume_approval(base_url=api, binding=ConsumeBinding(obj["id"], "sha256:" + "f" * 64),
|
|
token_provider=lambda: approval_token(cfg, scope="approval:consume"))
|
|
except DecisionError:
|
|
pass
|
|
else:
|
|
raise AssertionError("different digest accepted")
|
|
receipt["checks"]["different_digest_refused"] = True
|
|
try:
|
|
cli._require_lane_approval(cfg, entry, "apply")
|
|
except DecisionError:
|
|
pass
|
|
else:
|
|
raise AssertionError("spent claim accepted")
|
|
receipt["checks"]["spent_claim_refused"] = True
|
|
try:
|
|
operator_token("approval:consume")
|
|
except HTTPError as error:
|
|
assert error.code in (400, 401)
|
|
assert json.load(error)["error"] == "invalid_profile_usage"
|
|
else:
|
|
raise AssertionError("operator consume scope accepted")
|
|
receipt["checks"]["operator_consume_scope_denied_by_issuer"] = True
|
|
(root / "approval.secret").write_text("synthetic-wrong-secret")
|
|
try:
|
|
approval_token(cfg, scope="approval:read")
|
|
except BackendError:
|
|
pass
|
|
else:
|
|
raise AssertionError("wrong secret accepted")
|
|
receipt["checks"]["wrong_secret_refused"] = True
|
|
assert not (root / "approval.token").exists()
|
|
receipt["checks"]["no_access_token_file_created"] = True
|
|
|
|
# Independent input -> evaluator -> consumer: registry may
|
|
# override a caller's same-key facts without breaking replay.
|
|
request = json.loads((producer_files / "check_request_allow_rotate.json").read_text())
|
|
request["subject"]["attributes"] = {"roles": ["UntrustedCallerRole"]}
|
|
decision = pdp.decision(request)
|
|
assert decision["binding"]["subject"]["attributes"]["roles"] == ["Operator"]
|
|
assert decision["binding"]["submitted_request_digest"] == request_digest(request)
|
|
validate_decision_envelope(decision, request,
|
|
accepted_policy_packages={"secrets-engine.catalog-lane.lifecycle"}, accepted_policy_versions={"v2"})
|
|
receipt["checks"]["registry_override_accepts_exact_submission"] = True
|
|
replay = copy.deepcopy(request)
|
|
replay["subject"]["attributes"]["roles"] = ["Operator"]
|
|
try:
|
|
validate_decision_envelope(decision, replay,
|
|
accepted_policy_packages={"secrets-engine.catalog-lane.lifecycle"}, accepted_policy_versions={"v2"})
|
|
except DecisionError:
|
|
pass
|
|
else:
|
|
raise AssertionError("changed submission replayed")
|
|
receipt["checks"]["same_enriched_result_different_submission_refused"] = True
|
|
# Real dual-control policy: empty claim denies; fresh issued
|
|
# claim passes. No destructive backend is connected.
|
|
raw["approval"]["authorization_id"] = "synthetic-destroy"
|
|
destroy_entry = validate_entry(raw)
|
|
destroy_request = _expected_request(cfg, destroy_entry, "destroy")
|
|
denied = pdp.decision(destroy_request)
|
|
assert denied["effect"] == "deny" and denied["reason"] == "dual_control_required"
|
|
operator_request("/v1/approvals", {
|
|
"id": "synthetic-destroy", "binding": {"action": "secrets.kv.destroy", "target": {"id": destroy_entry.id, "stage": "prod"},
|
|
"actor": "service:approval-engine-operator", "principal": "synthetic-operator", "purpose": "disposable identity proof"},
|
|
"validity": {"not_before": (now - timedelta(minutes=1)).isoformat(), "expires_at": (now + timedelta(minutes=10)).isoformat()},
|
|
"pdp_digest": denied["binding"]["request_digest"], "pdp_path": True}, "approval:create")
|
|
operator_request("/v1/approvals/synthetic-destroy/entries", {}, "approval:approve")
|
|
(root / "approval.secret").write_text(values["secrets-engine-approval"])
|
|
cli._require_lane_approval(cfg, destroy_entry, "destroy")
|
|
assert engine.claim("synthetic-destroy")["consumed"]
|
|
receipt["checks"]["real_dual_control_denial_then_claim_check_consume"] = True
|
|
if args.exec_owner:
|
|
from secrets_engine.errors import DeliveryError
|
|
from secrets_engine.exec_owner import owner_digest
|
|
from types import SimpleNamespace
|
|
|
|
# Synthetic recipient and backend, real engine CLI, exact
|
|
# producer digest, JWT claim/consume, spawn and environment.
|
|
owner_dir = root / "exec-owner"
|
|
owner_dir.mkdir(mode=0o700)
|
|
executable = owner_dir / "dash"
|
|
shutil.copyfile("/bin/dash", executable)
|
|
executable.chmod(0o700)
|
|
script = owner_dir / "owner.sh"
|
|
private_write(script, 'test "$API_TOKEN" = synthetic-owner-value || exit 10\n'
|
|
'test -z "$SECRETS_ENGINE_APPROVAL_CLIENT_SECRET_FILE" || exit 11\n'
|
|
'test -z "$BAO_TOKEN" || exit 12\n'
|
|
'printf "owner-delivery-ok\\n"\n')
|
|
raw["approval"]["authorization_id"] = "synthetic-owner-delivery"
|
|
raw["delivery_config"] = {"exec_owner": {
|
|
"status": "configured", "owner": "synthetic-metered-owner",
|
|
"command": [str(executable), str(script)], "cwd": str(owner_dir),
|
|
"environment": {"PATH": "/usr/bin:/bin", "LANG": "C.UTF-8"},
|
|
"files": {str(p): {"sha256": hashlib.sha256(p.read_bytes()).hexdigest(), "private": True}
|
|
for p in [executable, script]},
|
|
}}
|
|
if args.human_control:
|
|
raw["approval"].update(model="decision", human_control=True,
|
|
decision_ref="synthetic-human-lane-review")
|
|
owner_entry = validate_entry(raw)
|
|
req = _expected_request(cfg, owner_entry, "exec", fields=("api_token",))
|
|
owner_request_digest = pdp.bind_request(req)
|
|
def issue_owner(ident, human_control):
|
|
# The requester declares intent on an unapproved object;
|
|
# only the subsequent human bind discharges the control.
|
|
return operator_request("/v1/approvals", {
|
|
"id": ident, "binding": {"action": "secrets.exec", "target": {"id": owner_entry.id, "stage": "prod"},
|
|
"actor": "service:approval-engine-operator", "principal": "synthetic-operator", "purpose": "disposable identity proof"},
|
|
"validity": {"not_before": (now - timedelta(minutes=1)).isoformat(), "expires_at": (now + timedelta(minutes=10)).isoformat()},
|
|
"pdp_digest": owner_request_digest, "pdp_path": True,
|
|
"human_control": human_control}, "approval:create")
|
|
issue_owner("synthetic-owner-delivery", args.human_control)
|
|
if args.human_control:
|
|
try:
|
|
operator_request("/v1/approvals/synthetic-owner-delivery/entries", {}, "approval:approve")
|
|
except HTTPError as error:
|
|
assert error.code == 403
|
|
assert engine.get("synthetic-owner-delivery").entries == []
|
|
else:
|
|
raise AssertionError("service bound a declared human control")
|
|
# Explicit local fixture: the positive human entry is seeded
|
|
# through the store, not a native human login/JWT claim.
|
|
engine.add_entry("synthetic-owner-delivery", "fixture:human",
|
|
principal_type="human", evidence_ref="synthetic-store-fixture")
|
|
issue_owner("synthetic-owner-undeclared", False)
|
|
operator_request("/v1/approvals/synthetic-owner-undeclared/entries", {}, "approval:approve")
|
|
receipt["checks"]["declared_control_refuses_real_keycape_service_bind"] = True
|
|
receipt["limitations"].append("positive human entry seeded in disposable store; no human JWT/PKCE proof")
|
|
else:
|
|
operator_request("/v1/approvals/synthetic-owner-delivery/entries", {}, "approval:approve")
|
|
backend_calls = []
|
|
class Backend:
|
|
@contextmanager
|
|
def approle_session(self, role):
|
|
assert role == owner_entry.role_name
|
|
yield SimpleNamespace(client=self)
|
|
def _run(self, command):
|
|
assert command == ["kv", "get", "-format=json", f"{owner_entry.mount}/{owner_entry.path}"]
|
|
return SimpleNamespace(returncode=0, stdout=json.dumps({"data": {"data": {"api_token": "synthetic-owner-value"}}}))
|
|
@contextmanager
|
|
def open_fixture_backend(*_args):
|
|
assert engine.claim("synthetic-owner-delivery")["consumed"]
|
|
backend_calls.append("after-consume")
|
|
yield Backend()
|
|
command = raw["delivery_config"]["exec_owner"]["command"]
|
|
cli_args = SimpleNamespace(catalog=owner_entry.id, field="api_token", mode="exec-env", command=command)
|
|
from secrets_engine.decisions import Decision
|
|
lane_review = Decision("synthetic-human-lane-review", "Synthetic lane review", "approved", None, "local-fixture")
|
|
with patch.object(cli, "get_entry", return_value=owner_entry), patch.object(cli, "_open_backend", open_fixture_backend), patch.object(cli, "resolve_decision", return_value=lane_review):
|
|
if args.human_control:
|
|
owner_entry.approval["authorization_id"] = "synthetic-owner-undeclared"
|
|
before_checks = len(pdp.calls)
|
|
try:
|
|
cli.cmd_exec(cfg, cli_args)
|
|
except DecisionError as error:
|
|
assert "human_control" in str(error)
|
|
assert not backend_calls and len(pdp.calls) == before_checks
|
|
assert not engine.claim("synthetic-owner-undeclared")["consumed"]
|
|
else:
|
|
raise AssertionError("undeclared control reached the protected action")
|
|
finally:
|
|
owner_entry.approval["authorization_id"] = "synthetic-owner-delivery"
|
|
receipt["checks"]["undeclared_control_refused_before_real_pdp_consume_backend"] = True
|
|
substituted = SimpleNamespace(**vars(cli_args))
|
|
substituted.command = ["/bin/echo", "substitute"]
|
|
try:
|
|
cli.cmd_exec(cfg, substituted)
|
|
except DeliveryError:
|
|
assert not engine.claim("synthetic-owner-delivery")["consumed"] and not backend_calls
|
|
else:
|
|
raise AssertionError("substitute recipient accepted")
|
|
original_env = raw["delivery_config"]["exec_owner"]["environment"]["LANG"]
|
|
raw["delivery_config"]["exec_owner"]["environment"]["LANG"] = "C"
|
|
try:
|
|
cli.cmd_exec(cfg, cli_args)
|
|
except DecisionError:
|
|
assert not engine.claim("synthetic-owner-delivery")["consumed"] and not backend_calls
|
|
else:
|
|
raise AssertionError("changed owner replay accepted")
|
|
raw["delivery_config"]["exec_owner"]["environment"]["LANG"] = original_env
|
|
output = io.StringIO()
|
|
with redirect_stdout(output), patch.dict(os.environ, {"BAO_TOKEN": "synthetic-parent-only", "SECRETS_ENGINE_APPROVAL_CLIENT_SECRET_FILE": str(root / "approval.secret")}):
|
|
assert cli.cmd_exec(cfg, cli_args) == 0
|
|
assert "owner-delivery-ok" in output.getvalue() and "synthetic-parent-only" not in output.getvalue()
|
|
assert backend_calls == ["after-consume"]
|
|
assert pdp.last_decision["binding"]["context"]["exec_owner_sha256"] == owner_digest(owner_entry)
|
|
receipt["checks"].update({
|
|
"exec_owner_substitution_refused_before_consume_backend": True,
|
|
"exec_owner_changed_environment_replay_refused_by_real_pdp_join": True,
|
|
"exec_owner_real_cli_consumes_before_fixture_backend": True,
|
|
"exec_owner_actual_child_excludes_parent_credentials": True,
|
|
"exec_owner_digest_preserved_by_real_evaluator": True,
|
|
})
|
|
if args.human_control:
|
|
assert pdp.last_decision["binding"]["context"]["human_control"] is True
|
|
assert engine.claim("synthetic-owner-delivery")["binding"]["human_control"] is True
|
|
receipt["checks"]["declared_human_intent_preserved_by_real_evaluator"] = True
|
|
receipt["checks"]["declared_human_fixture_consumed_before_owner_delivery"] = True
|
|
receipt["exec_owner_scope"] = "Synthetic recipient/backend with real KeyCape, Approval Engine, Flex Auth and Secrets Engine CLI; not native custody or human approval proof"
|
|
finally:
|
|
if api_server:
|
|
api_server.shutdown()
|
|
api_server.server_close()
|
|
if pdp:
|
|
pdp.stop()
|
|
if engine:
|
|
engine.close()
|
|
if created:
|
|
run(["docker", "stop", "--time=5", container])
|
|
server.shutdown()
|
|
server.server_close()
|
|
proxy_thread.join()
|
|
receipt.update(status="passed", cleanup_complete=True, finished_at=datetime.now(timezone.utc).isoformat())
|
|
private_write(args.receipt, json.dumps(receipt, indent=2) + "\n")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--keycape-source", required=True, type=Path)
|
|
parser.add_argument("--approval-engine-source", required=True, type=Path)
|
|
parser.add_argument("--flex-auth-source", required=True, type=Path)
|
|
parser.add_argument("--receipt", required=True, type=Path)
|
|
parser.add_argument("--exec-owner", action="store_true", help="also prove catalog-bound child delivery against the actual approval/PDP chain")
|
|
parser.add_argument("--human-control", action="store_true", help="require declared human control in the exec-owner exercise; positive human entry is an explicit local store fixture")
|
|
args = parser.parse_args()
|
|
if args.human_control and not args.exec_owner:
|
|
parser.error("--human-control requires --exec-owner")
|
|
if args.receipt.exists():
|
|
raise SystemExit("receipt path must be new")
|
|
try:
|
|
exercise(args)
|
|
except Exception as error:
|
|
# Never dump a request, client environment or credential-bearing frame.
|
|
print("Synthetic approval identity exercise failed: " + type(error).__name__, file=sys.stderr)
|
|
return 1
|
|
print("Synthetic approval identity exercise passed; temporary resources removed.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|