"""Types every evidence query speaks — rows, results, parameters. The whole point of the evidence layer is that facts are produced by deterministic SQL and nothing else. These types are where that is enforced structurally: a handler returns :class:`EvidenceRow` objects that each carry their own provenance, and there is no shape in which a handler can emit a fact without saying where it came from. Anything that later narrates a bundle (the M3 summary layer) is limited to referencing rows that exist here, so a narrated claim always has a record behind it. """ from __future__ import annotations import hashlib import json from collections.abc import Callable, Sequence from dataclasses import dataclass, field from datetime import UTC, date, datetime from typing import Any, overload from sqlalchemy.ext.asyncio import AsyncSession @dataclass(frozen=True, slots=True) class EvidenceRow: """One checkable fact. Exactly one provenance pair should be populated: * ``source_table`` + ``source_pk`` — the fact is a database row (a holding interval, a seat allocation, an approval step). * ``audit_log_id`` (+ ``source_trace_id``) — the fact only exists in the audit trail, which is the case for cloud-grant approvals: ``cmdb_cloud_access_grants`` has no ``approval_request_id``, so the approver is reconstructed from the ``cloud.grant.approve`` row. A row with neither is rejected at freeze time. That is deliberate: an unciteable "fact" in an evidence pack is worse than a missing one because a reviewer cannot tell it apart from a checkable one. """ subject_type: str payload: dict[str, Any] subject_id: str | None = None subject_label: str | None = None source_table: str | None = None source_pk: str | None = None audit_log_id: Any = None source_trace_id: Any = None occurred_at: datetime | None = None def __post_init__(self) -> None: """Normalise ``occurred_at`` to UTC-aware, always. MySQL/TiDB hand back NAIVE datetimes for ``DATETIME`` columns, while any value this code builds — ``datetime.combine(..., tzinfo=UTC)`` for an API key's validity dates, ``_parse_iso`` for a cloud grant's resurrection history — is aware. A handler that composes rosters therefore ends up sorting both kinds together and Python raises ``TypeError: can't compare offset-naive and offset-aware datetimes``, which surfaces as a 500 on the flagship person-footprint query for the most ordinary subject there is: a leaver who holds an API key AND anything else. Normalising per row rather than at each sort site means no future handler can reintroduce it, and the frozen evidence carries an explicit offset instead of a bare local-looking timestamp. """ if self.occurred_at is not None and self.occurred_at.tzinfo is None: object.__setattr__( self, "occurred_at", self.occurred_at.replace(tzinfo=UTC), ) def has_provenance(self) -> bool: return bool( (self.source_table and self.source_pk) or self.audit_log_id, ) @dataclass(frozen=True, slots=True) class DataGap: """A declared hole in the evidence. Handlers must emit one instead of silently returning fewer rows. The two that actually occur today: * the requested window starts before the point-in-time ledgers began recording (migration 484), so holdings that closed earlier are not represented; * a source could not be read at all. Rendered verbatim into the bundle and the evidence pack. An auditor who is told "this range predates the ledger" can still work; one who silently receives a short list cannot. """ code: str detail: str affected_source: str | None = None def as_dict(self) -> dict[str, Any]: return { "code": self.code, "detail": self.detail, "affected_source": self.affected_source, } @dataclass(slots=True) class EvidenceResult: """What a handler returns: rows plus any declared incompleteness.""" rows: list[EvidenceRow] = field(default_factory=list) data_gaps: list[DataGap] = field(default_factory=list) # A handler is a plain async callable. No ORM writes, no external calls, # no model invocation — evidence is SQL over Helix's own ledgers. EvidenceHandler = Callable[[AsyncSession, dict[str, Any]], Any] @dataclass(frozen=True, slots=True) class EvidenceParam: """One declared parameter of a query. The catalogue is the only place parameters are defined, so the HTTP layer, the MCP twin and the pre-compile cron all validate against one description instead of three drifting copies. """ name: str kind: str # person | permission | software | date | date_range | text required: bool = True label_zh: str = "" label_en: str = "" label_ja: str = "" # When omitted by the caller, materialise the run timestamp into the # stored params instead of leaving the bound implicit. Without this, # "who holds this role" asked twice would legitimately produce two # different hashes for identical params — and the whole point of the # hash is that a difference means the underlying data moved. Every # upper bound must therefore be explicit in the frozen bundle. fill_with_run_time: bool = False # Provenance stamps the platform writes, not questions a human # answers. They stay declared (the unknown-parameter gate rejects # anything undeclared, and they must survive into the frozen params) # but the workbench does not render an input for them — an auditor # given a free-text "offboarding process" box and a "leaver date" # picker on a query whose answer they do not affect can only mislead # themselves and stamp a bundle with a wrong provenance. internal: bool = False @dataclass(frozen=True, slots=True) class EvidenceQueryDefinition: """A question the platform can answer deterministically. ``version`` is part of the reproducibility contract and must be bumped whenever the handler's logic changes shape — two bundles with the same params but different versions are not comparable and the UI must never diff them. ``control_keys`` ties the query to the GRC controls it produces evidence for, which is what lets the monthly pre-compile cron work from the compliance catalogue rather than from a hand-kept list. """ key: str version: str handler: EvidenceHandler control_domain: str params: tuple[EvidenceParam, ...] = () control_keys: tuple[str, ...] = () title_zh: str = "" title_en: str = "" title_ja: str = "" description_zh: str = "" description_en: str = "" description_ja: str = "" def public(self) -> dict[str, Any]: """Catalogue projection for the UI / MCP listing.""" return { "key": self.key, "version": self.version, "control_domain": self.control_domain, "control_keys": list(self.control_keys), "title": { "zh-CN": self.title_zh, "en-US": self.title_en, "ja-JP": self.title_ja, }, "description": { "zh-CN": self.description_zh, "en-US": self.description_en, "ja-JP": self.description_ja, }, "params": [ { "name": p.name, "kind": p.kind, "required": p.required, "label": { "zh-CN": p.label_zh, "en-US": p.label_en, "ja-JP": p.label_ja, }, } for p in self.params if not p.internal ], } @overload def persisted_utc(value: datetime) -> datetime: ... @overload def persisted_utc(value: None) -> None: ... def persisted_utc(value: datetime | None) -> datetime | None: """Re-attach UTC to a timestamp read back from the database. Every ``occurred_at`` is normalised to UTC before it is written (:meth:`EvidenceRow.__post_init__`), but the column is ``DATETIME``, which stores a wall clock and no offset — so it reads back NAIVE. Rendering that value directly emits a bare local-looking timestamp, and re-hashing the persisted rows produces a different digest than the one frozen with the bundle, which makes the reproducibility claim on ``content_hash`` false exactly when someone tries to check it. Anything that reads rows back — the rows API, the evidence-pack CSV, a re-hash — goes through here, so "stored as UTC" and "rendered as UTC" cannot drift apart. """ if value is None: return None return value if value.tzinfo else value.replace(tzinfo=UTC) def canonical_params(params: dict[str, Any]) -> dict[str, Any]: """Normalise parameters so the same question hashes the same way. Sorted keys and ISO-8601 dates. Without this, ``{"at": date(...)}`` and ``{"at": "2026-05-01"}`` would produce different bundles for the same question and the reproducibility claim would be hollow. """ out: dict[str, Any] = {} for key in sorted(params): value = params[key] if isinstance(value, (datetime, date)): out[key] = value.isoformat() elif isinstance(value, (list, tuple)): out[key] = [ v.isoformat() if isinstance(v, (datetime, date)) else v for v in value ] else: out[key] = value return out def content_hash(rows: Sequence[EvidenceRow]) -> str: """sha256 over the rows, in emission order. Reproducing this from the persisted rows requires re-attaching UTC to ``occurred_at`` first (see :func:`persisted_utc`): the column is ``DATETIME`` and drops the offset, so a digest computed over the raw values read back will NOT match. Every reader in this codebase goes through that helper for exactly this reason. Covers provenance, ``occurred_at`` and payload — but NOT the bundle id or the run timestamp, so the same underlying facts hash identically across two runs. Because ``occurred_at`` is digested, the column that stores it must preserve the same precision (``PreciseDateTime`` on ``auditor_evidence_bundle_rows``); a whole-second column would round the value on insert and the digest would no longer be reproducible from the persisted rows. That equality is the useful signal: when a re-run of the same question at the same version yields a different hash, the underlying data moved between the two runs — which is itself an audit finding rather than a bug in the tool. """ digest = hashlib.sha256() for row in rows: digest.update( json.dumps( { "subject_type": row.subject_type, "subject_id": row.subject_id, "subject_label": row.subject_label, "source_table": row.source_table, "source_pk": row.source_pk, "audit_log_id": ( str(row.audit_log_id) if row.audit_log_id else None ), "occurred_at": ( row.occurred_at.isoformat() if row.occurred_at else None ), "payload": row.payload, }, sort_keys=True, default=str, ).encode("utf-8") ) digest.update(b"\x1e") return digest.hexdigest() __all__ = [ "DataGap", "EvidenceHandler", "EvidenceParam", "EvidenceQueryDefinition", "EvidenceResult", "EvidenceRow", "canonical_params", "content_hash", ]