Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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
52 changes: 27 additions & 25 deletions weight-custody-manifest/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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**.

Expand All @@ -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).
Expand All @@ -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

Expand Down
195 changes: 195 additions & 0 deletions weight-custody-manifest/channel_binding.py
Original file line number Diff line number Diff line change
@@ -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())
Loading