diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c1e782e..acf993c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -173,3 +173,15 @@ jobs: - name: Replay a SEV-SNP quote offline run: python snp_replay.py + + - name: Run the feature examples + run: | + for s in closed_model_e2e multi_stage_byom transparency_log post_quantum channel_binding revocation_kill_switch quote_verification; do + echo "== $s ==" + python "$s.py" + done + + - name: Run the model-signing provenance example + run: | + python -m pip install "weight-custody-manifest[model-signing]>=0.21.0" + python provenance_model_signing.py diff --git a/weight-custody-manifest/README.md b/weight-custody-manifest/README.md index 2023720..605b0db 100644 --- a/weight-custody-manifest/README.md +++ b/weight-custody-manifest/README.md @@ -2,7 +2,7 @@ The other examples in this repo govern what an agent may **do**. This one governs the model **weights** themselves: when a builder deploys a model into a customer's own or sovereign infrastructure, how do you prove the weights running in the enclave are exactly the ones the builder shipped, release the decryption key only against a valid hardware attestation, wipe it when custody lapses, and carry the lineage of every derivative? -That is the [Weight Custody Manifest](https://pypi.org/project/weight-custody-manifest/) (WCM): an open protocol and reference SDK. These four demos run its real code. +That is the [Weight Custody Manifest](https://pypi.org/project/weight-custody-manifest/) (WCM): an open protocol and reference SDK. These demos run its real code, roughly one per feature. For a public open-weight model, base-weight confidentiality is theater (anyone can download the weights), so the same flow does the work that still matters: **integrity and provenance** (is this the model the builder shipped, or a tampered copy), **license as a release condition**, **derivative custody and lineage**, and a **kill switch**. @@ -15,7 +15,7 @@ For a public open-weight model, base-weight confidentiality is theater (anyone c ```bash git clone https://github.com/agentrust-io/examples.git cd examples/weight-custody-manifest -pip install -r requirements.txt # weight-custody-manifest>=0.19.0, Python 3.11+ +pip install -r requirements.txt # weight-custody-manifest>=0.21.0, Python 3.11+ ``` The offline demos need nothing else. `real_open_model.py` needs run-local extras (below). @@ -36,47 +36,49 @@ The demos below go deeper on the same machinery. ## The demos -### 1. `open_model_e2e.py` -- the whole flow, end to end +Each script is one runnable feature with mock (software) attestation, so they run anywhere. `refuse_and_wipe.py` above is the 30-second version. -The full six-step Weight Custody Manifest flow on an open-weight model with a mock attestation provider: sign the joint manifest (builder + custodian), run the attestation-gated key release, hold the key under a wipe-on-lapse custody lease, enforce the license as a release condition, then fine-tune into a derivative and verify its lineage back to the base. Narrated, and honest about which steps are real work versus theater for a public model. Runs anywhere. +### Closed-weight vs open-weight -```bash -python open_model_e2e.py -``` +The same machinery, two trust settings. **Closed** weights (a frontier model deployed into someone else's infra) need secrecy. **Open** weights (Llama, Mistral, SmolLM) are already public, so the flow instead does integrity, license, and derivative custody. -### 2. `sovereign_self_custody.py` -- threshold release, no single point of trust +- **`closed_model_e2e.py`** -- the closed/frontier flow where secrecy is the job (`base_confidentiality: confidential`): sign, attestation-gated release of the real decryption key, wipe-on-lapse custody, and the honest hostile-owner caveat. +- **`open_model_e2e.py`** -- the full six-step flow on an open model, honest about which steps are real work versus theater for a public base (the base's secrecy is theater; integrity, license, derivative custody, and the kill switch are the point). -Sovereign self-custody with a 2-of-3 threshold split-key (SPEC 3.5): the release key is split with Shamir secret sharing so no single party can release the weights alone. Real WCM code, software attestation, runs anywhere. +### The four layers, feature by feature -```bash -python sovereign_self_custody.py -``` +- **`multi_stage_byom.py`** -- Layer 4 at depth (SPEC 3.8): a base to derivative to derivative chain, monotone rights (a derivative narrows but never widens), release gated on every upstream being logged, and a revocation cascading down the chain. +- **`revocation_kill_switch.py`** -- the kill switch: a serving session, a revocation, and the enclave stopping at the next cadence lapse (the key is zeroized, not suspended). +- **`channel_binding.py`** -- Layer 2 relay defense (CVE-2026-33697): the released key is sealed to the enclave's attested transport key, so a relayed attestation gets only ciphertext it cannot open. +- **`sovereign_self_custody.py`** -- 2-of-3 threshold release (SPEC 3.5): no single party, and no single forged quote, can assemble the key. -### 3. `snp_replay.py` -- replay a real attestation, offline +### Transparency, post-quantum, provenance -The CPU half of Layer 2's composite verification (SPEC 3.2), run offline against a recorded AMD SEV-SNP quote. It exercises exactly what a key broker runs before releasing a key: parse the SNP report, verify the VCEK to ASK to ARK certificate chain to a trusted AMD root, verify the report's own signature, and confirm REPORT_DATA binds the challenge nonce (anti-replay). A synthetic bundle is committed in `fixtures/` so it runs anywhere; pass a captured bundle to replay genuine silicon. +- **`transparency_log.py`** -- the append-only Merkle log (SPEC 3.7 / RFC 9162): inclusion and consistency proofs, and a monitor detecting a suppressed revocation. +- **`post_quantum.py`** -- sign and verify a manifest with ML-DSA-65 and with the Ed25519+ML-DSA-65 hybrid profile. +- **`provenance_model_signing.py`** -- interop with OpenSSF model-signing: a WCM manifest references a model-signing signature as its provenance, and `verify_provenance` cryptographically verifies it. Needs the extra: `pip install "weight-custody-manifest[model-signing]"`. -```bash -python snp_replay.py # committed synthetic bundle -python snp_replay.py path/to/snp_quote.json # a bundle captured on a real Azure SEV-SNP CVM -``` +### Verifying real attestations + +- **`snp_replay.py`** -- replay a recorded AMD SEV-SNP quote offline: parse the report, verify the VCEK to ASK to ARK chain to a trusted AMD root, the report signature, and the nonce binding. A synthetic bundle is committed in `fixtures/`; pass a captured bundle to replay genuine silicon. +- **`quote_verification.py`** -- the verifier machinery against a synthetic PKI: cert chain, report signature, and nonce binding, passing and failing (untrusted root, tampered report, nonce mismatch). This is the same machinery the SEV-SNP, TDX, and NVIDIA GPU verifiers plug into. -### 4. `real_open_model.py` -- run-local, over real weights +### Over real weights (run-local) -The same flow over a **real** open model's actual bytes: it downloads an open model (SmolLM2-135M by default, ~270MB), hashes its real safetensors into the manifest, includes a tamper demo (a one-byte-flipped fork no longer matches the manifest), and with `--infer` loads the model and generates so the certified serving stack is a real running model. Network and heavy, so it is run-local and excluded from CI. +- **`real_open_model.py`** -- the flow over a real open model's actual bytes: it downloads SmolLM2-135M (~270MB), hashes its real safetensors into the manifest, includes a tamper demo (a one-byte-flipped fork no longer matches), and with `--infer` loads the model and generates. Network and heavy, so it is run-local and excluded from CI. ```bash -pip install -r requirements-infer.txt -python real_open_model.py # download, hash, full flow, tamper demo -python real_open_model.py --infer # also load the model and generate -python real_open_model.py --local path/to/model.safetensors # skip the download +python closed_model_e2e.py # any of the offline examples +pip install "weight-custody-manifest[model-signing]" # for provenance_model_signing.py +pip install -r requirements-infer.txt # for real_open_model.py +python real_open_model.py ``` --- ## What runs in CI -The offline demos (`refuse_and_wipe.py`, `open_model_e2e.py`, `sovereign_self_custody.py`, `snp_replay.py`) run in CI against the published PyPI package and must exit 0. `real_open_model.py` is not in CI (it downloads a model). +Every offline example runs in CI against the published PyPI package and must exit 0: `refuse_and_wipe`, the closed/open e2e pair, `multi_stage_byom`, `revocation_kill_switch`, `channel_binding`, `sovereign_self_custody`, `transparency_log`, `post_quantum`, `quote_verification`, `snp_replay`, and `provenance_model_signing` (with the `[model-signing]` extra). Only `real_open_model.py` is excluded (it downloads a model). ## Reference diff --git a/weight-custody-manifest/channel_binding.py b/weight-custody-manifest/channel_binding.py new file mode 100644 index 0000000..1860b66 --- /dev/null +++ b/weight-custody-manifest/channel_binding.py @@ -0,0 +1,195 @@ +"""Channel binding: sealing the released key to the attested enclave (CVE-2026-33697). + + python channel_binding.py + +Real WCM code, no hardware. Nonce binding stops a REPLAYED quote, but not a +RELAYED one: a network adversary terminating the enclave's channel could forward +a live valid quote and divert the released key onto its own channel. WCM closes +that gap (SPEC 3.2 channel binding): the enclave folds a transport public key +into the quote's REPORT_DATA under the nonce, and the KBS seals the released key +to that transport key instead of returning it in the clear. This shows both +teeth: a relay that just forwards the release gets ciphertext it cannot open, and +a relay that substitutes its own transport key fails quote verification outright. + +The attestation quote here is signed by a synthetic PKI (real cryptography, not a +real vendor chain) so the verification is genuine end to end. +""" +from __future__ import annotations + +import base64 +import hashlib +import json +from datetime import datetime, timedelta, timezone + +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.x509.oid import NameOID + +from wcm import ( + CompositeEvidence, + CpuQuote, + Ed25519Signer, + JsonQuoteParser, + KeyBrokerService, + QuoteVerifier, + SealError, + TrustStore, + WeightCustodyManifest, + generate_ed25519, + generate_transport_keypair, + open_sealed, +) + +NOW = datetime(2026, 1, 1, tzinfo=timezone.utc) + + +def rule(title: str) -> None: + print(f"\n{'=' * 70}\n{title}\n{'=' * 70}") + + +def sha256(data: bytes) -> str: + return "sha256:" + hashlib.sha256(data).hexdigest() + + +def _pki(): + """A self-signed root CA and a leaf attestation-key cert it signs.""" + root_key = ec.generate_private_key(ec.SECP256R1()) + leaf_key = ec.generate_private_key(ec.SECP256R1()) + + def cert(subject, subj_key, issuer, issuer_key, ca): + b = ( + x509.CertificateBuilder() + .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, subject)])) + .issuer_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, issuer)])) + .public_key(subj_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(NOW - timedelta(days=1)) + .not_valid_after(NOW + timedelta(days=365)) + ) + if ca: + b = b.add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True) + return b.sign(issuer_key, hashes.SHA256()) + + root = cert("wcm-demo-root", root_key, "wcm-demo-root", root_key, True) + leaf = cert("attestation-key", leaf_key, "wcm-demo-root", root_key, False) + return root, leaf, leaf_key + + +def _quote(leaf, leaf_key, nonce_hex: str, transport_pub_hex: str) -> str: + """A quote whose REPORT_DATA binds sha256(nonce || transport_public_key).""" + digest = hashlib.sha256(bytes.fromhex(nonce_hex) + bytes.fromhex(transport_pub_hex)).digest() + report_body = digest + bytes(32) + b"measurement-and-tcb" + signature = leaf_key.sign(report_body, ec.ECDSA(hashes.SHA256())) + doc = { + "report_b64": base64.b64encode(report_body).decode(), + "signature_b64": base64.b64encode(signature).decode(), + "leaf_pem": leaf.public_bytes(serialization.Encoding.PEM).decode(), + "intermediates_pem": [], + "report_data_offset": 0, + } + return base64.b64encode(json.dumps(doc).encode()).decode() + + +def _manifest(): + serving = sha256(b"measured serving stack") + doc = { + "manifest_version": "0.1", + "weights_hash": sha256(b"the model weights"), + "builder": {"identity": "lab", "signing_key": "ed25519:lab"}, + "release_terms": { + "license": "deployment-agreement", + "permitted_derivatives": "none", + "permitted_environments": ["attested-enclave"], + }, + "release_policy": { + "required_assurance_tier": "hardware-attested", + "trusted_time_source": "secure-tsc", + "required_hw_platform": ["amd-sev-snp"], + "required_serving_image": { + "signer": "ed25519:lab", + "release_rule": "prefer-current", + "accepted_measurements": [{"measurement": serving, "status": "current"}], + }, + }, + "custody": { + "custodian": "customer", + "custodian_type": "customer-self-custody", + "kbs_image": {"measurement": sha256(b"kbs image"), "signer": "ed25519:lab"}, + "enclave_id": "did:example:enclave", + "attestation_cadence": "1h", + }, + } + b, c = generate_ed25519(), generate_ed25519() + m = WeightCustodyManifest.model_validate(doc) + return m.with_signatures([ + Ed25519Signer(b).sign(m.unsigned_dict(), role="builder", signer="lab"), + Ed25519Signer(c).sign(m.unsigned_dict(), role="custodian", signer="customer"), + ]), serving + + +def _evidence(manifest, serving, nonce, quote_b64, transport_pub_hex): + return CompositeEvidence( + cpu=CpuQuote( + platform="amd-sev-snp", + assurance_tier="hardware-attested", + serving_image_measurement=serving, + nonce_echo=nonce, + attestation_key_id="vcek:demo", + quote_b64=quote_b64, + transport_public_key=transport_pub_hex, + ) + ) + + +def main() -> int: + root, leaf, leaf_key = _pki() + manifest, serving = _manifest() + KEY = b"the-weight-decryption-key-32bytes" + + trust = TrustStore() + trust.add_root(root) + kbs = KeyBrokerService( + {manifest.weights_hash: KEY}, + now=lambda: NOW, + cpu_quote_verifier=QuoteVerifier(JsonQuoteParser(), trust), + require_channel_binding=True, + ) + + enclave_priv, enclave_pub = generate_transport_keypair() + attacker_priv, attacker_pub = generate_transport_keypair() + + rule("Step 1 - Legit enclave: the key releases, SEALED to its transport key") + ch = kbs.issue_challenge() + quote = _quote(leaf, leaf_key, ch.nonce, enclave_pub) # bound to the enclave key + decision = kbs.verify_and_release(manifest, _evidence(manifest, serving, ch.nonce, quote, enclave_pub)) + print("released :", decision.released) + print("raw key on the wire :", decision.key, " (never; sealed only)") + opened = open_sealed(decision.sealed_key, enclave_priv) + print("enclave opens the seal:", opened == KEY) + + rule("Step 2 - A relay forwards the sealed release: it gets only ciphertext") + try: + open_sealed(decision.sealed_key, attacker_priv) + print("attacker opened it : True (should not happen)") + except SealError: + print("attacker opens the seal: FAILS (SealError) - it is not the bound enclave") + + rule("Step 3 - A relay substitutes its own transport key: quote verification fails") + ch2 = kbs.issue_challenge() + quote2 = _quote(leaf, leaf_key, ch2.nonce, enclave_pub) # still bound to the enclave key + # ...but the relay claims the attacker key so the seal would land on its channel. + relayed = kbs.verify_and_release( + manifest, _evidence(manifest, serving, ch2.nonce, quote2, attacker_pub) + ) + print("released :", relayed.released, " (denied)") + reason = next((c.detail for c in relayed.failures if c.name == "cpu_quote_verified"), None) + print("why :", reason) + print() + print("Either way the relay is defeated: forward the seal and it is unopenable;") + print("swap the transport key and REPORT_DATA no longer matches, so nothing releases.") + return 0 if (decision.released and not relayed.released) else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/weight-custody-manifest/closed_model_e2e.py b/weight-custody-manifest/closed_model_e2e.py new file mode 100644 index 0000000..c97407c --- /dev/null +++ b/weight-custody-manifest/closed_model_e2e.py @@ -0,0 +1,138 @@ +"""End-to-end custody for a CLOSED-weight (frontier) model, where secrecy IS the job. + + python closed_model_e2e.py + +Real WCM code with a software (mock) attestation provider, so it runs anywhere +with no hardware. + +Closed vs open, the one distinction that drives everything +--------------------------------------------------------- +This is the counterpart to `open_model_e2e.py`. For a CLOSED frontier model the +base weights are a secret the builder is deploying into someone else's +infrastructure, so the whole point of attestation-gated release is secrecy: the +decryption key must reach only a builder-approved, measured serving stack, and +nothing else. (For an OPEN-weight model the base is already public, so the same +machinery instead does integrity, license, and derivative custody. See +`open_model_e2e.py`.) + +`base_confidentiality` is set to "confidential" and stated on the manifest, so a +verifier can see that this manifest really is protecting secrecy, not theater. +""" +from __future__ import annotations + +import hashlib + +from wcm import ( + Ed25519Signer, + EnclaveSession, + KeyBrokerService, + SoftwareProvider, + VerificationContext, + WeightCustodyManifest, + generate_ed25519, + verify_manifest, +) + + +def rule(title: str) -> None: + print(f"\n{'=' * 70}\n{title}\n{'=' * 70}") + + +def sha256(data: bytes) -> str: + return "sha256:" + hashlib.sha256(data).hexdigest() + + +def build_manifest(weights_hash: str, serving: str, org: str) -> dict: + return { + "manifest_version": "0.1", + "weights_hash": weights_hash, + "builder": {"identity": org, "signing_key": "ed25519:builder"}, + "release_terms": { + "license": "Frontier-Model-Deployment-Agreement (no redistribution)", + "permitted_derivatives": "none", + "derivatives": "none", + "permitted_environments": ["customer-attested-enclave"], + }, + "release_policy": { + "required_assurance_tier": "hardware-attested", + "trusted_time_source": "secure-tsc", + "required_hw_platform": ["amd-sev-snp", "nvidia-cc-gpu"], + "required_gpu_measurement": {"rim_pin": "nvidia-rim:golden"}, + "required_serving_image": { + "signer": "ed25519:builder", + "release_rule": "prefer-current", + "accepted_measurements": [{"measurement": serving, "status": "current"}], + }, + "attestation_revocation_check": "live-per-release, max-cache-age: short-window", + "revocation_authority": "builder-and-opaque-joint", + }, + "custody": { + "custodian": "customer-platform-team", + "custodian_type": "customer-self-custody", + "kbs_image": {"measurement": sha256(b"builder-signed KBS image"), "signer": "ed25519:builder"}, + "enclave_id": "did:example:customer-enclave-01", + "attestation_cadence": "30m", + }, + # The distinction this whole demo turns on: the base weights ARE secret. + "base_confidentiality": "confidential", + "deployment_model": "builder-to-customer", + } + + +def main() -> int: + builder, custodian = generate_ed25519(), generate_ed25519() + weights_hash = sha256(b"") + serving = sha256(b"vllm-0.6.3 + builder policy bundle (the approved serving stack)") + + rule("Step 0 - Certify: the manifest binds the secret weights + serving stack") + doc = build_manifest(weights_hash, serving, "frontier-lab") + manifest = WeightCustodyManifest.model_validate(doc) + manifest = manifest.with_signatures([ + Ed25519Signer(builder).sign(manifest.unsigned_dict(), role="builder", signer="frontier-lab"), + Ed25519Signer(custodian).sign(manifest.unsigned_dict(), role="custodian", signer="customer-platform-team"), + ]) + print("base_confidentiality :", manifest.base_confidentiality.value, " (secrecy is the job)") + print("deployment_model :", manifest.deployment_model.value) + print("signed jointly by builder + custodian") + + rule("Step 1 - Verify the manifest (both required roles signed)") + ctx = VerificationContext() + ctx.add_key(builder.public_bytes) + ctx.add_key(custodian.public_bytes) + result = verify_manifest(manifest, ctx) + print("manifest valid :", result.ok) + for note in result.notes: + print(" note:", note) + if not result.ok: + return 1 + + rule("Step 2 - Attestation gate: the decryption key releases only into the approved enclave") + kbs = KeyBrokerService({weights_hash: b"the-secret-weight-decryption-key"}) + challenge = kbs.issue_challenge() + evidence = SoftwareProvider().produce( + challenge, serving_image_measurement=serving, gpu_measurement="nvidia-rim:golden" + ) + decision = kbs.verify_and_release(manifest, evidence) + print("key released :", decision.released) + print("the key decrypts the secret weights ONLY under the builder-signed, measured") + print("serving stack. A different or tampered stack gets nothing (see refuse_and_wipe.py).") + + rule("Step 3 - Wipe-on-lapse custody (bounded exposure of the secret)") + session = EnclaveSession.from_release(manifest, decision) + session.use_key() + print("serving under a", doc["custody"]["attestation_cadence"], "cadence, time_floor =", session.time_floor.value) + print("miss the re-attestation and the key is zeroized, so worst-case exposure of") + print("the secret is one cadence window even if a revocation signal is blocked.") + + rule("The honest ceiling for a secret") + print("secrecy holds against software and remote adversaries. It does NOT hold against") + print("an operator who physically owns the hardware: TEE.fail / BadRAM extract keys") + print("from live memory (open question 8.8). Against that party WCM is accountability-") + print("grade (cost, detection, containment, mandatory physical hardening), not") + print("cryptographic custody. A frontier builder self-selects its counterparties on") + print("exactly this line.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/weight-custody-manifest/multi_stage_byom.py b/weight-custody-manifest/multi_stage_byom.py new file mode 100644 index 0000000..97a3d27 --- /dev/null +++ b/weight-custody-manifest/multi_stage_byom.py @@ -0,0 +1,147 @@ +"""Multi-stage BYOM: sequential re-custody down a derivative chain (SPEC 3.8). + + python multi_stage_byom.py + +Real WCM code, no hardware. A base model is fine-tuned, then that derivative is +fine-tuned again, each stage getting its own signed manifest that chains to its +parent by `derived_from`. `verify_lineage` enforces the three properties that +make a multi-stage pipeline safe: + + - monotone rights: a derivative may NARROW but never WIDEN the rights it + inherits (the `derivatives` policy and `permitted_environments`); + - upstream-logged gating: release is allowed only if every manifest up the + chain is present in the transparency log; + - revocation cascade: revoking any manifest invalidates everything downstream. +""" +from __future__ import annotations + +import hashlib + +from wcm import ( + Ed25519Signer, + WeightCustodyManifest, + generate_ed25519, + is_root, + verify_lineage, +) + + +def rule(title: str) -> None: + print(f"\n{'=' * 70}\n{title}\n{'=' * 70}") + + +def sha256(data: bytes) -> str: + return "sha256:" + hashlib.sha256(data).hexdigest() + + +def build_manifest( + *, + weights_hash: str, + org: str, + derivatives: str, + environments: list[str], + derived_from: str | None = None, + rights_holder: dict | None = None, +) -> dict: + serving = sha256(b"measured serving stack") + doc: dict = { + "manifest_version": "0.1", + "weights_hash": weights_hash, + "builder": {"identity": org, "signing_key": "ed25519:gov"}, + "release_terms": { + "license": "enterprise-governance", + "permitted_derivatives": derivatives, + "derivatives": derivatives, + "permitted_environments": environments, + }, + "release_policy": { + "required_assurance_tier": "hardware-attested", + "trusted_time_source": "secure-tsc", + "required_hw_platform": ["amd-sev-snp"], + "required_serving_image": { + "signer": "ed25519:gov", + "release_rule": "prefer-current", + "accepted_measurements": [{"measurement": serving, "status": "current"}], + }, + }, + "custody": { + "custodian": org, + "custodian_type": "customer-self-custody", + "kbs_image": {"measurement": sha256(b"kbs image"), "signer": "ed25519:gov"}, + "enclave_id": "did:example:gov-enclave", + "attestation_cadence": "1h", + }, + "base_confidentiality": "gated-open", + "deployment_model": "byom-symmetric", + } + if derived_from is not None: + doc["derived_from"] = derived_from + if rights_holder is not None: + doc["rights_holder"] = rights_holder + return doc + + +def sign(doc: dict, keys) -> WeightCustodyManifest: + m = WeightCustodyManifest.model_validate(doc) + return m.with_signatures([ + Ed25519Signer(keys[0]).sign(m.unsigned_dict(), role="builder", signer=doc["builder"]["identity"]), + Ed25519Signer(keys[1]).sign(m.unsigned_dict(), role="custodian", signer=doc["custody"]["custodian"]), + ]) + + +def show(label: str, result) -> None: + chain = " -> ".join(h.split(":")[1][:10] + "..." for h in result.chain) if result.chain else "(none)" + print(f"{label:<34} ok={result.ok} depth={result.depth} chain={chain}") + for err in getattr(result, "errors", []) or []: + print(" reason:", err) + + +def main() -> int: + keys = (generate_ed25519(), generate_ed25519()) + + rule("Build a 3-stage chain: base -> fine-tune -> fine-tune-of-fine-tune") + base_h = sha256(b"public base checkpoint") + d1_h = sha256(b"base + proprietary fine-tune A") + d2_h = sha256(b"derivative-1 + fine-tune B") + + # Rights narrow at every stage: unrestricted -> fine-tune-only -> none, and + # the permitted environments shrink. That is the monotone direction. + base = sign(build_manifest( + weights_hash=base_h, org="acme-gov", derivatives="unrestricted", + environments=["enclave-a", "enclave-b"]), keys) + d1 = sign(build_manifest( + weights_hash=d1_h, org="acme-gov", derivatives="fine-tune-only", + environments=["enclave-a"], derived_from=base_h, + rights_holder={"base": "meta", "derivative": "acme"}), keys) + d2 = sign(build_manifest( + weights_hash=d2_h, org="acme-gov", derivatives="none", + environments=["enclave-a"], derived_from=d1_h, + rights_holder={"base": "acme", "derivative": "acme"}), keys) + manifests = {base_h: base, d1_h: d1, d2_h: d2} + print("root is a base manifest:", is_root(base), " leaf derived_from:", d2.derived_from[:19], "...") + + rule("1. Monotone rights: the narrowing chain verifies") + show("leaf d2 (rights narrow):", verify_lineage(manifests, d2_h)) + + rule("2. A derivative that WIDENS its parent's rights is rejected") + # d1 allows only fine-tune-only; a child that re-opens to unrestricted widens. + bad_h = sha256(b"derivative that illegally re-widens") + bad = sign(build_manifest( + weights_hash=bad_h, org="acme-gov", derivatives="unrestricted", # widened! + environments=["enclave-a"], derived_from=d1_h, + rights_holder={"base": "acme", "derivative": "rogue"}), keys) + show("widened child:", verify_lineage({**manifests, bad_h: bad}, bad_h)) + + rule("3. Upstream-logged gating: release only if every ancestor is logged") + show("all three logged:", verify_lineage(manifests, d2_h, logged=[base_h, d1_h, d2_h])) + show("d1 missing from log:", verify_lineage(manifests, d2_h, logged=[base_h, d2_h])) + + rule("4. Revocation cascades down the chain") + show("revoke the base:", verify_lineage(manifests, d2_h, revoked=[base_h])) + print("revoking the base invalidates the leaf two stages down: one pull kills the") + print("whole lineage, which is what sequential re-custody has to guarantee.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/weight-custody-manifest/post_quantum.py b/weight-custody-manifest/post_quantum.py new file mode 100644 index 0000000..a5e9922 --- /dev/null +++ b/weight-custody-manifest/post_quantum.py @@ -0,0 +1,104 @@ +"""Post-quantum manifest signing: ML-DSA-65 and an Ed25519 + ML-DSA-65 hybrid. + + python post_quantum.py + +Real WCM code, no hardware. A weight-custody manifest is a long-lived artifact: +a signature that is safe today must still be safe when a model it governs is +still deployed years out. WCM supports FIPS 204 ML-DSA-65 (post-quantum) and a +hybrid that requires BOTH a classical Ed25519 signature and an ML-DSA-65 one, so +it stays valid unless an attacker breaks both. This signs and verifies a manifest +under each profile. +""" +from __future__ import annotations + +import hashlib + +from wcm import ( + HybridSigner, + MlDsa65Signer, + VerificationContext, + WeightCustodyManifest, + generate_hybrid, + generate_ml_dsa65, + verify_manifest, +) + + +def rule(title: str) -> None: + print(f"\n{'=' * 70}\n{title}\n{'=' * 70}") + + +def sha256(data: bytes) -> str: + return "sha256:" + hashlib.sha256(data).hexdigest() + + +def manifest_doc() -> dict: + serving = sha256(b"measured serving stack") + return { + "manifest_version": "0.1", + "weights_hash": sha256(b"the model weights"), + "builder": {"identity": "frontier-lab", "signing_key": "pq:builder"}, + "release_terms": { + "license": "deployment-agreement", + "permitted_derivatives": "fine-tune-only", + "permitted_environments": ["attested-enclave"], + }, + "release_policy": { + "required_assurance_tier": "hardware-attested", + "trusted_time_source": "secure-tsc", + "required_hw_platform": ["amd-sev-snp"], + "required_serving_image": { + "signer": "pq:builder", + "release_rule": "prefer-current", + "accepted_measurements": [{"measurement": serving, "status": "current"}], + }, + }, + "custody": { + "custodian": "customer-platform", + "custodian_type": "customer-self-custody", + "kbs_image": {"measurement": sha256(b"kbs image"), "signer": "pq:builder"}, + "enclave_id": "did:example:enclave", + "attestation_cadence": "1h", + }, + } + + +def main() -> int: + rule("Profile 1 - ML-DSA-65 (FIPS 204, post-quantum)") + b_key, c_key = generate_ml_dsa65(), generate_ml_dsa65() + m = WeightCustodyManifest.model_validate(manifest_doc()) + m = m.with_signatures([ + MlDsa65Signer(b_key).sign(m.unsigned_dict(), role="builder", signer="frontier-lab"), + MlDsa65Signer(c_key).sign(m.unsigned_dict(), role="custodian", signer="customer-platform"), + ]) + ctx = VerificationContext() + ctx.add_ml_dsa65_key(b_key.public_bytes) + ctx.add_ml_dsa65_key(c_key.public_bytes) + result = verify_manifest(m, ctx) + print("algorithm :", ", ".join(sorted({s.algorithm.value for s in m.signatures}))) + print("manifest verifies :", result.ok) + + rule("Profile 2 - Ed25519 + ML-DSA-65 hybrid (valid only if BOTH hold)") + b_hy, c_hy = generate_hybrid(), generate_hybrid() + m2 = WeightCustodyManifest.model_validate(manifest_doc()) + m2 = m2.with_signatures([ + HybridSigner(b_hy).sign(m2.unsigned_dict(), role="builder", signer="frontier-lab"), + HybridSigner(c_hy).sign(m2.unsigned_dict(), role="custodian", signer="customer-platform"), + ]) + ctx2 = VerificationContext() + ctx2.add_hybrid_key(b_hy.ed25519.public_bytes, b_hy.ml_dsa65.public_bytes) + ctx2.add_hybrid_key(c_hy.ed25519.public_bytes, c_hy.ml_dsa65.public_bytes) + result2 = verify_manifest(m2, ctx2) + print("algorithm :", ", ".join(sorted({s.algorithm.value for s in m2.signatures}))) + print("manifest verifies :", result2.ok) + + rule("Why both are offered") + print("ML-DSA-65 removes the quantum risk to the manifest signature outright.") + print("The hybrid is the conservative migration path: it keeps the classical") + print("guarantee AND adds the PQ one, so it stays valid unless an attacker breaks") + print("both schemes. Same manifest, same verifier, selectable per deployment.") + return 0 if (result.ok and result2.ok) else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/weight-custody-manifest/provenance_model_signing.py b/weight-custody-manifest/provenance_model_signing.py new file mode 100644 index 0000000..7518c74 --- /dev/null +++ b/weight-custody-manifest/provenance_model_signing.py @@ -0,0 +1,133 @@ +"""Provenance interop: WCM custody on top of an OpenSSF model-signing signature. + + pip install "weight-custody-manifest[model-signing]" + python provenance_model_signing.py + +Real WCM code, no hardware. OpenSSF model-signing (Sigstore) answers "who signed +this artifact and does it still match?"; WCM answers "release the key only into an +attested enclave, wipe it on lapse, and carry the lineage." They compose: a WCM +manifest carries a signed `provenance.model_signing` reference, and +`verify_provenance` cryptographically verifies that model-signing signature and +binds it to the manifest. WCM is the custody-and-release layer ON TOP of the +signing layer, not a competitor to it. + +This signs a model file with model-signing (using a local EC key, so no Sigstore +account or network is needed), records the model-signing digest in a WCM +manifest, and verifies the whole chain. +""" +from __future__ import annotations + +import hashlib +import pathlib +import tempfile + +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ec +from model_signing import signing + +from wcm import ( + Ed25519Signer, + WeightCustodyManifest, + generate_ed25519, + model_signing_digest, + verify_provenance, +) + + +def rule(title: str) -> None: + print(f"\n{'=' * 70}\n{title}\n{'=' * 70}") + + +def sha256(data: bytes) -> str: + return "sha256:" + hashlib.sha256(data).hexdigest() + + +def build_manifest(weights_hash: str, signed_digest: str) -> WeightCustodyManifest: + serving = sha256(b"measured serving stack") + doc = { + "manifest_version": "0.1", + "weights_hash": weights_hash, + "builder": {"identity": "frontier-lab", "signing_key": "ed25519:lab"}, + "release_terms": { + "license": "deployment-agreement", + "permitted_derivatives": "fine-tune-only", + "permitted_environments": ["attested-enclave"], + }, + "release_policy": { + "required_assurance_tier": "hardware-attested", + "trusted_time_source": "secure-tsc", + "required_hw_platform": ["amd-sev-snp"], + "required_serving_image": { + "signer": "ed25519:lab", + "release_rule": "prefer-current", + "accepted_measurements": [{"measurement": serving, "status": "current"}], + }, + }, + "custody": { + "custodian": "customer", + "custodian_type": "customer-self-custody", + "kbs_image": {"measurement": sha256(b"kbs image"), "signer": "ed25519:lab"}, + "enclave_id": "did:example:enclave", + "attestation_cadence": "1h", + }, + # The provenance reference, under the joint signature. + "provenance": { + "model_signing": { + "method": "openssf-model-signing", + "signed_digest": signed_digest, + "transparency": "sigstore-rekor", + "signer": "release-engineering@frontier-lab", + } + }, + } + b, c = generate_ed25519(), generate_ed25519() + m = WeightCustodyManifest.model_validate(doc) + return m.with_signatures([ + Ed25519Signer(b).sign(m.unsigned_dict(), role="builder", signer="frontier-lab"), + Ed25519Signer(c).sign(m.unsigned_dict(), role="custodian", signer="customer"), + ]) + + +def main() -> int: + with tempfile.TemporaryDirectory() as td: + d = pathlib.Path(td) + model_path = d / "model.safetensors" + model_path.write_bytes(b"") + sig_path = d / "model.sig" + priv_path = d / "signer.key" + pub_path = d / "signer.pub" + + rule("Step 1 - Sign the model with OpenSSF model-signing (local EC key)") + key = ec.generate_private_key(ec.SECP256R1()) + priv_path.write_bytes(key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + )) + pub_path.write_bytes(key.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo + )) + signing.Config().use_elliptic_key_signer(private_key=priv_path).sign(model_path, sig_path) + digest = model_signing_digest(model_path) + print("model-signing signature written:", sig_path.name) + print("model-signing digest :", digest[:26], "...") + + rule("Step 2 - Record it in a WCM manifest and verify the whole chain") + manifest = build_manifest(sha256(model_path.read_bytes()), digest) + result = verify_provenance(manifest, model_path, sig_path, public_key=pub_path) + print("provenance verified :", result.verified, + "" if result.verified else f"({result.reason})") + + rule("Step 3 - A manifest that claims the wrong digest fails the binding") + wrong = build_manifest(sha256(model_path.read_bytes()), sha256(b"a different artifact")) + bad = verify_provenance(wrong, model_path, sig_path, public_key=pub_path) + print("provenance verified :", bad.verified, f"({bad.reason})") + + print() + print("So the WCM manifest cannot claim a model-signing provenance it does not") + print("actually match: verify_provenance checks the signature AND the digest.") + return 0 if (result.verified and not bad.verified) else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/weight-custody-manifest/quote_verification.py b/weight-custody-manifest/quote_verification.py new file mode 100644 index 0000000..b48777a --- /dev/null +++ b/weight-custody-manifest/quote_verification.py @@ -0,0 +1,124 @@ +"""Quote verification: the KBS-side trust decision, and where real vendors plug in. + + python quote_verification.py + +Real WCM code, no hardware. Getting structured evidence to the gate is not the +trust decision. The trust decision is cryptographic: does the attestation quote +carry a valid hardware signature, does that key chain to a root the verifier +trusts, and is the quote bound to THIS challenge nonce? This drives +`QuoteVerifier` through the happy path and every failure mode against a synthetic +PKI (real cryptography), then points at the real vendor parsers that produce the +same `ParsedQuote` from genuine binary quotes. +""" +from __future__ import annotations + +import base64 +import hashlib +import json +from datetime import datetime, timedelta, timezone + +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.x509.oid import NameOID + +from wcm import ( + JsonQuoteParser, + QuoteVerifier, + SnpQuoteParser, + TrustStore, + build_gpu_verifier, + verify_tdx_quote, +) + +NOW = datetime(2026, 1, 1, tzinfo=timezone.utc) +NONCE = "ab" * 32 + + +def rule(title: str) -> None: + print(f"\n{'=' * 70}\n{title}\n{'=' * 70}") + + +def _pki(): + root_key = ec.generate_private_key(ec.SECP256R1()) + leaf_key = ec.generate_private_key(ec.SECP256R1()) + + def cert(subject, subj_key, issuer, issuer_key, ca): + b = ( + x509.CertificateBuilder() + .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, subject)])) + .issuer_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, issuer)])) + .public_key(subj_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(NOW - timedelta(days=1)) + .not_valid_after(NOW + timedelta(days=365)) + ) + if ca: + b = b.add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True) + return b.sign(issuer_key, hashes.SHA256()) + + root = cert("vendor-root", root_key, "vendor-root", root_key, True) + leaf = cert("attestation-key", leaf_key, "vendor-root", root_key, False) + return root, leaf, leaf_key + + +def _quote(leaf, leaf_key, nonce_hex: str) -> str: + report_body = hashlib.sha256(bytes.fromhex(nonce_hex)).digest() + bytes(32) + b"measurement" + signature = leaf_key.sign(report_body, ec.ECDSA(hashes.SHA256())) + doc = { + "report_b64": base64.b64encode(report_body).decode(), + "signature_b64": base64.b64encode(signature).decode(), + "leaf_pem": leaf.public_bytes(serialization.Encoding.PEM).decode(), + "intermediates_pem": [], + "report_data_offset": 0, + } + return base64.b64encode(json.dumps(doc).encode()).decode() + + +def _verifier(root) -> QuoteVerifier: + ts = TrustStore() + ts.add_root(root) + return QuoteVerifier(JsonQuoteParser(), ts) + + +def main() -> int: + root, leaf, leaf_key = _pki() + quote = _quote(leaf, leaf_key, NONCE) + v = _verifier(root) + + rule("Happy path - chain, signature, and nonce binding all check") + r = v.verify(quote, expected_nonce=NONCE, now=NOW) + print("verified:", r.verified, " leaf:", r.leaf_subject) + + rule("Untrusted root - the quote does not chain to a root we trust") + other_root, _, _ = _pki() + r = _verifier(other_root).verify(quote, expected_nonce=NONCE, now=NOW) + print("verified:", r.verified, " reason:", r.reason) + + rule("Tampered report - the signature no longer verifies") + doc = json.loads(base64.b64decode(quote)) + body = bytearray(base64.b64decode(doc["report_b64"])) + body[-1] ^= 0xFF + doc["report_b64"] = base64.b64encode(bytes(body)).decode() + tampered = base64.b64encode(json.dumps(doc).encode()).decode() + r = v.verify(tampered, expected_nonce=NONCE, now=NOW) + print("verified:", r.verified, " reason:", r.reason) + + rule("Nonce mismatch - a replayed quote is not bound to this challenge") + r = v.verify(quote, expected_nonce="cd" * 32, now=NOW) + print("verified:", r.verified, " reason:", r.reason) + + rule("Where the real vendors plug in") + print("The machinery above is vendor-agnostic. A real deployment swaps the") + print("JsonQuoteParser + synthetic root for a vendor parser + the vendor's real root:") + print(f" AMD SEV-SNP : {SnpQuoteParser.__name__} (VCEK -> ASK -> ARK), validated on live Azure") + print(f" Intel TDX : {verify_tdx_quote.__name__} (DCAP v4 -> Intel SGX Root CA), validated on GCP") + print(f" NVIDIA H100 : {build_gpu_verifier.__name__} (GPU report -> NVIDIA device root)") + print("None of this closes the key-extraction hole (open question 8.8): a") + print("physically-extracted key produces a genuinely valid signature. It raises") + print("the bar to a real hardware signature; it does not beat a hardware owner.") + return 0 if r is not None else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/weight-custody-manifest/requirements.txt b/weight-custody-manifest/requirements.txt index a2be462..e99a4f2 100644 --- a/weight-custody-manifest/requirements.txt +++ b/weight-custody-manifest/requirements.txt @@ -1,3 +1,4 @@ # The Weight Custody Manifest reference SDK, from PyPI. -# Covers the three offline demos (open_model_e2e, sovereign_self_custody, snp_replay). -weight-custody-manifest>=0.19.0 +# Covers the offline examples. provenance_model_signing.py additionally needs the +# [model-signing] extra: pip install "weight-custody-manifest[model-signing]". +weight-custody-manifest>=0.21.0 diff --git a/weight-custody-manifest/revocation_kill_switch.py b/weight-custody-manifest/revocation_kill_switch.py new file mode 100644 index 0000000..10cf262 --- /dev/null +++ b/weight-custody-manifest/revocation_kill_switch.py @@ -0,0 +1,133 @@ +"""The kill switch: revocation blocks new releases, wipe-on-lapse stops the current one. + + python revocation_kill_switch.py + +Real WCM code, no hardware. A recall (a safety issue, a license breach, a +compromised key) has to actually stop a model that is already serving. WCM does +that from two sides at once: + + - the KBS refuses any NEW release once the attestation key / manifest is + revoked (checked live per release); + - the enclave already holding a key zeroizes it at the next cadence lapse + unless it re-attests, and re-attestation now fails because of the revocation. + +So the worst case is one cadence window, even against a host that swallows every +push notification. +""" +from __future__ import annotations + +import hashlib +from datetime import datetime, timedelta, timezone + +from wcm import ( + Ed25519Signer, + EnclaveSession, + KeyBrokerService, + KeyWipedError, + SoftwareProvider, + WeightCustodyManifest, + generate_ed25519, +) + +NOW = datetime(2026, 1, 1, tzinfo=timezone.utc) +LIVE_KEY_ID = "vcek:live-0001" + + +def rule(title: str) -> None: + print(f"\n{'=' * 70}\n{title}\n{'=' * 70}") + + +def sha256(data: bytes) -> str: + return "sha256:" + hashlib.sha256(data).hexdigest() + + +def build_manifest(weights_hash: str, serving: str): + doc = { + "manifest_version": "0.1", + "weights_hash": weights_hash, + "builder": {"identity": "lab", "signing_key": "ed25519:lab"}, + "release_terms": { + "license": "deployment-agreement", + "permitted_derivatives": "none", + "permitted_environments": ["attested-enclave"], + }, + "release_policy": { + "required_assurance_tier": "hardware-attested", + "trusted_time_source": "secure-tsc", + "required_hw_platform": ["amd-sev-snp"], + "required_serving_image": { + "signer": "ed25519:lab", + "release_rule": "prefer-current", + "accepted_measurements": [{"measurement": serving, "status": "current"}], + }, + # Revocation is checked live on every release. + "attestation_revocation_check": "live-per-release, max-cache-age: short-window", + "revocation_authority": "builder-and-opaque-joint", + }, + "custody": { + "custodian": "customer", + "custodian_type": "customer-self-custody", + "kbs_image": {"measurement": sha256(b"kbs image"), "signer": "ed25519:lab"}, + "enclave_id": "did:example:enclave", + "attestation_cadence": "30m", + }, + } + b, c = generate_ed25519(), generate_ed25519() + m = WeightCustodyManifest.model_validate(doc) + return m.with_signatures([ + Ed25519Signer(b).sign(m.unsigned_dict(), role="builder", signer="lab"), + Ed25519Signer(c).sign(m.unsigned_dict(), role="custodian", signer="customer"), + ]) + + +def _release(kbs, manifest, serving): + ch = kbs.issue_challenge() + ev = SoftwareProvider().produce( + ch, serving_image_measurement=serving, gpu_measurement="nvidia-rim:golden", + attestation_key_id=LIVE_KEY_ID, + ) + return kbs.verify_and_release(manifest, ev) + + +def main() -> int: + serving = sha256(b"measured serving stack") + weights_hash = sha256(b"the model weights") + manifest = build_manifest(weights_hash, serving) + key = b"the-weight-decryption-key" + + rule("Step 1 - The model is attested, released, and serving") + kbs = KeyBrokerService({weights_hash: key}, now=lambda: NOW) + decision = _release(kbs, manifest, serving) + print("released :", decision.released) + session = EnclaveSession.from_release(manifest, decision, now=lambda: NOW) + session.use_key(now=NOW) + print("serving under a", "30m", "cadence") + + rule("Step 2 - A recall revokes the attestation key. New releases are refused.") + # The revocation authority marks the key revoked; the KBS enforces it live. + revoked_kbs = KeyBrokerService( + {weights_hash: key}, now=lambda: NOW, revoked_attestation_keys={LIVE_KEY_ID} + ) + denied = _release(revoked_kbs, manifest, serving) + print("new release after recall:", denied.released, " (refused)") + reason = next((c.detail for c in denied.failures if c.name == "attestation_revocation"), None) + print("why :", reason) + + rule("Step 3 - The enclave already serving zeroizes at the next cadence lapse") + later = NOW + timedelta(minutes=31) # past the 30m window, and re-attestation now fails + try: + session.use_key(now=later) + print("still serving : True (should not happen)") + except KeyWipedError: + print("cadence lapsed : key ZEROIZED (re-attestation would be refused anyway)") + print("state :", session.state.value) + + print() + print("Two sides of one kill switch: revocation stops the next release, wipe-on-lapse") + print("stops the running one. Anchoring the revocation in the transparency log") + print("(see transparency_log.py) makes it non-repudiable.") + return 0 if (decision.released and not denied.released and session.is_wiped) else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/weight-custody-manifest/transparency_log.py b/weight-custody-manifest/transparency_log.py new file mode 100644 index 0000000..9e1d013 --- /dev/null +++ b/weight-custody-manifest/transparency_log.py @@ -0,0 +1,83 @@ +"""Transparency log: inclusion, consistency, and detecting a suppressed revocation. + + python transparency_log.py + +Real WCM code, no hardware. WCM anchors manifests and revocations in an +append-only Merkle log (RFC 9162), so an operator cannot quietly serve a revoked +model or fork the history. This shows the three checks a monitor relies on: +an inclusion proof (an entry really is in the log), a consistency proof (the log +only ever appended, never rewrote), and a signed tree head (the log operator +actually vouched for this state). Then it shows the point of it all: a revocation +that is in the log cannot be hidden, and a forked log fails consistency. +""" +from __future__ import annotations + +import hashlib + +from wcm import ( + EntryType, + TransparencyLog, + generate_ed25519, + verify_inclusion, + verify_log_consistency, + verify_sth, +) + + +def rule(title: str) -> None: + print(f"\n{'=' * 70}\n{title}\n{'=' * 70}") + + +def sha256(data: bytes) -> str: + return "sha256:" + hashlib.sha256(data).hexdigest() + + +def main() -> int: + log_kp = generate_ed25519() + log = TransparencyLog(log_kp) + + rule("Step 1 - Append two manifests; take a signed tree head") + m1 = {"weights_hash": sha256(b"model-1"), "status": "in-force"} + m2 = {"weights_hash": sha256(b"model-2"), "status": "in-force"} + i1 = log.append(m1, entry_type=EntryType.manifest) + i2 = log.append(m2, entry_type=EntryType.manifest) + size1 = log.size + sth1 = log.signed_tree_head() + print("appended at indices :", i1, i2, " log size:", log.size) + print("STH signature valid :", verify_sth(sth1, log_kp.public_bytes)) + + rule("Step 2 - Inclusion proof: manifest 1 really is in the log") + proof = log.inclusion_proof(i1) + print("inclusion of m1 verifies:", verify_inclusion(m1, EntryType.manifest, proof, sth1)) + # A statement that was never logged does not verify against the same proof slot. + forged = {"weights_hash": sha256(b"never-logged"), "status": "in-force"} + print("inclusion of a forgery :", verify_inclusion(forged, EntryType.manifest, proof, sth1)) + + rule("Step 3 - Revoke model 1; the log only grows") + rev = {"weights_hash": m1["weights_hash"], "reason": "safety-recall"} + i3 = log.append(rev, entry_type=EntryType.revocation) + sth2 = log.signed_tree_head() + print("revocation appended at :", i3, " new size:", log.size) + + rule("Step 4 - Consistency proof: the new log is an append-only extension") + cproof = log.consistency_proof(size1) + print("old->new is consistent :", verify_log_consistency(sth1, sth2, cproof)) + + rule("Step 5 - A suppressed revocation is detectable") + # The operator would love to serve model 1 as if it were never revoked. But the + # revocation is in the append-only log, and a monitor can find it. + found_at = log.find(rev, entry_type=EntryType.revocation) + print("monitor finds the revocation at index:", found_at, "(cannot be hidden)") + # And a revocation that a party CLAIMS but never logged simply is not there, + # so 'I revoked it' is only credible if it is in the transparency log. + unlogged = {"weights_hash": m2["weights_hash"], "reason": "claimed-but-not-logged"} + print("an unlogged revocation claim -> :", log.find(unlogged, entry_type=EntryType.revocation), + "(not in the log, so not provable)") + print() + print("A forked or rewritten log would fail the Step 4 consistency check, so the") + print("operator cannot both keep serving and pretend the revocation never happened.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())