commit c8a04e144bdce0a7f533dd095f2a2add45e5b8db Author: Meanwhile Research Date: Sun Jul 19 16:31:58 2026 +1000 Initial public release: capability-licensing reference implementation + reproducible claim demos Clean-room reference implementation (Ed25519 certificate chain, AES-256-GCM envelope, three-rung revocation ladder, crypto-erasure) with runnable examples and 36 tests that reproduce each demonstrated claim on stand-in payloads. Co-Authored-By: Claude Opus 4.8 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..17cd2fc --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +.venv/ +__pycache__/ +*.pyc +.pytest_cache/ diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..b488a41 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Meanwhile Research + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..cc26189 --- /dev/null +++ b/Makefile @@ -0,0 +1,24 @@ +PYTHON ?= python3 +VENV := .venv + +.PHONY: all venv examples test clean + +all: examples test + +venv: + $(PYTHON) -m venv $(VENV) + $(VENV)/bin/pip install -r requirements.txt + @echo "Now run: PYTHON=$(VENV)/bin/python make all" + +examples: + $(PYTHON) examples/issue_certificate.py + $(PYTHON) examples/verify_output_provenance.py + $(PYTHON) examples/revoke_three_rungs.py + $(PYTHON) examples/crypto_erasure_undecryptable.py + +test: + $(PYTHON) -m pytest tests/ -q + +clean: + rm -rf $(VENV) .pytest_cache + find . -type d -name __pycache__ -exec rm -rf {} + diff --git a/README.md b/README.md new file mode 100644 index 0000000..490bbba --- /dev/null +++ b/README.md @@ -0,0 +1,293 @@ +# Capability licensing and accountable inference + +## A clean-room reference implementation + +This repository is a standalone, reproducible demonstration of the cryptographic +mechanisms behind the Meanwhile capability-licensing research line. Every primitive here +was written from scratch for this repository, on top of the standard Python +`cryptography` package, and shares no code with any production or research system. Every +checkable claim in this README is backed by a runnable example and a test you can execute +locally in minutes, on any machine, with no GPU and two pinned dependencies. + +The payloads throughout are stand-in blobs. Nothing in this repository is a real model. + +--- + +## The gap, in one sentence each + +A doctor, an engineer, a solicitor, an accountant, a pilot who reaches for a chat +interface to help make a consequential decision is a person the system has already +vetted: they hold a licence, they were examined to earn it, they can be audited, +suspended, and struck off, and they answer for what they do. The software sitting in the +middle of that same decision carries none of it: no record of which capability produced +which part of the answer, no way to trace an output back to a source, no authorisation +that says this system was ever cleared to do this task, and no attestation that the thing +that ran is the thing that was approved. + +Every professional in that list is licensed. The system in the middle of their decisions +is not. + +The research line is about closing that gap without waiting for the model vendors to do +it. The model is a commodity. The accountability layer around it is the value, and it can +be built on hardware the customer already controls. None of the individual mechanisms is +new: gated decrypt-at-load, cryptographic erasure, signed provenance, and hardware +attestation all have prior art. The claim is about the assembly, stated as an existence +proof of composition, never as invention of a part. + +This repository demonstrates the cryptographic core of that assembly, small enough to +audit in an afternoon. + +--- + +## What this repository demonstrates + +Four claims. Each is a runnable example that prints its evidence and a pytest module that +asserts it. + +| # | Claim | Example | Tests | +|---|-------|---------|-------| +| 1 | A three-tier certificate chain (self-signed root, organisational signer, leaf capability certificate, Ed25519 throughout) verifies offline against a trust store, and a tampered, expired, or wrongly-issued certificate FAILS verification. | `examples/issue_certificate.py` | `tests/test_certificates.py` | +| 2 | An output can be bound, by digest and signature, to the certificates of its composition (weight-source and context-source axes); the binding verifies VALID while the composition is intact and INVALID the moment a referenced certificate is revoked or crypto-erased. | `examples/verify_output_provenance.py` | `tests/test_provenance.py` | +| 3 | Revocation of an encrypted capability unit works at three graded rungs: soft (advisory list entry, reported by verifiers, reversible), access-gated (wrapping key withheld, decryption denied, recoverable), and crypto-erasure (wrapping key destroyed, permanent). | `examples/revoke_three_rungs.py` | `tests/test_revocation.py` | +| 4 | Cryptographic erasure defeats even an exfiltrated copy: an envelope-encrypted unit decrypts under a valid licence, a stolen byte-for-byte copy decrypts BEFORE erasure, and after the wrapping key is destroyed both the original and the stolen copy raise on every decrypt attempt. | `examples/crypto_erasure_undecryptable.py` | `tests/test_erasure.py` | + +Claim 4 is the strongest and cleanest in the set because it rests on standard +cryptography, not on model behaviour: AES-256-GCM ciphertext without its key is not +degraded or obfuscated, it is gone. + +--- + +## Reproduce every claim + +Requires Python 3.10 or later, no GPU, and network access only for the initial +`pip install`. + +```sh +python3 -m venv .venv +. .venv/bin/activate +pip install -r requirements.txt +./run_all.sh +``` + +`run_all.sh` runs all four examples (each prints its evidence and a final +`RESULT: PASS`) and then the full test suite. Everything can also be run individually: + +```sh +python examples/issue_certificate.py # claim 1 +python examples/verify_output_provenance.py # claim 2 +python examples/revoke_three_rungs.py # claim 3 +python examples/crypto_erasure_undecryptable.py # claim 4 +python -m pytest tests/ -v # all claims, asserted +``` + +A `Makefile` offers the same steps (`make venv`, then `PYTHON=.venv/bin/python make all`). + +--- + +## The four mechanisms, and what lives where + +The research line composes four mechanisms. This repository reimplements the +cryptographic core of the first three; the fourth, and everything requiring real model +weights or a GPU, is referenced to research result records instead, never asserted as +reproduced here. + +**Status vocabulary**, used consistently below: + +- **Demonstrated here** - runnable and asserted in this repository, on stand-in payloads. +- **Referenced** - a measured result in the research programme's registry, cited by + result identifier (RR-...). Those records are internal to the programme; they are cited + so the boundary between what this repository shows and what it merely reports is + explicit, not so this README can borrow their weight. +- **Open** - a named question not yet answered anywhere. + +### 1. The certificate-chain authority + +Think of the way a browser decides whether to trust a website: a root authority signs +intermediate authorities, they sign individual certificates, and any party can check the +chain without phoning home. Revoke once at the authority and every checker that consults +the revocation list stops trusting it. This repository applies that shape to model +capabilities: root, organisational signer, per-capability leaf, Ed25519 signatures over a +canonical encoding, offline chain verification against a caller-held trust store. + +- **Demonstrated here:** issue, chain verification, expiry, tamper detection, untrusted + issuer rejection, role separation (a leaf cannot issue certificates), and the authority + not retaining leaf private keys. +- **Referenced:** the same mechanism enforced in front of a real capability load on a + real inference engine, on an 8B open-weights base (RR-2026-07-16-C3). + +### 2. Signed output provenance + +The full research goal is per-source attribution on two axes: weight-source (which base +model and which capability-adapters produced each part of an output) and context-source +(which supplied or retrieved material informed it). This repository demonstrates the +certificate side of that: binding an output, by digest and Ed25519 signature, to the +exact certificates of its composition, and verifying that binding against the live state +of the authority. Revoke or erase a referenced source and the same output flips from +VALID to INVALID. Provenance here is not a static stamp; it is a claim checked against +the authority's current state. + +- **Demonstrated here:** output binding on both axes, digest and signature verification, + chain verification of every referenced certificate, invalidation on revocation and on + erasure, detection of a substituted certificate and of a tampered output. +- **Referenced:** the measured attribution results behind the weight-source axis. On a + dense merged model, per-unit attribution is a genuine null (held-out F1 0.296 against a + pre-registered 0.70 bar; RR-2026-07-15-WPROV01). When composition is routed, the + provenance is the routing decision itself: recovered at macro-F1 0.943 with about + +0.34% logging overhead on an 8B base (RR-2026-07-15-WPROV02). A prior-art survey + found the capability-adapter provenance axis unoccupied, with existing + content-credential standards as the natural outer envelope, not competitors + (RR-2026-07-15-WPROV03). +- **Open:** context-source attribution (design direction only; the axis appears here + only as a referenced certificate, not as measured attribution), harder capability + pairs without literal markers, more than two adapters, per-span attribution, and + engine-native routing logs. + +### 3. The three-rung revocation ladder + +Revoking a licence should have graded strength depending on how badly you need the +capability and its outputs gone: + +- **Soft (advisory).** An entry on a revocation list. Verifiers that consult the list + report the certificate revoked and a compliant key authority refuses release, but the + ciphertext and its wrapping key still exist. Reversible by removing the entry. +- **Access-gated.** The wrapping key is withheld at the authority. Decryption is denied + at use-time even for an otherwise valid licence. Recoverable: the authority can + re-release. +- **Cryptographic erasure.** The wrapping key is destroyed. The ciphertext, and every + copy of it anywhere, is permanently undecryptable. Irreversible by construction: the + reference implementation refuses to re-release or reinstate what no longer exists. + +The asymmetry is the honest shape of the ladder, not a rough edge to paper over: the +first two rungs are administrative states that depend on parties honouring the +authority, and only the third is final at the level of the cryptography itself. In the +research build, the reversibility of the first two rungs by a privileged administrator +is a live, recorded defect in the programme's issue tracker; this reference +implementation makes the same asymmetry explicit in its API. + +- **Demonstrated here:** all three rungs on an envelope-encrypted stand-in unit, + including the recovery paths (reinstate, re-release) and the refusal paths after + erasure. +- **Referenced:** the same ladder driven end to end against a real sealed adapter on a + real engine, where a valid licence produced the adapter's trained behaviour and each + rung denied the load with zero decrypt calls on the deny paths (RR-2026-07-16-C3). + +### 4. Attestation + +A provenance map or licence check the engine reports about ITSELF is only as trustworthy +as the engine. The research line's fourth mechanism is attestation: process isolation so +the orchestrating host never holds plaintext weights, and hardware-rooted measurement of +the code that touches plaintext. + +- **Demonstrated here:** nothing beyond the output-binding in claim 2. This repository + makes no attestation claims. +- **Referenced:** software process isolation with a default-deny effect surface + (host decrypt count zero, roughly 1.2 KB of output and attestation returning from + 349 MB of weights; RR-2026-07-16-F), explicitly NOT a hardware enclave; and + within-run TPM-rooted attestation, trust-on-first-use, still in flight after repeated + adversarial rescoping (RR-2026-07-16-F1). +- **Open / future work:** memory-sealing. Nothing in the research line yet protects + plaintext weights from a privileged host or reads of GPU memory. That requires + confidential-computing hardware (a memory-encrypting CPU enclave and a + confidential-computing GPU) and is stated strictly in the future tense. + +--- + +## Honest scope and bounds + +A sceptical reader should be able to read just this section and know where the edges are. + +1. **Stand-in payloads.** Every sealed unit here is a labelled random blob. No model + weights, no adapters, no trained behaviour. The claim demonstrated is about the + cryptography around a unit, which is indifferent to what the unit contains. +2. **Clean-room reference code, not the research build.** This repository was written + from scratch for public reproduction. It is a third artefact, distinct from both the + research build and any production system, and shares code with neither. Where the + research records demonstrate the same mechanism, that is stated as a reference, not + as identity of code. +3. **The referenced GPU demonstrations have their own boundary.** They drove a faithful + in-memory mirror of the platform-coupled orchestration (the real cryptographic + key-store code, with record-keeping reproduced in memory), not the live application, + and the governance primitives around the kill-switch (quorum, custody, restore-proof + logging) are covered by that programme's test suite rather than by the + demonstrations. Nothing referenced here is a production system, and none of it has + shipped to one. +4. **The ladder is asymmetric on purpose.** Soft and access-gated revocation are + reversible administrative states; only crypto-erasure is cryptographically final. + This repository's API enforces exactly that asymmetry. +5. **Erasure removes the sealed unit, not "the capability from the model".** Crypto- + erasure destroys a capability that was factored into a separate encrypted unit at + build time. Whether a capability can be factored so cleanly that no recoverable + residual remains in the base model is an open research question in the programme; on + current evidence, behavioural isolation held on a synthetic proxy while + representational isolation remains unproven. The honest phrasing is "removes the + licensed unit with a cryptographic guarantee". +6. **The erasure guarantee is scoped to key destruction.** If the wrapping KEY, rather + than the ciphertext, had been exfiltrated before erasure, destroying the authority's + copy would not help. Key custody (external key-stores, hardware modules, backup + semantics that cannot resurrect a destroyed key) is named open work. +7. **Containment, not alignment.** Everything here bounds what a system can DO and caps + the blast radius of a revocation, against an auditable, cooperating licensee. None of + it makes a model benevolent, and none of it stops a determined owner of the machine. +8. **Research artefact, not a production system.** The code favours readability over + hardening: keys live in process memory, there is no persistence, no side-channel + engineering, and no security audit. Do not deploy it. Its job is to make the + mechanisms checkable. + +--- + +## Claim provenance + +Every substantive claim above, mapped to its evidence class. **Demonstrated here** means +you can run it in this repository. **Referenced** means it rests on a named result +record internal to the research programme, cited by identifier only. **Open** means not +yet answered anywhere. + +| # | Claim | Status | Traces to | +|---|-------|--------|-----------| +| 1 | Certificate chain issues and verifies offline; tampered, expired, and wrongly-issued certificates fail. | Demonstrated here | `examples/issue_certificate.py`, `tests/test_certificates.py` | +| 2 | Output binding to composition certificates verifies VALID intact, INVALID on revocation or erasure of a referenced certificate. | Demonstrated here | `examples/verify_output_provenance.py`, `tests/test_provenance.py` | +| 3 | Three-rung revocation ladder, with recovery on the first two rungs and permanent refusal on the third. | Demonstrated here | `examples/revoke_three_rungs.py`, `tests/test_revocation.py` | +| 4 | After key destruction, both the original ciphertext and a pre-existing exfiltrated copy are permanently undecryptable; both decrypted fine before. | Demonstrated here | `examples/crypto_erasure_undecryptable.py`, `tests/test_erasure.py` | +| 5 | The same mechanisms enforced around a real 8B adapter on a real inference engine, with trained behaviour appearing under a valid licence and every rung denying cleanly. | Referenced | RR-2026-07-16-C3 | +| 6 | Dense-merged per-unit weight attribution is a null (F1 0.296 vs 0.70 bar); routed composition recovers attribution at macro-F1 0.943 for +0.34% overhead. | Referenced | RR-2026-07-15-WPROV01, RR-2026-07-15-WPROV02 | +| 7 | The capability-adapter provenance axis is unoccupied in prior art; content-credential standards are the outer envelope, not competitors. | Referenced | RR-2026-07-15-WPROV03 | +| 8 | Process isolation with a default-deny effect surface; not a hardware enclave. | Referenced | RR-2026-07-16-F | +| 9 | Within-run TPM-rooted attestation, trust-on-first-use; in flight, conservatively scoped. | Referenced | RR-2026-07-16-F1 | +| 10 | Memory-sealing and GPU/VRAM protection. | Open (future work, needs confidential-computing hardware) | - | +| 11 | Clean capability factorisation with no recoverable residual in the base model. | Open | - | +| 12 | Context-source attribution as a measured result. | Open (the axis appears here only as a certificate reference) | - | + +Nothing marked *Demonstrated here* claims more than its example and tests show. Nothing +marked *Referenced* is reproduced in this repository, and this repository's passing +tests lend it no additional weight. Nothing marked *Open* is asserted at all. + +--- + +## Repository layout + +``` +README.md this file +LICENSE MIT +requirements.txt two pinned dependencies: cryptography, pytest +run_all.sh run every example, then the test suite +Makefile the same, as make targets +src/capability_licensing/ + certificates.py Ed25519 three-tier authority + offline chain verification + envelope.py AES-256-GCM envelope encryption of capability units + revocation.py the revocation ledger (soft rung) + keystore.py wrapping-key custody (access-gated + erasure rungs) + provenance.py signed output-to-composition binding + verification +examples/ one runnable script per claim, printing PASS/evidence +tests/ pytest suite asserting every claim +``` + +## Requirements + +- Python 3.10 or later (developed and verified on 3.13) +- `cryptography` (pinned) for Ed25519 and AES-256-GCM +- `pytest` (pinned) for the test suite +- No GPU, no network access after install, no services, no containers + +## Licence + +MIT. See `LICENSE`. diff --git a/examples/crypto_erasure_undecryptable.py b/examples/crypto_erasure_undecryptable.py new file mode 100644 index 0000000..21b1f7f --- /dev/null +++ b/examples/crypto_erasure_undecryptable.py @@ -0,0 +1,162 @@ +#!/usr/bin/env python3 +"""Claim 4: cryptographic erasure makes ciphertext permanently undecryptable. + +The strongest claim in the set, resting on standard cryptography alone: + + 1. a payload (a STAND-IN blob, not a real model) is envelope-encrypted + with AES-256-GCM: fresh data key, wrapped by a capability wrapping key + held only by the key authority; + 2. a valid licence decrypts it; + 3. an EXFILTRATED COPY of the sealed unit is taken, and decrypts fine + BEFORE erasure (stolen ciphertext plus a licensed key release); + 4. the authority DESTROYS the wrapping key; + 5. both the original AND the exfiltrated copy are now permanently + undecryptable: every decrypt attempt raises. The escaped copy is + useless. + +Scope, stated plainly: the guarantee is about key destruction. If the KEY +had been exfiltrated before erasure, erasure would not help; key custody is +an external key-store / hardware concern and is named open work. +""" + +import copy +import hashlib +import os +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from capability_licensing import ( + CertificateAuthority, + KeyAuthority, + KeyDestroyedError, + RevocationList, + UnsealError, + unseal, +) + +failures = [] + + +def check(label: str, condition: bool, evidence: str) -> None: + marker = "ok " if condition else "FAIL" + print(f" [{marker}] {label}") + print(f" {evidence}") + if not condition: + failures.append(label) + + +def main() -> int: + print("=== Claim 4: crypto-erasure defeats even an exfiltrated copy ===\n") + + authority = CertificateAuthority() + root = authority.create_root("Reference Root Authority") + organisation = authority.issue_organisation(root.serial, "Reference Research Organisation") + licence, _licence_key = authority.issue_capability( + organisation.serial, + "demonstration-capability", + claims={"grant": "research demonstration only"}, + ) + keys = KeyAuthority( + trusted_roots=[root], + intermediates=[organisation], + revocation_list=RevocationList(), + ) + + payload = b"STAND-IN CAPABILITY UNIT (not a real model)\n" + os.urandom(256 * 1024) + payload_digest = hashlib.sha256(payload).hexdigest() + sealed = keys.seal_unit("unit-erasure-001", payload, licence.serial) + print(f" stand-in payload: {len(payload):,} bytes plaintext, " + f"sha256 {payload_digest[:16]}...") + print(f" sealed unit: {sealed.ciphertext_size():,} bytes AES-256-GCM " + "ciphertext; the only wrapping key lives at the key authority\n") + + # 1. A valid licence decrypts the original. + plaintext = keys.request_decrypt(sealed, licence) + check( + "a valid licence decrypts the sealed unit", + hashlib.sha256(plaintext).hexdigest() == payload_digest, + "decrypted plaintext digest matches the original payload", + ) + + # 2. An adversary takes a byte-for-byte copy of the sealed unit. + exfiltrated = copy.deepcopy(sealed) + check( + "an exfiltrated byte-for-byte copy of the sealed unit is taken", + exfiltrated == sealed and exfiltrated is not sealed, + "the copy is identical ciphertext, held outside the authority's control", + ) + + # 3. Before erasure, the exfiltrated copy decrypts (same ciphertext, and + # the wrapping key still exists at the authority). + copy_plaintext = keys.request_decrypt(exfiltrated, licence) + copy_decryptable_before_erase = ( + hashlib.sha256(copy_plaintext).hexdigest() == payload_digest + ) + check( + "BEFORE erasure the exfiltrated copy decrypts", + copy_decryptable_before_erase, + f"copy_decryptable_before_erase={copy_decryptable_before_erase} " + "(this is the honest baseline: the copy was a real threat)", + ) + + # 4. The authority destroys the wrapping key. + keys.destroy(sealed.unit_id) + print("\n >>> CRYPTO-ERASE: the authority zeroises and discards the only " + "copy of the wrapping key <<<\n") + + # 5. The ORIGINAL is now undecryptable: the request raises. + original_error = None + try: + keys.request_decrypt(sealed, licence) + except KeyDestroyedError as exc: + original_error = exc + check( + "AFTER erasure the ORIGINAL raises on decryption", + original_error is not None, + f"KeyDestroyedError: {original_error}", + ) + + # 6. The EXFILTRATED COPY is equally dead: same ciphertext, same + # destroyed key. + copy_error = None + try: + keys.request_decrypt(exfiltrated, licence) + except KeyDestroyedError as exc: + copy_error = exc + check( + "AFTER erasure the EXFILTRATED COPY raises on decryption", + copy_error is not None, + f"KeyDestroyedError: {copy_error}", + ) + + # 7. Ciphertext-level evidence: even bypassing the authority, unsealing + # the stolen copy with any other key fails AES-256-GCM + # authentication. Without the destroyed 256-bit wrapping key there is + # no path to the plaintext. + bypass_error = None + try: + unseal(exfiltrated, os.urandom(32)) + except UnsealError as exc: + bypass_error = exc + check( + "bypassing the authority with a guessed key fails authentication", + bypass_error is not None, + f"UnsealError: {bypass_error}", + ) + + print() + if failures: + print(f"RESULT: FAIL - {len(failures)} check(s) did not behave as claimed") + return 1 + print( + "RESULT: PASS - before erasure both the original and the exfiltrated " + "copy decrypted; after destroying the wrapping key, both raise on every " + "decrypt attempt. The escaped ciphertext is permanently useless." + ) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/examples/issue_certificate.py b/examples/issue_certificate.py new file mode 100644 index 0000000..f67a8a5 --- /dev/null +++ b/examples/issue_certificate.py @@ -0,0 +1,108 @@ +#!/usr/bin/env python3 +"""Claim 1: certificate-chain authority. + +Issue a three-tier chain (self-signed root, organisational signer, leaf +capability certificate) with Ed25519 signatures, verify the chain to the +trusted root, and show that a tampered, expired, or wrongly-issued +certificate FAILS verification. +""" + +import dataclasses +import sys +from datetime import timedelta +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from capability_licensing import CertificateAuthority, verify_chain +from capability_licensing.certificates import parse_iso + +failures = [] + + +def check(label: str, condition: bool, evidence: str) -> None: + marker = "ok " if condition else "FAIL" + print(f" [{marker}] {label}") + print(f" {evidence}") + if not condition: + failures.append(label) + + +def main() -> int: + print("=== Claim 1: certificate-chain authority (issue and verify) ===\n") + + authority = CertificateAuthority() + root = authority.create_root("Reference Root Authority") + organisation = authority.issue_organisation(root.serial, "Reference Research Organisation") + capability, _capability_key = authority.issue_capability( + organisation.serial, + "structured-document-summarisation", + claims={ + "capability": "structured-document-summarisation", + "grant": "research demonstration only", + }, + ) + + print(f" issued root: {root.serial} (self-signed, Ed25519)") + print(f" issued organisation: {organisation.serial} (signed by root)") + print(f" issued capability: {capability.serial} (signed by organisation)") + print() + + # 1. The intact chain verifies to the trusted root. + intact = verify_chain(capability, [organisation], [root]) + check( + "intact chain verifies to the trusted root", + intact.valid, + f"result: {'VALID' if intact.valid else 'INVALID'} - {intact.reason} " + f"(path: {' -> '.join(intact.path)})", + ) + + # 2. A tampered certificate fails: edit the grant terms after signing. + tampered = dataclasses.replace( + capability, + claims={**dict(capability.claims), "grant": "unlimited production use"}, + ) + tampered_result = verify_chain(tampered, [organisation], [root]) + check( + "tampered certificate (grant terms edited after signing) fails", + not tampered_result.valid and "does not verify" in tampered_result.reason, + f"result: INVALID - {tampered_result.reason}", + ) + + # 3. An expired certificate fails: verify one day after its not_after. + after_expiry = parse_iso(capability.not_after) + timedelta(days=1) + expired_result = verify_chain(capability, [organisation], [root], at=after_expiry) + check( + "expired certificate (checked one day after expiry) fails", + not expired_result.valid and "expired" in expired_result.reason, + f"result: INVALID - {expired_result.reason}", + ) + + # 4. A chain from a different, untrusted authority fails against our + # trust store, even though it is internally well-formed. + other_authority = CertificateAuthority() + other_root = other_authority.create_root("Unaccredited Authority") + other_org = other_authority.issue_organisation(other_root.serial, "Unaccredited Organisation") + other_leaf, _ = other_authority.issue_capability( + other_org.serial, "impersonated-capability", claims={} + ) + wrong_issuer_result = verify_chain(other_leaf, [other_org, other_root], [root]) + check( + "certificate chained to an untrusted root fails", + not wrong_issuer_result.valid and "not in the trust store" in wrong_issuer_result.reason, + f"result: INVALID - {wrong_issuer_result.reason}", + ) + + print() + if failures: + print(f"RESULT: FAIL - {len(failures)} check(s) did not behave as claimed") + return 1 + print( + "RESULT: PASS - the chain verifies offline against the trust store, and " + "tampering, expiry, and an untrusted issuer each fail verification" + ) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/examples/revoke_three_rungs.py b/examples/revoke_three_rungs.py new file mode 100644 index 0000000..d2b91b3 --- /dev/null +++ b/examples/revoke_three_rungs.py @@ -0,0 +1,178 @@ +#!/usr/bin/env python3 +"""Claim 3: the three-rung revocation ladder on an encrypted capability unit. + +The payload is a STAND-IN blob, not a real model. The three rungs, weakest to +strongest: + + (a) SOFT / advisory: a revocation-list entry; verifiers report revoked; the + ciphertext and its key still exist. Reversible. + (b) ACCESS-GATED: the authority withholds the wrapping key; decryption is + denied; recoverable if the authority re-releases. + (c) CRYPTO-ERASURE: the wrapping key is destroyed; decryption is + permanently impossible. Irreversible. +""" + +import hashlib +import os +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from capability_licensing import ( + CertificateAuthority, + KeyAuthority, + KeyDestroyedError, + KeyState, + KeyWithheldError, + LicenceInvalidError, + RevocationList, + RUNG_CRYPTO_ERASURE, + verify_chain, +) + +failures = [] + + +def check(label: str, condition: bool, evidence: str) -> None: + marker = "ok " if condition else "FAIL" + print(f" [{marker}] {label}") + print(f" {evidence}") + if not condition: + failures.append(label) + + +def main() -> int: + print("=== Claim 3: three-rung revocation on an encrypted capability unit ===\n") + + authority = CertificateAuthority() + root = authority.create_root("Reference Root Authority") + organisation = authority.issue_organisation(root.serial, "Reference Research Organisation") + licence, _licence_key = authority.issue_capability( + organisation.serial, + "demonstration-capability", + claims={"grant": "research demonstration only"}, + ) + + revocations = RevocationList() + keys = KeyAuthority( + trusted_roots=[root], + intermediates=[organisation], + revocation_list=revocations, + ) + + payload = b"STAND-IN CAPABILITY UNIT (not a real model)\n" + os.urandom(64 * 1024) + payload_digest = hashlib.sha256(payload).hexdigest() + sealed = keys.seal_unit("unit-demo-001", payload, licence.serial) + print(f" stand-in payload: {len(payload):,} bytes, sha256 {payload_digest[:16]}...") + print(f" sealed unit: {sealed.ciphertext_size():,} bytes AES-256-GCM ciphertext, " + f"licensed to {licence.serial}\n") + + # Baseline: a valid licence decrypts. + plaintext = keys.request_decrypt(sealed, licence) + check( + "baseline: a valid licence decrypts the unit", + hashlib.sha256(plaintext).hexdigest() == payload_digest, + "decrypted plaintext digest matches the original payload", + ) + + # ---- Rung (a): SOFT / advisory ------------------------------------- + print("\n --- rung (a): SOFT (advisory revocation-list entry) ---") + revocations.revoke(licence.serial, "licence terms breached") + chain_check = verify_chain( + licence, [organisation], [root], revocation_list=revocations + ) + check( + "verification reports the certificate revoked", + not chain_check.valid and "revoked" in chain_check.reason, + f"verify_chain: INVALID - {chain_check.reason}", + ) + denied_soft = None + try: + keys.request_decrypt(sealed, licence) + except LicenceInvalidError as exc: + denied_soft = exc + check( + "a compliant key authority honours the list and refuses decryption", + denied_soft is not None, + f"refused with LicenceInvalidError: {denied_soft}", + ) + check( + "advisory honesty: the ciphertext and wrapping key still exist", + keys.key_state(sealed.unit_id) is KeyState.RELEASED, + "key state is still 'released'; enforcement rests on verifiers honouring " + "the list, not on the cryptography", + ) + revocations.reinstate(licence.serial) + recovered_soft = keys.request_decrypt(sealed, licence) + check( + "soft revocation is reversible: reinstating restores decryption", + hashlib.sha256(recovered_soft).hexdigest() == payload_digest, + "after reinstatement the same licence decrypts again", + ) + + # ---- Rung (b): ACCESS-GATED ---------------------------------------- + print("\n --- rung (b): ACCESS-GATED (wrapping key withheld) ---") + keys.withhold(sealed.unit_id) + denied_gated = None + try: + keys.request_decrypt(sealed, licence) + except KeyWithheldError as exc: + denied_gated = exc + check( + "with the key withheld, decryption is denied even for a valid licence", + denied_gated is not None, + f"refused with KeyWithheldError: {denied_gated}", + ) + keys.re_release(sealed.unit_id) + recovered_gated = keys.request_decrypt(sealed, licence) + check( + "access-gating is recoverable: re-release restores decryption", + hashlib.sha256(recovered_gated).hexdigest() == payload_digest, + "after re-release the same licence decrypts again", + ) + + # ---- Rung (c): CRYPTO-ERASURE -------------------------------------- + print("\n --- rung (c): CRYPTO-ERASURE (wrapping key destroyed) ---") + keys.destroy(sealed.unit_id) + revocations.revoke( + licence.serial, "capability unit crypto-erased", rung=RUNG_CRYPTO_ERASURE + ) + denied_erased = None + try: + keys.request_decrypt(sealed, licence) + except KeyDestroyedError as exc: + denied_erased = exc + check( + "after erasure, every decrypt request fails permanently", + denied_erased is not None, + f"refused with KeyDestroyedError: {denied_erased}", + ) + unrecoverable = None + try: + keys.re_release(sealed.unit_id) + except KeyDestroyedError as exc: + unrecoverable = exc + check( + "erasure is irreversible: the authority cannot re-release a destroyed key", + unrecoverable is not None, + f"re-release refused with KeyDestroyedError: {unrecoverable}", + ) + + print() + print(" ladder asymmetry, stated honestly: rungs (a) and (b) are reversible") + print(" administrative states; only rung (c) is final at the level of the") + print(" cryptography itself.") + print() + if failures: + print(f"RESULT: FAIL - {len(failures)} check(s) did not behave as claimed") + return 1 + print( + "RESULT: PASS - soft flags and is honoured, access-gating denies and " + "recovers, crypto-erasure denies permanently" + ) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/examples/verify_output_provenance.py b/examples/verify_output_provenance.py new file mode 100644 index 0000000..aa0960f --- /dev/null +++ b/examples/verify_output_provenance.py @@ -0,0 +1,141 @@ +#!/usr/bin/env python3 +"""Claim 2: output provenance binding. + +Bind an output to the certificates of its composition (weight-source and +context-source axes), verify the binding VALID while the composition is +intact, and show it flips to INVALID when a referenced certificate is +revoked or crypto-erased. Also show a tampered output fails the digest check. +""" + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from capability_licensing import ( + AXIS_CONTEXT_SOURCE, + AXIS_WEIGHT_SOURCE, + CertificateAuthority, + RevocationList, + RUNG_CRYPTO_ERASURE, + bind_output, + verify_output_provenance, +) + +failures = [] + + +def check(label: str, condition: bool, evidence: str) -> None: + marker = "ok " if condition else "FAIL" + print(f" [{marker}] {label}") + print(f" {evidence}") + if not condition: + failures.append(label) + + +def main() -> int: + print("=== Claim 2: output provenance bound to composition certificates ===\n") + + authority = CertificateAuthority() + root = authority.create_root("Reference Root Authority") + organisation = authority.issue_organisation(root.serial, "Reference Research Organisation") + + weight_source, weight_key = authority.issue_capability( + organisation.serial, + "summarisation-adapter (weight-source)", + claims={"axis": AXIS_WEIGHT_SOURCE, "grant": "research demonstration only"}, + ) + context_source, _context_key = authority.issue_capability( + organisation.serial, + "licensed-reference-corpus (context-source)", + claims={"axis": AXIS_CONTEXT_SOURCE, "grant": "research demonstration only"}, + ) + + print(f" weight-source certificate: {weight_source.serial}") + print(f" context-source certificate: {context_source.serial}") + + output = ( + b"Stand-in generated output: a three-paragraph summary of the supplied " + b"reference material, produced by the licensed summarisation adapter." + ) + record = bind_output( + output, + sources=[ + (AXIS_WEIGHT_SOURCE, weight_source), + (AXIS_CONTEXT_SOURCE, context_source), + ], + signer_certificate=weight_source, + signer_private_key=weight_key, + ) + print(f" output bound: sha256 {record.output_sha256[:16]}..., " + f"signed by {record.signer_serial}\n") + + registry = authority.registry() + revocations = RevocationList() + + # 1. Intact composition verifies VALID. + intact = verify_output_provenance( + output, record, registry, [organisation], [root], revocations + ) + print(" checks performed on the intact composition:") + for item in intact.checks: + print(f" - {item}") + check( + "intact composition verifies VALID", + intact.valid, + f"result: {'VALID' if intact.valid else 'INVALID'} - {intact.reason}", + ) + + # 2. Tampered output fails the digest binding. + tampered_output = output + b" [edited after the fact]" + tampered = verify_output_provenance( + tampered_output, record, registry, [organisation], [root], revocations + ) + check( + "tampered output fails the digest binding", + not tampered.valid and "does not match the bound digest" in tampered.reason, + f"result: INVALID - {tampered.reason}", + ) + + # 3. Revoke the context-source certificate: the SAME record now verifies + # INVALID. Provenance is a live claim against authority state. + revocations.revoke(context_source.serial, "context licence terminated") + revoked = verify_output_provenance( + output, record, registry, [organisation], [root], revocations + ) + check( + "after revoking the context-source certificate, verification is INVALID", + not revoked.valid and "revoked" in revoked.reason, + f"result: INVALID - {revoked.reason}", + ) + revocations.reinstate(context_source.serial) + + # 4. Crypto-erase the weight-source capability: INVALID again, and the + # ledger records the strongest rung. + revocations.revoke( + weight_source.serial, + "capability unit crypto-erased by the authority", + rung=RUNG_CRYPTO_ERASURE, + ) + erased = verify_output_provenance( + output, record, registry, [organisation], [root], revocations + ) + check( + "after crypto-erasure of the weight-source capability, verification is INVALID", + not erased.valid and "crypto-erasure" in erased.reason, + f"result: INVALID - {erased.reason}", + ) + + print() + if failures: + print(f"RESULT: FAIL - {len(failures)} check(s) did not behave as claimed") + return 1 + print( + "RESULT: PASS - the binding verifies for an intact composition and is " + "INVALID for a tampered output or a revoked/erased source certificate" + ) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..5ee6477 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,2 @@ +[pytest] +testpaths = tests diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..1374d02 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +cryptography==49.0.0 +pytest==9.1.1 diff --git a/run_all.sh b/run_all.sh new file mode 100755 index 0000000..04c453f --- /dev/null +++ b/run_all.sh @@ -0,0 +1,29 @@ +#!/usr/bin/env bash +# Run every example, then the test suite. Exits non-zero if anything fails. +set -euo pipefail +cd "$(dirname "$0")" + +PYTHON="${PYTHON:-python3}" + +echo "==================================================================" +echo " Capability licensing reference implementation: run all claims" +echo "==================================================================" + +for example in \ + examples/issue_certificate.py \ + examples/verify_output_provenance.py \ + examples/revoke_three_rungs.py \ + examples/crypto_erasure_undecryptable.py +do + echo + "$PYTHON" "$example" +done + +echo +echo "==================================================================" +echo " Test suite" +echo "==================================================================" +"$PYTHON" -m pytest tests/ -q + +echo +echo "ALL EXAMPLES AND ALL TESTS PASSED" diff --git a/src/capability_licensing/__init__.py b/src/capability_licensing/__init__.py new file mode 100644 index 0000000..5986f40 --- /dev/null +++ b/src/capability_licensing/__init__.py @@ -0,0 +1,99 @@ +"""Clean-room reference implementation of capability licensing primitives. + +Four mechanisms, built from scratch on the ``cryptography`` package: + +* :mod:`.certificates` - an Ed25519 three-tier certificate authority + (root, organisational, capability) with offline chain verification; +* :mod:`.envelope` - AES-256-GCM envelope encryption of capability units + (a fresh data key per unit, wrapped by a per-unit wrapping key); +* :mod:`.revocation` and :mod:`.keystore` - the three-rung revocation + ladder: soft (advisory list entry), access-gated (key withheld, + recoverable), crypto-erasure (key destroyed, irreversible); +* :mod:`.provenance` - signed binding of an output to the certificates of + its composition, invalidated live by revocation or erasure. + +This package shares no code with any production system. Payloads throughout +are stand-in blobs, never real model weights. +""" + +from .certificates import ( + ROLE_CAPABILITY, + ROLE_ORGANISATION, + ROLE_ROOT, + Certificate, + CertificateAuthority, + VerificationResult, + verify_chain, +) +from .envelope import ( + KEY_BYTES, + SealedUnit, + UnsealError, + generate_wrapping_key, + seal, + unseal, +) +from .keystore import ( + KeyAuthority, + KeyAuthorityError, + KeyDestroyedError, + KeyState, + KeyWithheldError, + LicenceInvalidError, + UnknownUnitError, +) +from .provenance import ( + AXIS_CONTEXT_SOURCE, + AXIS_WEIGHT_SOURCE, + ProvenanceRecord, + ProvenanceResult, + SourceReference, + bind_output, + verify_output_provenance, +) +from .revocation import ( + RUNG_ACCESS_GATED, + RUNG_CRYPTO_ERASURE, + RUNG_SOFT, + IrreversibleRevocationError, + RevocationEntry, + RevocationList, +) + +__all__ = [ + "AXIS_CONTEXT_SOURCE", + "AXIS_WEIGHT_SOURCE", + "Certificate", + "CertificateAuthority", + "IrreversibleRevocationError", + "KEY_BYTES", + "KeyAuthority", + "KeyAuthorityError", + "KeyDestroyedError", + "KeyState", + "KeyWithheldError", + "LicenceInvalidError", + "ProvenanceRecord", + "ProvenanceResult", + "ROLE_CAPABILITY", + "ROLE_ORGANISATION", + "ROLE_ROOT", + "RUNG_ACCESS_GATED", + "RUNG_CRYPTO_ERASURE", + "RUNG_SOFT", + "RevocationEntry", + "RevocationList", + "SealedUnit", + "SourceReference", + "UnknownUnitError", + "UnsealError", + "VerificationResult", + "bind_output", + "generate_wrapping_key", + "seal", + "unseal", + "verify_chain", + "verify_output_provenance", +] + +__version__ = "0.1.0" diff --git a/src/capability_licensing/certificates.py b/src/capability_licensing/certificates.py new file mode 100644 index 0000000..722eb14 --- /dev/null +++ b/src/capability_licensing/certificates.py @@ -0,0 +1,378 @@ +"""Ed25519 certificate-chain authority. + +Clean-room reference implementation of a three-tier certificate authority for +capability licensing: a self-signed root, an organisational signing +certificate chained to the root, and leaf capability certificates. Built from +scratch on the ``cryptography`` package's Ed25519 primitives. No code is +shared with any production system. + +Certificates here are plain JSON-serialisable records signed over a canonical +byte encoding. That is deliberately simpler than X.509: the aim is a readable, +auditable demonstration of the chain-of-trust mechanism, not wire-format +compatibility with existing PKI tooling. +""" + +from __future__ import annotations + +import hashlib +import json +import os +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from typing import Dict, Iterable, Mapping, Optional, Tuple + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, + Ed25519PublicKey, +) +from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat + +ROLE_ROOT = "root" +ROLE_ORGANISATION = "organisation" +ROLE_CAPABILITY = "capability" + +# Which roles each role is permitted to issue. A root self-signs and signs +# organisational certificates; an organisational certificate signs capability +# leaves; a capability leaf signs nothing (it signs outputs, not certificates). +_MAY_ISSUE = { + ROLE_ROOT: (ROLE_ROOT, ROLE_ORGANISATION), + ROLE_ORGANISATION: (ROLE_CAPABILITY,), + ROLE_CAPABILITY: (), +} + +_MAX_CHAIN_DEPTH = 8 + + +def _utc_now() -> datetime: + return datetime.now(timezone.utc) + + +def _iso(moment: datetime) -> str: + return moment.astimezone(timezone.utc).isoformat(timespec="seconds") + + +def parse_iso(value: str) -> datetime: + """Parse an ISO 8601 timestamp as stored on a certificate.""" + return datetime.fromisoformat(value) + + +def _canonical_json(data: Mapping) -> bytes: + """Deterministic byte encoding: sorted keys, no whitespace, UTF-8.""" + return json.dumps(data, sort_keys=True, separators=(",", ":")).encode("utf-8") + + +@dataclass(frozen=True) +class Certificate: + """A signed capability-licensing certificate. + + ``signature_hex`` is an Ed25519 signature by the ISSUER's private key over + the canonical encoding of every other field (the to-be-signed portion). + For a self-signed root, issuer and subject are the same key. + """ + + serial: str + subject: str + issuer_serial: str + role: str + public_key_hex: str + not_before: str + not_after: str + claims: Mapping[str, str] = field(default_factory=dict) + signature_hex: str = "" + + def tbs_dict(self) -> Dict: + """The to-be-signed fields (everything except the signature).""" + return { + "serial": self.serial, + "subject": self.subject, + "issuer_serial": self.issuer_serial, + "role": self.role, + "public_key_hex": self.public_key_hex, + "not_before": self.not_before, + "not_after": self.not_after, + "claims": dict(self.claims), + } + + def tbs_bytes(self) -> bytes: + return _canonical_json(self.tbs_dict()) + + def to_dict(self) -> Dict: + data = self.tbs_dict() + data["signature_hex"] = self.signature_hex + return data + + def sha256(self) -> str: + """Digest over the full certificate, signature included. + + Used to bind other records (provenance, trust stores) to this exact + certificate content, so substituting a different certificate with the + same serial is detectable. + """ + return hashlib.sha256(_canonical_json(self.to_dict())).hexdigest() + + def public_key(self) -> Ed25519PublicKey: + return Ed25519PublicKey.from_public_bytes(bytes.fromhex(self.public_key_hex)) + + def is_self_signed(self) -> bool: + return self.issuer_serial == self.serial + + +@dataclass(frozen=True) +class VerificationResult: + """Outcome of a chain verification, with a human-readable reason.""" + + valid: bool + reason: str + path: Tuple[str, ...] = () + + def __bool__(self) -> bool: # pragma: no cover - convenience only + return self.valid + + +def _new_serial() -> str: + return "CERT-" + os.urandom(8).hex() + + +def _sign(private_key: Ed25519PrivateKey, data: bytes) -> str: + return private_key.sign(data).hex() + + +def _public_hex(private_key: Ed25519PrivateKey) -> str: + return private_key.public_key().public_bytes( + Encoding.Raw, PublicFormat.Raw + ).hex() + + +class CertificateAuthority: + """Issues and signs certificates for the three-tier chain. + + The authority retains only its own signing keys (root and organisational + tiers). A capability's private key is generated at issue time, handed to + the caller, and never stored: the authority cannot later sign outputs on + a licensee's behalf, and a compromise of the authority does not leak + licensee keys. + """ + + def __init__(self) -> None: + self._signing_keys: Dict[str, Ed25519PrivateKey] = {} + self._issued: Dict[str, Certificate] = {} + + # -- issuance --------------------------------------------------------- + + def create_root(self, subject: str, lifetime_days: int = 3650) -> Certificate: + """Create a self-signed root certificate and retain its signing key.""" + private_key = Ed25519PrivateKey.generate() + serial = _new_serial() + now = _utc_now() + unsigned = Certificate( + serial=serial, + subject=subject, + issuer_serial=serial, + role=ROLE_ROOT, + public_key_hex=_public_hex(private_key), + not_before=_iso(now), + not_after=_iso(now + timedelta(days=lifetime_days)), + claims={}, + ) + certificate = Certificate( + **{**unsigned.to_dict(), "signature_hex": _sign(private_key, unsigned.tbs_bytes())} + ) + self._signing_keys[serial] = private_key + self._issued[serial] = certificate + return certificate + + def issue_organisation( + self, root_serial: str, subject: str, lifetime_days: int = 1825 + ) -> Certificate: + """Issue an organisational signing certificate chained to the root.""" + certificate, private_key = self._issue( + issuer_serial=root_serial, + subject=subject, + role=ROLE_ORGANISATION, + claims={}, + lifetime_days=lifetime_days, + ) + # Organisational tier keys stay with the authority so it can issue + # capability leaves. + self._signing_keys[certificate.serial] = private_key + return certificate + + def issue_capability( + self, + organisation_serial: str, + subject: str, + claims: Mapping[str, str], + lifetime_days: int = 365, + ) -> Tuple[Certificate, Ed25519PrivateKey]: + """Issue a leaf capability certificate. + + Returns the certificate AND the leaf private key: the key belongs to + the licensee (for signing provenance records) and is not retained. + """ + certificate, private_key = self._issue( + issuer_serial=organisation_serial, + subject=subject, + role=ROLE_CAPABILITY, + claims=claims, + lifetime_days=lifetime_days, + ) + return certificate, private_key + + def _issue( + self, + issuer_serial: str, + subject: str, + role: str, + claims: Mapping[str, str], + lifetime_days: int, + ) -> Tuple[Certificate, Ed25519PrivateKey]: + issuer_certificate = self._issued.get(issuer_serial) + if issuer_certificate is None: + raise KeyError(f"unknown issuer serial: {issuer_serial}") + if role not in _MAY_ISSUE[issuer_certificate.role]: + raise ValueError( + f"role violation: a {issuer_certificate.role} certificate may not issue " + f"a {role} certificate" + ) + issuer_key = self._signing_keys.get(issuer_serial) + if issuer_key is None: + raise KeyError(f"no signing key held for issuer: {issuer_serial}") + subject_key = Ed25519PrivateKey.generate() + now = _utc_now() + unsigned = Certificate( + serial=_new_serial(), + subject=subject, + issuer_serial=issuer_serial, + role=role, + public_key_hex=_public_hex(subject_key), + not_before=_iso(now), + not_after=_iso(now + timedelta(days=lifetime_days)), + claims=dict(claims), + ) + certificate = Certificate( + **{**unsigned.to_dict(), "signature_hex": _sign(issuer_key, unsigned.tbs_bytes())} + ) + self._issued[certificate.serial] = certificate + return certificate, subject_key + + # -- lookup ----------------------------------------------------------- + + def certificate(self, serial: str) -> Certificate: + return self._issued[serial] + + def registry(self) -> Dict[str, Certificate]: + """A copy of every certificate this authority has issued.""" + return dict(self._issued) + + +def verify_chain( + leaf: Certificate, + intermediates: Iterable[Certificate], + trusted_roots: Iterable[Certificate], + at: Optional[datetime] = None, + revocation_list=None, +) -> VerificationResult: + """Verify a certificate chain from ``leaf`` up to a trusted root. + + Checks, per certificate on the path: + + * validity window covers ``at`` (defaults to now, UTC); + * not revoked, when a revocation list is supplied; + * the issuer exists, is permitted by role to issue this certificate, and + its Ed25519 signature over the to-be-signed bytes verifies; + * the chain terminates at a self-signed root that is byte-identical to a + certificate in the caller's trust store. + + Any verifier can run this with nothing but the certificates and the trust + store: no call home to the authority is needed. + """ + moment = at or _utc_now() + pool: Dict[str, Certificate] = {c.serial: c for c in intermediates} + roots: Dict[str, Certificate] = {c.serial: c for c in trusted_roots} + pool.update(roots) + + current = leaf + path = [] + seen = set() + for _ in range(_MAX_CHAIN_DEPTH): + if current.serial in seen: + return VerificationResult(False, "certificate chain contains a cycle", tuple(path)) + seen.add(current.serial) + path.append(current.serial) + + not_before = parse_iso(current.not_before) + not_after = parse_iso(current.not_after) + if moment < not_before: + return VerificationResult( + False, f"certificate not yet valid: {current.serial}", tuple(path) + ) + if moment > not_after: + return VerificationResult( + False, + f"certificate expired: {current.serial} (not_after {current.not_after})", + tuple(path), + ) + + if revocation_list is not None and revocation_list.is_revoked(current.serial): + entry = revocation_list.entry_for(current.serial) + detail = f" (rung: {entry.rung}; reason: {entry.reason})" if entry else "" + return VerificationResult( + False, f"certificate revoked: {current.serial}{detail}", tuple(path) + ) + + if current.is_self_signed(): + if current.role != ROLE_ROOT: + return VerificationResult( + False, + f"self-signed certificate {current.serial} has non-root role " + f"'{current.role}'", + tuple(path), + ) + trusted = roots.get(current.serial) + if trusted is None or trusted.sha256() != current.sha256(): + return VerificationResult( + False, + f"self-signed certificate {current.serial} is not in the trust " + "store (chain terminates at an untrusted authority)", + tuple(path), + ) + try: + current.public_key().verify( + bytes.fromhex(current.signature_hex), current.tbs_bytes() + ) + except InvalidSignature: + return VerificationResult( + False, f"root self-signature invalid: {current.serial}", tuple(path) + ) + return VerificationResult(True, "chain verifies to trusted root", tuple(path)) + + issuer = pool.get(current.issuer_serial) + if issuer is None: + return VerificationResult( + False, + f"issuer not found for {current.serial} " + f"(issuer serial {current.issuer_serial})", + tuple(path), + ) + if current.role not in _MAY_ISSUE.get(issuer.role, ()): + return VerificationResult( + False, + f"role violation: a {issuer.role} certificate may not issue a " + f"{current.role} certificate", + tuple(path), + ) + try: + issuer.public_key().verify( + bytes.fromhex(current.signature_hex), current.tbs_bytes() + ) + except (InvalidSignature, ValueError): + return VerificationResult( + False, + f"signature on {current.serial} does not verify against issuer " + f"{issuer.serial} (tampered certificate or wrong issuer)", + tuple(path), + ) + current = issuer + + return VerificationResult(False, "certificate chain too deep", tuple(path)) diff --git a/src/capability_licensing/envelope.py b/src/capability_licensing/envelope.py new file mode 100644 index 0000000..6c3a8f9 --- /dev/null +++ b/src/capability_licensing/envelope.py @@ -0,0 +1,122 @@ +"""AES-256-GCM envelope encryption for capability units. + +A capability unit (in this repository, always a STAND-IN payload blob, never a +real model) is sealed with a fresh random 256-bit data key, and that data key +is itself encrypted (wrapped) under a per-unit wrapping key held by the key +authority. Destroying the wrapping key therefore renders the sealed unit, and +every copy of it anywhere, permanently undecryptable: this is cryptographic +erasure. + +Both layers use AES-256-GCM, an authenticated mode: decryption with the wrong +key does not yield garbage, it FAILS closed with an authentication error. The +unit identifier is bound in as associated data so a ciphertext cannot be +replayed under a different unit's identity. +""" + +from __future__ import annotations + +import hashlib +import os +from dataclasses import dataclass + +from cryptography.exceptions import InvalidTag +from cryptography.hazmat.primitives.ciphers.aead import AESGCM + +KEY_BYTES = 32 # AES-256 +NONCE_BYTES = 12 # 96-bit GCM nonce + + +class UnsealError(Exception): + """Raised when AES-256-GCM authentication fails during unsealing. + + This is the failure mode for a wrong key, a missing key substituted with a + guess, or tampered ciphertext. There is no partial decryption. + """ + + +@dataclass(frozen=True) +class SealedUnit: + """An envelope-encrypted capability unit. + + Everything in this record is safe to copy, transmit, or exfiltrate: none + of it can be decrypted without the wrapping key, which lives only at the + key authority. + """ + + unit_id: str + capability_serial: str # the capability certificate this unit is licensed under + payload_nonce_hex: str + payload_ciphertext_hex: str # AES-256-GCM over the payload, fresh data key + wrap_nonce_hex: str + wrapped_data_key_hex: str # the data key, encrypted under the wrapping key + payload_sha256: str # digest of the plaintext, for integrity display only + + def ciphertext_size(self) -> int: + return len(self.payload_ciphertext_hex) // 2 + + +def generate_wrapping_key() -> bytes: + """A fresh random 256-bit wrapping key.""" + return AESGCM.generate_key(bit_length=256) + + +def seal( + unit_id: str, + payload: bytes, + wrapping_key: bytes, + capability_serial: str, +) -> SealedUnit: + """Envelope-encrypt ``payload`` into a sealed unit. + + A fresh data key encrypts the payload; the wrapping key encrypts the data + key. The caller should discard the wrapping key immediately (the key + authority holds the only copy in the intended deployment shape). + """ + if len(wrapping_key) != KEY_BYTES: + raise ValueError(f"wrapping key must be {KEY_BYTES} bytes") + associated_data = unit_id.encode("utf-8") + + data_key = AESGCM.generate_key(bit_length=256) + payload_nonce = os.urandom(NONCE_BYTES) + payload_ciphertext = AESGCM(data_key).encrypt(payload_nonce, payload, associated_data) + + wrap_nonce = os.urandom(NONCE_BYTES) + wrapped_data_key = AESGCM(bytes(wrapping_key)).encrypt( + wrap_nonce, data_key, associated_data + ) + + return SealedUnit( + unit_id=unit_id, + capability_serial=capability_serial, + payload_nonce_hex=payload_nonce.hex(), + payload_ciphertext_hex=payload_ciphertext.hex(), + wrap_nonce_hex=wrap_nonce.hex(), + wrapped_data_key_hex=wrapped_data_key.hex(), + payload_sha256=hashlib.sha256(payload).hexdigest(), + ) + + +def unseal(unit: SealedUnit, wrapping_key: bytes) -> bytes: + """Decrypt a sealed unit: unwrap the data key, then decrypt the payload. + + Raises :class:`UnsealError` if the wrapping key is wrong or the ciphertext + has been tampered with. Without the exact wrapping key there is no path to + the plaintext: the data key is recoverable only through the wrap layer. + """ + associated_data = unit.unit_id.encode("utf-8") + try: + data_key = AESGCM(bytes(wrapping_key)).decrypt( + bytes.fromhex(unit.wrap_nonce_hex), + bytes.fromhex(unit.wrapped_data_key_hex), + associated_data, + ) + return AESGCM(data_key).decrypt( + bytes.fromhex(unit.payload_nonce_hex), + bytes.fromhex(unit.payload_ciphertext_hex), + associated_data, + ) + except InvalidTag as exc: + raise UnsealError( + f"AES-256-GCM authentication failed for unit '{unit.unit_id}': " + "wrong or missing wrapping key, or tampered ciphertext" + ) from exc diff --git a/src/capability_licensing/keystore.py b/src/capability_licensing/keystore.py new file mode 100644 index 0000000..6ed2460 --- /dev/null +++ b/src/capability_licensing/keystore.py @@ -0,0 +1,200 @@ +"""The key authority: wrapping-key custody and the two stronger revocation rungs. + +The authority is the only holder of each sealed unit's wrapping key. Sealing +happens through the authority so the key never leaves it. Decryption is a +request TO the authority: it checks the presented licence (the capability +certificate chain, against its trust store and revocation list) and the key's +state before unsealing. + +Key states implement the two stronger rungs of the revocation ladder: + +* RELEASED: a valid licence decrypts. +* WITHHELD (access-gated rung): decryption denied; recoverable, the authority + can re-release. +* DESTROYED (crypto-erasure rung): the key material has been zeroised and + discarded; every decrypt request fails permanently, for the original sealed + unit and for any exfiltrated copy alike. + +Honesty note on scope: destroying a Python ``bytearray`` stands in for key +destruction in an external key-store or hardware security module. The +property demonstrated is the logical one that matters: once the only copy of +the wrapping key is gone, no future request can ever succeed, because the +information needed to decrypt no longer exists anywhere. Key custody +(external key-store, backup semantics that cannot resurrect a destroyed key) +is named open work in the research line, not solved by this reference code. +""" + +from __future__ import annotations + +import os +from datetime import datetime, timezone +from enum import Enum +from typing import Dict, Iterable, List, Optional, Tuple + +from .certificates import Certificate, verify_chain +from .envelope import KEY_BYTES, SealedUnit, seal, unseal +from .revocation import RevocationList + + +class KeyState(str, Enum): + RELEASED = "released" + WITHHELD = "withheld" + DESTROYED = "destroyed" + + +class KeyAuthorityError(Exception): + """Base class for key-authority refusals.""" + + +class UnknownUnitError(KeyAuthorityError): + pass + + +class LicenceInvalidError(KeyAuthorityError): + """The presented certificate chain failed verification (or is revoked).""" + + +class KeyWithheldError(KeyAuthorityError): + """Access-gated rung: the key exists but the authority withholds it.""" + + +class KeyDestroyedError(KeyAuthorityError): + """Crypto-erasure rung: the key no longer exists; decryption is impossible.""" + + +class KeyAuthority: + """Holds wrapping keys and enforces licence checks at decrypt time.""" + + def __init__( + self, + trusted_roots: Iterable[Certificate], + intermediates: Iterable[Certificate] = (), + revocation_list: Optional[RevocationList] = None, + ) -> None: + self._trusted_roots: Tuple[Certificate, ...] = tuple(trusted_roots) + self._intermediates: Tuple[Certificate, ...] = tuple(intermediates) + self._revocation_list = revocation_list + self._keys: Dict[str, bytearray] = {} + self._states: Dict[str, KeyState] = {} + self._events: List[Tuple[str, str, str]] = [] + + # -- internal --------------------------------------------------------- + + def _log(self, unit_id: str, event: str) -> None: + stamp = datetime.now(timezone.utc).isoformat(timespec="seconds") + self._events.append((stamp, unit_id, event)) + + def _require_known(self, unit_id: str) -> None: + if unit_id not in self._states: + raise UnknownUnitError(f"no key registered for unit '{unit_id}'") + + # -- sealing ---------------------------------------------------------- + + def seal_unit(self, unit_id: str, payload: bytes, capability_serial: str) -> SealedUnit: + """Seal a payload under a fresh wrapping key held only by this authority.""" + if unit_id in self._states: + raise KeyAuthorityError(f"unit '{unit_id}' already registered") + wrapping_key = bytearray(os.urandom(KEY_BYTES)) + sealed = seal(unit_id, payload, bytes(wrapping_key), capability_serial) + self._keys[unit_id] = wrapping_key + self._states[unit_id] = KeyState.RELEASED + self._log(unit_id, "sealed (wrapping key generated and retained)") + return sealed + + # -- state ------------------------------------------------------------ + + def key_state(self, unit_id: str) -> KeyState: + self._require_known(unit_id) + return self._states[unit_id] + + def audit_log(self) -> Tuple[Tuple[str, str, str], ...]: + return tuple(self._events) + + # -- the revocation ladder, rungs two and three ------------------------ + + def withhold(self, unit_id: str) -> None: + """Access-gated rung: deny key release. Recoverable via re_release().""" + self._require_known(unit_id) + if self._states[unit_id] is KeyState.DESTROYED: + raise KeyDestroyedError( + f"unit '{unit_id}' was crypto-erased; there is no key to withhold" + ) + self._states[unit_id] = KeyState.WITHHELD + self._log(unit_id, "key withheld (access-gated revocation)") + + def re_release(self, unit_id: str) -> None: + """Recover from the access-gated rung. Impossible after erasure.""" + self._require_known(unit_id) + if self._states[unit_id] is KeyState.DESTROYED: + raise KeyDestroyedError( + f"unit '{unit_id}' was crypto-erased; the wrapping key no longer " + "exists and cannot be re-released" + ) + self._states[unit_id] = KeyState.RELEASED + self._log(unit_id, "key re-released") + + def destroy(self, unit_id: str) -> None: + """Crypto-erasure rung: zeroise and discard the only copy of the key. + + After this call the sealed unit, and every copy of its ciphertext + anywhere, is permanently undecryptable. There is no undo. + """ + self._require_known(unit_id) + if self._states[unit_id] is KeyState.DESTROYED: + raise KeyDestroyedError(f"unit '{unit_id}' is already crypto-erased") + key = self._keys.pop(unit_id) + for index in range(len(key)): + key[index] = 0 + self._states[unit_id] = KeyState.DESTROYED + self._log(unit_id, "wrapping key DESTROYED (crypto-erasure; irreversible)") + + # -- licensed decryption ---------------------------------------------- + + def request_decrypt( + self, + unit: SealedUnit, + presented_certificate: Certificate, + presented_chain: Iterable[Certificate] = (), + at: Optional[datetime] = None, + ) -> bytes: + """Decrypt a sealed unit for a holder of a valid capability licence. + + Refusals, in order of finality: + + * :class:`KeyDestroyedError` if the key was destroyed (permanent); + * :class:`KeyWithheldError` if the key is access-gated (recoverable); + * :class:`LicenceInvalidError` if the presented certificate does not + match the unit's licence, fails chain verification, or is revoked + on the authority's revocation list (the soft rung, honoured here). + """ + self._require_known(unit.unit_id) + state = self._states[unit.unit_id] + if state is KeyState.DESTROYED: + raise KeyDestroyedError( + f"the wrapping key for unit '{unit.unit_id}' was destroyed " + "(crypto-erasure); this ciphertext is permanently undecryptable" + ) + if state is KeyState.WITHHELD: + raise KeyWithheldError( + f"the wrapping key for unit '{unit.unit_id}' is withheld " + "(access-gated revocation); decryption is denied but recoverable " + "if the authority re-releases the key" + ) + + if presented_certificate.serial != unit.capability_serial: + raise LicenceInvalidError( + f"presented certificate {presented_certificate.serial} does not " + f"match the unit's licence {unit.capability_serial}" + ) + chain = tuple(presented_chain) + self._intermediates + result = verify_chain( + presented_certificate, + chain, + self._trusted_roots, + at=at, + revocation_list=self._revocation_list, + ) + if not result.valid: + raise LicenceInvalidError(f"licence verification failed: {result.reason}") + + return unseal(unit, bytes(self._keys[unit.unit_id])) diff --git a/src/capability_licensing/provenance.py b/src/capability_licensing/provenance.py new file mode 100644 index 0000000..ceb9935 --- /dev/null +++ b/src/capability_licensing/provenance.py @@ -0,0 +1,204 @@ +"""Output provenance: bind an output to the certificates of its composition. + +A produced output is bound, by digest and by Ed25519 signature, to the +certificates of the sources that composed it, on two axes: + +* weight-source: the capability (for example a fine-tuned adapter) whose + weights produced the output; +* context-source: the licensed context or retrieved material that informed it. + +The binding is signed by the producing capability's private key, so the +record cannot be forged or edited, and each referenced certificate is pinned +by content digest, so a certificate cannot be silently swapped for another +with the same serial. + +Verification checks the output digest, the signature, and the full chain of +every referenced certificate against the trust store AND the revocation +ledger. The consequence, demonstrated in the examples and tests: an output +verifies VALID while its composition is intact, and flips to INVALID the +moment any referenced certificate is revoked or crypto-erased. Provenance is +not a static stamp; it is a live claim against the current state of the +authority. + +Scope note (honest bound): this is OUTPUT-LEVEL binding. Per-unit-of-text +attribution of which source produced which span is a separate measured +research result, referenced from the README by result id, and is not +reimplemented here. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Dict, Iterable, Mapping, Optional, Sequence, Tuple + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from .certificates import Certificate, verify_chain +from .revocation import RevocationList + +AXIS_WEIGHT_SOURCE = "weight-source" +AXIS_CONTEXT_SOURCE = "context-source" + + +@dataclass(frozen=True) +class SourceReference: + """A composition source, pinned to an exact certificate by digest.""" + + axis: str + serial: str + cert_sha256: str + + +@dataclass(frozen=True) +class ProvenanceRecord: + output_sha256: str + sources: Tuple[SourceReference, ...] + signer_serial: str + issued_at: str + signature_hex: str = "" + + def tbs_dict(self) -> Dict: + return { + "output_sha256": self.output_sha256, + "sources": [ + {"axis": s.axis, "serial": s.serial, "cert_sha256": s.cert_sha256} + for s in self.sources + ], + "signer_serial": self.signer_serial, + "issued_at": self.issued_at, + } + + def tbs_bytes(self) -> bytes: + return json.dumps(self.tbs_dict(), sort_keys=True, separators=(",", ":")).encode( + "utf-8" + ) + + +@dataclass(frozen=True) +class ProvenanceResult: + valid: bool + reason: str + checks: Tuple[str, ...] = () + + def __bool__(self) -> bool: # pragma: no cover - convenience only + return self.valid + + +def bind_output( + output: bytes, + sources: Sequence[Tuple[str, Certificate]], + signer_certificate: Certificate, + signer_private_key: Ed25519PrivateKey, + at: Optional[datetime] = None, +) -> ProvenanceRecord: + """Create a signed provenance record binding ``output`` to its sources. + + ``sources`` is a sequence of ``(axis, certificate)`` pairs, one per + composition source. The signer is the producing capability (normally the + weight-source), holding its own leaf private key: the authority does not + sign provenance, the capability does, and any verifier checks it against + the capability's certificate. + """ + moment = at or datetime.now(timezone.utc) + unsigned = ProvenanceRecord( + output_sha256=hashlib.sha256(output).hexdigest(), + sources=tuple( + SourceReference(axis=axis, serial=cert.serial, cert_sha256=cert.sha256()) + for axis, cert in sources + ), + signer_serial=signer_certificate.serial, + issued_at=moment.astimezone(timezone.utc).isoformat(timespec="seconds"), + ) + signature = signer_private_key.sign(unsigned.tbs_bytes()).hex() + return ProvenanceRecord( + output_sha256=unsigned.output_sha256, + sources=unsigned.sources, + signer_serial=unsigned.signer_serial, + issued_at=unsigned.issued_at, + signature_hex=signature, + ) + + +def verify_output_provenance( + output: bytes, + record: ProvenanceRecord, + registry: Mapping[str, Certificate], + intermediates: Iterable[Certificate], + trusted_roots: Iterable[Certificate], + revocation_list: Optional[RevocationList] = None, + at: Optional[datetime] = None, +) -> ProvenanceResult: + """Verify an output against its provenance record and the live authority state. + + Returns VALID only if every check passes: + + 1. the output's digest matches the bound digest; + 2. the signer's certificate is known and its signature over the record + verifies; + 3. the signer's chain verifies to a trusted root and is not revoked; + 4. every referenced source certificate is known, byte-identical to the + one bound at composition time, chain-valid, and not revoked or erased. + """ + checks = [] + intermediates = tuple(intermediates) + trusted_roots = tuple(trusted_roots) + + def fail(reason: str) -> ProvenanceResult: + return ProvenanceResult(False, reason, tuple(checks)) + + digest = hashlib.sha256(output).hexdigest() + if digest != record.output_sha256: + return fail( + "output does not match the bound digest (tampered or substituted output)" + ) + checks.append(f"output digest matches ({digest[:16]}...)") + + signer = registry.get(record.signer_serial) + if signer is None: + return fail(f"signer certificate not found: {record.signer_serial}") + try: + signer.public_key().verify( + bytes.fromhex(record.signature_hex), record.tbs_bytes() + ) + except (InvalidSignature, ValueError): + return fail( + f"provenance signature does not verify against signer {signer.serial} " + "(forged or edited record)" + ) + checks.append(f"record signature by {signer.serial} verifies") + + signer_chain = verify_chain( + signer, intermediates, trusted_roots, at=at, revocation_list=revocation_list + ) + if not signer_chain.valid: + return fail(f"signer chain invalid: {signer_chain.reason}") + checks.append("signer chain verifies to trusted root") + + for source in record.sources: + certificate = registry.get(source.serial) + if certificate is None: + return fail(f"{source.axis} certificate not found: {source.serial}") + if certificate.sha256() != source.cert_sha256: + return fail( + f"{source.axis} certificate {source.serial} does not match the " + "content bound at composition time (substituted certificate)" + ) + chain = verify_chain( + certificate, + intermediates, + trusted_roots, + at=at, + revocation_list=revocation_list, + ) + if not chain.valid: + return fail( + f"{source.axis} certificate {source.serial} failed verification: " + f"{chain.reason}" + ) + checks.append(f"{source.axis} {source.serial}: intact, chain valid, not revoked") + + return ProvenanceResult(True, "output provenance verifies", tuple(checks)) diff --git a/src/capability_licensing/revocation.py b/src/capability_licensing/revocation.py new file mode 100644 index 0000000..3e41b33 --- /dev/null +++ b/src/capability_licensing/revocation.py @@ -0,0 +1,96 @@ +"""The revocation ledger: the soft (advisory) rung of the revocation ladder. + +Three rungs, in increasing strength: + +* SOFT (advisory), this module: an entry on a revocation list. Compliant + verifiers and authorities refuse the certificate, but the encrypted + material and its wrapping key still exist. Reversible: remove the entry. +* ACCESS-GATED, :mod:`capability_licensing.keystore`: the authority withholds + the wrapping key, so decryption is denied at use-time. Reversible: the + authority re-releases the key. +* CRYPTO-ERASURE, :mod:`capability_licensing.keystore`: the wrapping key is + destroyed. Irreversible: no administrator action brings the plaintext back. + +The asymmetry is deliberate and honest: the first two rungs depend on parties +honouring the authority; only the third is final at the level of the +cryptography itself. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Dict, Optional, Tuple + +RUNG_SOFT = "soft" +RUNG_ACCESS_GATED = "access-gated" +RUNG_CRYPTO_ERASURE = "crypto-erasure" + +_RUNGS = (RUNG_SOFT, RUNG_ACCESS_GATED, RUNG_CRYPTO_ERASURE) + + +class IrreversibleRevocationError(Exception): + """Raised on an attempt to reinstate a crypto-erased certificate. + + The ledger records erasure as an irreversible event: even if the entry + were deleted, the wrapping key no longer exists, so reinstatement would + assert something that cannot be true. + """ + + +@dataclass(frozen=True) +class RevocationEntry: + serial: str + reason: str + rung: str + revoked_at: str + + +class RevocationList: + """A certificate revocation ledger consulted by verifiers. + + This is the classic certificate-revocation-list shape: revoke once at the + authority and every verifier that consults the list refuses the + certificate. The list itself enforces nothing; it is advisory. Its + strength comes from verifiers and key authorities honouring it. + """ + + def __init__(self) -> None: + self._entries: Dict[str, RevocationEntry] = {} + + def revoke(self, serial: str, reason: str, rung: str = RUNG_SOFT) -> RevocationEntry: + if rung not in _RUNGS: + raise ValueError(f"unknown revocation rung: {rung!r} (expected one of {_RUNGS})") + entry = RevocationEntry( + serial=serial, + reason=reason, + rung=rung, + revoked_at=datetime.now(timezone.utc).isoformat(timespec="seconds"), + ) + self._entries[serial] = entry + return entry + + def reinstate(self, serial: str) -> None: + """Remove a revocation entry (reversibility of the advisory rung). + + Refused for crypto-erasure entries: that rung is not reversible, and + the ledger must not pretend otherwise. + """ + entry = self._entries.get(serial) + if entry is None: + raise KeyError(f"no revocation entry for {serial}") + if entry.rung == RUNG_CRYPTO_ERASURE: + raise IrreversibleRevocationError( + f"certificate {serial} was crypto-erased; erasure is irreversible " + "and the entry cannot be reinstated" + ) + del self._entries[serial] + + def is_revoked(self, serial: str) -> bool: + return serial in self._entries + + def entry_for(self, serial: str) -> Optional[RevocationEntry]: + return self._entries.get(serial) + + def entries(self) -> Tuple[RevocationEntry, ...]: + return tuple(self._entries.values()) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..8b9132a --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,56 @@ +"""Shared test fixtures: a full authority chain and a sealed stand-in unit.""" + +import os +import sys +from pathlib import Path +from types import SimpleNamespace + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from capability_licensing import ( # noqa: E402 + CertificateAuthority, + KeyAuthority, + RevocationList, +) + + +@pytest.fixture() +def chain(): + """A root, an organisational signer, and a capability leaf with its key.""" + authority = CertificateAuthority() + root = authority.create_root("Test Root Authority") + organisation = authority.issue_organisation(root.serial, "Test Organisation") + capability, capability_key = authority.issue_capability( + organisation.serial, + "test-capability", + claims={"grant": "test use only"}, + ) + return SimpleNamespace( + authority=authority, + root=root, + organisation=organisation, + capability=capability, + capability_key=capability_key, + ) + + +@pytest.fixture() +def sealed_setup(chain): + """A key authority with one sealed stand-in unit licensed to the leaf.""" + revocations = RevocationList() + keys = KeyAuthority( + trusted_roots=[chain.root], + intermediates=[chain.organisation], + revocation_list=revocations, + ) + payload = b"STAND-IN CAPABILITY UNIT (not a real model)\n" + os.urandom(4096) + sealed = keys.seal_unit("unit-test-001", payload, chain.capability.serial) + return SimpleNamespace( + chain=chain, + revocations=revocations, + keys=keys, + payload=payload, + sealed=sealed, + ) diff --git a/tests/test_certificates.py b/tests/test_certificates.py new file mode 100644 index 0000000..5c76b3a --- /dev/null +++ b/tests/test_certificates.py @@ -0,0 +1,97 @@ +"""Claim 1: the certificate-chain authority behaves as stated.""" + +import dataclasses +from datetime import timedelta + +import pytest + +from capability_licensing import CertificateAuthority, verify_chain +from capability_licensing.certificates import parse_iso + + +def test_intact_chain_verifies_to_trusted_root(chain): + result = verify_chain(chain.capability, [chain.organisation], [chain.root]) + assert result.valid + assert result.path == ( + chain.capability.serial, + chain.organisation.serial, + chain.root.serial, + ) + + +def test_root_is_self_signed_and_verifies_alone(chain): + result = verify_chain(chain.root, [], [chain.root]) + assert result.valid + + +def test_tampered_certificate_fails(chain): + tampered = dataclasses.replace( + chain.capability, + claims={**dict(chain.capability.claims), "grant": "unlimited production use"}, + ) + result = verify_chain(tampered, [chain.organisation], [chain.root]) + assert not result.valid + assert "does not verify" in result.reason + + +def test_tampered_subject_fails(chain): + tampered = dataclasses.replace(chain.capability, subject="renamed-capability") + result = verify_chain(tampered, [chain.organisation], [chain.root]) + assert not result.valid + + +def test_tampered_intermediate_fails(chain): + tampered_org = dataclasses.replace( + chain.organisation, subject="Hostile Organisation" + ) + result = verify_chain(chain.capability, [tampered_org], [chain.root]) + assert not result.valid + + +def test_expired_certificate_fails(chain): + after_expiry = parse_iso(chain.capability.not_after) + timedelta(days=1) + result = verify_chain( + chain.capability, [chain.organisation], [chain.root], at=after_expiry + ) + assert not result.valid + assert "expired" in result.reason + + +def test_not_yet_valid_certificate_fails(chain): + before_validity = parse_iso(chain.capability.not_before) - timedelta(days=1) + result = verify_chain( + chain.capability, [chain.organisation], [chain.root], at=before_validity + ) + assert not result.valid + assert "not yet valid" in result.reason + + +def test_chain_to_untrusted_root_fails(chain): + other = CertificateAuthority() + other_root = other.create_root("Unaccredited Authority") + other_org = other.issue_organisation(other_root.serial, "Unaccredited Organisation") + other_leaf, _ = other.issue_capability(other_org.serial, "impostor", claims={}) + result = verify_chain(other_leaf, [other_org, other_root], [chain.root]) + assert not result.valid + assert "not in the trust store" in result.reason + + +def test_missing_intermediate_fails(chain): + result = verify_chain(chain.capability, [], [chain.root]) + assert not result.valid + assert "issuer not found" in result.reason + + +def test_capability_may_not_issue_certificates(chain): + with pytest.raises(ValueError, match="role violation"): + chain.authority._issue( + issuer_serial=chain.capability.serial, + subject="illicit-sub-capability", + role="capability", + claims={}, + lifetime_days=1, + ) + + +def test_authority_does_not_retain_leaf_keys(chain): + assert chain.capability.serial not in chain.authority._signing_keys diff --git a/tests/test_erasure.py b/tests/test_erasure.py new file mode 100644 index 0000000..78f8000 --- /dev/null +++ b/tests/test_erasure.py @@ -0,0 +1,90 @@ +"""Claim 4: crypto-erasure makes both the original and an exfiltrated copy +permanently undecryptable.""" + +import copy +import os + +import pytest + +from capability_licensing import ( + KeyDestroyedError, + UnsealError, + generate_wrapping_key, + seal, + unseal, +) + + +def test_exfiltrated_copy_decrypts_before_erasure(sealed_setup): + exfiltrated = copy.deepcopy(sealed_setup.sealed) + plaintext = sealed_setup.keys.request_decrypt( + exfiltrated, sealed_setup.chain.capability + ) + assert plaintext == sealed_setup.payload + + +def test_after_erasure_both_copies_are_undecryptable(sealed_setup): + exfiltrated = copy.deepcopy(sealed_setup.sealed) + + # Honest baseline: the copy is decryptable before erasure. + assert ( + sealed_setup.keys.request_decrypt(exfiltrated, sealed_setup.chain.capability) + == sealed_setup.payload + ) + + sealed_setup.keys.destroy(sealed_setup.sealed.unit_id) + + with pytest.raises(KeyDestroyedError): + sealed_setup.keys.request_decrypt( + sealed_setup.sealed, sealed_setup.chain.capability + ) + with pytest.raises(KeyDestroyedError): + sealed_setup.keys.request_decrypt( + exfiltrated, sealed_setup.chain.capability + ) + + +def test_ciphertext_is_useless_without_the_exact_wrapping_key(): + wrapping_key = generate_wrapping_key() + payload = b"STAND-IN CAPABILITY UNIT (not a real model)\n" + os.urandom(2048) + sealed = seal("unit-raw-001", payload, wrapping_key, "CERT-example") + + # The exact key decrypts. + assert unseal(sealed, wrapping_key) == payload + + # Any other key fails AES-256-GCM authentication: no partial decryption, + # no garbage plaintext, a hard failure. + with pytest.raises(UnsealError): + unseal(sealed, os.urandom(32)) + + # A single flipped bit in the key also fails: recovery requires the + # destroyed key exactly, not something close to it. + near_miss = bytearray(wrapping_key) + near_miss[0] ^= 0x01 + with pytest.raises(UnsealError): + unseal(sealed, bytes(near_miss)) + + +def test_tampered_ciphertext_fails_authentication(): + import dataclasses + + wrapping_key = generate_wrapping_key() + payload = b"STAND-IN CAPABILITY UNIT (not a real model)\n" + os.urandom(2048) + sealed = seal("unit-raw-002", payload, wrapping_key, "CERT-example") + + tampered_hex = bytearray(sealed.payload_ciphertext_hex.encode()) + tampered_hex[0] = ord("f") if tampered_hex[0] != ord("f") else ord("0") + tampered = dataclasses.replace( + sealed, payload_ciphertext_hex=tampered_hex.decode() + ) + with pytest.raises(UnsealError): + unseal(tampered, wrapping_key) + + +def test_sealed_unit_reports_payload_digest(sealed_setup): + import hashlib + + assert ( + sealed_setup.sealed.payload_sha256 + == hashlib.sha256(sealed_setup.payload).hexdigest() + ) diff --git a/tests/test_provenance.py b/tests/test_provenance.py new file mode 100644 index 0000000..e07abae --- /dev/null +++ b/tests/test_provenance.py @@ -0,0 +1,129 @@ +"""Claim 2: output provenance binding verifies and invalidates as stated.""" + +import dataclasses + +import pytest + +from capability_licensing import ( + AXIS_CONTEXT_SOURCE, + AXIS_WEIGHT_SOURCE, + RevocationList, + RUNG_CRYPTO_ERASURE, + bind_output, + verify_output_provenance, +) + +OUTPUT = b"Stand-in generated output bound to its composition certificates." + + +@pytest.fixture() +def bound(chain): + """A second (context-source) certificate and a signed provenance record.""" + context_source, _ = chain.authority.issue_capability( + chain.organisation.serial, + "licensed-reference-corpus", + claims={"axis": AXIS_CONTEXT_SOURCE}, + ) + record = bind_output( + OUTPUT, + sources=[ + (AXIS_WEIGHT_SOURCE, chain.capability), + (AXIS_CONTEXT_SOURCE, context_source), + ], + signer_certificate=chain.capability, + signer_private_key=chain.capability_key, + ) + return chain, context_source, record + + +def _verify(bound, output=OUTPUT, revocations=None): + chain, _context_source, record = bound + return verify_output_provenance( + output, + record, + chain.authority.registry(), + [chain.organisation], + [chain.root], + revocations or RevocationList(), + ) + + +def test_intact_composition_is_valid(bound): + result = _verify(bound) + assert result.valid, result.reason + # One digest check, one signature check, one signer-chain check, and one + # check per referenced source. + assert len(result.checks) == 5 + + +def test_tampered_output_is_invalid(bound): + result = _verify(bound, output=OUTPUT + b" [edited]") + assert not result.valid + assert "does not match the bound digest" in result.reason + + +def test_forged_record_signature_is_invalid(bound): + chain, context_source, record = bound + forged = dataclasses.replace(record, output_sha256="0" * 64) + result = verify_output_provenance( + OUTPUT, + forged, + chain.authority.registry(), + [chain.organisation], + [chain.root], + ) + assert not result.valid + + +def test_revoked_source_certificate_invalidates_the_output(bound): + _chain, context_source, _record = bound + revocations = RevocationList() + revocations.revoke(context_source.serial, "context licence terminated") + result = _verify(bound, revocations=revocations) + assert not result.valid + assert "revoked" in result.reason + assert context_source.serial in result.reason + + +def test_erased_source_certificate_invalidates_the_output(bound): + chain, _context_source, _record = bound + revocations = RevocationList() + revocations.revoke( + chain.capability.serial, + "capability unit crypto-erased", + rung=RUNG_CRYPTO_ERASURE, + ) + result = _verify(bound, revocations=revocations) + assert not result.valid + assert "crypto-erasure" in result.reason + + +def test_reinstating_a_soft_revocation_restores_validity(bound): + _chain, context_source, _record = bound + revocations = RevocationList() + revocations.revoke(context_source.serial, "temporary hold") + assert not _verify(bound, revocations=revocations).valid + revocations.reinstate(context_source.serial) + assert _verify(bound, revocations=revocations).valid + + +def test_substituted_source_certificate_is_detected(bound): + chain, context_source, record = bound + # An attacker replaces the registry's copy of the context-source + # certificate with a different one under the same serial. + substituted, _ = chain.authority.issue_capability( + chain.organisation.serial, "look-alike-corpus", claims={} + ) + registry = chain.authority.registry() + registry[context_source.serial] = dataclasses.replace( + substituted, serial=context_source.serial + ) + result = verify_output_provenance( + OUTPUT, + record, + registry, + [chain.organisation], + [chain.root], + ) + assert not result.valid + assert "substituted certificate" in result.reason diff --git a/tests/test_revocation.py b/tests/test_revocation.py new file mode 100644 index 0000000..fb66e9b --- /dev/null +++ b/tests/test_revocation.py @@ -0,0 +1,136 @@ +"""Claim 3: the three-rung revocation ladder behaves as stated.""" + +import pytest + +from capability_licensing import ( + IrreversibleRevocationError, + KeyDestroyedError, + KeyState, + KeyWithheldError, + LicenceInvalidError, + RevocationList, + RUNG_CRYPTO_ERASURE, + verify_chain, +) + + +def test_baseline_valid_licence_decrypts(sealed_setup): + plaintext = sealed_setup.keys.request_decrypt( + sealed_setup.sealed, sealed_setup.chain.capability + ) + assert plaintext == sealed_setup.payload + + +# ---- rung (a): SOFT / advisory ------------------------------------------ + + +def test_soft_revocation_reported_by_verifiers(sealed_setup): + chain = sealed_setup.chain + sealed_setup.revocations.revoke(chain.capability.serial, "terms breached") + result = verify_chain( + chain.capability, + [chain.organisation], + [chain.root], + revocation_list=sealed_setup.revocations, + ) + assert not result.valid + assert "revoked" in result.reason + + +def test_soft_revocation_denies_at_a_compliant_authority(sealed_setup): + chain = sealed_setup.chain + sealed_setup.revocations.revoke(chain.capability.serial, "terms breached") + with pytest.raises(LicenceInvalidError, match="revoked"): + sealed_setup.keys.request_decrypt(sealed_setup.sealed, chain.capability) + + +def test_soft_revocation_is_advisory_key_still_exists(sealed_setup): + sealed_setup.revocations.revoke( + sealed_setup.chain.capability.serial, "terms breached" + ) + assert sealed_setup.keys.key_state(sealed_setup.sealed.unit_id) is KeyState.RELEASED + + +def test_soft_revocation_is_reversible(sealed_setup): + chain = sealed_setup.chain + sealed_setup.revocations.revoke(chain.capability.serial, "terms breached") + sealed_setup.revocations.reinstate(chain.capability.serial) + plaintext = sealed_setup.keys.request_decrypt(sealed_setup.sealed, chain.capability) + assert plaintext == sealed_setup.payload + + +# ---- rung (b): ACCESS-GATED --------------------------------------------- + + +def test_withheld_key_denies_even_a_valid_licence(sealed_setup): + sealed_setup.keys.withhold(sealed_setup.sealed.unit_id) + with pytest.raises(KeyWithheldError): + sealed_setup.keys.request_decrypt( + sealed_setup.sealed, sealed_setup.chain.capability + ) + + +def test_withheld_key_is_recoverable_by_re_release(sealed_setup): + sealed_setup.keys.withhold(sealed_setup.sealed.unit_id) + sealed_setup.keys.re_release(sealed_setup.sealed.unit_id) + plaintext = sealed_setup.keys.request_decrypt( + sealed_setup.sealed, sealed_setup.chain.capability + ) + assert plaintext == sealed_setup.payload + + +# ---- rung (c): CRYPTO-ERASURE ------------------------------------------- + + +def test_destroyed_key_denies_permanently(sealed_setup): + sealed_setup.keys.destroy(sealed_setup.sealed.unit_id) + with pytest.raises(KeyDestroyedError): + sealed_setup.keys.request_decrypt( + sealed_setup.sealed, sealed_setup.chain.capability + ) + + +def test_destroyed_key_cannot_be_re_released(sealed_setup): + sealed_setup.keys.destroy(sealed_setup.sealed.unit_id) + with pytest.raises(KeyDestroyedError): + sealed_setup.keys.re_release(sealed_setup.sealed.unit_id) + + +def test_destroyed_key_cannot_be_withheld_or_destroyed_again(sealed_setup): + sealed_setup.keys.destroy(sealed_setup.sealed.unit_id) + with pytest.raises(KeyDestroyedError): + sealed_setup.keys.withhold(sealed_setup.sealed.unit_id) + with pytest.raises(KeyDestroyedError): + sealed_setup.keys.destroy(sealed_setup.sealed.unit_id) + + +def test_erasure_ledger_entry_cannot_be_reinstated(): + revocations = RevocationList() + revocations.revoke("CERT-example", "unit erased", rung=RUNG_CRYPTO_ERASURE) + with pytest.raises(IrreversibleRevocationError): + revocations.reinstate("CERT-example") + + +# ---- licence checks at the authority ------------------------------------ + + +def test_wrong_certificate_is_refused(sealed_setup): + chain = sealed_setup.chain + other_leaf, _ = chain.authority.issue_capability( + chain.organisation.serial, "some-other-capability", claims={} + ) + with pytest.raises(LicenceInvalidError, match="does not match"): + sealed_setup.keys.request_decrypt(sealed_setup.sealed, other_leaf) + + +def test_audit_log_records_the_ladder(sealed_setup): + keys = sealed_setup.keys + unit_id = sealed_setup.sealed.unit_id + keys.withhold(unit_id) + keys.re_release(unit_id) + keys.destroy(unit_id) + events = [event for _stamp, logged_unit, event in keys.audit_log() if logged_unit == unit_id] + assert any("sealed" in event for event in events) + assert any("withheld" in event for event in events) + assert any("re-released" in event for event in events) + assert any("DESTROYED" in event for event in events)