meanwhile/research
meanwhile/research
3
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Initial public release: capability-licensing reference implementation + reproducible claim demos

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Meanwhile Research 2026-07-19 16:31:58 +10:00
commit c8a04e144b
22 changed files with 2571 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
.gitignore vendored Normal file
View file

@ -0,0 +1,4 @@
.venv/
__pycache__/
*.pyc
.pytest_cache/

21
LICENSE Normal file
View file

@ -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.

24
Makefile Normal file
View file

@ -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 {} +

293
README.md Normal file
View file

@ -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`.

162
examples/crypto_erasure_undecryptable.py Normal file
View file

@ -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())

108
examples/issue_certificate.py Normal file
View file

@ -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())

178
examples/revoke_three_rungs.py Normal file
View file

@ -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())

141
examples/verify_output_provenance.py Normal file
View file

@ -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())

2
pytest.ini Normal file
View file

@ -0,0 +1,2 @@
[pytest]
testpaths = tests

2
requirements.txt Normal file
View file

@ -0,0 +1,2 @@
cryptography==49.0.0
pytest==9.1.1

29
run_all.sh Executable file
View file

@ -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"

99
src/capability_licensing/__init__.py Normal file
View file

@ -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"

378
src/capability_licensing/certificates.py Normal file
View file

@ -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))

122
src/capability_licensing/envelope.py Normal file
View file

@ -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

200
src/capability_licensing/keystore.py Normal file
View file

@ -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]))

204
src/capability_licensing/provenance.py Normal file
View file

@ -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))

96
src/capability_licensing/revocation.py Normal file
View file

@ -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())

56
tests/conftest.py Normal file
View file

@ -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,
)

97
tests/test_certificates.py Normal file
View file

@ -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

90
tests/test_erasure.py Normal file
View file

@ -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()
)

129
tests/test_provenance.py Normal file
View file

@ -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

136
tests/test_revocation.py Normal file
View file

@ -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)