From c9bc6456fe217a5b34005b0507a06e434de4b98c Mon Sep 17 00:00:00 2001 From: quantamixsol Date: Tue, 4 Aug 2026 22:36:01 +0200 Subject: [PATCH] =?UTF-8?q?feat(compliance):=20CR-010.R3=20=E2=80=94=20com?= =?UTF-8?q?pliance=20packs=20as=20data,=20x-sox=20first?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cherry-picked from private 3798bb86 (private PR #331, merged 7c1cf8b5). Ships the SOX/COSO financial-controls pack and the data-driven pack format that makes the NEXT framework authorable without Python. * NEW graqle/pct/extensions/x_sox.py — frozen XSoxExtension mirroring the x_ai_eu.py pattern. Binds an AI-assisted decision to a NAMED internal control, a financial-statement assertion, and a reporting period. Preparer/reviewer are HMAC-SHA256 pseudonym tokens or the explicit "omit" sentinel; raw identifiers are rejected at construction, because these values travel inside long-lived, widely-shared audit artifacts with no redaction path. * NEW graqle/compliance/packs/ — pack.yaml + schema.json per framework, discovered by directory scan over importlib.resources. NIST AI RMF and ISO 42001 become two data files each. NO entry-point discovery: that is R10, and a test enforces the boundary. * NEW graqle/compliance/management_review_gate.py — SOX-vocabulary counterpart of the Article 14 human-oversight gate. STRICTLY ADDITIVE: article_14_gate.py is byte-for-byte unmodified (0 diff lines). * NEW graqle/compliance/_gate_helpers.py — public re-exports of the three shared validators. Works without engine changes because taxonomy.py:127 already accepts ^x-[a-z0-9_-]{1,64}$, so this pack's four x-sox-* claim limits validate with zero taxonomy edit. VERIFIED ON THIS PUBLIC BASE (597a7c49, not on the private branch): 133 R3 tests; 913 passed/4 skipped (test_compliance + test_pct); 772 passed/6 skipped (test_governance + test_tamper_evidence); TS-screen 0; article_14_gate.py 0 diff lines; public wheel builds with zero duplicate entries and ships all 4 pack files. All 8 R3 source files are byte-identical to private/master. Refs: CR-010.R3, plan_54c13aec, lesson_20260804T121740 Co-Authored-By: Claude Opus 5 --- graqle/compliance/__init__.py | 15 + graqle/compliance/_gate_helpers.py | 46 ++ graqle/compliance/management_review_gate.py | 242 +++++++++++ graqle/compliance/packs/__init__.py | 36 ++ graqle/compliance/packs/_loader.py | 378 ++++++++++++++++ graqle/compliance/packs/x_sox/pack.yaml | 166 +++++++ graqle/compliance/packs/x_sox/schema.json | 118 +++++ graqle/pct/extensions/x_sox.py | 391 +++++++++++++++++ pyproject.toml | 7 + .../test_packs/test_management_review_gate.py | 260 +++++++++++ .../test_packs/test_pack_loader.py | 410 ++++++++++++++++++ .../test_packs/test_x_sox_extension.py | 266 ++++++++++++ 12 files changed, 2335 insertions(+) create mode 100644 graqle/compliance/_gate_helpers.py create mode 100644 graqle/compliance/management_review_gate.py create mode 100644 graqle/compliance/packs/__init__.py create mode 100644 graqle/compliance/packs/_loader.py create mode 100644 graqle/compliance/packs/x_sox/pack.yaml create mode 100644 graqle/compliance/packs/x_sox/schema.json create mode 100644 graqle/pct/extensions/x_sox.py create mode 100644 tests/test_compliance/test_packs/test_management_review_gate.py create mode 100644 tests/test_compliance/test_packs/test_pack_loader.py create mode 100644 tests/test_compliance/test_packs/test_x_sox_extension.py diff --git a/graqle/compliance/__init__.py b/graqle/compliance/__init__.py index 765c5e6c..1ee4407a 100644 --- a/graqle/compliance/__init__.py +++ b/graqle/compliance/__init__.py @@ -5,6 +5,13 @@ banner + machine-readable ``ai_disclosure`` field for MCP envelopes. * :mod:`graqle.compliance.robustness` — Article 15 machine-readable robustness attestation for deployer compliance pipelines. + * :mod:`graqle.compliance.management_review_gate` — SOX/COSO + management-review-control gate (CR-010.R3). The financial-controls + counterpart of the Article 14 human-oversight gate: same mechanics, + different vocabulary and a distinct error code. + * :mod:`graqle.compliance.packs` — compliance frameworks expressed as + data (``pack.yaml`` + ``schema.json``), so a new framework needs no + Python and no engine change. All modules in this package are READ-ONLY and SIDE-EFFECT-FREE except for the banner emit (which writes to stderr exactly once per session). @@ -20,6 +27,11 @@ maybe_emit_session_banner, reset_session_banner_state, ) +from graqle.compliance.management_review_gate import ( + MANAGEMENT_REVIEW_ERROR_CODE, + ManagementReviewGateResult, + check_management_review, +) from graqle.compliance.robustness import ( Defence, MeasurableClaim, @@ -31,11 +43,14 @@ "AIDisclosure", "ComplianceEnvelope", "Defence", + "MANAGEMENT_REVIEW_ERROR_CODE", + "ManagementReviewGateResult", "MeasurableClaim", "RobustnessAttestation", "build_ai_disclosure", "build_compliance_envelope", "build_robustness_attestation", + "check_management_review", "is_eu_ai_act_mode_on", "is_ai_disclosure_suppressed", "maybe_emit_session_banner", diff --git a/graqle/compliance/_gate_helpers.py b/graqle/compliance/_gate_helpers.py new file mode 100644 index 00000000..ddbbbdf3 --- /dev/null +++ b/graqle/compliance/_gate_helpers.py @@ -0,0 +1,46 @@ +"""Shared validation helpers for compliance review gates (CR-010.R3). + +Both review gates — :mod:`graqle.compliance.article_14_gate` (EU AI Act +Article 14) and :mod:`graqle.compliance.management_review_gate` (SOX/COSO) — +need the same three primitives: coerce an arming flag, validate a +confidence, validate a threshold. + +**Why this module exists as a re-export rather than the definition site.** +The obvious refactor is to move the three helpers out of ``article_14_gate`` +and have both gates import them from here. That was rejected: +``article_14_gate`` is a live EU AI Act enforcement path with four +production consumers (``mcp_dev_server.py`` at three sites, +``switch_status.py``), and moving code out of it — even code that is +module-private — means editing a file whose contract those consumers pin, +for the benefit of a brand-new module. The risk sits entirely on the +regulated path and the benefit sits entirely on the new one. + +So the canonical definitions stay in ``article_14_gate`` and this module +re-exports them under stable, public (non-underscore) names. New callers +import from here and are insulated from the private spelling. If the +helpers are ever genuinely relocated, this module is the single place that +changes. + +The coupling is pinned by ``test_gate_helper_contract`` so that a rename in +``article_14_gate`` fails a test rather than surfacing as an ``ImportError`` +at process start — which is the failure mode that made this worth +addressing at all. +""" + +from __future__ import annotations + +from graqle.compliance.article_14_gate import ( + _coerce_bool as coerce_arming_flag, +) +from graqle.compliance.article_14_gate import ( + _validate_confidence as validate_confidence, +) +from graqle.compliance.article_14_gate import ( + _validate_threshold as validate_threshold, +) + +__all__ = [ + "coerce_arming_flag", + "validate_confidence", + "validate_threshold", +] diff --git a/graqle/compliance/management_review_gate.py b/graqle/compliance/management_review_gate.py new file mode 100644 index 00000000..b2668ea6 --- /dev/null +++ b/graqle/compliance/management_review_gate.py @@ -0,0 +1,242 @@ +"""Management-review-control gate — SOX/COSO vocabulary (CR-010.R3). + +The EU AI Act calls it *human oversight* (Article 14). SOX calls it a +*management review control*. The mechanics are the same: before an +automated outcome is relied upon, a competent human must look at it, and +the fact that they looked must be evidenced. + +This module is the SOX-vocabulary counterpart of +:mod:`graqle.compliance.article_14_gate`. It exists as a **sibling module +rather than a modification** of that one, deliberately: + +``article_14_gate`` is a live EU AI Act enforcement path with four +production consumers pinning its contract — ``mcp_dev_server.py`` at three +sites (two of which hardcode the string ``"ARTICLE_14_HUMAN_REVIEW_REQUIRED"``) +and ``switch_status.py``, which republishes the default threshold and the +refusal error code into a public status envelope. Renaming its fields or +generalising its error code to cover SOX would be a silent breaking change +across those consumers for no functional gain. So the Article 14 surface is +left byte-for-byte alone and the shared *logic* is reused by import. + +What is shared vs. what differs +------------------------------- +Shared, by import (not by copy): confidence/threshold validation, the +placeholder-vs-calibrated threshold markers, and the comparison rule that +a confidence **exactly at** the threshold ALLOWS — refusal fires only +strictly below it. + +Different: the error code (``MANAGEMENT_REVIEW_REQUIRED``), the vocabulary +in the reason string, the arming signal (an explicit argument rather than +``GRAQLE_EU_AI_ACT_MODE``, because SOX applicability is a property of the +*control*, not of a global deployment mode), and the ability to name the +control the refusal belongs to. + +Threshold calibration +--------------------- +Like the Article 14 gate, this gate's default threshold is a **placeholder** +and says so in every result via ``threshold_status``. Wiring the existing +calibration subsystem into both gates is deliberately out of scope for this +CR (see CR-010.R3 § 4.4): it changes behaviour on a live enforcement path +and belongs in its own reviewable slice. + +References: + - CR-010.R3 § 4.3 — additive generalisation + - Sarbanes-Oxley Act § 404; COSO Internal Control — Integrated Framework + - Companion: :mod:`graqle.compliance.article_14_gate` + - Companion: :mod:`graqle.pct.extensions.x_sox` +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from graqle.compliance._gate_helpers import ( + coerce_arming_flag, + validate_confidence, + validate_threshold, +) +from graqle.compliance.article_14_gate import ( + DEFAULT_HUMAN_REVIEW_THRESHOLD, + THRESHOLD_STATUS_CALIBRATED, + THRESHOLD_STATUS_PLACEHOLDER, +) + +__all__ = [ + "DEFAULT_MANAGEMENT_REVIEW_THRESHOLD", + "MANAGEMENT_REVIEW_ERROR_CODE", + "MANAGEMENT_REVIEW_VOCABULARY", + "ManagementReviewGateResult", + "THRESHOLD_STATUS_CALIBRATED", + "THRESHOLD_STATUS_PLACEHOLDER", + "check_management_review", +] + +#: Typed error code for a management-review refusal. Distinct from +#: ``ARTICLE_14_HUMAN_REVIEW_REQUIRED`` so a downstream handler can tell +#: which regime refused — an auditor asking "why was this blocked?" gets a +#: different answer under SOX than under the EU AI Act. +MANAGEMENT_REVIEW_ERROR_CODE: str = "MANAGEMENT_REVIEW_REQUIRED" + +#: Default control vocabulary label carried in the result. +MANAGEMENT_REVIEW_VOCABULARY: str = "management_review" + +#: Default threshold. Deliberately aliased to the Article 14 default rather +#: than redeclared: one placeholder value, one place to replace when the +#: calibration work lands. PLACEHOLDER — not evidence-derived. +DEFAULT_MANAGEMENT_REVIEW_THRESHOLD: float = DEFAULT_HUMAN_REVIEW_THRESHOLD + + +@dataclass(frozen=True) +class ManagementReviewGateResult: + """Outcome of the management-review-control gate check. + + Field names and ordering intentionally mirror + :class:`~graqle.compliance.article_14_gate.Article14GateResult` for the + first five fields, so code that handles one shape handles the other. + The SOX-specific fields are appended, never interleaved. + + Attributes: + allowed: True iff the automated path may proceed without review. + confidence: The confidence the gate evaluated. + threshold: The threshold it compared against. + threshold_status: :data:`THRESHOLD_STATUS_PLACEHOLDER` until a + calibrated value is wired in. + reason: Human-readable reason. Empty when ``allowed`` is True. + control_id: The named internal control this decision falls under, + when known. Carried so a refusal is traceable to a control + without a second lookup. + control_vocabulary: Which vocabulary produced this result. Defaults + to :data:`MANAGEMENT_REVIEW_VOCABULARY`. + """ + + allowed: bool + confidence: float + threshold: float + threshold_status: str = THRESHOLD_STATUS_PLACEHOLDER + reason: str = "" + control_id: str | None = None + control_vocabulary: str = MANAGEMENT_REVIEW_VOCABULARY + + def to_refusal_envelope(self) -> dict[str, Any]: + """Build the structured refusal envelope for the tool response. + + Mirrors ``Article14GateResult.to_refusal_envelope`` — including + raising on an allowed result, because an "allowed refusal" is a + caller bug that should surface loudly rather than emit a + contradictory envelope. + + Raises: + RuntimeError: If called on an ``allowed=True`` result. + """ + if self.allowed: + raise RuntimeError( + "to_refusal_envelope() called on an allowed gate result; " + "the caller should check `.allowed` before envelope build." + ) + envelope: dict[str, Any] = { + "success": False, + "error_code": MANAGEMENT_REVIEW_ERROR_CODE, + "error": self.reason, + "control_vocabulary": self.control_vocabulary, + "confidence": round(float(self.confidence), 4), + "threshold": round(float(self.threshold), 4), + "threshold_status": self.threshold_status, + "next_action": "present_to_management_reviewer", + } + if self.control_id is not None: + envelope["control_id"] = self.control_id + return envelope + + +def check_management_review( + *, + confidence: float, + management_review_required: Any = None, + threshold: float | None = None, + threshold_status: str = THRESHOLD_STATUS_PLACEHOLDER, + control_id: str | None = None, + action_label: str = "decision", +) -> ManagementReviewGateResult: + """Evaluate the management-review-control gate. + + The gate is ARMED when ``management_review_required`` is truthy (per the + same coercion the Article 14 gate uses). Unlike the Article 14 gate, + there is **no environment-variable arming path**: SOX applicability is a + property of the specific control a decision falls under, not of a + deployment-wide mode, so arming this gate globally via an env var would + misrepresent scope. + + When ARMED and ``confidence < threshold``, the gate REFUSES: the result + has ``allowed=False`` and the caller should return + :meth:`ManagementReviewGateResult.to_refusal_envelope` as its response. + + A confidence **exactly at** the threshold ALLOWS, matching the Article 14 + gate: "set threshold=0.75 → 0.749 refused, 0.75 allowed". + + Args: + confidence: Confidence in [0.0, 1.0]. NaN, infinity, and + out-of-range values raise rather than producing a surprising + allow or refuse. + management_review_required: Arming signal. Truthy arms the gate. + threshold: Optional threshold override. When ``None``, uses + :data:`DEFAULT_MANAGEMENT_REVIEW_THRESHOLD` — a **placeholder**, + not an evidence-derived value. + threshold_status: Marker for whether the threshold is calibrated or + placeholder. + control_id: Optional named internal control, carried into the result + and the refusal envelope. + action_label: Short label used in the refusal reason string. + + Returns: + ManagementReviewGateResult: ``allowed=True`` if the gate is disarmed + or confidence meets the threshold; otherwise ``allowed=False`` with a + populated ``reason``. + + Raises: + ValueError: If ``confidence`` or ``threshold`` is NaN, infinite, or + outside [0.0, 1.0]. + """ + eff_confidence = validate_confidence(confidence) + eff_threshold = ( + validate_threshold(threshold) + if threshold is not None + else DEFAULT_MANAGEMENT_REVIEW_THRESHOLD + ) + + if not coerce_arming_flag(management_review_required): + return ManagementReviewGateResult( + allowed=True, + confidence=eff_confidence, + threshold=eff_threshold, + threshold_status=threshold_status, + control_id=control_id, + ) + + if eff_confidence < eff_threshold: + control_phrase = ( + f" under control {control_id}" if control_id else "" + ) + return ManagementReviewGateResult( + allowed=False, + confidence=eff_confidence, + threshold=eff_threshold, + threshold_status=threshold_status, + control_id=control_id, + reason=( + f"Management review control refused {action_label!s}" + f"{control_phrase}: confidence " + f"{round(eff_confidence, 4)} is below threshold " + f"{round(eff_threshold, 4)} (status: {threshold_status}). " + f"Present the proposed outcome to a management reviewer " + f"before relying on it." + ), + ) + + return ManagementReviewGateResult( + allowed=True, + confidence=eff_confidence, + threshold=eff_threshold, + threshold_status=threshold_status, + control_id=control_id, + ) diff --git a/graqle/compliance/packs/__init__.py b/graqle/compliance/packs/__init__.py new file mode 100644 index 00000000..f03f3d27 --- /dev/null +++ b/graqle/compliance/packs/__init__.py @@ -0,0 +1,36 @@ +"""Compliance packs — regulatory frameworks expressed as data (CR-010.R3). + +Each pack is a directory of two data files (``pack.yaml`` + ``schema.json``) +describing one framework's extension namespace, its contributed claim +limits, and its field vocabulary. Adding a framework requires no Python and +no engine change. + +Shipped packs: + * ``x_sox`` — Sarbanes-Oxley / COSO internal controls. + +See :mod:`graqle.compliance.packs._loader` for the loading contract and the +rationale behind directory-scan discovery (rather than entry points, which +are CR-010.R10). +""" + +from __future__ import annotations + +from graqle.compliance.packs._loader import ( + PACK_MANIFEST_FILENAME, + PACK_SCHEMA_FILENAME, + CompliancePack, + CompliancePackError, + discover_packs, + load_all_packs, + load_pack, +) + +__all__ = [ + "PACK_MANIFEST_FILENAME", + "PACK_SCHEMA_FILENAME", + "CompliancePack", + "CompliancePackError", + "discover_packs", + "load_all_packs", + "load_pack", +] diff --git a/graqle/compliance/packs/_loader.py b/graqle/compliance/packs/_loader.py new file mode 100644 index 00000000..e7b5390a --- /dev/null +++ b/graqle/compliance/packs/_loader.py @@ -0,0 +1,378 @@ +"""Data-driven compliance-pack loader (CR-010.R3). + +A **compliance pack** is a regulatory framework expressed as data rather +than code: a directory holding exactly two files. + +:: + + graqle/compliance/packs// + pack.yaml # the pack content + schema.json # the pack's OWN schema, which pack.yaml must satisfy + +The point of the pack format is that adding the *next* framework — NIST AI +RMF, ISO/IEC 42001, SOC 2, HIPAA — requires **no Python and no engine +change**: drop in two data files and the pack loads. The typed dataclasses +in :mod:`graqle.pct.extensions` remain available as ergonomic bootstraps +for the frameworks that have them, but they are not the mechanism. + +Why each pack ships its own schema +---------------------------------- +Proof-spec v1.0 permits unknown top-level members, so a bundle carrying an +unrecognised extension **passes v1.0 validation without that extension +being checked at all** (``proof-spec/v1.0/SPEC.md`` § 8.1). A green v1.0 +validation is therefore not assurance about fields v1.0 never defined. Each +pack closes that gap for its own namespace by publishing a schema and being +validated against it here, in addition to v1.0. + +Discovery is a directory scan — deliberately +-------------------------------------------- +This module uses ``importlib.resources`` over the packaged ``packs/`` +directory and nothing else. It does **not** use ``importlib.metadata``, +``pkg_resources``, or entry points: third-party installable packs are a +separate, larger question (plugin discovery, trust, and versioning of +code shipped by someone else), tracked as CR-010.R10. Keeping discovery to +first-party packaged data means a pack cannot arrive from an untrusted +distribution as a side effect of installing an unrelated library. + +Failure posture: **fail closed, loudly** +---------------------------------------- +Every load error raises :class:`CompliancePackError`. A malformed pack is +never skipped with a warning. A silently-skipped compliance pack is a +governance regression indistinguishable from "this framework does not +apply" — the same reasoning that rejected ``skipif``-when-file-absent for +the anti-fabrication gate in CR-B10.5. + +References: + - CR-010.R3 — Compliance packs as data + - ``graqle/pct/schema/proof-spec/v1.0/SPEC.md`` § 8.1 — extension posture + - :mod:`graqle.compliance.claim_limits.taxonomy` — the ``x-`` namespace regex +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from typing import Any + +import jsonschema +import yaml + +from graqle.compliance.claim_limits.taxonomy import is_valid_claim_limit + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +#: Filenames a pack directory must contain. +PACK_MANIFEST_FILENAME: str = "pack.yaml" +PACK_SCHEMA_FILENAME: str = "schema.json" + +#: Keys required at the top level of ``pack.yaml``. +_REQUIRED_MANIFEST_KEYS: tuple[str, ...] = ( + "namespace", + "pack_version", + "framework", +) + +#: Namespaces a pack may not claim, mapped to the pack that owns each. +#: +#: These are first-party namespaces whose vocabulary is defined in code and +#: relied upon by governance surfaces. The extension regex alone does not +#: protect them — ``x-ai-eu`` is a perfectly well-formed ``x-`` namespace — +#: so without this check an operator-authored pack could declare +#: ``namespace: x-ai-eu`` and shadow the EU AI Act vocabulary that +#: compliance reporting reads. Reserving them is cheap; discovering the +#: shadowing during an audit is not. +RESERVED_NAMESPACES: dict[str, str] = { + "x-ai-eu": "graqle.pct.extensions.x_ai_eu (EU AI Act)", + "x-sox": "graqle/compliance/packs/x_sox (SOX/COSO)", +} + +#: The packaged directory permitted to declare each reserved namespace. +#: ``x-ai-eu`` maps to ``None`` because it is owned by a code module, not by +#: a pack directory — no pack may claim it. +_RESERVED_OWNER_DIRS: dict[str, str | None] = { + "x-ai-eu": None, + "x-sox": "x_sox", +} + + +class CompliancePackError(Exception): + """Raised when a compliance pack is missing, malformed, or invalid. + + One exception type for every failure mode so a caller loading untrusted + or operator-authored packs has a single thing to catch. + """ + + +# --------------------------------------------------------------------------- +# Pack value object +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class CompliancePack: + """A loaded, validated compliance pack. + + Attributes: + namespace: The ``x-``-prefixed extension namespace, e.g. ``"x-sox"``. + Guaranteed to satisfy the taxonomy's extension regex. + pack_version: The pack's own version string, independent of both the + SDK version and the proof-spec version. + framework: Human-readable framework name, e.g. + ``"Sarbanes-Oxley Act / COSO Internal Control"``. + claim_limits: Namespaced claim-limit values this pack contributes. + Every entry is guaranteed to pass + :func:`~graqle.compliance.claim_limits.taxonomy.is_valid_claim_limit`. + fields: The field definitions from the manifest, keyed by field name. + manifest: The full parsed ``pack.yaml``, for callers needing a key + this dataclass does not surface. + schema: The parsed ``schema.json`` the manifest was validated against. + """ + + namespace: str + pack_version: str + framework: str + claim_limits: tuple[str, ...] = () + fields: dict[str, Any] = field(default_factory=dict) + manifest: dict[str, Any] = field(default_factory=dict) + schema: dict[str, Any] = field(default_factory=dict) + + def qualify(self, field_name: str) -> str: + """Return ``field_name`` prefixed with this pack's namespace. + + ``pack.qualify("control_id") -> "x-sox:control_id"`` — the same + ``{namespace}:{field}`` shape the typed extension dataclasses emit + from ``to_pct_extension_dict()``. + """ + return f"{self.namespace}:{field_name}" + + +# --------------------------------------------------------------------------- +# Loading +# --------------------------------------------------------------------------- + + +def _packs_root() -> Any: + """Return the packaged ``packs/`` directory as a traversable. + + Uses ``importlib.resources`` rather than ``Path(__file__).parents[N]``: + the latter silently resolves to a *different* directory once installed + into site-packages, which is a latent wheel bug rather than a visible + failure. + """ + from importlib.resources import files as _files + + return _files("graqle.compliance.packs") + + +def _read_pack_text(resource: Any) -> str: + """Read a pack file as text, tolerating a UTF-8 BOM. + + Decoded via ``read_bytes()`` + ``utf-8-sig`` rather than + ``read_text(encoding="utf-8")`` for two measured reasons: + + 1. A BOM is fatal to ``json.loads`` (``Unexpected UTF-8 BOM``) even + though ``yaml.safe_load`` tolerates one — so a ``schema.json`` saved + by a Windows editor would fail as "not valid JSON" while its + ``pack.yaml`` sibling loaded fine. ``utf-8-sig`` strips the BOM when + present and is a no-op when it is not. + 2. On some Python/zipimport combinations the ``encoding`` argument to + ``Traversable.read_text`` has been unreliable, falling back to the + platform default (``cp1252`` on Windows). Decoding explicitly removes + that dependency. + """ + return resource.read_bytes().decode("utf-8-sig") + + +def _parse_manifest(raw: str, pack_name: str) -> dict[str, Any]: + """Parse and shape-check ``pack.yaml`` content.""" + try: + manifest = yaml.safe_load(raw) + except yaml.YAMLError as exc: + raise CompliancePackError( + f"compliance pack {pack_name!r}: {PACK_MANIFEST_FILENAME} is not " + f"valid YAML: {exc}" + ) from exc + + if not isinstance(manifest, dict): + raise CompliancePackError( + f"compliance pack {pack_name!r}: {PACK_MANIFEST_FILENAME} must " + f"contain a mapping at the top level, got " + f"{type(manifest).__name__}." + ) + + missing = [k for k in _REQUIRED_MANIFEST_KEYS if k not in manifest] + if missing: + raise CompliancePackError( + f"compliance pack {pack_name!r}: {PACK_MANIFEST_FILENAME} is " + f"missing required key(s): {', '.join(sorted(missing))}." + ) + return manifest + + +def _validate_namespace(namespace: Any, pack_name: str) -> str: + """Validate the pack namespace against the taxonomy extension rule. + + The loader registers *through* the existing taxonomy rather than around + it: a pack namespace is exactly an operator-extension claim-limit value, + so the single regex in ``taxonomy.py`` remains the one authority on what + an ``x-`` namespace may look like. + """ + if not isinstance(namespace, str): + raise CompliancePackError( + f"compliance pack {pack_name!r}: namespace must be a string, got " + f"{type(namespace).__name__}." + ) + if not is_valid_claim_limit(namespace): + raise CompliancePackError( + f"compliance pack {pack_name!r}: namespace {namespace!r} is not a " + f"valid extension namespace — it must match " + f"^x-[a-z0-9_-]{{1,64}}$ (lowercase, 'x-' prefixed)." + ) + # A reserved namespace may only be claimed by the first-party pack that + # owns it. `owner_pack` is the packaged directory name; any other pack + # declaring the same namespace would shadow a governance vocabulary. + owner = RESERVED_NAMESPACES.get(namespace) + if owner is not None and pack_name != _RESERVED_OWNER_DIRS.get(namespace): + raise CompliancePackError( + f"compliance pack {pack_name!r}: namespace {namespace!r} is " + f"RESERVED by {owner} and may not be declared by another pack. " + f"Shadowing a first-party namespace would silently replace the " + f"vocabulary that compliance reporting reads." + ) + return namespace + + +def load_pack(pack_name: str) -> CompliancePack: + """Load and validate one packaged compliance pack by directory name. + + Args: + pack_name: The pack directory name, e.g. ``"x_sox"``. + + Returns: + CompliancePack: The loaded, validated pack. + + Raises: + CompliancePackError: If the directory or either file is absent, the + YAML/JSON is malformed, a required key is missing, the namespace + is not a valid ``x-`` extension namespace, a contributed claim + limit is invalid, or ``pack.yaml`` does not validate against the + pack's own ``schema.json``. + """ + root = _packs_root() + pack_dir = root / pack_name + + if not pack_dir.is_dir(): + raise CompliancePackError( + f"compliance pack {pack_name!r} not found — expected a directory " + f"at graqle/compliance/packs/{pack_name}/." + ) + + manifest_file = pack_dir / PACK_MANIFEST_FILENAME + schema_file = pack_dir / PACK_SCHEMA_FILENAME + for required in (manifest_file, schema_file): + if not required.is_file(): + raise CompliancePackError( + f"compliance pack {pack_name!r} is incomplete — missing " + f"{required.name}. A pack must ship both " + f"{PACK_MANIFEST_FILENAME} and {PACK_SCHEMA_FILENAME}." + ) + + manifest = _parse_manifest(_read_pack_text(manifest_file), pack_name) + + try: + schema = json.loads(_read_pack_text(schema_file)) + except json.JSONDecodeError as exc: + raise CompliancePackError( + f"compliance pack {pack_name!r}: {PACK_SCHEMA_FILENAME} is not " + f"valid JSON: {exc}" + ) from exc + + # The pack must satisfy its own published schema. This is the check that + # proof-spec v1.0 structurally cannot perform for an extension (§ 8.1). + try: + jsonschema.validate(instance=manifest, schema=schema) + except jsonschema.ValidationError as exc: + location = "/".join(str(p) for p in exc.absolute_path) or "" + raise CompliancePackError( + f"compliance pack {pack_name!r}: {PACK_MANIFEST_FILENAME} does not " + f"validate against its own {PACK_SCHEMA_FILENAME} at {location}: " + f"{exc.message}" + ) from exc + except jsonschema.SchemaError as exc: + raise CompliancePackError( + f"compliance pack {pack_name!r}: {PACK_SCHEMA_FILENAME} is not a " + f"valid JSON Schema: {exc.message}" + ) from exc + + namespace = _validate_namespace(manifest["namespace"], pack_name) + + raw_limits = manifest.get("claim_limits") or [] + if not isinstance(raw_limits, list): + raise CompliancePackError( + f"compliance pack {pack_name!r}: claim_limits must be a list, got " + f"{type(raw_limits).__name__}." + ) + invalid = [ + v for v in raw_limits if not isinstance(v, str) or not is_valid_claim_limit(v) + ] + if invalid: + raise CompliancePackError( + f"compliance pack {pack_name!r}: invalid claim_limits " + f"{invalid!r} — each must be a canonical claim limit or match " + f"^x-[a-z0-9_-]{{1,64}}$." + ) + + raw_fields = manifest.get("fields") or {} + if not isinstance(raw_fields, dict): + raise CompliancePackError( + f"compliance pack {pack_name!r}: fields must be a mapping, got " + f"{type(raw_fields).__name__}." + ) + + return CompliancePack( + namespace=namespace, + pack_version=str(manifest["pack_version"]), + framework=str(manifest["framework"]), + claim_limits=tuple(raw_limits), + fields=dict(raw_fields), + manifest=manifest, + schema=schema, + ) + + +def discover_packs() -> tuple[str, ...]: + """Return the names of every packaged pack directory, sorted. + + A directory counts as a pack if it contains a ``pack.yaml``. Sorted so + the order is deterministic across platforms and filesystems. + """ + root = _packs_root() + names: list[str] = [] + for entry in root.iterdir(): + if not entry.is_dir(): + continue + if (entry / PACK_MANIFEST_FILENAME).is_file(): + names.append(entry.name) + return tuple(sorted(names)) + + +def load_all_packs() -> dict[str, CompliancePack]: + """Load every discoverable pack, keyed by namespace. + + Raises: + CompliancePackError: If any pack fails to load (fail-closed — one + bad pack is an error, never a silent omission), or if two packs + declare the same namespace. + """ + packs: dict[str, CompliancePack] = {} + for name in discover_packs(): + pack = load_pack(name) + if pack.namespace in packs: + raise CompliancePackError( + f"duplicate compliance-pack namespace {pack.namespace!r} — " + f"declared by more than one pack directory." + ) + packs[pack.namespace] = pack + return packs diff --git a/graqle/compliance/packs/x_sox/pack.yaml b/graqle/compliance/packs/x_sox/pack.yaml new file mode 100644 index 00000000..262effab --- /dev/null +++ b/graqle/compliance/packs/x_sox/pack.yaml @@ -0,0 +1,166 @@ +# Compliance pack — Sarbanes-Oxley / COSO internal controls (CR-010.R3) +# +# This file is DATA. Adding the next framework (NIST AI RMF, ISO/IEC 42001, +# SOC 2, HIPAA) means writing a sibling directory with a pack.yaml and a +# schema.json — no Python and no engine change. +# +# The loader validates this file against the schema.json beside it, because +# proof-spec v1.0 permits unknown members and therefore cannot check an +# extension namespace itself (proof-spec/v1.0/SPEC.md § 8.1). + +namespace: x-sox +pack_version: "1.0" +framework: Sarbanes-Oxley Act / COSO Internal Control — Integrated Framework + +description: >- + Binds an AI-assisted decision to a named internal control, a + financial-statement assertion, and a reporting period, so that a decision + taken during a financial close is auditable offline. The + management-review fields are the SOX-vocabulary counterpart of the EU AI + Act Article 14 human-oversight gate: same mechanics, different words. + +references: + - Sarbanes-Oxley Act § 302 (corporate responsibility for financial reports) + - Sarbanes-Oxley Act § 404 (management assessment of internal controls) + - COSO Internal Control — Integrated Framework (2013) + - COSO Enterprise Risk Management (2017) + +# Claim limits this pack contributes. Each matches the operator-extension +# regex in graqle/compliance/claim_limits/taxonomy.py, so they are accepted +# by is_valid_claim_limit() with no taxonomy edit — that is precisely what +# makes a pack authorable as data. +claim_limits: + - x-sox-not-an-audit-opinion + - x-sox-management-review-required + - x-sox-icfr-scope + - x-sox-preliminary-unreviewed + +# Field vocabulary. Mirrors graqle/pct/extensions/x_sox.py, which is the +# typed bootstrap for this pack. +fields: + control_id: + type: string + required: true + description: >- + The operator's named internal control (e.g. FCC-1042). Free-form, + because control catalogues are operator-specific, but never empty. + + assertion: + type: enum + required: true + values: + - existence + - completeness + - accuracy + - cutoff + - valuation + - rights_and_obligations + - presentation_and_disclosure + description: The financial-statement assertion the control supports. + + reporting_period_start: + type: date + required: true + description: ISO-8601 YYYY-MM-DD. Date-only — a reporting period is a + calendar concept, and timestamps would invite timezone ambiguity at a + period boundary. + + reporting_period_end: + type: date + required: true + description: ISO-8601 YYYY-MM-DD. Must not precede reporting_period_start. + + management_review_status: + type: enum + required: true + values: + - not_required + - pending + - completed + - waived_with_reason + description: >- + Management-review-control status. The SOX counterpart of the Article 14 + human-oversight mode. + + control_framework: + type: enum + required: false + values: + - coso_2013 + - coso_erm_2017 + - custom + description: COSO framework revision the control is expressed under. + + fiscal_period_label: + type: string + required: false + description: >- + Operator's own label, e.g. FY2026-Q3. Carried verbatim and never + parsed, because fiscal calendars are not uniform across filers. + + preparer_token: + type: pseudonym + required: false + default: omit + pii_classification: pseudonymous + description: >- + HMAC-SHA256 hex token (64 lowercase hex chars) or the sentinel "omit". + Raw identifiers are rejected at construction. The identity-to-token + mapping stays with the operator and is never written to a pack, a + token, or an evidence artifact. + + reviewer_token: + type: pseudonym + required: false + default: omit + pii_classification: pseudonymous + description: >- + As preparer_token. An auditor can check that preparer and reviewer + tokens differ (segregation of duties) without learning either identity. + + management_review_ref: + type: uri + required: false + required_when: + field: management_review_status + in: [completed, waived_with_reason] + description: >- + Evidence pointer for the review. Required when review is asserted as + completed or waived — a review claimed without evidence is + unverifiable. + + control_operating_effectiveness: + type: enum + required: false + values: + - effective + - deficiency + - significant_deficiency + - material_weakness + description: SOX 404 operating-effectiveness conclusion, escalating. + + icfr_scope: + type: boolean + required: false + description: Whether the control is in scope for internal control over + financial reporting. + + policy_version: + type: string + required: false + description: >- + Content-addressed SHA-256 of the active baseline-doc at issuance. + Mirrors x-ai-eu:policy_version so drift between an issued token and the + baseline it was signed against is detectable. + +# Notes for auditors and pack authors. +notes: + leaf_hash: >- + x-sox fields are carried, not signed. The Merkle leaf is computed over a + frozen 5-field allowlist (LEAF_HASH_FIELDS) and every other key is + projected away, so no field in this pack can alter a leaf hash or a + signed preimage. Adding a field here is additive and cannot invalidate a + previously-issued bundle. + validation: >- + Validating a bundle against proof-spec v1.0 does NOT validate this + namespace. Validate against this pack's schema.json in addition. diff --git a/graqle/compliance/packs/x_sox/schema.json b/graqle/compliance/packs/x_sox/schema.json new file mode 100644 index 00000000..646e9474 --- /dev/null +++ b/graqle/compliance/packs/x_sox/schema.json @@ -0,0 +1,118 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://graqle.com/schema/compliance-pack/x-sox/1.0/pack.schema.json", + "title": "GraQle compliance pack — x-sox (SOX / COSO)", + "description": "Schema for the x-sox compliance pack manifest. Proof-spec v1.0 permits unknown members and therefore cannot validate an extension namespace (SPEC.md section 8.1); this schema closes that gap for x-sox and MUST be applied in addition to v1.0.", + "type": "object", + "required": ["namespace", "pack_version", "framework", "fields"], + "properties": { + "namespace": { + "type": "string", + "const": "x-sox", + "description": "Fixed for this pack. The loader independently re-checks it against the taxonomy extension regex." + }, + "pack_version": { + "type": "string", + "pattern": "^[0-9]+\\.[0-9]+$", + "description": "Pack version, independent of both the SDK version and the proof-spec version." + }, + "framework": { + "type": "string", + "minLength": 1 + }, + "description": { + "type": "string" + }, + "references": { + "type": "array", + "items": { "type": "string", "minLength": 1 } + }, + "claim_limits": { + "type": "array", + "items": { + "type": "string", + "pattern": "^x-[a-z0-9_-]{1,64}$", + "description": "Must match the operator-extension namespace regex in graqle/compliance/claim_limits/taxonomy.py." + }, + "uniqueItems": true + }, + "fields": { + "type": "object", + "minProperties": 1, + "description": "Field vocabulary, keyed by field name.", + "required": [ + "control_id", + "assertion", + "reporting_period_start", + "reporting_period_end", + "management_review_status" + ], + "additionalProperties": { + "type": "object", + "required": ["type", "required"], + "properties": { + "type": { + "type": "string", + "enum": [ + "string", + "enum", + "date", + "uri", + "boolean", + "pseudonym" + ] + }, + "required": { "type": "boolean" }, + "description": { "type": "string" }, + "values": { + "type": "array", + "items": { "type": "string", "minLength": 1 }, + "minItems": 1, + "uniqueItems": true + }, + "default": {}, + "pii_classification": { + "type": "string", + "enum": ["none", "pseudonymous", "personal"], + "description": "Declared so an offline auditor knows the field semantics without holding the operator's identity mapping." + }, + "required_when": { + "type": "object", + "required": ["field", "in"], + "properties": { + "field": { "type": "string", "minLength": 1 }, + "in": { + "type": "array", + "items": { "type": "string" }, + "minItems": 1 + } + }, + "additionalProperties": false + } + }, + "allOf": [ + { + "if": { "properties": { "type": { "const": "enum" } }, "required": ["type"] }, + "then": { "required": ["values"] } + }, + { + "if": { "properties": { "type": { "const": "pseudonym" } }, "required": ["type"] }, + "then": { + "required": ["pii_classification"], + "properties": { + "pii_classification": { "const": "pseudonymous" } + }, + "description": "A pseudonym field must declare itself pseudonymous, so a pack cannot quietly carry raw identifiers." + } + } + ], + "additionalProperties": false + } + }, + "notes": { + "type": "object", + "additionalProperties": { "type": "string" } + } + }, + "additionalProperties": false +} diff --git a/graqle/pct/extensions/x_sox.py b/graqle/pct/extensions/x_sox.py new file mode 100644 index 00000000..4340c108 --- /dev/null +++ b/graqle/pct/extensions/x_sox.py @@ -0,0 +1,391 @@ +"""PCT extension namespace ``x-sox`` — SOX / COSO internal controls. + +NEW extension namespace, authored by Quantamix Solutions per CR-010.R3. +Mirrors the OPSF naming convention ``x-{framework}:{field}`` and the +existing sibling :mod:`graqle.pct.extensions.x_ai_eu`. + +Where ``x-ai-eu`` surfaces an operator's *EU AI Act* posture, ``x-sox`` +surfaces an operator's *financial-controls* posture: which named internal +control an AI-assisted decision was performed under, which +financial-statement assertion it supports, the reporting period it falls +in, and whether the required management review happened. + +The motivating use case is an AI-assisted financial close. A decision +emitted during close must be bindable to a **named control** and a +**reporting period**, and an auditor must be able to check that binding +**offline** — without GraQle, and without the operator's HR directory. + +This module exports: + - :data:`X_SOX_NAMESPACE` — the canonical namespace prefix ``"x-sox"``. + - :class:`XSoxExtension` — frozen dataclass for the 13-field payload. + - :func:`is_pseudonym_token` — the preparer/reviewer token predicate. + - :data:`PSEUDONYM_OMITTED` — the explicit "not provided" sentinel. + +Relationship to the signed proof (read before adding fields) +------------------------------------------------------------ +Extension fields are **carried, not trusted**. The Merkle leaf is computed +over a frozen 5-field allowlist +(:data:`graqle.governance.tamper_evidence.leaf_input_schema.LEAF_HASH_FIELDS`) +and ``project_leaf_input()`` drops every other key, so **nothing in this +module can alter a leaf hash or a signed preimage**. Two consequences, +both deliberate: + +1. Adding a field here is a MINOR, additive change. It cannot invalidate a + previously-issued bundle. +2. Validating a bundle against proof-spec v1.0 does **not** validate this + namespace (v1.0 permits unknown members — see ``proof-spec/v1.0/SPEC.md`` + § 8.1). This namespace therefore ships its own schema, in + ``graqle/compliance/packs/x_sox/schema.json``, and must be validated + against it *in addition* to v1.0. + +Why preparer/reviewer are pseudonym tokens +------------------------------------------ +SOX evidence names people. Those names would otherwise travel inside +long-lived, widely-shared audit artifacts (exported evidence files, PCT +payloads handed to external auditors) with no practical redaction path +once distributed. So this module refuses to carry them: the fields accept +a 64-char lowercase-hex HMAC-SHA256 token or the explicit sentinel +``"omit"``, and **reject raw identifiers at construction time**. + +The identity-to-token mapping stays with the operator. It is never written +to a pack, a token, or an evidence artifact. GraQle never needs it: an +auditor checks that *the same* preparer token differs from *the same* +reviewer token (segregation of duties) without learning either identity. +Operators SHOULD salt per reporting period so tokens cannot be correlated +across periods. + +References: + - CR-010.R3 — Compliance packs as data (``x-sox`` first) + - GRAQLE_SDK_ENTERPRISE_REQUIREMENTS.md:138-158 — source requirement + - Sarbanes-Oxley Act § 302 / § 404; COSO Internal Control — Integrated + Framework (2013); COSO ERM (2017) + - Companion module: :mod:`graqle.pct.extensions.x_ai_eu` + - Extension posture: ``graqle/pct/schema/proof-spec/v1.0/SPEC.md`` § 8.1 +""" + +from __future__ import annotations + +import re +from dataclasses import asdict, dataclass +from datetime import date as _date +from typing import Any, Literal + +# --------------------------------------------------------------------------- +# Namespace constant +# --------------------------------------------------------------------------- + +#: Canonical OPSF-style namespace prefix for the SOX/COSO extension. +X_SOX_NAMESPACE: str = "x-sox" + + +# --------------------------------------------------------------------------- +# Pseudonym discipline +# --------------------------------------------------------------------------- + +#: Explicit "not provided" sentinel for the preparer/reviewer fields. +#: +#: A sentinel rather than ``None`` so that "deliberately omitted" is +#: distinguishable in an audit artifact from "field forgotten". An auditor +#: reading ``"omit"`` knows the operator made a choice. +PSEUDONYM_OMITTED: str = "omit" + +#: A preparer/reviewer token is exactly a lowercase hex SHA-256/HMAC digest. +#: Uppercase is rejected so the same identity cannot produce two distinct +#: token spellings (which would silently defeat segregation-of-duties checks). +_PSEUDONYM_TOKEN_RE: re.Pattern[str] = re.compile(r"^[a-f0-9]{64}$") + +#: Fields subject to the pseudonym rule. +_PSEUDONYM_FIELDS: tuple[str, ...] = ("preparer_token", "reviewer_token") + + +def is_pseudonym_token(value: str) -> bool: + """Return True iff ``value`` is an acceptable pseudonym token. + + Acceptable means: the sentinel :data:`PSEUDONYM_OMITTED`, or a 64-char + lowercase-hex digest (HMAC-SHA256 of the identity under an + operator-held, per-period salt). + + Args: + value: Candidate token. + + Returns: + bool: ``True`` if acceptable, ``False`` otherwise. + + Raises: + TypeError: If ``value`` is not a ``str``. + """ + if not isinstance(value, str): + raise TypeError( + f"is_pseudonym_token expects str, got {type(value).__name__}" + ) + if value == PSEUDONYM_OMITTED: + return True + return bool(_PSEUDONYM_TOKEN_RE.match(value)) + + +# --------------------------------------------------------------------------- +# Enum literals — closed vocabularies +# --------------------------------------------------------------------------- + +#: COSO framework revision the control is expressed under. +SoxControlFramework = Literal[ + "coso_2013", + "coso_erm_2017", + "custom", +] + +#: Financial-statement assertions (the classic audit assertion set). +SoxAssertion = Literal[ + "existence", + "completeness", + "accuracy", + "cutoff", + "valuation", + "rights_and_obligations", + "presentation_and_disclosure", +] + +#: Management-review-control status. This is the SOX-vocabulary counterpart +#: of the EU AI Act Article 14 human-oversight mode: same mechanics +#: (a human must look before the outcome is relied upon), different words. +SoxManagementReviewStatus = Literal[ + "not_required", + "pending", + "completed", + "waived_with_reason", +] + +#: SOX 404 operating-effectiveness conclusion, in escalating severity. +SoxOperatingEffectiveness = Literal[ + "effective", + "deficiency", + "significant_deficiency", + "material_weakness", +] + +#: Review statuses that REQUIRE a corroborating evidence pointer. A claim +#: that review completed (or was waived) is unfalsifiable without one, which +#: is precisely the kind of self-attestation an auditor must reject. +_REVIEW_STATUSES_REQUIRING_REF: frozenset[str] = frozenset( + {"completed", "waived_with_reason"} +) + +#: ISO-8601 calendar date, ``YYYY-MM-DD``. Deliberately date-only: a +#: reporting period is a calendar concept, and admitting timestamps would +#: invite timezone ambiguity into a period boundary. +_ISO_DATE_RE: re.Pattern[str] = re.compile(r"^\d{4}-\d{2}-\d{2}$") + + +# --------------------------------------------------------------------------- +# Extension dataclass +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class XSoxExtension: + """The SOX/COSO PCT extension payload. + + All validation happens at construction time in :meth:`__post_init__`, + so an instance that exists is an instance that is safe to emit. + + Attributes: + control_id: REQUIRED. The operator's named internal control (e.g. + ``"FCC-1042"``). Free-form because control catalogues are + operator-specific, but must be non-empty and non-whitespace. + assertion: REQUIRED. One of :data:`SoxAssertion`. + reporting_period_start: REQUIRED. ISO-8601 ``YYYY-MM-DD``. + reporting_period_end: REQUIRED. ISO-8601 ``YYYY-MM-DD``. Must not + precede ``reporting_period_start``. + management_review_status: REQUIRED. One of + :data:`SoxManagementReviewStatus`. + control_framework: OPTIONAL. One of :data:`SoxControlFramework`. + fiscal_period_label: OPTIONAL. Operator's label, e.g. ``"FY2026-Q3"``. + Carried verbatim; never parsed (fiscal calendars are not uniform). + preparer_token: OPTIONAL pseudonym token. Defaults to + :data:`PSEUDONYM_OMITTED`. + reviewer_token: OPTIONAL pseudonym token. Defaults to + :data:`PSEUDONYM_OMITTED`. + management_review_ref: CONDITIONAL — REQUIRED when + ``management_review_status`` is ``completed`` or + ``waived_with_reason``. URI to the review evidence. + control_operating_effectiveness: OPTIONAL. One of + :data:`SoxOperatingEffectiveness`. + icfr_scope: OPTIONAL. Whether the control is in scope for internal + control over financial reporting. + policy_version: OPTIONAL. Content-addressed SHA-256 of the active + baseline-doc at issuance, mirroring + ``x-ai-eu:policy_version``, so drift between the token and the + baseline it was signed against is detectable. + + Raises: + ValueError: On any violated constraint (see :meth:`__post_init__`). + TypeError: If a field carries the wrong Python type. + """ + + control_id: str + assertion: SoxAssertion + reporting_period_start: str + reporting_period_end: str + management_review_status: SoxManagementReviewStatus + control_framework: SoxControlFramework | None = None + fiscal_period_label: str | None = None + preparer_token: str = PSEUDONYM_OMITTED + reviewer_token: str = PSEUDONYM_OMITTED + management_review_ref: str | None = None + control_operating_effectiveness: SoxOperatingEffectiveness | None = None + icfr_scope: bool | None = None + policy_version: str | None = None + + def __post_init__(self) -> None: + """Validate every constraint that makes the payload auditable. + + Enforced, in order: + + 1. ``control_id`` is a non-empty, non-whitespace string — the whole + point of the namespace is binding to a *named* control. + 2. Both reporting-period dates are ISO-8601 ``YYYY-MM-DD``, are real + calendar dates, and the period does not run backwards. Shape and + calendar validity are separate checks: the regex accepts + ``"2026-13-45"``, so the value is also parsed with + ``date.fromisoformat()`` before comparison. + 3. Preparer/reviewer values are pseudonym tokens, never raw + identifiers. + 4. ``management_review_ref`` is present and non-empty when the + status asserts that review happened or was waived. + + Empty and whitespace-only strings are treated as missing throughout + (matching :class:`~graqle.pct.extensions.x_ai_eu.XAiEuExtension`): + a blank ``management_review_ref`` would pass a naive truthy check + while being operationally indistinguishable from "absent" to an + auditor. + """ + # 1. control_id — the binding target. + if not isinstance(self.control_id, str): + raise TypeError( + f"control_id must be str, got {type(self.control_id).__name__}" + ) + if not self.control_id.strip(): + raise ValueError( + "control_id is required and must be non-empty, non-whitespace " + "— an x-sox payload exists to bind a decision to a NAMED control." + ) + + # 2. Reporting period — well-formed, a REAL calendar date, and not + # inverted. The regex alone is insufficient: it constrains shape, + # not calendar validity, so "2026-13-45" satisfies it. Parsing + # with date.fromisoformat() is what rejects an impossible date + # before it can be signed into an audit record. + parsed: dict[str, _date] = {} + for field_name in ("reporting_period_start", "reporting_period_end"): + value = getattr(self, field_name) + if not isinstance(value, str): + raise TypeError( + f"{field_name} must be str, got {type(value).__name__}" + ) + # Shape first, so the error message names the expected format + # rather than leaking a stdlib parser message. fromisoformat() + # also accepts forms this field does not permit (e.g. "20260701" + # on 3.11+, and full timestamps), so the regex is still load-bearing. + if not _ISO_DATE_RE.match(value): + raise ValueError( + f"{field_name} must be an ISO-8601 date (YYYY-MM-DD), " + f"got {value!r}." + ) + try: + parsed[field_name] = _date.fromisoformat(value) + except ValueError as exc: + raise ValueError( + f"{field_name} is not a real calendar date: {value!r} " + f"({exc})." + ) from exc + if parsed["reporting_period_end"] < parsed["reporting_period_start"]: + raise ValueError( + f"reporting_period_end ({self.reporting_period_end}) must not " + f"precede reporting_period_start ({self.reporting_period_start})." + ) + + # 3. Pseudonym discipline — reject raw identifiers before they can + # reach any audit artifact. + for field_name in _PSEUDONYM_FIELDS: + value = getattr(self, field_name) + if not isinstance(value, str): + raise TypeError( + f"{field_name} must be str, got {type(value).__name__}" + ) + if not is_pseudonym_token(value): + raise ValueError( + f"{field_name} must be a 64-character lowercase-hex " + f"HMAC-SHA256 token or the sentinel {PSEUDONYM_OMITTED!r}; " + f"raw identifiers (names, emails, employee IDs) are " + f"prohibited because this value travels inside long-lived, " + f"widely-shared audit artifacts with no redaction path." + ) + + # 4. A completed/waived review must point at its evidence. + ref = self.management_review_ref + ref_missing = ref is None or (isinstance(ref, str) and not ref.strip()) + if self.management_review_status in _REVIEW_STATUSES_REQUIRING_REF and ref_missing: + raise ValueError( + f"management_review_ref is required (non-empty, non-whitespace) " + f"when management_review_status is " + f"{self.management_review_status!r} — a review asserted without " + f"an evidence pointer is unverifiable." + ) + + def to_pct_extension_dict(self) -> dict[str, Any]: + """Convert to the ``{"x-sox:": }`` shape. + + Returns a dict ready to be placed inside the PCT payload's + ``extensions`` field. Fields that are ``None`` are omitted so the + payload stays minimal. + + The pseudonym sentinel :data:`PSEUDONYM_OMITTED` **is** emitted when + set explicitly — "the operator chose not to name a preparer" is + itself audit-relevant, and silently dropping it would erase that + distinction. + """ + out: dict[str, Any] = {} + for key, value in asdict(self).items(): + if value is None: + continue + if isinstance(value, list) and not value: + continue + out[f"{X_SOX_NAMESPACE}:{key}"] = value + return out + + @classmethod + def from_pct_extension_dict(cls, ext: dict[str, Any]) -> "XSoxExtension": + """Parse a ``{"x-sox:": ...}`` dict back to a dataclass. + + Keys outside this namespace, and unknown keys within it, are + ignored — forward-compatibility with future namespace revisions, + matching ``XAiEuExtension.from_pct_extension_dict``. + + The result is constructed through the normal ``__init__``, so a + payload that would violate any constraint in :meth:`__post_init__` + raises rather than yielding an invalid instance. Round-tripping is + therefore validating, not merely mechanical. + + Raises: + ValueError: If required fields are absent from ``ext`` or any + constraint is violated. + """ + prefix = f"{X_SOX_NAMESPACE}:" + known_fields = {f for f in cls.__dataclass_fields__} + kwargs: dict[str, Any] = {} + for key, value in ext.items(): + if not key.startswith(prefix): + continue + field_name = key[len(prefix) :] + if field_name in known_fields: + kwargs[field_name] = value + try: + return cls(**kwargs) + except TypeError as exc: + # Missing REQUIRED fields surface as a TypeError from __init__; + # re-raise as ValueError so callers parsing untrusted payloads + # have one exception type to catch for "this payload is bad". + raise ValueError( + f"x-sox extension payload is missing required fields or has " + f"unusable values: {exc}" + ) from exc diff --git a/pyproject.toml b/pyproject.toml index c9553d95..60fc5c0f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -204,6 +204,13 @@ artifacts = [ "graqle/pct/schema/proof-spec/**/*.md", "graqle/pct/schema/conformance/*.json", "graqle/pct/schema/conformance/fixtures/*", + # CR-010.R3: compliance packs are DATA — a framework is a pack.yaml plus a + # schema.json, so adding NIST AI RMF / ISO 42001 needs no Python. These MUST + # ship in the wheel or the packs silently vanish from an installed SDK and + # only first-party imports keep working. Loaded via importlib.resources + # (never Path(__file__).parents[N], which resolves elsewhere once installed). + "graqle/compliance/packs/**/*.yaml", + "graqle/compliance/packs/**/*.json", ] [tool.hatch.build.targets.wheel] diff --git a/tests/test_compliance/test_packs/test_management_review_gate.py b/tests/test_compliance/test_packs/test_management_review_gate.py new file mode 100644 index 00000000..04730f1b --- /dev/null +++ b/tests/test_compliance/test_packs/test_management_review_gate.py @@ -0,0 +1,260 @@ +"""Tests for graqle.compliance.management_review_gate (CR-010.R3). + +Two jobs: + 1. Verify the new SOX-vocabulary gate behaves correctly. + 2. Verify — as regression — that the live Article 14 EU AI Act surface is + completely UNCHANGED (AC-5). That second job is the point of the CR's + additive design and must fail loudly if anyone later "unifies" the two. +""" + +from __future__ import annotations + +import dataclasses + +import pytest + +from graqle.compliance.article_14_gate import ( + ARTICLE_14_CLAUSES, + Article14GateResult, + DEFAULT_HUMAN_REVIEW_THRESHOLD, + THRESHOLD_STATUS_CALIBRATED, + THRESHOLD_STATUS_PLACEHOLDER, + check_article_14_human_review, +) +from graqle.compliance.management_review_gate import ( + DEFAULT_MANAGEMENT_REVIEW_THRESHOLD, + MANAGEMENT_REVIEW_ERROR_CODE, + ManagementReviewGateResult, + check_management_review, +) + + +class TestArming: + def test_disarmed_allows_regardless_of_confidence(self): + result = check_management_review(confidence=0.01) + assert result.allowed is True + assert result.reason == "" + + def test_armed_below_threshold_refuses(self): + result = check_management_review( + confidence=0.10, management_review_required=True + ) + assert result.allowed is False + assert "below threshold" in result.reason + + def test_armed_above_threshold_allows(self): + result = check_management_review( + confidence=0.99, management_review_required=True + ) + assert result.allowed is True + + def test_confidence_exactly_at_threshold_allows(self): + # Matches the Article 14 rule: refusal fires strictly below. + result = check_management_review( + confidence=0.75, management_review_required=True, threshold=0.75 + ) + assert result.allowed is True + + def test_just_below_threshold_refuses(self): + result = check_management_review( + confidence=0.749, management_review_required=True, threshold=0.75 + ) + assert result.allowed is False + + def test_no_env_var_arming_path(self, monkeypatch): + # SOX applicability is a property of the control, not of a global + # deployment mode — the EU env var must NOT arm this gate. + monkeypatch.setenv("GRAQLE_EU_AI_ACT_MODE", "1") + result = check_management_review(confidence=0.01) + assert result.allowed is True + + +class TestRefusalEnvelope: + def test_envelope_shape(self): + result = check_management_review( + confidence=0.10, + management_review_required=True, + control_id="FCC-1042", + ) + env = result.to_refusal_envelope() + assert env["success"] is False + assert env["error_code"] == MANAGEMENT_REVIEW_ERROR_CODE + assert env["error_code"] == "MANAGEMENT_REVIEW_REQUIRED" + assert env["control_id"] == "FCC-1042" + assert env["control_vocabulary"] == "management_review" + assert env["next_action"] == "present_to_management_reviewer" + assert env["threshold_status"] == THRESHOLD_STATUS_PLACEHOLDER + + def test_control_id_omitted_when_absent(self): + result = check_management_review( + confidence=0.10, management_review_required=True + ) + assert "control_id" not in result.to_refusal_envelope() + + def test_envelope_on_allowed_result_raises(self): + result = check_management_review(confidence=0.99) + with pytest.raises(RuntimeError, match="allowed gate result"): + result.to_refusal_envelope() + + def test_control_id_appears_in_reason(self): + result = check_management_review( + confidence=0.10, + management_review_required=True, + control_id="REV-7781", + ) + assert "REV-7781" in result.reason + + +class TestValidation: + @pytest.mark.parametrize("bad", [float("nan"), float("inf"), -0.1, 1.1]) + def test_invalid_confidence_raises(self, bad): + with pytest.raises(ValueError): + check_management_review(confidence=bad, management_review_required=True) + + @pytest.mark.parametrize("bad", [float("nan"), float("inf"), -0.1, 1.1]) + def test_invalid_threshold_raises(self, bad): + with pytest.raises(ValueError): + check_management_review( + confidence=0.5, management_review_required=True, threshold=bad + ) + + +class TestThresholdStatus: + def test_default_is_placeholder(self): + # Assert the MARKER, never the float — so wiring calibration later + # cannot silently pass a stale expectation. + result = check_management_review(confidence=0.9) + assert result.threshold_status == THRESHOLD_STATUS_PLACEHOLDER + + def test_calibrated_marker_propagates(self): + result = check_management_review( + confidence=0.9, threshold_status=THRESHOLD_STATUS_CALIBRATED + ) + assert result.threshold_status == THRESHOLD_STATUS_CALIBRATED + + def test_default_threshold_aliases_article_14(self): + # One placeholder value, one place to replace. + assert DEFAULT_MANAGEMENT_REVIEW_THRESHOLD == DEFAULT_HUMAN_REVIEW_THRESHOLD + + +class TestImmutability: + def test_result_is_frozen(self): + result = check_management_review(confidence=0.9) + with pytest.raises(Exception): + result.allowed = False # type: ignore[misc] + + +class TestArticle14ContractUnchanged: + """AC-5 regression: the live EU AI Act surface must not move. + + These tests exist to fail if anyone later "unifies" the two gates by + renaming fields or generalising the error code. Four production + consumers pin this contract: mcp_dev_server.py (3 sites) and + switch_status.py. + """ + + def test_error_code_string_unchanged(self): + result = check_article_14_human_review( + confidence=0.10, human_review_required=True + ) + assert result.to_refusal_envelope()["error_code"] == ( + "ARTICLE_14_HUMAN_REVIEW_REQUIRED" + ) + + def test_default_threshold_value_unchanged(self): + assert DEFAULT_HUMAN_REVIEW_THRESHOLD == 0.75 + + def test_field_names_and_order_unchanged(self): + names = [f.name for f in dataclasses.fields(Article14GateResult)] + assert names == [ + "allowed", + "confidence", + "threshold", + "threshold_status", + "reason", + ] + + def test_clauses_unchanged(self): + assert ARTICLE_14_CLAUSES == ("14(4)(c)", "14(4)(d)") + + def test_next_action_unchanged(self): + result = check_article_14_human_review( + confidence=0.10, human_review_required=True + ) + env = result.to_refusal_envelope() + assert env["next_action"] == "present_diff_to_human_reviewer" + + def test_two_gates_are_distinct_types(self): + # Not an alias of one another — a SOX refusal must never be mistaken + # for an EU AI Act refusal by an isinstance check. + assert ManagementReviewGateResult is not Article14GateResult + + def test_error_codes_are_distinct(self): + sox = check_management_review( + confidence=0.10, management_review_required=True + ).to_refusal_envelope() + eu = check_article_14_human_review( + confidence=0.10, human_review_required=True + ).to_refusal_envelope() + assert sox["error_code"] != eu["error_code"] + + def test_switch_status_probe_envelope_unchanged(self): + from graqle.compliance.switch_status import _probe_article_14_gate + + probe = _probe_article_14_gate() + assert probe["default_threshold"] == 0.75 + assert probe["refusal_error_code"] == "ARTICLE_14_HUMAN_REVIEW_REQUIRED" + + +class TestGateHelperContract: + """Sentinel F-1 (PARTIALLY VALID): pin the cross-module helper coupling. + + The SOX gate reuses three validation helpers that live in + ``article_14_gate`` under underscore-prefixed names, which carry no + stability contract. ``graqle.compliance._gate_helpers`` re-exports them + under public names so callers are insulated from the private spelling, + and these tests make a rename fail HERE — as a named test failure — + rather than as an ImportError at process start, which was the actual + problem worth fixing. + """ + + def test_public_helper_names_resolve(self): + from graqle.compliance import _gate_helpers + + for name in ("coerce_arming_flag", "validate_confidence", "validate_threshold"): + assert callable(getattr(_gate_helpers, name)), name + + def test_helpers_are_the_article_14_implementations(self): + # Same objects — shared logic, not a divergent copy that could drift. + from graqle.compliance import _gate_helpers, article_14_gate + + assert _gate_helpers.coerce_arming_flag is article_14_gate._coerce_bool + assert _gate_helpers.validate_confidence is article_14_gate._validate_confidence + assert _gate_helpers.validate_threshold is article_14_gate._validate_threshold + + def test_both_gates_share_validation_behaviour(self): + # Divergent validation between the two regimes would be a + # correctness bug, not merely a style difference. + for bad in (float("nan"), float("inf"), -0.1, 1.1): + with pytest.raises(ValueError): + check_article_14_human_review( + confidence=bad, human_review_required=True + ) + with pytest.raises(ValueError): + check_management_review( + confidence=bad, management_review_required=True + ) + + def test_sox_gate_does_not_import_private_names_directly(self): + import pathlib + + from graqle.compliance import management_review_gate + + text = pathlib.Path(management_review_gate.__file__).read_text( + encoding="utf-8" + ) + # The import block must go through _gate_helpers, not reach into + # article_14_gate's private surface. + assert "_coerce_bool" not in text + assert "_validate_confidence" not in text + assert "_validate_threshold" not in text diff --git a/tests/test_compliance/test_packs/test_pack_loader.py b/tests/test_compliance/test_packs/test_pack_loader.py new file mode 100644 index 00000000..00bcc591 --- /dev/null +++ b/tests/test_compliance/test_packs/test_pack_loader.py @@ -0,0 +1,410 @@ +"""Tests for graqle.compliance.packs (CR-010.R3). + +Covers the data-driven pack loader: happy path, the fail-closed posture on +every malformation, and AC-3 — that a framework pack is authorable as +schema + YAML with no Python and no engine change. +""" + +from __future__ import annotations + +import json +import shutil + +import pytest +import yaml + +from graqle.compliance.claim_limits.taxonomy import is_valid_claim_limit +from graqle.compliance.packs import ( + CompliancePack, + CompliancePackError, + discover_packs, + load_all_packs, + load_pack, +) +from graqle.compliance.packs import _loader + + +class TestShippedSoxPack: + def test_x_sox_is_discoverable(self): + assert "x_sox" in discover_packs() + + def test_loads(self): + pack = load_pack("x_sox") + assert isinstance(pack, CompliancePack) + assert pack.namespace == "x-sox" + assert pack.pack_version == "1.0" + assert "Sarbanes-Oxley" in pack.framework + + def test_declares_required_control_fields(self): + pack = load_pack("x_sox") + for required in ( + "control_id", + "assertion", + "reporting_period_start", + "reporting_period_end", + "management_review_status", + ): + assert required in pack.fields, required + + def test_qualify_matches_extension_emit_shape(self): + assert load_pack("x_sox").qualify("control_id") == "x-sox:control_id" + + def test_load_all_keys_by_namespace(self): + assert "x-sox" in load_all_packs() + + def test_pseudonym_fields_declare_pii_classification(self): + # An offline auditor must be able to tell a pseudonymous field from a + # plain string without holding the operator's identity mapping. + pack = load_pack("x_sox") + for name in ("preparer_token", "reviewer_token"): + assert pack.fields[name]["pii_classification"] == "pseudonymous" + + +class TestClaimLimitIntegration: + """The load-bearing fact behind AC-3: no taxonomy edit is needed.""" + + def test_pack_claim_limits_pass_existing_taxonomy_unmodified(self): + pack = load_pack("x_sox") + assert pack.claim_limits, "pack contributes no claim limits" + for value in pack.claim_limits: + assert is_valid_claim_limit(value), value + + +@pytest.fixture() +def temp_pack_root(tmp_path, monkeypatch): + """Point the loader at a temporary packs root. + + Lets the negative cases build deliberately-broken packs without writing + into the installed package. + """ + monkeypatch.setattr(_loader, "_packs_root", lambda: tmp_path) + return tmp_path + + +def _write_pack(root, name, manifest, schema=None): + """Write a pack dir; defaults to the real x-sox schema.""" + pack_dir = root / name + pack_dir.mkdir(parents=True, exist_ok=True) + (pack_dir / "pack.yaml").write_text( + yaml.safe_dump(manifest), encoding="utf-8" + ) + if schema is None: + real = _loader._packs_root # noqa: SLF001 — restored real root below + del real + from importlib.resources import files as _files + + src = _files("graqle.compliance.packs") / "x_sox" / "schema.json" + schema_text = src.read_text(encoding="utf-8") + else: + schema_text = json.dumps(schema) + (pack_dir / "schema.json").write_text(schema_text, encoding="utf-8") + return pack_dir + + +def _valid_manifest(**overrides): + # Deliberately NOT a reserved namespace: x-sox and x-ai-eu are reserved + # for their first-party owners, so fixtures use a neutral test namespace. + manifest = { + "namespace": "x-testpack", + "pack_version": "1.0", + "framework": "Test framework", + "fields": { + "control_id": {"type": "string", "required": True}, + "assertion": { + "type": "enum", + "required": True, + "values": ["existence"], + }, + "reporting_period_start": {"type": "date", "required": True}, + "reporting_period_end": {"type": "date", "required": True}, + "management_review_status": { + "type": "enum", + "required": True, + "values": ["not_required"], + }, + }, + } + manifest.update(overrides) + return manifest + + +class TestAuthorableAsDataOnly: + """AC-3: a pack is loadable from data alone — no Python, no engine change.""" + + def test_pack_created_only_from_data_files_loads(self, temp_pack_root): + # Written as data alone — no Python, no engine change. + schema = { + "type": "object", + "required": ["namespace", "pack_version", "framework", "fields"], + "properties": {"pack_version": {"type": "string"}}, + } + _write_pack(temp_pack_root, "x_demo", _valid_manifest(), schema=schema) + pack = load_pack("x_demo") + assert pack.namespace == "x-testpack" + assert pack.fields["control_id"]["required"] is True + + def test_new_namespace_needs_no_taxonomy_edit(self, temp_pack_root): + manifest = _valid_manifest( + namespace="x-nist-ai-rmf", + claim_limits=["x-nist-govern-1-1"], + ) + schema = { + "type": "object", + "required": ["namespace", "pack_version", "framework", "fields"], + "properties": {"namespace": {"type": "string"}}, + } + _write_pack(temp_pack_root, "x_nist", manifest, schema=schema) + pack = load_pack("x_nist") + assert pack.namespace == "x-nist-ai-rmf" + assert is_valid_claim_limit("x-nist-govern-1-1") + + +class TestFailClosed: + """A malformed pack raises. It is never skipped with a warning.""" + + def test_missing_pack_raises(self, temp_pack_root): + with pytest.raises(CompliancePackError, match="not found"): + load_pack("does_not_exist") + + def test_missing_schema_file_raises(self, temp_pack_root): + pack_dir = temp_pack_root / "x_broken" + pack_dir.mkdir() + (pack_dir / "pack.yaml").write_text( + yaml.safe_dump(_valid_manifest()), encoding="utf-8" + ) + with pytest.raises(CompliancePackError, match="incomplete|schema.json"): + load_pack("x_broken") + + def test_missing_manifest_file_raises(self, temp_pack_root): + pack_dir = temp_pack_root / "x_broken" + pack_dir.mkdir() + (pack_dir / "schema.json").write_text("{}", encoding="utf-8") + with pytest.raises(CompliancePackError, match="not found|incomplete"): + load_pack("x_broken") + + def test_malformed_yaml_raises(self, temp_pack_root): + pack_dir = temp_pack_root / "x_broken" + pack_dir.mkdir() + (pack_dir / "pack.yaml").write_text( + "namespace: [unclosed\n", encoding="utf-8" + ) + (pack_dir / "schema.json").write_text("{}", encoding="utf-8") + with pytest.raises(CompliancePackError, match="not valid YAML"): + load_pack("x_broken") + + def test_malformed_json_schema_raises(self, temp_pack_root): + pack_dir = temp_pack_root / "x_broken" + pack_dir.mkdir() + (pack_dir / "pack.yaml").write_text( + yaml.safe_dump(_valid_manifest()), encoding="utf-8" + ) + (pack_dir / "schema.json").write_text("{not json", encoding="utf-8") + with pytest.raises(CompliancePackError, match="not valid JSON"): + load_pack("x_broken") + + def test_yaml_scalar_top_level_raises(self, temp_pack_root): + pack_dir = temp_pack_root / "x_broken" + pack_dir.mkdir() + (pack_dir / "pack.yaml").write_text("just a string\n", encoding="utf-8") + (pack_dir / "schema.json").write_text("{}", encoding="utf-8") + with pytest.raises(CompliancePackError, match="mapping"): + load_pack("x_broken") + + @pytest.mark.parametrize("missing", ["namespace", "pack_version", "framework"]) + def test_missing_required_key_raises(self, temp_pack_root, missing): + manifest = _valid_manifest() + del manifest[missing] + schema = {"type": "object"} + _write_pack(temp_pack_root, "x_broken", manifest, schema=schema) + with pytest.raises(CompliancePackError, match="missing required key"): + load_pack("x_broken") + + @pytest.mark.parametrize( + "bad_ns", + [ + "X-SOX", # uppercase + "sox", # no x- prefix + "x-", # prefix only + "x-" + "a" * 65, # over the 64-char limit + "x-sox!", # illegal char + "x-sox space", # space + ], + ) + def test_invalid_namespace_raises(self, temp_pack_root, bad_ns): + manifest = _valid_manifest(namespace=bad_ns) + schema = {"type": "object"} + _write_pack(temp_pack_root, "x_broken", manifest, schema=schema) + with pytest.raises(CompliancePackError, match="not a valid extension namespace"): + load_pack("x_broken") + + def test_manifest_violating_own_schema_raises(self, temp_pack_root): + # The check proof-spec v1.0 structurally cannot perform (SPEC.md 8.1). + manifest = _valid_manifest(pack_version="not-a-version") + _write_pack(temp_pack_root, "x_broken", manifest) + with pytest.raises(CompliancePackError, match="does not validate"): + load_pack("x_broken") + + def test_invalid_schema_raises(self, temp_pack_root): + manifest = _valid_manifest() + _write_pack( + temp_pack_root, + "x_broken", + manifest, + schema={"type": "not-a-real-type"}, + ) + with pytest.raises(CompliancePackError, match="not a valid JSON Schema"): + load_pack("x_broken") + + def test_invalid_claim_limit_raises(self, temp_pack_root): + manifest = _valid_manifest(claim_limits=["not_x_prefixed"]) + schema = {"type": "object"} + _write_pack(temp_pack_root, "x_broken", manifest, schema=schema) + with pytest.raises(CompliancePackError, match="invalid claim_limits"): + load_pack("x_broken") + + def test_claim_limits_wrong_type_raises(self, temp_pack_root): + manifest = _valid_manifest(claim_limits="x-sox-oops") + schema = {"type": "object"} + _write_pack(temp_pack_root, "x_broken", manifest, schema=schema) + with pytest.raises(CompliancePackError, match="claim_limits must be a list"): + load_pack("x_broken") + + @pytest.mark.parametrize("reserved", ["x-ai-eu", "x-sox"]) + def test_reserved_namespace_cannot_be_shadowed(self, temp_pack_root, reserved): + """Sentinel F-5 (CONFIRMED): x-ai-eu is a well-formed x- namespace. + + The extension regex alone does not protect first-party vocabularies, + so an operator pack could otherwise declare namespace: x-ai-eu and + silently replace the EU AI Act definitions compliance reporting reads. + """ + manifest = _valid_manifest(namespace=reserved) + schema = {"type": "object"} + _write_pack(temp_pack_root, "x_impostor", manifest, schema=schema) + with pytest.raises(CompliancePackError, match="RESERVED"): + load_pack("x_impostor") + + def test_owning_pack_may_declare_its_reserved_namespace(self, temp_pack_root): + # x_sox owns x-sox — the reservation must not lock out the owner. + manifest = _valid_manifest(namespace="x-sox") + schema = {"type": "object"} + _write_pack(temp_pack_root, "x_sox", manifest, schema=schema) + assert load_pack("x_sox").namespace == "x-sox" + + def test_shipped_sox_pack_still_loads_under_reservation(self): + # Guards against the reservation breaking the real packaged pack. + assert load_pack("x_sox").namespace == "x-sox" + + def test_duplicate_namespace_raises(self, temp_pack_root): + schema = {"type": "object"} + _write_pack(temp_pack_root, "x_one", _valid_manifest(), schema=schema) + _write_pack(temp_pack_root, "x_two", _valid_manifest(), schema=schema) + with pytest.raises(CompliancePackError, match="duplicate"): + load_all_packs() + + def test_directory_without_manifest_is_not_a_pack(self, temp_pack_root): + (temp_pack_root / "__pycache__").mkdir() + assert "__pycache__" not in discover_packs() + + +class TestEncodingRobustness: + """graq_predict chain 5 (CONFIRMED): a UTF-8 BOM is fatal to json.loads. + + yaml.safe_load tolerates a BOM but json.loads raises + "Unexpected UTF-8 BOM", so a schema.json saved by a Windows editor would + fail as "not valid JSON" while its pack.yaml sibling loaded fine. + """ + + def test_bom_in_both_files_is_tolerated(self, temp_pack_root): + pack_dir = temp_pack_root / "x_bom" + pack_dir.mkdir() + manifest = yaml.safe_dump(_valid_manifest()) + schema = json.dumps({"type": "object"}) + # utf-8-sig writes the BOM, exactly as a Windows editor would. + (pack_dir / "pack.yaml").write_text(manifest, encoding="utf-8-sig") + (pack_dir / "schema.json").write_text(schema, encoding="utf-8-sig") + pack = load_pack("x_bom") + assert pack.namespace == "x-testpack" + + def test_non_ascii_content_round_trips(self, temp_pack_root): + manifest = _valid_manifest(framework="Directive — “quoted” 指令") + _write_pack(temp_pack_root, "x_uni", manifest, schema={"type": "object"}) + assert "指令" in load_pack("x_uni").framework + + +class TestNoEagerLoadingAtImport: + """graq_predict chains 3+4: a pack error must never break the package. + + If load_all_packs() were called at module scope, a single malformed + pack.yaml would raise CompliancePackError during + `import graqle.compliance` and take down every consumer of the package — + turning a recoverable data problem into an unrecoverable import failure. + Loading is lazy by design; this test pins that. + """ + + def test_importing_compliance_does_not_load_packs(self): + import subprocess + import sys + + # Fresh interpreter: if any module-scope call existed, patching the + # loader to explode would surface it as a non-zero exit. + code = ( + "import graqle.compliance.packs._loader as L\n" + "def boom(*a, **k): raise AssertionError('eager load at import')\n" + "L.load_all_packs = boom\n" + "L.load_pack = boom\n" + "import importlib\n" + "importlib.import_module('graqle.compliance')\n" + "print('OK')\n" + ) + proc = subprocess.run( + [sys.executable, "-c", code], capture_output=True, text=True + ) + assert "OK" in proc.stdout, proc.stderr[-500:] + + def test_loader_has_no_module_scope_load_calls(self): + import pathlib + + text = pathlib.Path(_loader.__file__).read_text(encoding="utf-8") + for line in text.splitlines(): + # A load call at column 0 would run at import time. + assert not line.startswith("load_all_packs("), line + assert not line.startswith("PACK_REGISTRY = load"), line + + +class TestWheelSafety: + """AC-7: pack data must resolve the same way it will from a wheel.""" + + def test_packs_root_is_a_traversable_not_a_derived_path(self): + """The root must come from the package, not from __file__ arithmetic. + + Asserted behaviourally rather than by grepping the source: the + module docstring legitimately *mentions* Path(__file__).parents[N] + while explaining why it is wrong, so a text search reports a false + positive on the explanation itself. + """ + from importlib.resources import files as _files + + root = _loader._packs_root() # noqa: SLF001 — asserting the seam + expected = _files("graqle.compliance.packs") + # Same resolved location, obtained the importlib way. + assert str(root) == str(expected) + assert (root / "x_sox" / "pack.yaml").is_file() + + def test_loader_does_not_import_entry_point_machinery(self): + """AC-8: entry-point plugin discovery is R10 scope, not R3.""" + source_path = shutil.os.path.join( + shutil.os.path.dirname(_loader.__file__), "_loader.py" + ) + text = open(source_path, encoding="utf-8").read() + # Match real imports, not prose in the docstring explaining the choice. + assert "import importlib.metadata" not in text + assert "from importlib.metadata" not in text + assert "import pkg_resources" not in text + assert "entry_points(" not in text + + def test_resolves_through_package_traversable(self): + from importlib.resources import files as _files + + root = _files("graqle.compliance.packs") + assert (root / "x_sox" / "pack.yaml").is_file() + assert (root / "x_sox" / "schema.json").is_file() diff --git a/tests/test_compliance/test_packs/test_x_sox_extension.py b/tests/test_compliance/test_packs/test_x_sox_extension.py new file mode 100644 index 00000000..ee8ca343 --- /dev/null +++ b/tests/test_compliance/test_packs/test_x_sox_extension.py @@ -0,0 +1,266 @@ +"""Tests for graqle.pct.extensions.x_sox (CR-010.R3). + +Covers the typed x-sox bootstrap: round-tripping, the pseudonym discipline, +conditional-field enforcement, and the reporting-period constraints. +""" + +from __future__ import annotations + +import pytest + +from graqle.pct.extensions.x_sox import ( + PSEUDONYM_OMITTED, + X_SOX_NAMESPACE, + XSoxExtension, + is_pseudonym_token, +) + +# A valid 64-char lowercase-hex token (HMAC-SHA256 shaped). +TOKEN_A = "a" * 64 +TOKEN_B = "b3f1" + "0" * 60 + + +def _minimal(**overrides): + """Build a minimally-valid extension, with optional overrides.""" + kwargs = { + "control_id": "FCC-1042", + "assertion": "completeness", + "reporting_period_start": "2026-07-01", + "reporting_period_end": "2026-09-30", + "management_review_status": "not_required", + } + kwargs.update(overrides) + return XSoxExtension(**kwargs) + + +class TestNamespace: + def test_namespace_constant(self): + assert X_SOX_NAMESPACE == "x-sox" + + def test_emitted_keys_are_all_namespaced(self): + ext = _minimal() + emitted = ext.to_pct_extension_dict() + assert emitted, "expected a non-empty payload" + for key in emitted: + assert key.startswith("x-sox:"), key + + +class TestRoundTrip: + def test_round_trip_is_lossless(self): + original = _minimal( + control_framework="coso_2013", + fiscal_period_label="FY2026-Q3", + preparer_token=TOKEN_A, + reviewer_token=TOKEN_B, + management_review_status="completed", + management_review_ref="https://audit.example/reviews/9912", + control_operating_effectiveness="effective", + icfr_scope=True, + policy_version="c" * 64, + ) + restored = XSoxExtension.from_pct_extension_dict( + original.to_pct_extension_dict() + ) + assert restored == original + + def test_minimal_round_trip(self): + original = _minimal() + restored = XSoxExtension.from_pct_extension_dict( + original.to_pct_extension_dict() + ) + assert restored == original + + def test_none_fields_are_omitted(self): + emitted = _minimal().to_pct_extension_dict() + assert "x-sox:control_framework" not in emitted + assert "x-sox:icfr_scope" not in emitted + + def test_omit_sentinel_is_emitted_not_dropped(self): + # "the operator chose not to name a preparer" is audit-relevant and + # must survive serialisation — it is not the same as "absent". + emitted = _minimal().to_pct_extension_dict() + assert emitted["x-sox:preparer_token"] == PSEUDONYM_OMITTED + + def test_false_and_zero_are_emitted_not_skipped(self): + """Sentinel F-2 (REFUTED): only None/empty-list are skipped. + + A falsy-but-present value must survive. `icfr_scope=False` means + "explicitly out of ICFR scope", which an auditor must be able to + distinguish from "never assessed". The skip predicate is + `value is None`, deliberately NOT `if value`. + """ + emitted = _minimal(icfr_scope=False).to_pct_extension_dict() + assert emitted["x-sox:icfr_scope"] is False + restored = XSoxExtension.from_pct_extension_dict(emitted) + assert restored.icfr_scope is False + + def test_explicit_none_survives_round_trip(self): + """Sentinel F-4 (REFUTED): from_ filters by key, never by value.""" + payload = _minimal().to_pct_extension_dict() + payload["x-sox:management_review_ref"] = None + restored = XSoxExtension.from_pct_extension_dict(payload) + assert restored.management_review_ref is None + + def test_foreign_and_unknown_keys_ignored(self): + payload = _minimal().to_pct_extension_dict() + payload["x-ai-eu:article_6_classification"] = "non_high_risk" + payload["x-sox:some_future_field"] = "ignored" + restored = XSoxExtension.from_pct_extension_dict(payload) + assert restored == _minimal() + + def test_missing_required_field_raises_value_error(self): + # Callers parsing untrusted payloads get ONE exception type. + with pytest.raises(ValueError): + XSoxExtension.from_pct_extension_dict({"x-sox:control_id": "FCC-1"}) + + +class TestControlBinding: + """AC-2: bind a decision to a named control and a reporting period.""" + + def test_binds_named_control_and_period(self): + emitted = _minimal(control_id="REV-7781").to_pct_extension_dict() + assert emitted["x-sox:control_id"] == "REV-7781" + assert emitted["x-sox:reporting_period_start"] == "2026-07-01" + assert emitted["x-sox:reporting_period_end"] == "2026-09-30" + + @pytest.mark.parametrize("bad", ["", " ", "\t", "\n"]) + def test_empty_control_id_rejected(self, bad): + with pytest.raises(ValueError, match="control_id"): + _minimal(control_id=bad) + + def test_non_string_control_id_rejected(self): + with pytest.raises(TypeError): + _minimal(control_id=1042) + + +class TestReportingPeriod: + @pytest.mark.parametrize( + "bad", + [ + "2026-7-1", # not zero-padded + "07/01/2026", # wrong format + "2026-07-01T00:00Z", # timestamp, not a date + "", + "not-a-date", + "20260701", + ], + ) + def test_malformed_dates_rejected(self, bad): + with pytest.raises(ValueError, match="ISO-8601"): + _minimal(reporting_period_start=bad) + + @pytest.mark.parametrize( + "bad", + [ + "2026-13-45", # month 13, day 45 + "2026-99-99", + "2026-00-10", # month 0 + "2026-02-30", # Feb 30 never exists + "2025-02-29", # 2025 is not a leap year + "2026-04-31", # April has 30 days + ], + ) + def test_calendar_impossible_dates_rejected(self, bad): + """Sentinel F-3 (CONFIRMED): the regex checks shape, not calendar. + + Without date.fromisoformat() these pass validation and get signed + into an audit record carrying a nonsensical reporting period. + """ + with pytest.raises(ValueError, match="not a real calendar date"): + _minimal(reporting_period_start=bad, reporting_period_end="2026-12-31") + + def test_leap_day_accepted_in_leap_year(self): + ext = _minimal( + reporting_period_start="2028-02-29", + reporting_period_end="2028-03-31", + ) + assert ext.reporting_period_start == "2028-02-29" + + def test_inverted_period_rejected(self): + with pytest.raises(ValueError, match="must not precede"): + _minimal( + reporting_period_start="2026-09-30", + reporting_period_end="2026-07-01", + ) + + def test_single_day_period_allowed(self): + ext = _minimal( + reporting_period_start="2026-07-01", + reporting_period_end="2026-07-01", + ) + assert ext.reporting_period_end == "2026-07-01" + + +class TestPseudonymDiscipline: + """AC-4: raw identifiers must never reach an audit artifact.""" + + def test_valid_token_accepted(self): + assert is_pseudonym_token(TOKEN_A) + assert is_pseudonym_token(PSEUDONYM_OMITTED) + + @pytest.mark.parametrize( + "raw", + [ + "jane.doe@example.com", # email + "Jane Doe", # display name + "EMP-00417", # employee ID + "550e8400-e29b-41d4-a716-446655440000", # UUID + "a" * 63, # one char short + "a" * 65, # one char long + "A" * 64, # uppercase hex + "g" * 64, # non-hex chars + "", # empty + " ", # whitespace + "omitted", # near-miss on the sentinel + "OMIT", # wrong case sentinel + ], + ) + def test_raw_identifiers_rejected(self, raw): + assert is_pseudonym_token(raw) is False + with pytest.raises(ValueError, match="prohibited|HMAC-SHA256"): + _minimal(preparer_token=raw) + with pytest.raises(ValueError, match="prohibited|HMAC-SHA256"): + _minimal(reviewer_token=raw) + + def test_non_string_token_rejected(self): + with pytest.raises(TypeError): + is_pseudonym_token(12345) + with pytest.raises(TypeError): + _minimal(preparer_token=None) + + def test_segregation_of_duties_checkable_without_identities(self): + # An auditor can compare tokens without learning who they are. + ext = _minimal(preparer_token=TOKEN_A, reviewer_token=TOKEN_B) + assert ext.preparer_token != ext.reviewer_token + + +class TestManagementReviewLinkage: + @pytest.mark.parametrize("status", ["completed", "waived_with_reason"]) + def test_ref_required_when_review_asserted(self, status): + with pytest.raises(ValueError, match="management_review_ref"): + _minimal(management_review_status=status) + + @pytest.mark.parametrize("status", ["completed", "waived_with_reason"]) + @pytest.mark.parametrize("blank", ["", " ", "\n"]) + def test_blank_ref_treated_as_missing(self, status, blank): + with pytest.raises(ValueError, match="management_review_ref"): + _minimal(management_review_status=status, management_review_ref=blank) + + @pytest.mark.parametrize("status", ["not_required", "pending"]) + def test_ref_not_required_otherwise(self, status): + ext = _minimal(management_review_status=status) + assert ext.management_review_ref is None + + def test_ref_accepted_when_provided(self): + ext = _minimal( + management_review_status="completed", + management_review_ref="https://audit.example/r/1", + ) + assert ext.management_review_ref == "https://audit.example/r/1" + + +class TestImmutability: + def test_frozen(self): + ext = _minimal() + with pytest.raises(Exception): + ext.control_id = "OTHER" # type: ignore[misc]