aegis/engines/veracity.py
Rama main · réplica de lectura del repositorio original.
# Copyright (c) 2026 Juan Luna. All rights reserved.
# SPDX-License-Identifier: LicenseRef-Aegis-Commercial-1.0
# Licensed under the Aegis Commercial License 1.0; see LICENSE and NOTICE.
"""Veracity engine — evidence commitment and inclusion proofs, without the proxy.
Wraps :class:`~aegis.core.crypto_audit.CryptographicAuditLedger` and
:class:`~aegis.core.crypto_shredder.CryptoShredder` so an application already
running its own gateway can commit evidence and issue portable proofs by
library call:
engine = VeracityEngine(wal_path="/var/lib/aegis/evidence.jsonl",
signing_key=os.environ["AEGIS_SIGNING_KEY"])
record = engine.commit_evidence_record(
state_id="req-1", request_bytes=b"...", response_bytes=b"...")
proof = engine.generate_inclusion_proof(record.leaf_index)
engine.verify_evidence_proof(record.leaf_digest, proof, trusted_root)
Boundaries carried over unchanged
---------------------------------
This is a facade. It adds no cryptographic property and relaxes none, so every
boundary on the underlying components still applies:
- Proofs are **non-zero-knowledge** inclusion proofs for a disclosed leaf
against a **separately trusted root** (``CLM-006``). Passing the proof's own
root back as ``trusted_root`` proves internal consistency and nothing else.
- One process, one WAL path. The single-writer lock refuses a second writer
rather than serialising it; this facade does not change that.
- ``fsync`` returning is a process-observed acknowledgement, not a power-loss
guarantee on any particular device.
- Crypto-shredding here is the unwired mechanism described in ``CLM-068``: it
is **not** wired into the ledger's own commit path, so
:meth:`crypto_shred` destroys a key for payloads *this facade sealed*, and
establishes nothing about media sanitisation or any legal obligation.
"""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any
from aegis.core.crypto_audit import CryptographicAuditLedger
from aegis.core.crypto_shredder import CryptoShredder, SealedPayload
from aegis.core.mmr import MMRInclusionProofV1
from aegis.engines import require_module
from aegis.licensing.validator import LicenseEntitlement
_MODULE_NAME = "veracity"
@dataclass(frozen=True, slots=True)
class EvidenceRecord:
"""What a commit produced, enough to verify it later without the ledger.
``proof`` is the node's own self-contained inclusion proof, captured at
commit time. It stays verifiable after a peak-set restore has summarised
the leaf away, which is why it is carried here rather than fetched later.
"""
state_id: str
node_hash: str
merkle_root: str
leaf_index: int
leaf_digest: str
proof: dict[str, Any] | None = None
class VeracityEngine:
"""Evidence commitment and portable inclusion proofs as a library."""
def __init__(
self,
wal_path: str,
*,
signing_key: str = "",
shredder_vault_path: str | None = None,
entitlement: LicenseEntitlement | None = None,
**ledger_kwargs: Any,
) -> None:
self._entitlement = require_module(_MODULE_NAME, entitlement=entitlement)
self._ledger = CryptographicAuditLedger(wal_path, signing_key=signing_key, **ledger_kwargs)
self._shredder = CryptoShredder(shredder_vault_path)
@property
def ledger(self) -> CryptographicAuditLedger:
"""The underlying ledger, for callers needing the full surface."""
return self._ledger
@property
def entitlement(self) -> LicenseEntitlement | None:
return self._entitlement
# ── commitment ──────────────────────────────────────────────────────
def commit_evidence_record(
self,
*,
state_id: str,
request_bytes: bytes,
response_bytes: bytes | None = None,
tenant_id: str = "default",
model: str = "unknown",
endpoint: str = "library",
**extra: Any,
) -> EvidenceRecord:
"""Commit one evidence node and return its identifiers.
The write is durable before this returns, in the same sense the gateway
means it: appended, flushed and ``fsync``-ed. See the module docstring
on what ``fsync`` does and does not establish.
"""
node = self._ledger.commit_forensic(
state_id=state_id,
request_bytes=request_bytes,
response_bytes=response_bytes,
tenant_id=tenant_id,
model=model,
endpoint=endpoint,
**extra,
)
return self._record_from_node(node)
def _record_from_node(self, node: Any) -> EvidenceRecord:
"""Read the identifiers the node already carries.
``mmr_leaf_hash`` and ``mmr_proof`` are written by the ledger at commit
time. Recomputing a leaf digest here would be a second, independent
guess at the ledger's leaf construction, and a wrong one would produce
proofs that never verify.
"""
proof: Mapping[str, Any] | None = getattr(node, "mmr_proof", None)
return EvidenceRecord(
state_id=str(getattr(node, "state_id", "")),
node_hash=str(getattr(node, "node_hash", "")),
merkle_root=str(getattr(node, "merkle_root", "")),
leaf_index=int(proof["leaf_index"]) if proof else -1,
leaf_digest=str(getattr(node, "mmr_leaf_hash", "")),
proof=dict(proof) if proof else None,
)
# ── proofs ──────────────────────────────────────────────────────────
def generate_inclusion_proof(self, leaf_index: int) -> dict[str, Any]:
"""Return the portable ``aegis-mmr-inclusion-v1`` proof for a leaf.
Returned as a plain dict so it can be serialised straight onto the
wire. Raises ``MMRHistoricalLeafUnavailableError`` for a leaf summarised
away by a peak-set restore; every committed node also carries its own
self-contained copy, which is the durable one.
"""
proof = self._ledger._mmr.get_portable_inclusion_proof(leaf_index)
as_dict: dict[str, Any] = proof.to_dict()
return as_dict
def verify_evidence_proof(
self,
leaf_digest: str,
proof: Mapping[str, Any] | MMRInclusionProofV1,
trusted_root: str,
) -> bool:
"""Verify a portable proof against a root obtained independently.
Accepts the proof as the dataclass or as the dict it serialises to, so
a proof that arrived over the wire needs no unwrapping by the caller.
``trusted_root`` must come from somewhere other than the proof itself —
supplying ``proof["root"]`` back is circular and shows only that the
envelope is internally consistent.
"""
envelope = (
proof
if isinstance(proof, MMRInclusionProofV1)
else MMRInclusionProofV1.from_dict(dict(proof))
)
result: bool = self._ledger._mmr.verify_portable_inclusion_hash(
leaf_digest, envelope, trusted_root
)
return result
def current_root(self) -> str:
root: str = self._ledger._mmr.get_root_hash()
return root
def verify_integrity(self) -> tuple[bool, int | None]:
"""Walk the retained window. Not a full-history verification."""
return self._ledger.verify_integrity()
# ── cryptographic erasure ───────────────────────────────────────────
def seal_payload(self, subject_id: str, plaintext: bytes) -> SealedPayload:
"""Seal a payload under ``subject_id``'s key; commit its ``commitment``."""
return self._shredder.seal(subject_id, plaintext)
def open_payload(self, sealed: SealedPayload) -> bytes:
return self._shredder.open(sealed)
def crypto_shred(self, subject_id: str) -> bool:
"""Destroy a subject's key. Returns whether one was present.
The ledger does not move: roots, peaks and previously issued proofs are
bit-for-bit unchanged. This makes the plaintext unrecoverable **to a
holder of the ciphertext** and says nothing about the physical medium,
vault backups, or any regulatory obligation (``CLM-068``).
"""
return self._shredder.erase(subject_id)
def close(self) -> None:
self._shredder.close()
self._ledger.close()
def __enter__(self) -> VeracityEngine:
return self
def __exit__(self, *exc_info: object) -> None:
self.close()
__all__ = ["EvidenceRecord", "VeracityEngine"]