"""Hosted Breach/Compliance Record publication (WP-0006-T09). Publishes the Licensor's own breach and termination determinations for a Phase, per `specs/TargetRevenueSourceLicense-V1C1.md` §7.4 and `specs/TechnicalSpecificationDocument.md` §4.1. This module never decides whether a breach occurred, was cured, or was terminated — it only records, signs, and publishes the Licensor's assertion of one of those facts, exactly like `attestation.py` records an already-true conversion rather than deciding one. The distinction between `alleged` and `determined` (and the further `cured`/`terminated` outcomes) is the Licensor's call, or a Commercial Use Agreement's dispute process's call — never this module's. Naming: default anonymized (Phase and category only, no party identified). Named disclosure requires the caller to affirmatively assert that the applicable Commercial Use Agreement's naming/disclosure clause (License §7.4, CUA V1C1 §9) authorizes it — this module records that assertion, it does not, and cannot, verify the underlying CUA text itself. """ from __future__ import annotations from typing import TYPE_CHECKING, Any from . import hashing if TYPE_CHECKING: from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from psycopg import Connection from .registry import Licensor _EVENT_TYPES = frozenset({"alleged", "cured", "determined", "terminated"}) _REQUIRED_FIELDS = ("id", "case_id", "category", "event_at") def _validate(event_input: dict[str, Any]) -> list[str]: errors = [] for field in _REQUIRED_FIELDS: if not event_input.get(field): errors.append(f"{field}: required") event_type = event_input.get("event_type") if event_type not in _EVENT_TYPES: errors.append( f"event_type: must be one of {sorted(_EVENT_TYPES)}, got {event_type!r}" ) anonymized = event_input.get("anonymized", True) named_holder = event_input.get("named_entitlement_holder") authorized = event_input.get("named_disclosure_authorized_under_cua", False) if not anonymized: if not named_holder: errors.append( "named_entitlement_holder: required when anonymized is false" ) if not authorized: errors.append( "named_disclosure_authorized_under_cua: must be true to publish a " "named record — the Trust Service records this assertion, it does " "not verify the underlying Commercial Use Agreement text " "(License V1C1 §7.4)" ) elif named_holder: errors.append( "named_entitlement_holder: must not be set when anonymized is true " "(default)" ) return errors def publish_breach_event( conn: "Connection", licensor: "Licensor", phase_id: str, event_input: dict[str, Any], signing_key: "Ed25519PrivateKey", ) -> dict[str, Any]: """Validate, sign, and append one breach/compliance lifecycle event. Never mutates a prior event for the same `case_id` — resolution (cure, determination, termination) is always a new, separately dated event, per License V1C1 §7.4's "update the record promptly upon resolution," which this module implements as append, not edit (`migrations/0004_breach_records.sql` grants trf_app no UPDATE/DELETE on this table at all). """ from .registry import RegistrationError, get_phase_manifest manifest = get_phase_manifest(conn, phase_id) if manifest is None: raise RegistrationError(f"phase {phase_id!r} is not registered") manifest_licensor = conn.execute( "SELECT licensor_id FROM phase_manifests WHERE phase_id = %s", (phase_id,) ).fetchone()[0] if manifest_licensor != licensor.licensor_id: raise RegistrationError( "this token is not authorized to publish breach records for this Phase" ) errors = _validate(event_input) if errors: raise RegistrationError("breach event rejected: " + "; ".join(errors)) anonymized = event_input.get("anonymized", True) record = { "id": event_input["id"], "case_id": event_input["case_id"], "phase": phase_id, "event_type": event_input["event_type"], "category": event_input["category"], "event_at": event_input["event_at"], "anonymized": anonymized, "named_entitlement_holder": event_input.get("named_entitlement_holder") if not anonymized else None, "evidence_reference": event_input.get("evidence_reference"), } signed = {**record, "signature": hashing.sign_record(record, signing_key)} from psycopg.errors import UniqueViolation try: conn.execute( """ INSERT INTO breach_records (record_id, case_id, phase_id, event_type, category, event_at, anonymized, named_entitlement_holder, evidence_reference, published_by, signature) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( signed["id"], signed["case_id"], phase_id, signed["event_type"], signed["category"], signed["event_at"], signed["anonymized"], signed["named_entitlement_holder"], signed["evidence_reference"], licensor.licensor_id, signed["signature"], ), ) except UniqueViolation as exc: raise RegistrationError(f"breach record id {signed['id']!r} already exists") from exc return signed def get_breach_records(conn: "Connection", phase_id: str) -> list[dict[str, Any]]: """Public read: the Phase's full breach/compliance event history. Anonymized events never carry `named_entitlement_holder` in the first place (enforced at publish time and by the table's own CHECK constraint); this function does not need to additionally strip it, but the caller must never bypass this function to read the table directly for a public-facing response. """ rows = conn.execute( """ SELECT record_id, case_id, event_type, category, event_at, anonymized, named_entitlement_holder, evidence_reference, signature FROM breach_records WHERE phase_id = %s ORDER BY sequence ASC """, (phase_id,), ).fetchall() columns = ( "id", "case_id", "event_type", "category", "event_at", "anonymized", "named_entitlement_holder", "evidence_reference", "signature", ) return [dict(zip(columns, row)) for row in rows]