From 890f3711020197e1df0643ffc865c80fa2d03461 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 7 Nov 2026 06:57:29 +0800 Subject: [PATCH] feat: add inert P01 authorization and provenance contracts --- docs/ibkr-runtime-rollout.md | 43 ++++ scripts/recovery_authorization_gate.py | 208 ++++++++++++++++++++ tests/test_recovery_authorization_gate.py | 226 ++++++++++++++++++++++ 3 files changed, 477 insertions(+) create mode 100644 scripts/recovery_authorization_gate.py create mode 100644 tests/test_recovery_authorization_gate.py diff --git a/docs/ibkr-runtime-rollout.md b/docs/ibkr-runtime-rollout.md index f2a825ba..b968dd7d 100644 --- a/docs/ibkr-runtime-rollout.md +++ b/docs/ibkr-runtime-rollout.md @@ -814,3 +814,46 @@ checksum belongs in an issue, PR, public log or job summary. A rerun or preexist local export directory is refused; concurrent approved reads cannot overwrite one another. An interrupted/failed run uploads nothing. Pricing, recovery and runtime qualification remain unproved and separately gated. + + +## Inert P01 local authorization and provenance contracts + +`scripts/recovery_authorization_gate.py` is an unreferenced, TEST_ONLY local +simulation. It has no command-line execution entry, workflow caller, transport, +provider adapter, provisioner, private channel or durable control store. Its only +accepted schema is `TEST_ONLY_P01_V1`, with literal `` and visibly +FAKE identities. It does not implement or assert a supported deployment protocol. + +The pure validator compares authority, envelope and fake authenticated-metadata +observations against separate harness pins. It checks exact repository, workflow, +ref, run, first attempt, run/workflow/operator/reader/source revisions, image, +Environment, target, recipient/certificate pin, nonce, generation, request, +artifact and mode. Digests bind the opaque TEST_ONLY ciphertext, envelope and +payload separately. Time uses deterministic integer seconds in a fake epoch; +unknown fields, duplicate JSON keys, non-finite values, absent C0 proofs, +caller approval booleans, tampering, stale evidence and wrong producers deny. +Opaque fake ciphertext is not a cryptographic validity or decryption test. +Even the consistent result is `SIMULATION_ONLY`, with `execution_authorized`, +`authentic_provenance` and `private_delivery` all false. + +`FakeGate` models one intent, terminal unknown results, nonce consumption and +latest-decision/deadline checks before and after a simulated step. Fake clients +and reconstructed gate objects share an in-memory `FakeStore`; a thread lock +models atomic creation within that process only. No cross-process or cross-host +durability, authenticated issuer, policy enforcement or real restart reliability +is established. A missing claim, PLAN_ONLY/disabled/superseded decision, expiry, +clock regression, changed current main/operator or lost intent blocks. Results +never remove the intent or renew an allowance. This model cannot perform an +operation or implement compensation. + +The installed strict operator==runSHA and main trust contract, traffic/health, +resourceVersion and single distinct compensation behavior remain unchanged. +No operator revision is selected or rebound here. Real execution remains blocked +on appointed configuration provisioner and Environment administrator; target, +policy and identity ownership proof; authenticated recipient and certificate +custody; a supported private channel and receipt store; an independently approved +authentic issuer/control-store protocol; separately approved version selection; +fresh runtime and budget evidence; and fresh, exact action authorities and +absolute deadlines. C0 ownership and C2 private-channel prerequisites remain +unresolved. Offline tests, source publication and merge cannot satisfy them or +reactivate any historical recovery authority or deadline. diff --git a/scripts/recovery_authorization_gate.py b/scripts/recovery_authorization_gate.py new file mode 100644 index 00000000..7a41090d --- /dev/null +++ b/scripts/recovery_authorization_gate.py @@ -0,0 +1,208 @@ +"""TEST_ONLY P01 local contracts. See AGENTS.md before changing this module. + +No provider protocol, cryptography, transport, persistent adapter or runtime entry. +FakeTrust is deliberately unauthenticated. Even a consistent simulation NEVER +establishes authentic authority, producer identity, delivery or execution readiness. +""" +from __future__ import annotations + +from copy import deepcopy +import hashlib +import json +import re +from threading import Lock + + +C0_ROLES = ("configuration", "admin", "inventory", "recipient", "channel", + "issuer", "ownership", "policy", "control_store") +_BINDINGS = frozenset(("project", "repository", "workflow", "ref", "run", "attempt", + "run_sha", "workflow_sha", "operator", "reader", "source", "image", "environment", + "target", "recipient", "cert_pin", "nonce", "request", "generation", "artifact", "mode")) +_SCHEMA = "TEST_ONLY_P01_V1" + + +class Denied(ValueError): + """Fixed diagnostic: never retain or interpolate input values.""" + + def __init__(self): + super().__init__("TEST_ONLY validation denied") + + +def _require(condition): + if not condition: + raise Denied() + + +def encode(value): + return json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() + + +def digest(value): + return hashlib.sha256(value).hexdigest() + + +def parse_document(raw): + """Bounded JSON object parser; schema-specific validation remains mandatory.""" + def pairs(items): + result = {} + for key, value in items: + _require(key not in result) + result[key] = value + return result + def invalid(_): + raise Denied() + try: + _require(type(raw) in (str, bytes) and len(raw) <= 65536) + value = json.loads(raw, object_pairs_hook=pairs, parse_constant=invalid) + _require(type(value) is dict) + return value + except (ValueError, TypeError, RecursionError): + raise Denied() from None + + +def _shape(value, fields): + _require(type(value) is dict and value.keys() == set(fields)) + + +def _integer(value): + return type(value) is int and value >= 0 + + +def _hex(value, size): + return (type(value) is str and re.fullmatch(r"[a-f0-9]{%d}" % size, value) + and value != "0" * size) + + +class FakeTrust: + """Harness-supplied FAKE pins, NOT an authenticated authority provider.""" + + def __init__(self, bindings, c0): + self.bindings, self.c0 = deepcopy(bindings), deepcopy(c0) + self.current_main = bindings.get("run_sha") + + +def _validate_bindings(b, trust): + _shape(b, _BINDINGS) + _require(b == trust.bindings and b["project"] == "") + for key in ("repository", "workflow", "run", "environment", "target", "recipient", + "request", "artifact"): + _require(type(b[key]) is str and b[key].startswith("FAKE") and len(b[key]) <= 120) + _require(b["ref"] == "refs/heads/main" and type(b["attempt"]) is int + and b["attempt"] == 1 and type(b["generation"]) is int and b["generation"] > 0) + _require(b["mode"] in ("snapshot", "receipt") and type(b["nonce"]) is str + and b["nonce"].startswith("TEST_ONLY_") and len(b["nonce"]) <= 120) + for key in ("run_sha", "workflow_sha", "operator", "reader", "source"): + _require(_hex(b[key], 40)) + _require(b["operator"] == b["run_sha"] == b["workflow_sha"] == b["reader"] + == trust.current_main) + _require(type(b["image"]) is str and b["image"].startswith("sha256:") + and _hex(b["image"][7:], 64) and _hex(b["cert_pin"], 64)) + + +def validate_simulation(*, authority, envelope, metadata, ciphertext, payload, trust, now): + """Reject inconsistent fake declarations; never return an operational PASS. + + Metadata is a visibly fake stub of authenticated provider observations. No + field or caller boolean can turn it into genuine provider/owner evidence. + The fake ciphertext is opaque bytes; CMS validity is not claimed or tested. + """ + try: + _require(type(trust) is FakeTrust and _integer(now)) + _shape(authority, ("schema", "bindings", "c0", "issuer", "claim", "decision", + "issued", "deadline")) + _shape(envelope, ("schema", "bindings", "captured", "expires", "receipt", + "payload_digest")) + _shape(metadata, ("schema", "bindings", "issuer", "conclusion", "created", + "ciphertext_digest", "envelope_digest")) + for document in (authority, envelope, metadata): + _require(document["schema"] == _SCHEMA) + _validate_bindings(document["bindings"], trust) + _shape(authority["c0"], C0_ROLES) + _shape(trust.c0, C0_ROLES) + for role in C0_ROLES: + _require(type(authority["c0"][role]) is str + and authority["c0"][role] == trust.c0[role] == "FAKE_PROOF_" + role) + _require(authority["issuer"] == "FAKE_ISSUER" and authority["claim"] == "FAKE_CLAIM" + and authority["decision"] == "EXECUTE") + _require(metadata["issuer"] == "FAKE_PRODUCER" and metadata["conclusion"] == "success") + for value in (authority["issued"], authority["deadline"], envelope["captured"], + envelope["expires"], metadata["created"]): + _require(_integer(value)) + _require(authority["issued"] <= envelope["captured"] <= metadata["created"] <= now + < envelope["expires"] <= authority["deadline"] + and now - envelope["captured"] <= 120 + and 0 < authority["deadline"] - authority["issued"] <= 3600) + _require(envelope["receipt"] == "TEST_ONLY_READ_VERIFIED") + for value in (ciphertext, payload): + _require(type(value) is bytes and value.startswith(b"TEST_ONLY_") + and len(value) <= 65536) + _require(envelope["payload_digest"] == digest(payload) + and metadata["ciphertext_digest"] == digest(ciphertext) + and metadata["envelope_digest"] == digest(encode(envelope))) + except (KeyError, TypeError, ValueError, AttributeError, RecursionError): + raise Denied() from None + return dict(status="SIMULATION_ONLY", execution_authorized=False, + authentic_provenance=False, private_delivery=False) + + +class FakeStore: + """In-memory test store shared by fake clients/restarts, no real adapter. + + The lock models an atomic create in one process only. It is not proof of + cross-process persistence, provider policy, ownership or distributed CAS. + """ + + def __init__(self): + self.lock = Lock() + self.intents = {} + self.results = {} + self.seen_nonces = set() + self.last_time = {} + + +class FakeGate: + """No actions are performed: only simulated intent/result states are kept.""" + + def __init__(self, bundle, store, *, now): + _require(type(store) is FakeStore) + self.bundle, self.store, self.now = bundle, store, now + self.key = None + self.intent = None + + def _current(self): + # Revalidate live harness values at EACH boundary, never a cached allow. + now = self.now() + validate_simulation(**self.bundle, now=now) + return now + + def begin(self): + with self.store.lock: + now = self._current() + b = self.bundle["authority"]["bindings"] + key = (b["generation"], b["request"]) + nonce = (b["recipient"], b["cert_pin"], b["nonce"]) + _require(self.key is None and key not in self.store.intents + and nonce not in self.store.seen_nonces) + intent = encode(self.bundle["authority"]) + self.store.intents[key] = intent + self.store.seen_nonces.add(nonce) + self.store.last_time[key] = now + self.key, self.intent = key, intent + return "TEST_ONLY_INTENT" + + def finish(self, outcome): + with self.store.lock: + _require(self.key is not None and self.key not in self.store.results + and self.store.intents.get(self.key) == self.intent) + _require(type(outcome) is str and outcome in ("success", "failure", "unknown")) + try: + now = self._current() + _require(now >= self.store.last_time[self.key] + and encode(self.bundle["authority"]) == self.intent) + except (Denied, KeyError, TypeError): + result = "TEST_ONLY_BLOCKED_AFTER_INTENT" + else: + result = "TEST_ONLY_UNKNOWN" if outcome == "unknown" else "TEST_ONLY_RECORDED" + # Expiry/failure/unknown cannot remove intent or recover its allowance. + self.store.results[self.key] = result + return result diff --git a/tests/test_recovery_authorization_gate.py b/tests/test_recovery_authorization_gate.py new file mode 100644 index 00000000..320d76b7 --- /dev/null +++ b/tests/test_recovery_authorization_gate.py @@ -0,0 +1,226 @@ +"""TEST_ONLY: deterministic P01 simulations, never authenticated provider evidence.""" +from concurrent.futures import ThreadPoolExecutor +from copy import deepcopy +import json + +import pytest + +from scripts import recovery_authorization_gate as gate + + +@pytest.fixture +def bundle(): + """All identities are FAKE; no keys, credentials, transport or runtime data.""" + bindings = dict(project="", repository="FAKE/repository", + workflow="FAKE/reader.yml", ref="refs/heads/main", + run="FAKE_RUN", attempt=1, run_sha="b" * 40, + workflow_sha="b" * 40, operator="b" * 40, reader="b" * 40, + source="a" * 40, image="sha256:" + "c" * 64, + environment="FAKE_ENVIRONMENT", target="FAKE_TARGET", + recipient="FAKE_RECIPIENT", cert_pin="d" * 64, + nonce="TEST_ONLY_NONCE", request="FAKE_REQUEST", + generation=1, artifact="FAKE_ARTIFACT", mode="receipt") + c0 = {role: "FAKE_PROOF_" + role for role in gate.C0_ROLES} + authority = dict(schema="TEST_ONLY_P01_V1", bindings=bindings, c0=c0, + issuer="FAKE_ISSUER", claim="FAKE_CLAIM", decision="EXECUTE", + issued=100, deadline=200) + envelope = dict(schema="TEST_ONLY_P01_V1", bindings=deepcopy(bindings), + captured=110, expires=190, receipt="TEST_ONLY_READ_VERIFIED", + payload_digest=gate.digest(b"TEST_ONLY_PAYLOAD")) + ciphertext = b"TEST_ONLY_VALID_CIPHERTEXT" + metadata = dict(schema="TEST_ONLY_P01_V1", bindings=deepcopy(bindings), + issuer="FAKE_PRODUCER", conclusion="success", + created=111, ciphertext_digest=gate.digest(ciphertext), + envelope_digest=gate.digest(gate.encode(envelope))) + trust = gate.FakeTrust(deepcopy(bindings), deepcopy(c0)) + return dict(authority=authority, envelope=envelope, metadata=metadata, + ciphertext=ciphertext, payload=b"TEST_ONLY_PAYLOAD", trust=trust) + + +def validate(b): + return gate.validate_simulation(**b, now=120) + + +def denied(b): + with pytest.raises(gate.Denied, match="^TEST_ONLY validation denied$"): + validate(b) + + +def test_consistent_fake_evidence_never_authorizes_execution(bundle): + assert validate(bundle) == {"status": "SIMULATION_ONLY", "execution_authorized": False, + "authentic_provenance": False, "private_delivery": False} + + +@pytest.mark.parametrize("role", gate.C0_ROLES) +def test_missing_c0_proof_denies(bundle, role): + del bundle["authority"]["c0"][role] + denied(bundle) + + +@pytest.mark.parametrize("value", [True, False, "", "approved", None]) +def test_caller_boolean_or_label_is_not_ownership(bundle, value): + bundle["authority"]["c0"]["recipient"] = value + denied(bundle) + + +@pytest.mark.parametrize("field", ["repository", "workflow", "ref", "run", "attempt", + "run_sha", "workflow_sha", "operator", "reader", "source", "image", "environment", + "target", "recipient", "cert_pin", "nonce", "request", "generation", "artifact", "mode"]) +def test_every_producer_envelope_binding_is_pinned(bundle, field): + for layer in ("metadata", "envelope", "authority"): + bad = deepcopy(bundle) + bad[layer]["bindings"][field] = "FAKE_ATTACKER" + denied(bad) + + +def test_valid_attacker_ciphertext_is_not_producer_provenance(bundle): + bundle["ciphertext"] = b"TEST_ONLY_VALID_ATTACKER_CIPHERTEXT" + bundle["metadata"]["ciphertext_digest"] = gate.digest(bundle["ciphertext"]) + bundle["metadata"]["issuer"] = "FAKE_ATTACKER" + denied(bundle) + + +@pytest.mark.parametrize("layer,field,value", [ + ("metadata", "conclusion", "unknown"), ("metadata", "schema", "unknown"), + ("metadata", "issuer", True), ("metadata", "ciphertext_digest", "0" * 64), + ("metadata", "envelope_digest", "0" * 64), ("metadata", "created", 90), + ("metadata", "created", 121), ("envelope", "captured", 0), + ("envelope", "captured", 121), ("envelope", "expires", 120), + ("envelope", "payload_digest", "0" * 64), ("envelope", "receipt", "APPROVED"), + ("authority", "issuer", "FAKE_ATTACKER"), ("authority", "claim", ""), + ("authority", "decision", "PLAN_ONLY"), ("authority", "decision", "DISABLED"), + ("authority", "decision", "SUPERSEDED"), ("authority", "issued", 121), + ("authority", "deadline", 120), ("authority", "deadline", True), +]) +def test_tamper_stale_unknown_or_disabled_denies(bundle, layer, field, value): + bundle[layer][field] = value + denied(bundle) + + +@pytest.mark.parametrize("layer", ["authority", "metadata", "envelope"]) +def test_missing_unknown_schema_fields_and_provenance_deny(bundle, layer): + for value in (None, {}, {**bundle[layer], "approved": True}): + bad = deepcopy(bundle) + bad[layer] = value + denied(bad) + + +def test_strict_json_rejects_duplicate_unknown_and_nonfinite_values(): + for raw in ('{"a":1,"a":2}', '{"a":NaN}', '[]', '{broken'): + with pytest.raises(gate.Denied): + gate.parse_document(raw) + assert gate.parse_document('{"schema":"TEST_ONLY"}') == {"schema": "TEST_ONLY"} + + +def test_fixed_diagnostics_never_echo_private_input(bundle, capsys): + bundle["metadata"]["issuer"] = "FAKE_PRIVATE_DO_NOT_ECHO" + denied(bundle) + assert capsys.readouterr() == ("", "") + + +def new_model(bundle, store=None): + return gate.FakeGate(bundle, store or gate.FakeStore(), now=lambda: 120) + + +def test_missing_claim_and_provenance_cannot_create_intent(bundle): + bundle["authority"]["claim"] = None + model = new_model(bundle) + with pytest.raises(gate.Denied): + model.begin() + assert model.store.intents == {} + + +def test_duplicate_and_restart_with_unknown_outcome_never_retry(bundle): + store = gate.FakeStore() + first = new_model(bundle, store) + assert first.begin() == "TEST_ONLY_INTENT" + with pytest.raises(gate.Denied): + new_model(bundle, store).begin() + assert first.finish("unknown") == "TEST_ONLY_UNKNOWN" + with pytest.raises(gate.Denied): + new_model(bundle, store).begin() + + +def test_two_fake_clients_have_one_intent_winner(bundle): + store = gate.FakeStore() + def begin(_): + try: + return new_model(bundle, store).begin() + except gate.Denied: + return "DENIED" + with ThreadPoolExecutor(max_workers=2) as pool: + outcomes = list(pool.map(begin, range(2))) + assert sorted(outcomes) == ["DENIED", "TEST_ONLY_INTENT"] + + +@pytest.mark.parametrize("change", ["PLAN_ONLY", "DISABLED", "SUPERSEDED", "expiry", + "operator", "main", "claim", "owner"]) +def test_latest_state_after_intent_blocks_next_step(bundle, change): + clock = [120] + model = gate.FakeGate(bundle, gate.FakeStore(), now=lambda: clock[0]) + model.begin() + if change == "expiry": clock[0] = 200 + elif change == "operator": bundle["authority"]["bindings"]["operator"] = "e" * 40 + elif change == "main": bundle["trust"].current_main = "e" * 40 + elif change == "claim": bundle["authority"]["claim"] = "FAKE_OTHER_CLAIM" + elif change == "owner": bundle["authority"]["c0"]["issuer"] = "FAKE_OTHER_OWNER" + else: bundle["authority"]["decision"] = change + assert model.finish("success") == "TEST_ONLY_BLOCKED_AFTER_INTENT" + with pytest.raises(gate.Denied): model.begin() + + +def test_expiry_before_intent_and_backward_clock_deny(bundle): + for now in (99, 200, True): + with pytest.raises(gate.Denied): + gate.FakeGate(bundle, gate.FakeStore(), now=lambda: now).begin() + clock = [120] + model = gate.FakeGate(bundle, gate.FakeStore(), now=lambda: clock[0]) + model.begin() + clock[0] = 119 + assert model.finish("success") == "TEST_ONLY_BLOCKED_AFTER_INTENT" + + +def test_lost_intent_and_duplicate_result_deny(bundle): + model = new_model(bundle) + model.begin() + model.store.intents.clear() # TEST_ONLY simulated integrity loss; never a retry. + with pytest.raises(gate.Denied): model.finish("success") + model = new_model(bundle) + model.begin() + assert model.finish("success") == "TEST_ONLY_RECORDED" + with pytest.raises(gate.Denied): model.finish("success") + + +def test_schema_rejects_real_targets_and_attempt_boolean(bundle): + for field, value in (("project", "real-target"), ("attempt", True), + ("ref", "refs/tags/FAKE"), ("run_sha", "0" * 40)): + for layer in ("authority", "metadata", "envelope"): + bundle[layer]["bindings"][field] = value + bundle["trust"].bindings[field] = value + denied(bundle) + + +def test_parser_accepts_fake_roundtrip(bundle): + raw = json.dumps(bundle["authority"]) + assert gate.parse_document(raw) == bundle["authority"] + + +def test_replayed_nonce_cannot_gain_new_request_or_generation(bundle): + store = gate.FakeStore() + first = new_model(bundle, store) + first.begin() + first.finish("success") + # A new fake card and producer tuple do not make old evidence fresh. + for layer in ("authority", "envelope", "metadata"): + bundle[layer]["bindings"]["request"] = "FAKE_NEW_REQUEST" + bundle[layer]["bindings"]["generation"] = 2 + bundle["trust"].bindings.update(request="FAKE_NEW_REQUEST", generation=2) + bundle["metadata"]["envelope_digest"] = gate.digest(gate.encode(bundle["envelope"])) + with pytest.raises(gate.Denied): new_model(bundle, store).begin() + + +def test_fake_store_clock_integrity_loss_is_fixed_deny(bundle): + model = new_model(bundle) + model.begin() + model.store.last_time.clear() + assert model.finish("success") == "TEST_ONLY_BLOCKED_AFTER_INTENT"