diff --git a/.github/workflows/audit-evidence-review-quality.yml b/.github/workflows/audit-evidence-review-quality.yml new file mode 100644 index 000000000..12edd8f80 --- /dev/null +++ b/.github/workflows/audit-evidence-review-quality.yml @@ -0,0 +1,74 @@ +name: Audit Evidence Review Quality + +on: + pull_request: + branches: + - develop + paths: + - "packages/audit-evidence-review/**" + - ".github/requirements/foundation-test.txt" + - ".github/workflows/audit-evidence-review-quality.yml" + - "docs/adr/0102-governed-audit-evidence-review.md" + - "docs/doctoring/audit-evidence-review-references.md" + - "docs/traceability/audit-evidence-review.md" + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: audit-evidence-review-quality-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +jobs: + unit: + name: Audit evidence review contract and 100% coverage + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - name: Checkout exact candidate + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.event.pull_request.head.sha || github.sha }} + persist-credentials: false + - name: Prove exact candidate checkout + env: + ORGMETRA_EXPECTED_HEAD_SHA: ${{ github.event.pull_request.head.sha || github.sha }} + run: test "$(git rev-parse HEAD)" = "$ORGMETRA_EXPECTED_HEAD_SHA" + - name: Set up Python + uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.14" + check-latest: false + - name: Install reviewed test and build toolchain + run: | + python -m pip install --require-hashes --no-deps --only-binary=:all: -r .github/requirements/foundation-test.txt + printf '%s\n' 'setuptools==84.0.0 --hash=sha256:51a52592b3b99e102b609654876bd65f19f999935166d1352678931132b0c670' > /tmp/orgmetra-audit-evidence-review-build.txt + python -m pip install --require-hashes --no-deps --only-binary=:all: -r /tmp/orgmetra-audit-evidence-review-build.txt + python -m pip check + - name: Compile audit evidence review package + run: python -m compileall -q packages/audit-evidence-review/src packages/audit-evidence-review/tests + - name: Build and install exact package artifact + run: | + rm -rf /tmp/orgmetra-audit-evidence-review-build-tree /tmp/orgmetra-audit-evidence-review-dist /tmp/orgmetra-audit-evidence-review-venv + cp -a packages/audit-evidence-review /tmp/orgmetra-audit-evidence-review-build-tree + mkdir -p /tmp/orgmetra-audit-evidence-review-dist + python -m pip wheel --no-deps --no-build-isolation --wheel-dir /tmp/orgmetra-audit-evidence-review-dist /tmp/orgmetra-audit-evidence-review-build-tree + test "$(find /tmp/orgmetra-audit-evidence-review-dist -maxdepth 1 -type f -name '*.whl' | wc -l)" -eq 1 + python -m venv /tmp/orgmetra-audit-evidence-review-venv + /tmp/orgmetra-audit-evidence-review-venv/bin/python -m pip install --require-hashes --no-deps --only-binary=:all: -r "$GITHUB_WORKSPACE/.github/requirements/foundation-test.txt" + wheel_path="$(find /tmp/orgmetra-audit-evidence-review-dist -maxdepth 1 -type f -name '*.whl' -print -quit)" + wheel_sha="$(sha256sum "$wheel_path" | awk '{print $1}')" + printf 'orgmetra-audit-evidence-review[test] @ file://%s --hash=sha256:%s\n' "$wheel_path" "$wheel_sha" > /tmp/orgmetra-audit-evidence-review-install.txt + /tmp/orgmetra-audit-evidence-review-venv/bin/python -m pip install --require-hashes --no-deps -r /tmp/orgmetra-audit-evidence-review-install.txt + /tmp/orgmetra-audit-evidence-review-venv/bin/python -m pip check + - name: Test installed audit evidence review artifact with exact statement and branch coverage + env: + COVERAGE_FILE: /tmp/orgmetra-audit-evidence-review.coverage + run: | + cd /tmp + /tmp/orgmetra-audit-evidence-review-venv/bin/python -m pytest -c "$GITHUB_WORKSPACE/packages/audit-evidence-review/pyproject.toml" "$GITHUB_WORKSPACE/packages/audit-evidence-review/tests" + - name: Require clean checkout + run: | + git diff --exit-code + test -z "$(git status --porcelain)" diff --git a/docs/adr/0102-governed-audit-evidence-review.md b/docs/adr/0102-governed-audit-evidence-review.md new file mode 100644 index 000000000..9e23723d0 --- /dev/null +++ b/docs/adr/0102-governed-audit-evidence-review.md @@ -0,0 +1,31 @@ +# ADR 0102: Governed audit evidence review boundary + +Status: Proposed + +## Context + +Protected `develop` already persists a PII-minimized CloudEvents-compatible canonical audit envelope and SHA-256 digest in `audit_event_record` under forced tenant RLS. The repository did not yet define an executable read-side trust boundary for audit review. A caller could otherwise implement ad hoc reads that fetch evidence before authorization, fail to bind authorization to the exact tenant/request/purpose, return unbounded history, or trust persisted bytes without rechecking their digest, row identity, and PII-minimized envelope shape. + +NIST SP 800-53 Rev. 5 AU-6 calls for review and analysis of system audit records; the current NIST control catalog update reviewed for this decision is 5.2.0. CloudEvents v1.0.2 defines the stable event context contract used by Orgmetra's existing audit envelope. These sources motivate reviewability and interoperability; they do not make this package a certification or a complete security-monitoring system. + +## Decision + +Introduce an Orgmetra-owned, transport-neutral `audit-evidence-review` package with four explicit boundaries: + +1. `AuditEvidenceQuery` fixes tenant scope, pseudonymous request/requester correlations, the closed `audit_evidence_review` purpose, a maximum 90-day system-recorded interval, and a maximum 200-row page. +2. `AuditEvidenceReadAuthority` must be invoked before any audit-store read. Its authorization result must exactly match tenant, query reference, requester and purpose and must explicitly permit the read. +3. `AuditEvidenceRowReader` is the read-only storage protocol. The included `PostgresAuditEvidenceRowReader` reads only the existing Orgmetra `public.audit_event_record` relation. It enters a read-only transaction, proves `CURRENT_USER` is neither `SUPERUSER` nor `BYPASSRLS`, sets transaction-local `orgmetra.tenant_record_id`, and executes a static parameterized query over the authorized half-open system-recorded interval with deterministic ordering and the authorized limit. Deployment composition owns TLS, credentials, pooling and the least-privileged login role. +4. Every returned `PersistedAuditEvidenceRow` re-verifies valid bounded UTF-8, standard finite JSON values, exact canonical JSON bytes, SHA-256, the protected-main PII-minimized top-level/nested envelope key sets and value types, CloudEvents version/media type, persisted tenant/event identity, authorized time window, result count and deterministic ordering before evidence reaches the reviewer. + +The package returns the already PII-minimized canonical audit envelope and does not query application-table HR values or another service database. A privileged storage rewrite cannot make extra top-level HR fields or nested rating-like fields acceptable merely by recomputing a matching digest. The package does not grant or imply employment-decision authority. An empty page means the authorized bounded reader returned no rows; an adapter failure must remain a failure rather than being converted to an empty page. + +## Consequences + +The buyer gains an executable, testable control boundary for audit review without turning audit evidence into a shadow HR system of record. The PostgreSQL adapter makes least-privilege role state, tenant RLS binding, read-only transaction semantics, bounded parameterized SQL and stable ordering executable rather than leaving them only as host documentation. Host composition still retains responsibility for identity/policy resolution, credentials/TLS/pooling, observability and auditing the review itself. A customer-facing audit UI/API may be added later only if it preserves authorization-before-read and exact evidence verification. + +## Alternatives rejected + +- **Expose `canonical_event_json` directly from arbitrary SQL.** Rejected because authorization ordering, bounds and integrity verification would be host-specific and easy to omit. +- **Let a privileged or RLS-bypass database role perform audit review.** Rejected because it would make tenant isolation depend on application predicates alone rather than the existing forced-RLS defense in depth. +- **Copy HR fields into an audit-search index.** Rejected because it creates a second sensitive system of record and weakens minimization/retention boundaries. +- **Treat a successful database read as proof of evidence integrity.** Rejected because read-time digest, exact-envelope-shape and row-identity verification provide defense in depth against corruption or privileged storage tampering. diff --git a/docs/doctoring/audit-evidence-review-references.md b/docs/doctoring/audit-evidence-review-references.md new file mode 100644 index 000000000..c8fbb19e2 --- /dev/null +++ b/docs/doctoring/audit-evidence-review-references.md @@ -0,0 +1,15 @@ +# Audit evidence review references + +Reviewed 2026-08-24 against current authoritative sources. + +## References (APA 7) + +Cloud Native Computing Foundation. (2022). *CloudEvents specification (Version 1.0.2)*. https://github.com/cloudevents/spec/tree/v1.0.2 + +Joint Task Force. (2020). *Security and privacy controls for information systems and organizations (NIST Special Publication 800-53, Revision 5; updated release catalog 5.2.0)*. National Institute of Standards and Technology. https://doi.org/10.6028/NIST.SP.800-53r5 + +## Design use + +CloudEvents v1.0.2 remains the latest stable CloudEvents core release reviewed for the existing Orgmetra audit envelope. The package therefore rechecks `specversion = 1.0` and the existing JSON media-type contract rather than adopting a working draft. + +NIST SP 800-53 Rev. 5 AU-6 identifies audit-record review, analysis and reporting as an organizational control capability. Orgmetra uses that as a design input for purpose-bound reviewability and accountable access. The reference does not imply NIST certification, FedRAMP authorization, SOC 2 attestation, or that this bounded package implements the entire AU family. diff --git a/docs/traceability/audit-evidence-review.md b/docs/traceability/audit-evidence-review.md new file mode 100644 index 000000000..925e70bcc --- /dev/null +++ b/docs/traceability/audit-evidence-review.md @@ -0,0 +1,27 @@ +# Audit evidence review traceability + +## Protected-main truth + +`develop@9e3e4847510e1e612b48474ba42b177b8ed824df` already owns immutable, PII-minimized audit/outbox persistence: `AuditOutboxEvent` produces deterministic CloudEvents-compatible canonical JSON and SHA-256, and `database/migrations/0003_audit_outbox_persistence.sql` stores it in forced-RLS `audit_event_record` with append-only mutation guards. Protected main does not yet contain this review package or its PostgreSQL reader. + +## Active-PR truth + +PR #102 adds the transport-neutral `orgmetra-audit-evidence-review` boundary and its Orgmetra-owned PostgreSQL reader. The package requires authorization before store access; exact authorization scope; bounded system-recorded interval and result count; exact runtime trust primitives; standard finite JSON value types; SHA-256 and canonical-JSON revalidation; row/event/tenant binding; in-window tenant-only rows; and strict `(recorded_at, audit_event_record_id)` order. It contains no HR application-table query and no employment-decision authority. + +`PostgresAuditEvidenceRowReader` is the concrete read-only adapter for the existing `public.audit_event_record` relation. Before audit evidence is selected it opens `SET TRANSACTION READ ONLY`, verifies `CURRENT_USER` resolves to a PostgreSQL role with both `rolsuper=false` and `rolbypassrls=false`, and binds the exact tenant through transaction-local `orgmetra.tenant_record_id`. The evidence query is static and parameterized, applies the authorized half-open `recorded_at` interval and limit, and orders by `recorded_at, audit_event_record_id`. Deployment composition owns TLS, credentials, pooling and the actual least-privileged login; this adapter does not query another service database or an HR application table. + +The first contract head `592cd8f097cb1d45cd901b85f542b590f9ebb546` intentionally had no production package and is the test-first predecessor. Its workflows were queued before follow-up commits, so it is not claimed as terminal hosted RED evidence and no predecessor status is transferred. + +Fresh self-review then found a distinct privacy-integrity defect after the first implementation: canonical JSON plus a recomputed digest could carry additional top-level HR data or extra nested `data` fields because read-time verification checked the digest, CloudEvents version, event id and tenant but not the exact existing PII-minimized envelope shape. RED regression `1aa81b250669373ffd63c2b398925773c6cf967c` requires both widened shapes to fail closed and also requires malformed unencodable text to become a stable validation failure. Root repair `fb668be543d93c2bc7a1d0a930d5f889a916fd7b` binds read-time verification to the exact protected-main audit key sets and hashes a single verified UTF-8 byte sequence. + +A second runtime-integrity self-review found that Python frozen dataclasses can still be rewritten with `object.__setattr__` after successful construction. Without boundary revalidation, a once-valid query could have its limit widened, a once-valid authorization could have `permitted` rewritten to a truthy non-boolean, and a once-valid persisted row could have both canonical JSON and digest replaced after construction. RED regression `b9385dbe9c9501dd1748cca6261c4d4c118ac1dd` fixes those attack paths as acceptance tests. Root repair `5005c9f58c3f35c6a08b8134362478418d011bcc` reconstructs live query, authorization, and row fields through their governed constructors and returns detached verified row snapshots, so authorization/store access and returned evidence never rely on a merely once-valid mutable Python object. + +The PostgreSQL host-adapter slice began with regression head `df289f8506e170d290340c5d3dab32a31812e2ce`, which requires a public `PostgresAuditEvidenceRowReader`, read-only transaction entry, least-privileged role proof, transaction-local tenant binding, a static parameterized bounded audit query, query revalidation before opening a connection, and fail-closed unexpected row shape. Its initial hosted workflows queued before implementation, so no terminal RED is claimed. Root implementation began at `b7269c6b9fc091ab141c85e98c9c2c00580295bf`; exact-current-head evidence must be evaluated fresh after documentation and public-contract alignment. + +## Planned host surfaces + +A customer-facing audit UI/API may consume the verified review page through an application boundary. That surface is still planned and must not bypass `read_audit_evidence()` authorization ordering or the PostgreSQL reader's least-privilege/RLS contract. + +## Out of scope + +This PR does not create a second audit store, copy HR values into an index, query another service's application tables, mutate any dedicated-writer dependency repository, authorize employment decisions, provide SIEM correlation, or claim NIST/SOC 2 certification. diff --git a/packages/audit-evidence-review/CHANGELOG.md b/packages/audit-evidence-review/CHANGELOG.md new file mode 100644 index 000000000..1262e489f --- /dev/null +++ b/packages/audit-evidence-review/CHANGELOG.md @@ -0,0 +1,13 @@ +# Changelog + +## Unreleased + +- Add a purpose-bound audit evidence review request with a 90-day system-recorded window and 200-row maximum. +- Require authoritative exact-scope authorization before the audit-store reader is invoked. +- Add `PostgresAuditEvidenceRowReader` for the existing forced-RLS `public.audit_event_record`: it uses a read-only transaction, verifies the current PostgreSQL role is `NOSUPERUSER NOBYPASSRLS`, binds transaction-local tenant context, applies the authorized half-open recorded-time window and limit with parameterized SQL, and returns governed row snapshots only. +- Revalidate live query, authorization, and persisted-row fields into detached governed snapshots before authority/store use or evidence return, so post-construction `object.__setattr__` mutation cannot widen request bounds, forge permission, or bypass evidence checks. +- Re-verify persisted canonical JSON, SHA-256 digest, CloudEvents version/media type, tenant/event identity, review window, result count, and deterministic `(recorded_at, audit_event_record_id)` ordering before evidence is returned. +- Fail closed if stored canonical evidence is widened beyond the existing PII-minimized audit envelope/data key sets, even when a privileged rewrite recomputes a matching SHA-256 digest. +- Normalize unencodable canonical text to a stable validation failure before hashing or parsing. +- Keep HR application values, cross-service SQL, and employment-decision authority outside the review contract. +- Add exact statement and branch coverage over the installed wheel plus clean-checkout enforcement. diff --git a/packages/audit-evidence-review/README.md b/packages/audit-evidence-review/README.md new file mode 100644 index 000000000..2d24b8c91 --- /dev/null +++ b/packages/audit-evidence-review/README.md @@ -0,0 +1,29 @@ +# Orgmetra Audit Evidence Review + +This package is the purpose-bound review boundary for Orgmetra's existing immutable `audit_event_record` evidence. It makes one ordering guarantee explicit: **authorization happens before any audit-store read**. It then re-validates the live query and authorization objects, re-verifies persisted evidence into detached row snapshots, and only then returns evidence to an authorized reviewer. + +## What it does + +`AuditEvidenceQuery` carries only tenant scope, an opaque review correlation, a pseudonymous requester correlation, the closed purpose `audit_evidence_review`, a bounded system-recorded interval, and a maximum result count. The interval is capped at 90 days and a page at 200 rows. `read_audit_evidence()` reconstructs the live query through this governed constructor before calling either authority or storage, so a once-valid frozen dataclass cannot be widened later with `object.__setattr__`. + +`AuditEvidenceReadAuthority` is a host-owned protocol. A production host must resolve the authenticated actor and purpose through the authoritative policy/identity boundary and return an exact-scope `AuditEvidenceReadAuthorization`. Constructing that dataclass does **not** itself prove authorization. The returned object is also reconstructed through its governed constructor before permission or scope is trusted. + +`AuditEvidenceRowReader` is the read-only storage protocol. `PostgresAuditEvidenceRowReader` is the included Orgmetra PostgreSQL implementation for the existing audit store. It opens `SET TRANSACTION READ ONLY`, proves the current login is neither `SUPERUSER` nor `BYPASSRLS`, binds the exact `orgmetra.tenant_record_id` transaction-local setting, reads only `public.audit_event_record`, constrains `recorded_at` to the authorized half-open interval, orders by `(recorded_at, audit_event_record_id)`, and applies the authorized limit through bound parameters. Deployment composition still owns connection pooling, TLS, credentials, and selection of the least-privileged application role. The adapter does not query HR application tables or any other service database. + +`PersistedAuditEvidenceRow` performs defense-in-depth verification after the read: the canonical text must be valid UTF-8 within the byte budget; exact bytes must match the lower-case SHA-256 digest; those bytes must round-trip through Orgmetra's deterministic JSON form; the top-level and nested `data` keys must still match the existing PII-minimized audit envelope (with only the governed optional confirmation extension); CloudEvents `specversion`/media type must match the governed envelope; and event id and tenant in canonical evidence must match the persisted row columns. `read_audit_evidence()` reconstructs each live reader row and returns the reconstructed snapshot rather than the reader-owned object, so a row that was valid only at construction time cannot bypass current checks. A privileged rewrite cannot widen the envelope with extra HR fields merely by recomputing a matching digest. + +## Privacy and decision boundary + +The review result exposes only the canonical audit envelope already designed to be PII-minimized. It does not read Person names, contact details, compensation, ratings, candidate content, free-form feedback, credentials, or model output from application tables. Opaque actor/resource references remain correlation data and still require purpose-bound access control, retention, export control, and audit handling. + +This package grants **no** employment-decision authority and makes no decision. It does not make LLM output authoritative. High-impact employment decisions remain subject to their separate human-confirmation boundaries and immutable evidence contracts. + +## Failure behavior + +The boundary fails closed before returning evidence when authorization is absent, denied, malformed, scope-mismatched, or post-construction-mutated; when the reader returns a mutable/non-tuple collection, too many rows, a non-governed or post-construction-mutated row, cross-tenant evidence, out-of-window evidence, non-monotonic row order, invalid/unencodable or noncanonical JSON, a widened envelope/data shape, or a digest/identity mismatch. The PostgreSQL adapter additionally fails closed before tenant context or evidence access when the current database role is a superuser or has `BYPASSRLS`, and it revalidates a directly supplied query before opening a connection. + +A storage adapter failure propagates to the host; callers must not reinterpret missing evidence as an empty successful review. + +## Current status + +This package, including `PostgresAuditEvidenceRowReader`, is **active-PR evidence**, not protected-main shipped truth until its pull request is integrated. A customer-facing audit UI/API remains a follow-on host surface and must preserve authorization-before-read, tenant RLS, bounded evidence review, and exact evidence verification. diff --git a/packages/audit-evidence-review/pyproject.toml b/packages/audit-evidence-review/pyproject.toml new file mode 100644 index 000000000..462ef783b --- /dev/null +++ b/packages/audit-evidence-review/pyproject.toml @@ -0,0 +1,24 @@ +[build-system] +requires = ["setuptools>=69"] +build-backend = "setuptools.build_meta" + +[project] +name = "orgmetra-audit-evidence-review" +version = "0.1.0" +description = "Purpose-bound immutable audit-evidence review contracts for Orgmetra." +requires-python = ">=3.12" + +[project.optional-dependencies] +test = ["pytest>=8.3", "pytest-cov>=5.0"] + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = [ + "--cov=orgmetra_audit_evidence_review", + "--cov-branch", + "--cov-report=term-missing", + "--cov-fail-under=100", +] diff --git a/packages/audit-evidence-review/src/orgmetra_audit_evidence_review/__init__.py b/packages/audit-evidence-review/src/orgmetra_audit_evidence_review/__init__.py new file mode 100644 index 000000000..58f7d3d70 --- /dev/null +++ b/packages/audit-evidence-review/src/orgmetra_audit_evidence_review/__init__.py @@ -0,0 +1,23 @@ +"""Public audit evidence review contracts.""" + +from .postgres import PostgresAuditEvidenceRowReader +from .review import ( + AuditEvidenceQuery, + AuditEvidenceReadAuthorization, + AuditEvidenceReadAuthority, + AuditEvidenceReviewPage, + AuditEvidenceRowReader, + PersistedAuditEvidenceRow, + read_audit_evidence, +) + +__all__ = [ + "AuditEvidenceQuery", + "AuditEvidenceReadAuthorization", + "AuditEvidenceReadAuthority", + "AuditEvidenceReviewPage", + "AuditEvidenceRowReader", + "PersistedAuditEvidenceRow", + "PostgresAuditEvidenceRowReader", + "read_audit_evidence", +] diff --git a/packages/audit-evidence-review/src/orgmetra_audit_evidence_review/postgres.py b/packages/audit-evidence-review/src/orgmetra_audit_evidence_review/postgres.py new file mode 100644 index 000000000..51f44a2b2 --- /dev/null +++ b/packages/audit-evidence-review/src/orgmetra_audit_evidence_review/postgres.py @@ -0,0 +1,116 @@ +"""Least-privileged PostgreSQL reader for governed Orgmetra audit evidence. + +The adapter reads only Orgmetra's immutable ``public.audit_event_record`` relation. +It opens a read-only transaction, proves the current database role is neither a +superuser nor an RLS-bypass role, binds the exact tenant into PostgreSQL's +transaction-local RLS context, and issues one bounded parameterized query. It +never reads HR application tables or another service's database. +""" + +from __future__ import annotations + +from contextlib import AbstractContextManager +from dataclasses import dataclass +from typing import Any, Callable + +from .review import AuditEvidenceQuery, PersistedAuditEvidenceRow + +PostgresConnectionFactory = Callable[[], AbstractContextManager[Any]] + +_READ_ONLY_SQL = "SET TRANSACTION READ ONLY" +_ROLE_GUARD_SQL = """ +SELECT 1 +FROM pg_catalog.pg_roles +WHERE rolname = CURRENT_USER + AND rolsuper IS FALSE + AND rolbypassrls IS FALSE +""".strip() +_TENANT_CONTEXT_SQL = "SELECT pg_catalog.set_config('orgmetra.tenant_record_id', %s, true)" +_AUDIT_EVIDENCE_SQL = """ +SELECT + tenant_record_id, + audit_event_record_id, + canonical_event_json, + event_envelope_digest, + recorded_at +FROM public.audit_event_record +WHERE tenant_record_id = %s + AND recorded_at >= %s + AND recorded_at < %s +ORDER BY recorded_at ASC, audit_event_record_id ASC +LIMIT %s +""".strip() + + +@dataclass(frozen=True, slots=True) +class PostgresAuditEvidenceRowReader: + """Read bounded audit evidence through forced RLS and a least-privileged role. + + ``connection_factory`` must return a DB-API-compatible connection context + manager. Deployment composition owns pooling, TLS, credentials, and the + actual ``NOSUPERUSER NOBYPASSRLS`` login role; this adapter verifies that + role property again before it establishes tenant context or reads evidence. + """ + + connection_factory: PostgresConnectionFactory + + def __post_init__(self) -> None: + """Reject an unusable connection factory before any audit read is attempted.""" + if not callable(self.connection_factory): + raise TypeError("connection_factory must be callable") + + def read_rows(self, query: AuditEvidenceQuery) -> tuple[PersistedAuditEvidenceRow, ...]: + """Return one ordered, tenant-bound page of revalidated immutable audit rows.""" + if type(query) is not AuditEvidenceQuery: + raise TypeError("query must be an exact AuditEvidenceQuery.") + verified_query = AuditEvidenceQuery( + tenant_record_id=query.tenant_record_id, + query_reference=query.query_reference, + requester_reference=query.requester_reference, + purpose_code=query.purpose_code, + recorded_from=query.recorded_from, + recorded_before=query.recorded_before, + limit=query.limit, + ) + + with self.connection_factory() as connection: + with connection.cursor() as cursor: + cursor.execute(_READ_ONLY_SQL) + cursor.execute(_ROLE_GUARD_SQL) + if cursor.fetchone() is None: + raise PermissionError( + "audit evidence reads require a NOSUPERUSER NOBYPASSRLS database role." + ) + cursor.execute(_TENANT_CONTEXT_SQL, (str(verified_query.tenant_record_id),)) + cursor.execute( + _AUDIT_EVIDENCE_SQL, + ( + verified_query.tenant_record_id, + verified_query.recorded_from, + verified_query.recorded_before, + verified_query.limit, + ), + ) + rows = cursor.fetchmany(verified_query.limit) + + verified_rows: list[PersistedAuditEvidenceRow] = [] + for row in rows: + if not isinstance(row, tuple) or len(row) != 5: + raise ValueError("database row must contain exactly five audit evidence columns.") + ( + tenant_record_id, + audit_event_record_id, + canonical_event_json, + event_envelope_digest, + recorded_at, + ) = row + verified_rows.append( + PersistedAuditEvidenceRow( + tenant_record_id=tenant_record_id, + audit_event_record_id=audit_event_record_id, + canonical_event_json=canonical_event_json, + event_envelope_digest=event_envelope_digest, + recorded_at=recorded_at, + ) + ) + return tuple(verified_rows) diff --git a/packages/audit-evidence-review/src/orgmetra_audit_evidence_review/review.py b/packages/audit-evidence-review/src/orgmetra_audit_evidence_review/review.py new file mode 100644 index 000000000..e917a7b4c --- /dev/null +++ b/packages/audit-evidence-review/src/orgmetra_audit_evidence_review/review.py @@ -0,0 +1,334 @@ +"""Purpose-bound integrity verification for Orgmetra audit evidence review.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from hashlib import sha256 +import json +import re +from typing import NoReturn, Protocol +from uuid import UUID + +_MAX_UUID_INT = (1 << 128) - 1 +_REFERENCE_PATTERN = re.compile( + r"^(?P[a-z][a-z0-9_]*):(?P[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})$" +) +_DIGEST_PATTERN = re.compile(r"^[0-9a-f]{64}$") +_PURPOSE = "audit_evidence_review" +_MAX_WINDOW = timedelta(days=90) +_MAX_LIMIT = 200 +_MAX_CANONICAL_BYTES = 32768 +_BASE_EVENT_KEYS = frozenset( + { + "data", + "datacontenttype", + "id", + "orgmetraactor", + "orgmetraevidence", + "orgmetrapurpose", + "orgmetrareason", + "orgmetratenant", + "source", + "specversion", + "subject", + "time", + "type", + } +) +_CONFIRMATION_KEY = "orgmetraconfirmation" +_DATA_KEYS = frozenset({"high_impact", "result_code"}) +_EVENT_STRING_KEYS = frozenset( + { + "datacontenttype", + "id", + "orgmetraactor", + "orgmetraevidence", + "orgmetrapurpose", + "orgmetrareason", + "orgmetratenant", + "source", + "specversion", + "subject", + "time", + "type", + } +) + + +def _validate_operational_uuid(name: str, value: UUID) -> None: + """Require a canonical non-sentinel UUID owned by an authoritative Orgmetra boundary.""" + if type(value) is not UUID or value.int in (0, _MAX_UUID_INT): + raise ValueError(f"{name} must be a non-sentinel UUID.") + + +def _validate_reference(name: str, value: str, namespace: str) -> None: + """Require an exact built-in string carrying one packet-owned opaque UUIDv4 reference.""" + if type(value) is not str or len(value) > 96: + raise ValueError(f"{name} must be a bounded string.") + match = _REFERENCE_PATTERN.fullmatch(value) + if match is None or match.group("namespace") != namespace: + raise ValueError(f"{name} must use the {namespace}: UUIDv4 namespace.") + parsed = UUID(match.group("uuid")) + if str(parsed) != match.group("uuid") or parsed.version != 4: + raise ValueError(f"{name} must use a canonical UUIDv4 suffix.") + + +def _freeze_timestamp(name: str, value: datetime) -> datetime: + """Detach a caller-owned timezone provider and return a built-in UTC instant.""" + if type(value) is not datetime or value.tzinfo is None: + raise ValueError(f"{name} must be an aware datetime.") + try: + offset = value.utcoffset() + except Exception as error: # noqa: BLE001 - caller timezone code is untrusted. + raise ValueError(f"{name} timezone offset could not be resolved.") from error + if type(offset) is not timedelta: + raise ValueError(f"{name} must resolve to a concrete UTC offset.") + try: + return (value.replace(tzinfo=None) - offset).replace(tzinfo=timezone.utc) + except (OverflowError, ValueError) as error: + raise ValueError(f"{name} cannot be normalized to UTC.") from error + + +def _reject_nonfinite_json_number(value: str) -> NoReturn: + """Reject JSON extensions that encode non-finite numbers instead of valid JSON.""" + raise ValueError(f"non-finite JSON number {value} is not permitted") + + +@dataclass(frozen=True, slots=True) +class AuditEvidenceQuery: + """One bounded audit-review request that carries no HR application values.""" + + tenant_record_id: UUID + query_reference: str + requester_reference: str + purpose_code: str + recorded_from: datetime + recorded_before: datetime + limit: int = 100 + + def __post_init__(self) -> None: + """Validate scope and freeze the requested system-recorded interval to UTC.""" + _validate_operational_uuid("tenant_record_id", self.tenant_record_id) + _validate_reference("query_reference", self.query_reference, "audit_review") + _validate_reference("requester_reference", self.requester_reference, "actor") + if type(self.purpose_code) is not str or self.purpose_code != _PURPOSE: + raise ValueError("purpose_code must be audit_evidence_review.") + frozen_from = _freeze_timestamp("recorded_from", self.recorded_from) + frozen_before = _freeze_timestamp("recorded_before", self.recorded_before) + object.__setattr__(self, "recorded_from", frozen_from) + object.__setattr__(self, "recorded_before", frozen_before) + if frozen_before <= frozen_from: + raise ValueError("recorded_before must be later than recorded_from.") + if frozen_before - frozen_from > _MAX_WINDOW: + raise ValueError("audit review window must not exceed 90 days.") + if type(self.limit) is not int or not 1 <= self.limit <= _MAX_LIMIT: + raise ValueError("limit must be an integer from 1 through 200.") + + +@dataclass(frozen=True, slots=True) +class AuditEvidenceReadAuthorization: + """Authoritative decision proving one requester may read one exact audit query.""" + + tenant_record_id: UUID + query_reference: str + requester_reference: str + purpose_code: str + permitted: bool + + def __post_init__(self) -> None: + """Validate authorization evidence without granting authority by construction.""" + _validate_operational_uuid("tenant_record_id", self.tenant_record_id) + _validate_reference("query_reference", self.query_reference, "audit_review") + _validate_reference("requester_reference", self.requester_reference, "actor") + if type(self.purpose_code) is not str or self.purpose_code != _PURPOSE: + raise ValueError("purpose_code must be audit_evidence_review.") + if type(self.permitted) is not bool: + raise ValueError("permitted must be a boolean.") + + +@dataclass(frozen=True, slots=True) +class PersistedAuditEvidenceRow: + """One immutable audit-store row reverified before review use.""" + + tenant_record_id: UUID + audit_event_record_id: UUID + canonical_event_json: str + event_envelope_digest: str + recorded_at: datetime + + def __post_init__(self) -> None: + """Verify row identity, exact governed envelope bytes, digest, and recorded time.""" + _validate_operational_uuid("tenant_record_id", self.tenant_record_id) + _validate_operational_uuid("audit_event_record_id", self.audit_event_record_id) + if type(self.canonical_event_json) is not str: + raise ValueError("canonical_event_json must be a string.") + try: + canonical_bytes = self.canonical_event_json.encode("utf-8") + except UnicodeEncodeError as error: + raise ValueError("canonical_event_json must be valid UTF-8 text.") from error + if len(canonical_bytes) > _MAX_CANONICAL_BYTES: + raise ValueError("canonical_event_json exceeds the 32768-byte review budget.") + if type(self.event_envelope_digest) is not str or _DIGEST_PATTERN.fullmatch( + self.event_envelope_digest + ) is None: + raise ValueError("event_envelope_digest must be a lower-case SHA-256 digest.") + if sha256(canonical_bytes).hexdigest() != self.event_envelope_digest: + raise ValueError("audit event digest does not match persisted canonical bytes.") + try: + document = json.loads( + self.canonical_event_json, + parse_constant=_reject_nonfinite_json_number, + ) + except (ValueError, UnicodeError) as error: + raise ValueError("canonical_event_json must contain valid UTF-8 JSON.") from error + if type(document) is not dict: + raise ValueError("canonical_event_json must contain one JSON object.") + if ( + json.dumps(document, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + != self.canonical_event_json + ): + raise ValueError("canonical_event_json is not in canonical Orgmetra JSON form.") + event_keys = frozenset(document) + if event_keys not in (_BASE_EVENT_KEYS, _BASE_EVENT_KEYS | {_CONFIRMATION_KEY}): + raise ValueError("canonical_event_json does not match the governed audit envelope shape.") + event_data = document.get("data") + if type(event_data) is not dict or frozenset(event_data) != _DATA_KEYS: + raise ValueError("canonical_event_json does not match the governed audit data shape.") + if any(type(document.get(key)) is not str for key in _EVENT_STRING_KEYS): + raise ValueError("canonical_event_json does not match the governed audit value types.") + if type(event_data["result_code"]) is not str or type(event_data["high_impact"]) is not bool: + raise ValueError("canonical_event_json does not match the governed audit value types.") + if _CONFIRMATION_KEY in document and type(document[_CONFIRMATION_KEY]) is not str: + raise ValueError("canonical_event_json does not match the governed audit value types.") + if document.get("specversion") != "1.0" or document.get("datacontenttype") != "application/json": + raise ValueError("audit event must use the governed CloudEvents 1.0 JSON contract.") + if document.get("id") != str(self.audit_event_record_id): + raise ValueError("audit event id does not match the persisted row identity.") + if document.get("orgmetratenant") != str(self.tenant_record_id): + raise ValueError("audit event tenant does not match the persisted row scope.") + object.__setattr__(self, "recorded_at", _freeze_timestamp("recorded_at", self.recorded_at)) + + +@dataclass(frozen=True, slots=True) +class AuditEvidenceReviewPage: + """Verified evidence returned for one exact authorized review query.""" + + query_reference: str + records: tuple[PersistedAuditEvidenceRow, ...] + + def __post_init__(self) -> None: + """Keep review output immutable and bound to a canonical query reference.""" + _validate_reference("query_reference", self.query_reference, "audit_review") + if type(self.records) is not tuple: + raise ValueError("records must be an immutable tuple.") + if any(type(row) is not PersistedAuditEvidenceRow for row in self.records): + raise ValueError("records must contain exact persisted audit evidence rows.") + + +class AuditEvidenceReadAuthority(Protocol): + """Host authorization boundary invoked before any audit-store read.""" + + def authorize(self, query: AuditEvidenceQuery) -> AuditEvidenceReadAuthorization: + """Return an exact-scope authorization decision for the review query.""" + + +class AuditEvidenceRowReader(Protocol): + """Read-only adapter for the existing Orgmetra immutable audit store.""" + + def read_rows(self, query: AuditEvidenceQuery) -> tuple[PersistedAuditEvidenceRow, ...]: + """Return rows already constrained by tenant, time window, order, and limit.""" + + +def _snapshot_query(query: AuditEvidenceQuery) -> AuditEvidenceQuery: + """Revalidate live query fields and return a detached governed snapshot.""" + return AuditEvidenceQuery( + tenant_record_id=query.tenant_record_id, + query_reference=query.query_reference, + requester_reference=query.requester_reference, + purpose_code=query.purpose_code, + recorded_from=query.recorded_from, + recorded_before=query.recorded_before, + limit=query.limit, + ) + + +def _snapshot_authorization( + authorization: AuditEvidenceReadAuthorization, +) -> AuditEvidenceReadAuthorization: + """Revalidate authority output so post-construction mutation cannot widen permission.""" + return AuditEvidenceReadAuthorization( + tenant_record_id=authorization.tenant_record_id, + query_reference=authorization.query_reference, + requester_reference=authorization.requester_reference, + purpose_code=authorization.purpose_code, + permitted=authorization.permitted, + ) + + +def _snapshot_row(row: PersistedAuditEvidenceRow) -> PersistedAuditEvidenceRow: + """Reverify current persisted-row fields and detach evidence from the reader object.""" + return PersistedAuditEvidenceRow( + tenant_record_id=row.tenant_record_id, + audit_event_record_id=row.audit_event_record_id, + canonical_event_json=row.canonical_event_json, + event_envelope_digest=row.event_envelope_digest, + recorded_at=row.recorded_at, + ) + + +def read_audit_evidence( + *, + query: AuditEvidenceQuery, + authority: AuditEvidenceReadAuthority, + reader: AuditEvidenceRowReader, +) -> AuditEvidenceReviewPage: + """Authorize first, then verify a bounded ordered page from the immutable audit store.""" + if type(query) is not AuditEvidenceQuery: + raise TypeError("query must be an exact AuditEvidenceQuery.") + verified_query = _snapshot_query(query) + authorization = authority.authorize(_snapshot_query(verified_query)) + if type(authorization) is not AuditEvidenceReadAuthorization: + raise TypeError("authority must return AuditEvidenceReadAuthorization.") + verified_authorization = _snapshot_authorization(authorization) + expected_scope = ( + verified_query.tenant_record_id, + verified_query.query_reference, + verified_query.requester_reference, + verified_query.purpose_code, + ) + actual_scope = ( + verified_authorization.tenant_record_id, + verified_authorization.query_reference, + verified_authorization.requester_reference, + verified_authorization.purpose_code, + ) + if actual_scope != expected_scope or not verified_authorization.permitted: + raise PermissionError("audit evidence review is not authorized for the exact query scope.") + + rows = reader.read_rows(_snapshot_query(verified_query)) + if type(rows) is not tuple: + raise TypeError("reader must return an immutable tuple.") + if len(rows) > verified_query.limit: + raise ValueError("reader returned more audit rows than the authorized limit.") + + previous_key: tuple[datetime, int] | None = None + verified_rows: list[PersistedAuditEvidenceRow] = [] + for row in rows: + if type(row) is not PersistedAuditEvidenceRow: + raise TypeError("reader returned an ungoverned audit evidence row.") + verified_row = _snapshot_row(row) + if verified_row.tenant_record_id != verified_query.tenant_record_id: + raise PermissionError("reader returned cross-tenant audit evidence.") + if not verified_query.recorded_from <= verified_row.recorded_at < verified_query.recorded_before: + raise ValueError("reader returned audit evidence outside the authorized time window.") + key = (verified_row.recorded_at, verified_row.audit_event_record_id.int) + if previous_key is not None and key <= previous_key: + raise ValueError("reader must return audit evidence in strict recorded-time/id order.") + previous_key = key + verified_rows.append(verified_row) + + return AuditEvidenceReviewPage( + query_reference=verified_query.query_reference, + records=tuple(verified_rows), + ) diff --git a/packages/audit-evidence-review/tests/test_postgres_reader.py b/packages/audit-evidence-review/tests/test_postgres_reader.py new file mode 100644 index 000000000..f0a50f349 --- /dev/null +++ b/packages/audit-evidence-review/tests/test_postgres_reader.py @@ -0,0 +1,202 @@ +"""Executable contract for the least-privileged PostgreSQL audit reader.""" + +from __future__ import annotations + +from datetime import datetime, timezone +from hashlib import sha256 +import json +import unittest +from uuid import UUID + +from orgmetra_audit_evidence_review import ( + AuditEvidenceQuery, + PostgresAuditEvidenceRowReader, +) + +TENANT = UUID("10000000-0000-7000-8000-000000000001") +EVENT = UUID("11111111-1111-4111-8111-111111111111") +QUERY = "audit_review:22222222-2222-4222-8222-222222222222" +REQUESTER = "actor:33333333-3333-4333-8333-333333333333" +FROM = datetime(2026, 8, 1, tzinfo=timezone.utc) +BEFORE = datetime(2026, 9, 1, tzinfo=timezone.utc) +RECORDED = datetime(2026, 8, 20, 12, tzinfo=timezone.utc) + + +def _query() -> AuditEvidenceQuery: + """Return one bounded tenant-scoped audit query.""" + return AuditEvidenceQuery( + tenant_record_id=TENANT, + query_reference=QUERY, + requester_reference=REQUESTER, + purpose_code="audit_evidence_review", + recorded_from=FROM, + recorded_before=BEFORE, + limit=50, + ) + + +def _canonical_event() -> str: + """Return one canonical PII-minimized persisted audit event.""" + document = { + "data": {"high_impact": False, "result_code": "updated"}, + "datacontenttype": "application/json", + "id": str(EVENT), + "orgmetraactor": REQUESTER, + "orgmetraevidence": "v1", + "orgmetrapurpose": "people_record_update", + "orgmetrareason": "authorized_change", + "orgmetratenant": str(TENANT), + "source": "urn:orgmetra:people_api", + "specversion": "1.0", + "subject": "person:44444444-4444-4444-8444-444444444444", + "time": "2026-08-20T11:59:00Z", + "type": "orgmetra.people.updated", + } + return json.dumps(document, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + +def _row() -> tuple[object, ...]: + """Return one driver-shaped persisted audit row.""" + canonical = _canonical_event() + return ( + TENANT, + EVENT, + canonical, + sha256(canonical.encode("utf-8")).hexdigest(), + RECORDED, + ) + + +class _Cursor: + """Capture SQL while exposing deterministic role and audit rows.""" + + def __init__(self, *, role_allowed: bool = True, rows: list[tuple[object, ...]] | None = None) -> None: + self.role_allowed = role_allowed + self.rows = list(rows or []) + self.executions: list[tuple[str, tuple[object, ...] | None]] = [] + self.fetchmany_sizes: list[int] = [] + + def __enter__(self) -> _Cursor: + return self + + def __exit__(self, exc_type: object, exc_value: object, traceback: object) -> None: + return None + + def execute(self, sql: str, parameters: tuple[object, ...] | None = None) -> None: + """Record one parameterized statement.""" + self.executions.append((sql, parameters)) + + def fetchone(self) -> tuple[int] | None: + """Return a row only for a NOSUPERUSER/NOBYPASSRLS role.""" + return (1,) if self.role_allowed else None + + def fetchmany(self, size: int) -> list[tuple[object, ...]]: + """Return at most the requested bounded number of audit rows.""" + self.fetchmany_sizes.append(size) + return self.rows[:size] + + +class _Connection: + """Provide one transaction-scoped cursor.""" + + def __init__(self, cursor: _Cursor) -> None: + self.cursor_instance = cursor + self.enter_count = 0 + self.exit_count = 0 + + def __enter__(self) -> _Connection: + self.enter_count += 1 + return self + + def __exit__(self, exc_type: object, exc_value: object, traceback: object) -> None: + self.exit_count += 1 + return None + + def cursor(self) -> _Cursor: + """Return the deterministic cursor.""" + return self.cursor_instance + + +class PostgresAuditEvidenceRowReaderTests(unittest.TestCase): + """Prove read-only role enforcement, tenant RLS binding, and bounded reads.""" + + def _reader( + self, + *, + role_allowed: bool = True, + rows: list[tuple[object, ...]] | None = None, + ) -> tuple[PostgresAuditEvidenceRowReader, _Connection, _Cursor]: + cursor = _Cursor(role_allowed=role_allowed, rows=rows) + connection = _Connection(cursor) + return PostgresAuditEvidenceRowReader(lambda: connection), connection, cursor + + def test_reads_verified_rows_inside_read_only_tenant_bound_transaction(self) -> None: + reader, connection, cursor = self._reader(rows=[_row()]) + + rows = reader.read_rows(_query()) + + self.assertEqual((connection.enter_count, connection.exit_count), (1, 1)) + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0].tenant_record_id, TENANT) + self.assertEqual(rows[0].audit_event_record_id, EVENT) + self.assertEqual(cursor.executions[0], ("SET TRANSACTION READ ONLY", None)) + self.assertIn("pg_catalog.pg_roles", cursor.executions[1][0]) + self.assertEqual(cursor.executions[1][1], None) + self.assertEqual( + cursor.executions[2], + ( + "SELECT pg_catalog.set_config('orgmetra.tenant_record_id', %s, true)", + (str(TENANT),), + ), + ) + audit_sql, audit_parameters = cursor.executions[3] + self.assertIn("public.audit_event_record", audit_sql) + self.assertIn("ORDER BY recorded_at ASC, audit_event_record_id ASC", audit_sql) + self.assertNotIn(str(TENANT), audit_sql) + self.assertEqual(audit_parameters, (TENANT, FROM, BEFORE, 50)) + self.assertEqual(cursor.fetchmany_sizes, [50]) + + def test_privileged_role_fails_closed_before_tenant_context_or_audit_read(self) -> None: + reader, _, cursor = self._reader(role_allowed=False, rows=[_row()]) + + with self.assertRaisesRegex(PermissionError, "NOSUPERUSER NOBYPASSRLS"): + reader.read_rows(_query()) + + self.assertEqual(len(cursor.executions), 2) + self.assertEqual(cursor.fetchmany_sizes, []) + + def test_mutated_query_is_revalidated_before_opening_database_connection(self) -> None: + calls = 0 + + def connection_factory() -> _Connection: + nonlocal calls + calls += 1 + return _Connection(_Cursor()) + + query = _query() + object.__setattr__(query, "limit", 999) + reader = PostgresAuditEvidenceRowReader(connection_factory) + + with self.assertRaisesRegex(ValueError, "limit must be an integer"): + reader.read_rows(query) + + self.assertEqual(calls, 0) + + def test_unexpected_driver_row_shape_fails_closed(self) -> None: + reader, _, _ = self._reader(rows=[(TENANT, EVENT)]) + + with self.assertRaisesRegex(ValueError, "five audit evidence columns"): + reader.read_rows(_query()) + + def test_factory_and_query_runtime_types_are_rejected_before_database_use(self) -> None: + with self.assertRaisesRegex(TypeError, "connection_factory must be callable"): + PostgresAuditEvidenceRowReader(None) # type: ignore[arg-type] + + reader, connection, _ = self._reader() + with self.assertRaisesRegex(TypeError, "exact AuditEvidenceQuery"): + reader.read_rows(object()) # type: ignore[arg-type] + self.assertEqual(connection.enter_count, 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/packages/audit-evidence-review/tests/test_privacy_integrity.py b/packages/audit-evidence-review/tests/test_privacy_integrity.py new file mode 100644 index 000000000..33b188852 --- /dev/null +++ b/packages/audit-evidence-review/tests/test_privacy_integrity.py @@ -0,0 +1,146 @@ +"""Privacy regressions for read-time audit-envelope integrity verification.""" + +from __future__ import annotations + +from datetime import datetime, timezone +from hashlib import sha256 +import json +from uuid import UUID + +import pytest + +from orgmetra_audit_evidence_review import PersistedAuditEvidenceRow + +TENANT = UUID("10000000-0000-7000-8000-000000000001") +EVENT = UUID("11111111-1111-4111-8111-111111111111") +RECORDED = datetime(2026, 8, 20, 12, tzinfo=timezone.utc) + + +def _document() -> dict[str, object]: + """Return the exact existing PII-minimized audit envelope shape.""" + return { + "data": {"high_impact": False, "result_code": "updated"}, + "datacontenttype": "application/json", + "id": str(EVENT), + "orgmetraactor": "actor:bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb", + "orgmetraevidence": "v1", + "orgmetrapurpose": "people_record_update", + "orgmetrareason": "authorized_change", + "orgmetratenant": str(TENANT), + "source": "urn:orgmetra:people_api", + "specversion": "1.0", + "subject": "person:cccccccc-cccc-4ccc-8ccc-cccccccccccc", + "time": "2026-08-20T11:59:00Z", + "type": "orgmetra.people.updated", + } + + +def _row(canonical: str) -> PersistedAuditEvidenceRow: + """Build a row whose digest matches the supplied bytes exactly.""" + return PersistedAuditEvidenceRow( + tenant_record_id=TENANT, + audit_event_record_id=EVENT, + canonical_event_json=canonical, + event_envelope_digest=sha256(canonical.encode("utf-8")).hexdigest(), + recorded_at=RECORDED, + ) + + +def test_read_time_verification_accepts_governed_optional_confirmation_extension() -> None: + """High-impact evidence may carry the single existing confirmation extension.""" + document = _document() + document["data"] = {"high_impact": True, "result_code": "confirmed"} + document["orgmetraconfirmation"] = "confirmation:dddddddd-dddd-4ddd-8ddd-dddddddddddd" + canonical = json.dumps(document, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + assert _row(canonical).canonical_event_json == canonical + + +def test_read_time_verification_rejects_extra_top_level_hr_payload() -> None: + """A recomputed digest cannot legitimize a widened envelope carrying HR payload.""" + document = _document() + document["employee_name"] = "should-never-enter-audit-envelope" + canonical = json.dumps(document, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + with pytest.raises(ValueError, match="governed audit envelope shape"): + _row(canonical) + + +def test_read_time_verification_rejects_extra_data_payload() -> None: + """The nested data object stays limited to result and high-impact metadata.""" + document = _document() + data = document["data"] + assert isinstance(data, dict) + data["rating"] = 5 + canonical = json.dumps(document, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + with pytest.raises(ValueError, match="governed audit data shape"): + _row(canonical) + + +def test_read_time_verification_rejects_non_object_data_payload() -> None: + """The audit `data` member must remain the governed metadata object, not another JSON type.""" + document = _document() + document["data"] = [] + canonical = json.dumps(document, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + with pytest.raises(ValueError, match="governed audit data shape"): + _row(canonical) + + +@pytest.mark.parametrize( + "data", + [ + {"high_impact": 1, "result_code": "updated"}, + {"high_impact": False, "result_code": 7}, + ], +) +def test_read_time_verification_rejects_noncanonical_data_value_types( + data: dict[str, object], +) -> None: + """The governed audit data values remain an exact boolean and result-code string.""" + document = _document() + document["data"] = data + canonical = json.dumps(document, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + with pytest.raises(ValueError, match="governed audit value types"): + _row(canonical) + + +def test_read_time_verification_rejects_nonfinite_json_numbers() -> None: + """JSON NaN and Infinity extensions cannot enter canonical evidence.""" + document = _document() + document["data"] = {"high_impact": False, "result_code": float("nan")} + canonical = json.dumps(document, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + with pytest.raises(ValueError, match="valid UTF-8 JSON"): + _row(canonical) + + +@pytest.mark.parametrize( + ("member", "value"), + [("source", 7), ("orgmetraconfirmation", 7)], +) +def test_read_time_verification_rejects_noncanonical_event_value_types( + member: str, value: object, +) -> None: + """CloudEvents members and optional confirmation remain exact strings.""" + document = _document() + document[member] = value + canonical = json.dumps(document, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + with pytest.raises(ValueError, match="governed audit value types"): + _row(canonical) + + +def test_unencodable_text_fails_closed_as_validation_error() -> None: + """Malformed caller text does not escape as an implementation encoding exception.""" + malformed = '{"bad":"' + chr(0xD800) + '"}' + with pytest.raises(ValueError, match="valid UTF-8 text"): + PersistedAuditEvidenceRow( + tenant_record_id=TENANT, + audit_event_record_id=EVENT, + canonical_event_json=malformed, + event_envelope_digest="0" * 64, + recorded_at=RECORDED, + ) diff --git a/packages/audit-evidence-review/tests/test_review.py b/packages/audit-evidence-review/tests/test_review.py new file mode 100644 index 000000000..85744622a --- /dev/null +++ b/packages/audit-evidence-review/tests/test_review.py @@ -0,0 +1,105 @@ +"""Executable contract for purpose-bound immutable audit evidence review.""" + +from __future__ import annotations + +from datetime import datetime, timezone +from hashlib import sha256 +import json +from uuid import UUID + +from orgmetra_audit_evidence_review import ( + AuditEvidenceQuery, + AuditEvidenceReadAuthorization, + PersistedAuditEvidenceRow, + read_audit_evidence, +) + +TENANT = UUID("10000000-0000-7000-8000-000000000001") +EVENT = UUID("11111111-1111-4111-8111-111111111111") +QUERY = "audit_review:22222222-2222-4222-8222-222222222222" +REQUESTER = "actor:33333333-3333-4333-8333-333333333333" +FROM = datetime(2026, 8, 1, tzinfo=timezone.utc) +BEFORE = datetime(2026, 9, 1, tzinfo=timezone.utc) +RECORDED = datetime(2026, 8, 20, 12, tzinfo=timezone.utc) + + +def _canonical_event() -> str: + """Return one canonical PII-minimized audit envelope from the existing store contract.""" + document = { + "data": {"high_impact": False, "result_code": "updated"}, + "datacontenttype": "application/json", + "id": str(EVENT), + "orgmetraactor": REQUESTER, + "orgmetraevidence": "v1", + "orgmetrapurpose": "people_record_update", + "orgmetrareason": "authorized_change", + "orgmetratenant": str(TENANT), + "source": "urn:orgmetra:people_api", + "specversion": "1.0", + "subject": "person:44444444-4444-4444-8444-444444444444", + "time": "2026-08-20T11:59:00Z", + "type": "orgmetra.people.updated", + } + return json.dumps(document, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + +class _Authority: + """Return exact authorization evidence for the requested audit review.""" + + def __init__(self) -> None: + """Track whether authorization happened before any row read.""" + self.called = False + + def authorize(self, query: AuditEvidenceQuery) -> AuditEvidenceReadAuthorization: + """Authorize exactly the governed query scope.""" + self.called = True + return AuditEvidenceReadAuthorization( + tenant_record_id=query.tenant_record_id, + query_reference=query.query_reference, + requester_reference=query.requester_reference, + purpose_code=query.purpose_code, + permitted=True, + ) + + +class _Reader: + """Expose one persisted audit row only after authorization.""" + + def __init__(self, authority: _Authority) -> None: + """Bind the fake reader to the authority call-order probe.""" + self.authority = authority + + def read_rows(self, query: AuditEvidenceQuery) -> tuple[PersistedAuditEvidenceRow, ...]: + """Return one digest-verified row and prove authorization preceded the read.""" + assert self.authority.called is True + canonical = _canonical_event() + return ( + PersistedAuditEvidenceRow( + tenant_record_id=query.tenant_record_id, + audit_event_record_id=EVENT, + canonical_event_json=canonical, + event_envelope_digest=sha256(canonical.encode("utf-8")).hexdigest(), + recorded_at=RECORDED, + ), + ) + + +def test_authorized_review_verifies_persisted_evidence_before_returning_it() -> None: + """Authorized review returns only digest-bound tenant evidence from the requested window.""" + query = AuditEvidenceQuery( + tenant_record_id=TENANT, + query_reference=QUERY, + requester_reference=REQUESTER, + purpose_code="audit_evidence_review", + recorded_from=FROM, + recorded_before=BEFORE, + limit=50, + ) + authority = _Authority() + page = read_audit_evidence(query=query, authority=authority, reader=_Reader(authority)) + + assert page.query_reference == QUERY + assert page.records[0].audit_event_record_id == EVENT + assert page.records[0].event_envelope_digest == sha256( + _canonical_event().encode("utf-8") + ).hexdigest() diff --git a/packages/audit-evidence-review/tests/test_runtime_integrity.py b/packages/audit-evidence-review/tests/test_runtime_integrity.py new file mode 100644 index 000000000..a871509d2 --- /dev/null +++ b/packages/audit-evidence-review/tests/test_runtime_integrity.py @@ -0,0 +1,216 @@ +"""Runtime-integrity regressions for the audit-review trust boundary.""" + +from __future__ import annotations + +from datetime import datetime, timezone +from hashlib import sha256 +import json +from uuid import UUID + +import pytest + +from orgmetra_audit_evidence_review import ( + AuditEvidenceQuery, + AuditEvidenceReadAuthorization, + PersistedAuditEvidenceRow, + read_audit_evidence, +) + +TENANT = UUID("10000000-0000-7000-8000-000000000001") +OTHER_TENANT = UUID("20000000-0000-7000-8000-000000000002") +EVENT = UUID("11111111-1111-4111-8111-111111111111") +QUERY_REF = "audit_review:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" +REQUESTER = "actor:bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb" +START = datetime(2026, 8, 1, tzinfo=timezone.utc) +END = datetime(2026, 9, 1, tzinfo=timezone.utc) +RECORDED = datetime(2026, 8, 20, 12, tzinfo=timezone.utc) + + +def _query() -> AuditEvidenceQuery: + """Create one valid governed review query.""" + return AuditEvidenceQuery( + tenant_record_id=TENANT, + query_reference=QUERY_REF, + requester_reference=REQUESTER, + purpose_code="audit_evidence_review", + recorded_from=START, + recorded_before=END, + limit=50, + ) + + +def _authorization(query: AuditEvidenceQuery) -> AuditEvidenceReadAuthorization: + """Create one exact-scope permitted authorization decision.""" + return AuditEvidenceReadAuthorization( + tenant_record_id=query.tenant_record_id, + query_reference=query.query_reference, + requester_reference=query.requester_reference, + purpose_code=query.purpose_code, + permitted=True, + ) + + +def _canonical_event() -> str: + """Create one valid PII-minimized canonical audit envelope.""" + document = { + "data": {"high_impact": False, "result_code": "updated"}, + "datacontenttype": "application/json", + "id": str(EVENT), + "orgmetraactor": REQUESTER, + "orgmetraevidence": "v1", + "orgmetrapurpose": "people_record_update", + "orgmetrareason": "authorized_change", + "orgmetratenant": str(TENANT), + "source": "urn:orgmetra:people_api", + "specversion": "1.0", + "subject": "person:cccccccc-cccc-4ccc-8ccc-cccccccccccc", + "time": "2026-08-20T11:59:00Z", + "type": "orgmetra.people.updated", + } + return json.dumps(document, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + +def _row() -> PersistedAuditEvidenceRow: + """Create one valid persisted audit row.""" + canonical = _canonical_event() + return PersistedAuditEvidenceRow( + tenant_record_id=TENANT, + audit_event_record_id=EVENT, + canonical_event_json=canonical, + event_envelope_digest=sha256(canonical.encode("utf-8")).hexdigest(), + recorded_at=RECORDED, + ) + + +class _Authority: + """Return one configured authorization object.""" + + def __init__(self, decision: AuditEvidenceReadAuthorization) -> None: + """Store the decision and count calls.""" + self.decision = decision + self.calls = 0 + + def authorize(self, query: AuditEvidenceQuery) -> AuditEvidenceReadAuthorization: + """Return the configured decision.""" + self.calls += 1 + return self.decision + + +class _Reader: + """Return configured persisted rows and count calls.""" + + def __init__(self, rows: tuple[PersistedAuditEvidenceRow, ...]) -> None: + """Store immutable rows.""" + self.rows = rows + self.calls = 0 + + def read_rows(self, query: AuditEvidenceQuery) -> tuple[PersistedAuditEvidenceRow, ...]: + """Return the configured rows.""" + self.calls += 1 + return self.rows + + +def test_post_construction_query_mutation_is_revalidated_before_authority_or_store_access() -> None: + """A caller cannot widen a validated query by rewriting frozen fields after issuance.""" + query = _query() + object.__setattr__(query, "limit", 1000) + authority = _Authority(_authorization(_query())) + reader = _Reader(()) + + with pytest.raises(ValueError, match="limit"): + read_audit_evidence(query=query, authority=authority, reader=reader) + + assert authority.calls == 0 + assert reader.calls == 0 + + +def test_authority_callback_cannot_mutate_the_authoritative_query() -> None: + """The host callback receives a snapshot, so its object-level mutation cannot widen the read.""" + query = _query() + captured: list[AuditEvidenceQuery] = [] + + class MutatingAuthority: + """Attempt to rewrite every query field exposed to the authorization callback.""" + + def authorize(self, callback_query: AuditEvidenceQuery) -> AuditEvidenceReadAuthorization: + """Mutate only the callback snapshot and return authorization for the original query.""" + object.__setattr__(callback_query, "tenant_record_id", OTHER_TENANT) + object.__setattr__(callback_query, "recorded_from", datetime(2026, 1, 1, tzinfo=timezone.utc)) + object.__setattr__(callback_query, "limit", 200) + return _authorization(query) + + class CapturingReader: + """Capture the query that reaches the store boundary.""" + + def read_rows(self, reader_query: AuditEvidenceQuery) -> tuple[PersistedAuditEvidenceRow, ...]: + """Return no rows after recording the authoritative query snapshot.""" + captured.append(reader_query) + return () + + page = read_audit_evidence( + query=query, + authority=MutatingAuthority(), + reader=CapturingReader(), + ) + + assert page.records == () + assert captured[0].tenant_record_id == TENANT + assert captured[0].recorded_from == START + assert captured[0].limit == 50 + + +def test_reader_callback_cannot_mutate_the_authoritative_query() -> None: + """The store callback receives a snapshot, so returned-page checks keep the authorized scope.""" + query = _query() + + class MutatingReader: + """Attempt to widen the query after the store callback receives it.""" + + def read_rows(self, callback_query: AuditEvidenceQuery) -> tuple[PersistedAuditEvidenceRow, ...]: + """Rewrite callback-only fields before returning one valid row.""" + object.__setattr__(callback_query, "query_reference", "audit_review:cccccccc-cccc-4ccc-8ccc-cccccccccccc") + object.__setattr__(callback_query, "recorded_from", datetime(2020, 1, 1, tzinfo=timezone.utc)) + object.__setattr__(callback_query, "limit", 0) + return (_row(),) + + page = read_audit_evidence( + query=query, + authority=_Authority(_authorization(query)), + reader=MutatingReader(), + ) + + assert page.query_reference == QUERY_REF + assert len(page.records) == 1 + + +def test_post_construction_authorization_mutation_cannot_turn_non_boolean_into_permission() -> None: + """Authorization output is revalidated rather than trusting a once-valid mutable Python object.""" + query = _query() + decision = _authorization(query) + object.__setattr__(decision, "permitted", "yes") + authority = _Authority(decision) + reader = _Reader(()) + + with pytest.raises(ValueError, match="permitted"): + read_audit_evidence(query=query, authority=authority, reader=reader) + + assert authority.calls == 1 + assert reader.calls == 0 + + +def test_post_construction_row_mutation_is_reverified_before_evidence_is_returned() -> None: + """A reader cannot mutate canonical bytes and their digest after row construction to widen evidence.""" + query = _query() + persisted = _row() + document = json.loads(persisted.canonical_event_json) + document["employee_name"] = "should-never-enter-audit-envelope" + widened = json.dumps(document, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + object.__setattr__(persisted, "canonical_event_json", widened) + object.__setattr__(persisted, "event_envelope_digest", sha256(widened.encode("utf-8")).hexdigest()) + + with pytest.raises(ValueError, match="governed audit envelope shape"): + read_audit_evidence( + query=query, + authority=_Authority(_authorization(query)), + reader=_Reader((persisted,)), + ) diff --git a/packages/audit-evidence-review/tests/test_validation.py b/packages/audit-evidence-review/tests/test_validation.py new file mode 100644 index 000000000..64a0d0da2 --- /dev/null +++ b/packages/audit-evidence-review/tests/test_validation.py @@ -0,0 +1,447 @@ +"""Adversarial validation and authority-boundary regressions for audit review.""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone, tzinfo +from hashlib import sha256 +import json +from uuid import UUID + +import pytest + +from orgmetra_audit_evidence_review import ( + AuditEvidenceQuery, + AuditEvidenceReadAuthorization, + AuditEvidenceReviewPage, + PersistedAuditEvidenceRow, + read_audit_evidence, +) + +TENANT = UUID("10000000-0000-7000-8000-000000000001") +OTHER_TENANT = UUID("20000000-0000-7000-8000-000000000002") +EVENT1 = UUID("11111111-1111-4111-8111-111111111111") +EVENT2 = UUID("22222222-2222-4222-8222-222222222222") +QUERY_REF = "audit_review:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" +REQUESTER = "actor:bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb" +START = datetime(2026, 8, 1, tzinfo=timezone.utc) +END = datetime(2026, 9, 1, tzinfo=timezone.utc) +RECORDED = datetime(2026, 8, 20, 12, tzinfo=timezone.utc) + + +def canonical_event(*, tenant: UUID = TENANT, event: UUID = EVENT1) -> str: + """Build exact canonical JSON matching the existing immutable audit store contract.""" + document = { + "data": {"high_impact": False, "result_code": "updated"}, + "datacontenttype": "application/json", + "id": str(event), + "orgmetraactor": REQUESTER, + "orgmetraevidence": "v1", + "orgmetrapurpose": "people_record_update", + "orgmetrareason": "authorized_change", + "orgmetratenant": str(tenant), + "source": "urn:orgmetra:people_api", + "specversion": "1.0", + "subject": "person:cccccccc-cccc-4ccc-8ccc-cccccccccccc", + "time": "2026-08-20T11:59:00Z", + "type": "orgmetra.people.updated", + } + return json.dumps(document, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + +def row( + *, + tenant: UUID = TENANT, + event: UUID = EVENT1, + recorded_at: datetime = RECORDED, + canonical: str | None = None, + digest: str | None = None, +) -> PersistedAuditEvidenceRow: + """Build one valid persisted row unless an explicit adversarial value is supplied.""" + text = canonical_event(tenant=tenant, event=event) if canonical is None else canonical + actual_digest = sha256(text.encode("utf-8")).hexdigest() if digest is None else digest + return PersistedAuditEvidenceRow( + tenant_record_id=tenant, + audit_event_record_id=event, + canonical_event_json=text, + event_envelope_digest=actual_digest, + recorded_at=recorded_at, + ) + + +def query(**changes: object) -> AuditEvidenceQuery: + """Build a valid bounded query and permit focused overrides.""" + values: dict[str, object] = { + "tenant_record_id": TENANT, + "query_reference": QUERY_REF, + "requester_reference": REQUESTER, + "purpose_code": "audit_evidence_review", + "recorded_from": START, + "recorded_before": END, + "limit": 100, + } + values.update(changes) + return AuditEvidenceQuery(**values) # type: ignore[arg-type] + + +def authorization(q: AuditEvidenceQuery, **changes: object) -> AuditEvidenceReadAuthorization: + """Build matching authorization evidence and permit focused overrides.""" + values: dict[str, object] = { + "tenant_record_id": q.tenant_record_id, + "query_reference": q.query_reference, + "requester_reference": q.requester_reference, + "purpose_code": q.purpose_code, + "permitted": True, + } + values.update(changes) + return AuditEvidenceReadAuthorization(**values) # type: ignore[arg-type] + + +class Authority: + """Configurable fake authority used to prove authorization ordering and scope.""" + + def __init__(self, decision: object) -> None: + """Store the decision and count calls.""" + self.decision = decision + self.calls = 0 + + def authorize(self, q: AuditEvidenceQuery) -> object: + """Return the configured decision.""" + self.calls += 1 + return self.decision + + +class Reader: + """Configurable fake immutable audit-store reader.""" + + def __init__(self, rows: object) -> None: + """Store rows and count calls.""" + self.rows = rows + self.calls = 0 + + def read_rows(self, q: AuditEvidenceQuery) -> object: + """Return configured rows.""" + self.calls += 1 + return self.rows + + +class RaisingTimezone(tzinfo): + """Timezone provider that fails when asked for an offset.""" + + def utcoffset(self, dt: datetime | None) -> timedelta | None: + """Raise to prove caller timezone code cannot escape the boundary.""" + raise RuntimeError("boom") + + def dst(self, dt: datetime | None) -> timedelta | None: + """Return no DST adjustment.""" + return None + + +class MissingOffsetTimezone(tzinfo): + """Timezone provider that cannot resolve a concrete offset.""" + + def utcoffset(self, dt: datetime | None) -> timedelta | None: + """Return no offset.""" + return None + + def dst(self, dt: datetime | None) -> timedelta | None: + """Return no DST adjustment.""" + return None + + +class ForgedStr(str): + """Hostile string subtype used to prove exact-runtime trust boundaries.""" + + +class ForgedInt(int): + """Hostile integer subtype used to prove bool/subclass rejection.""" + + +class ForgedUUID(UUID): + """UUID subtype used to prove authoritative identities require exact runtime UUID.""" + + +class ForgedDateTime(datetime): + """Datetime subtype used to prove time evidence cannot override behavior.""" + + +def test_query_normalizes_fixed_offset_and_accepts_boundary_limit() -> None: + """Query stores detached UTC timestamps and the maximum bounded page size.""" + item = query( + recorded_from=datetime(2026, 8, 1, 9, tzinfo=timezone(timedelta(hours=9))), + recorded_before=datetime(2026, 10, 30, 9, tzinfo=timezone(timedelta(hours=9))), + limit=200, + ) + assert item.recorded_from == datetime(2026, 8, 1, tzinfo=timezone.utc) + assert item.recorded_before == datetime(2026, 10, 30, tzinfo=timezone.utc) + + +@pytest.mark.parametrize("bad_uuid", [UUID(int=0), UUID(int=(1 << 128) - 1)]) +def test_query_rejects_reserved_tenant_uuid(bad_uuid: UUID) -> None: + """Reserved UUID sentinels cannot identify a tenant.""" + with pytest.raises(ValueError, match="tenant_record_id"): + query(tenant_record_id=bad_uuid) + + +def test_query_rejects_uuid_subtype() -> None: + """Caller-defined UUID runtime types cannot enter trusted tenant scope.""" + with pytest.raises(ValueError, match="tenant_record_id"): + query(tenant_record_id=ForgedUUID(str(TENANT))) + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("query_reference", "bad"), + ("query_reference", "audit_review:aaaaaaaa-aaaa-1aaa-8aaa-aaaaaaaaaaaa"), + ("query_reference", "audit_review:" + "a" * 100), + ("requester_reference", "reviewer:bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"), + ("requester_reference", ForgedStr(REQUESTER)), + ], +) +def test_query_rejects_invalid_references(field: str, value: str) -> None: + """Packet-owned query and requester references are bounded canonical UUIDv4 text.""" + with pytest.raises(ValueError, match=field): + query(**{field: value}) + + +@pytest.mark.parametrize("purpose", ["other", ForgedStr("audit_evidence_review")]) +def test_query_rejects_noncanonical_purpose(purpose: str) -> None: + """Only exact built-in audit-review purpose text is trusted.""" + with pytest.raises(ValueError, match="purpose_code"): + query(purpose_code=purpose) + + +@pytest.mark.parametrize( + ("value", "message"), + [ + (datetime(2026, 8, 1), "aware"), + (ForgedDateTime(2026, 8, 1, tzinfo=timezone.utc), "aware"), + (datetime(2026, 8, 1, tzinfo=RaisingTimezone()), "could not"), + (datetime(2026, 8, 1, tzinfo=MissingOffsetTimezone()), "concrete"), + ], +) +def test_query_rejects_untrusted_time(value: datetime, message: str) -> None: + """System-recorded query bounds reject ambiguous or caller-controlled runtime time.""" + with pytest.raises(ValueError, match=message): + query(recorded_from=value) + + +def test_query_normalizes_timestamp_overflow_to_value_error() -> None: + """Offset arithmetic overflow fails closed with a stable validation error.""" + with pytest.raises(ValueError, match="cannot be normalized"): + query(recorded_from=datetime.min.replace(tzinfo=timezone(timedelta(hours=14)))) + + +@pytest.mark.parametrize( + ("changes", "message"), + [ + ({"recorded_before": START}, "later"), + ({"recorded_before": datetime(2026, 11, 1, tzinfo=timezone.utc)}, "90 days"), + ({"limit": 0}, "1 through 200"), + ({"limit": 201}, "1 through 200"), + ({"limit": True}, "1 through 200"), + ({"limit": ForgedInt(10)}, "1 through 200"), + ], +) +def test_query_rejects_unbounded_or_invalid_window(changes: dict[str, object], message: str) -> None: + """Review requests remain time- and row-bounded.""" + with pytest.raises(ValueError, match=message): + query(**changes) + + +def test_authorization_validates_scope_and_boolean() -> None: + """Authorization evidence uses the same strict scope vocabulary as the query.""" + item = query() + assert authorization(item).permitted is True + with pytest.raises(ValueError, match="tenant_record_id"): + authorization(item, tenant_record_id=UUID(int=0)) + with pytest.raises(ValueError, match="query_reference"): + authorization(item, query_reference="bad") + with pytest.raises(ValueError, match="requester_reference"): + authorization(item, requester_reference="bad") + with pytest.raises(ValueError, match="purpose_code"): + authorization(item, purpose_code="other") + with pytest.raises(ValueError, match="permitted"): + authorization(item, permitted=1) + + +def test_row_accepts_canonical_digest_bound_evidence_and_normalizes_recorded_time() -> None: + """A valid row is reverified and its system-recorded time is detached to UTC.""" + item = row(recorded_at=datetime(2026, 8, 20, 21, tzinfo=timezone(timedelta(hours=9)))) + assert item.recorded_at == RECORDED + + +@pytest.mark.parametrize( + ("changes", "message"), + [ + ({"tenant_record_id": UUID(int=0)}, "tenant_record_id"), + ({"audit_event_record_id": UUID(int=0)}, "audit_event_record_id"), + ({"canonical_event_json": 1}, "canonical_event_json must be a string"), + ({"event_envelope_digest": "x"}, "SHA-256"), + ({"event_envelope_digest": ForgedStr("0" * 64)}, "SHA-256"), + ], +) +def test_row_rejects_invalid_primitive_evidence(changes: dict[str, object], message: str) -> None: + """Persisted row trust primitives are validated before evidence use.""" + values: dict[str, object] = { + "tenant_record_id": TENANT, + "audit_event_record_id": EVENT1, + "canonical_event_json": canonical_event(), + "event_envelope_digest": sha256(canonical_event().encode()).hexdigest(), + "recorded_at": RECORDED, + } + values.update(changes) + with pytest.raises(ValueError, match=message): + PersistedAuditEvidenceRow(**values) # type: ignore[arg-type] + + +def test_row_rejects_oversized_canonical_bytes() -> None: + """Audit review does not accept an unbounded canonical envelope.""" + oversized = "x" * 32769 + with pytest.raises(ValueError, match="32768-byte"): + PersistedAuditEvidenceRow( + tenant_record_id=TENANT, + audit_event_record_id=EVENT1, + canonical_event_json=oversized, + event_envelope_digest="0" * 64, + recorded_at=RECORDED, + ) + + +def test_row_rejects_digest_mismatch() -> None: + """Stored digest must bind the exact canonical UTF-8 bytes.""" + with pytest.raises(ValueError, match="digest"): + row(digest="0" * 64) + + +@pytest.mark.parametrize( + ("canonical", "message"), + [ + ("{", "valid UTF-8 JSON"), + ("[]", "one JSON object"), + ('{ "a":1}', "canonical Orgmetra JSON"), + ], +) +def test_row_rejects_noncanonical_json(canonical: str, message: str) -> None: + """Persisted bytes must parse to exactly one canonical JSON object.""" + with pytest.raises(ValueError, match=message): + row(canonical=canonical) + + +def test_row_rejects_wrong_cloudevent_contract() -> None: + """Read-time verification preserves the existing CloudEvents envelope version/media type.""" + document = json.loads(canonical_event()) + document["specversion"] = "0.3" + text = json.dumps(document, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + with pytest.raises(ValueError, match="CloudEvents"): + row(canonical=text) + document["specversion"] = "1.0" + document["datacontenttype"] = "text/plain" + text = json.dumps(document, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + with pytest.raises(ValueError, match="CloudEvents"): + row(canonical=text) + + +def test_row_rejects_mismatched_event_and_tenant_bindings() -> None: + """Canonical event identity and tenant scope must match persisted row columns.""" + with pytest.raises(ValueError, match="event id"): + row(event=EVENT2, canonical=canonical_event(event=EVENT1)) + with pytest.raises(ValueError, match="event tenant"): + row(tenant=OTHER_TENANT, canonical=canonical_event(tenant=TENANT)) + + +def test_row_rejects_invalid_recorded_time() -> None: + """Persisted system-recorded time receives the same fail-closed timestamp validation.""" + with pytest.raises(ValueError, match="aware"): + row(recorded_at=datetime(2026, 8, 20, 12)) + + +def test_review_page_validates_reference_and_record_collection() -> None: + """Review output remains immutable and governed by exact runtime record types.""" + item = row() + assert AuditEvidenceReviewPage(QUERY_REF, (item,)).records == (item,) + with pytest.raises(ValueError, match="query_reference"): + AuditEvidenceReviewPage("bad", (item,)) + with pytest.raises(ValueError, match="immutable tuple"): + AuditEvidenceReviewPage(QUERY_REF, [item]) # type: ignore[arg-type] + with pytest.raises(ValueError, match="exact persisted"): + AuditEvidenceReviewPage(QUERY_REF, (object(),)) # type: ignore[arg-type] + + +def test_read_requires_exact_query_and_authorization_types_before_store_access() -> None: + """Ungoverned query or authority evidence cannot trigger an audit-store read.""" + item = query() + reader = Reader(()) + with pytest.raises(TypeError, match="exact AuditEvidenceQuery"): + read_audit_evidence(query=object(), authority=Authority(authorization(item)), reader=reader) # type: ignore[arg-type] + assert reader.calls == 0 + with pytest.raises(TypeError, match="authority"): + read_audit_evidence(query=item, authority=Authority(object()), reader=reader) + assert reader.calls == 0 + + +@pytest.mark.parametrize( + "changes", + [ + {"tenant_record_id": OTHER_TENANT}, + {"query_reference": "audit_review:dddddddd-dddd-4ddd-8ddd-dddddddddddd"}, + {"requester_reference": "actor:eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee"}, + {"purpose_code": "audit_evidence_review", "permitted": False}, + ], +) +def test_read_denies_mismatched_or_denied_authorization_before_store_access(changes: dict[str, object]) -> None: + """Authorization must exactly bind tenant, request, requester, purpose, and permission.""" + item = query() + decision = authorization(item, **changes) + reader = Reader(()) + with pytest.raises(PermissionError, match="not authorized"): + read_audit_evidence(query=item, authority=Authority(decision), reader=reader) + assert reader.calls == 0 + + +def test_read_requires_tuple_and_limit_bound() -> None: + """Reader output must be immutable and may not exceed the authorized count.""" + item = query(limit=1) + auth = Authority(authorization(item)) + with pytest.raises(TypeError, match="immutable tuple"): + read_audit_evidence(query=item, authority=auth, reader=Reader([])) + with pytest.raises(ValueError, match="authorized limit"): + read_audit_evidence(query=item, authority=auth, reader=Reader((row(), row(event=EVENT2)))) + + +def test_read_rejects_ungoverned_cross_tenant_and_out_of_window_rows() -> None: + """Every returned row is rechecked for runtime type, tenant, and recorded-time scope.""" + item = query() + auth = Authority(authorization(item)) + with pytest.raises(TypeError, match="ungoverned"): + read_audit_evidence(query=item, authority=auth, reader=Reader((object(),))) + with pytest.raises(PermissionError, match="cross-tenant"): + read_audit_evidence(query=item, authority=auth, reader=Reader((row(tenant=OTHER_TENANT),))) + with pytest.raises(ValueError, match="outside"): + read_audit_evidence( + query=item, + authority=auth, + reader=Reader((row(recorded_at=datetime(2026, 9, 1, tzinfo=timezone.utc)),)), + ) + + +def test_read_requires_strict_recorded_time_id_order() -> None: + """Stable keyset order prevents duplicate or regressing evidence pages.""" + item = query() + auth = Authority(authorization(item)) + with pytest.raises(ValueError, match="strict"): + read_audit_evidence(query=item, authority=auth, reader=Reader((row(event=EVENT2), row(event=EVENT1)))) + + +def test_read_returns_verified_authorized_page() -> None: + """Authorized evidence reaches the caller only after row and ordering verification.""" + item = query() + first = row(event=EVENT1) + second = row(event=EVENT2, recorded_at=RECORDED + timedelta(seconds=1)) + page = read_audit_evidence( + query=item, + authority=Authority(authorization(item)), + reader=Reader((first, second)), + ) + assert page.query_reference == QUERY_REF + assert page.records == (first, second)