research/src/capability_licensing/revocation.py

"""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 export_state(self) -> list:
        """Serialise the ledger to a JSON-safe list of entry dicts."""
        return [
            {
                "serial": entry.serial,
                "reason": entry.reason,
                "rung": entry.rung,
                "revoked_at": entry.revoked_at,
            }
            for entry in self._entries.values()
        ]

    @classmethod
    def restore(cls, entries: list) -> "RevocationList":
        """Rebuild a ledger previously serialised with :meth:`export_state`."""
        ledger = cls()
        for entry_data in entries:
            ledger._entries[entry_data["serial"]] = RevocationEntry(**entry_data)
        return ledger

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