research/examples/revoke_three_rungs.py

#!/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())