From f3b098946bc430055430b15ae38154872b111b99 Mon Sep 17 00:00:00 2001 From: anthonyfuller <38588965+anthfuller@users.noreply.github.com> Date: Sun, 20 Sep 2026 16:28:03 -0400 Subject: [PATCH 1/2] feat: add evidence integrity and deterministic replay --- .github/workflows/f7las-ci.yml | 17 ++ ROADMAP.md | 2 +- docs/F7-LAS-QA.md | 11 + examples/canonical-workflow/README.md | 44 +++ schemas/contracts/README.md | 14 +- scripts/validate-contracts.py | 402 +------------------------ src/canonical/evidence.py | 119 ++++++++ src/canonical/replay.py | 84 ++++++ src/canonical/validation.py | 412 ++++++++++++++++++++++++++ tests/test_evidence_replay.py | 181 +++++++++++ 10 files changed, 884 insertions(+), 402 deletions(-) create mode 100644 src/canonical/evidence.py create mode 100644 src/canonical/replay.py create mode 100644 src/canonical/validation.py create mode 100644 tests/test_evidence_replay.py diff --git a/.github/workflows/f7las-ci.yml b/.github/workflows/f7las-ci.yml index 468bd20..405fc1b 100644 --- a/.github/workflows/f7las-ci.yml +++ b/.github/workflows/f7las-ci.yml @@ -66,6 +66,23 @@ jobs: run: 'pytest -q tests/test_behavioral_scenarios.py ' + - name: Generate canonical evidence + run: | + python -m src.canonical.cli \ + --input examples/canonical-workflow/request.json \ + --output "$RUNNER_TEMP/f7las-canonical-evidence.json" \ + --opa-binary "$OPA_BIN" + - name: Verify canonical evidence integrity + run: | + python -m src.canonical.evidence \ + --evidence "$RUNNER_TEMP/f7las-canonical-evidence.json" + - name: Replay canonical evidence deterministically + run: | + python -m src.canonical.replay \ + --input examples/canonical-workflow/request.json \ + --evidence "$RUNNER_TEMP/f7las-canonical-evidence.json" \ + --output "$RUNNER_TEMP/f7las-replayed-evidence.json" \ + --opa-binary "$OPA_BIN" - name: Run unit tests run: 'pytest -q --ignore=tests/test_behavioral_scenarios.py diff --git a/ROADMAP.md b/ROADMAP.md index 1a96e80..23f4cbc 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -15,7 +15,7 @@ The target is an **Executable Reference Implementation**: one offline, determini 3. **Canonical Python + OPA path** — one offline, fail-closed Layers 1–7 workflow is implemented for the bounded synthetic action. 4. **Approval binding** — synthetic approval is bound to the exact request/action digests, scope, complete policy reference, authority, and expiry in the canonical path. 5. **Behavioral scenarios** — an executable canonical matrix tests permitted, denied, malformed, unauthorized, unavailable, timed-out, tampered, expired, obligation, and recovery paths. -6. **Evidence and replay** — correlate records, detect tampering, and reproduce deterministic outcomes. +6. **Evidence and replay** — canonical evidence can be independently verified for complete correlation and tampering, then replayed from the admitted input to reproduce the exact canonical outcome. 7. **Supply chain and CI** — pin dependencies and actions; add integrity, vulnerability, secret, and SBOM checks. 8. **Documentation reconciliation** — execute every documented command in a clean environment. 9. **F7-LAS-specific diagrams** — replace diagrams only after execution semantics are frozen. diff --git a/docs/F7-LAS-QA.md b/docs/F7-LAS-QA.md index 9916d78..ce80109 100644 --- a/docs/F7-LAS-QA.md +++ b/docs/F7-LAS-QA.md @@ -50,6 +50,17 @@ PDP/PEP boundaries. The current Layer 6 demonstration is a synthetic in-process executor that makes no network calls; it is not an OS/container sandbox or an enforced network-isolation boundary. +## What do evidence verification and replay prove? + +The canonical verifier independently checks the complete record chain, +cross-record bindings, policy and action digests, and final audit correlation. +Deterministic replay reruns the same admitted input through the synthetic +workflow and requires the complete canonical evidence document to match. These +checks detect inconsistent mutation, and a previously trusted evidence-set +digest exposes later replacement. Replay reproduces this bounded outcome. The +checks do not provide a digital signature, trusted timestamp, external +attestation, or replay of real-world side effects. + ## Does F7-LAS expose internal model reasoning? No. Architecture and evidence use auditable plans, decisions, source references, proposed actions, policy outcomes, approvals, results, and evaluations. Private chain-of-thought is neither required nor treated as an audit artifact. diff --git a/examples/canonical-workflow/README.md b/examples/canonical-workflow/README.md index 0b39b13..d76de82 100644 --- a/examples/canonical-workflow/README.md +++ b/examples/canonical-workflow/README.md @@ -60,3 +60,47 @@ OPA_BIN=opa pytest -q tests/test_behavioral_scenarios.py These are deterministic reference-workflow scenarios, not claims of production fault injection, infrastructure recovery, or OS/container isolation. + +## Evidence integrity + +Verify a canonical workflow output independently of the producer: + +```bash +python -m src.canonical.evidence \ + --evidence /tmp/f7las-canonical-records.json +``` + +Verification rejects duplicate JSON keys; schema or cross-record violations; +broken record chains, references, action/output digests, or policy bindings; +incomplete or reordered final audit sources; and audit summaries inconsistent +with the policy decision or execution result. The reported evidence-set digest +detects later mutation when compared with a previously trusted copy. It is not +a signature, proof of origin, trusted timestamp, or external attestation; a +party that can replace both evidence and its expected digest can construct a +different self-consistent set. + +## Deterministic replay + +Replay the same admitted input through the canonical workflow and require the +complete RFC 8785 canonical evidence document to match: + +```bash +python -m src.canonical.replay \ + --input examples/canonical-workflow/request.json \ + --evidence /tmp/f7las-canonical-records.json \ + --output /tmp/f7las-replayed-records.json \ + --opa-binary opa +``` + +Replay first verifies the reviewed evidence, reruns the fixed synthetic action +and offline policy evaluation, verifies the new evidence, and then compares the +complete canonical documents. A reproduced denial is a successful replay; it +does not become an allow decision. Exit status `4` indicates invalid evidence +or a replay mismatch. The output path may not overwrite the input or reviewed +evidence. + +Replay covers this deterministic, side-effect-free reference workflow only. It +does not reproduce external systems, network calls, operating-system state, +human identity proofing, or real-world side effects. Reproduction requires the +reviewed input, repository policy identified by its digest, compatible Python +dependencies, and the pinned OPA behavior. diff --git a/schemas/contracts/README.md b/schemas/contracts/README.md index 6681930..e52e146 100644 --- a/schemas/contracts/README.md +++ b/schemas/contracts/README.md @@ -59,6 +59,7 @@ record: SHA-256("F7-LAS:record::1.0.0\n" || JCS(record without reco action: SHA-256("F7-LAS:action:1.0.0\n" || JCS(security-relevant action projection)) output: SHA-256("F7-LAS:output:1.0.0\n" || JCS(output)) policy: SHA-256("F7-LAS:policy:1.0.0\n" || JCS(policy document)) +evidence-set: SHA-256("F7-LAS:evidence-set:1.0.0\n" || JCS(complete evidence document)) ``` Digests use `sha256:<64 lowercase hexadecimal characters>`. Object key order is @@ -71,9 +72,18 @@ the exact `action_digest`. Milestone 2 defined these contracts. The canonical workflow emits and validates them around a real offline OPA decision and binds its deterministic synthetic -approval through PDP and PEP enforcement. This does not claim an interactive +approval through PDP and PEP enforcement. The canonical evidence verifier adds +a stricter single-action profile: exactly one record of each type, a final audit +event that binds every preceding record in order, and an audit summary that +matches the decision and result. This does not claim an interactive human-approval service, identity proofing, OS/container sandbox containment, -enforced network isolation, or replay. +enforced network isolation, signed evidence, or external attestation. + +The evidence-set digest detects later mutation only when it is compared with a +previously trusted copy. Because the records are not signed, a party able to +replace both the evidence and its expected digest can construct another +self-consistent set. Deterministic replay adds an independent comparison to the +reviewed input and repository policy, but it is not proof of provenance. ## Validate diff --git a/scripts/validate-contracts.py b/scripts/validate-contracts.py index c33ee86..fb3c5ff 100644 --- a/scripts/validate-contracts.py +++ b/scripts/validate-contracts.py @@ -1,411 +1,15 @@ #!/usr/bin/env python3 -"""Validate F7-LAS canonical records and their cross-record relationships.""" +"""Validate the repository's canonical contract examples.""" -from __future__ import annotations - -import json import sys -from collections import Counter -from datetime import datetime from pathlib import Path -from typing import Any, Iterable - -import rfc8785 -from jsonschema import Draft202012Validator, FormatChecker REPO_ROOT = Path(__file__).resolve().parents[1] if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) -from src.canonical.contracts import ( - calculate_action_digest, - calculate_output_digest, - calculate_record_digest, - digest_payload, -) - - -SCHEMA_PATH = REPO_ROOT / "schemas" / "contracts" / "f7las-records-v1.schema.json" -EXAMPLES_DIR = REPO_ROOT / "schemas" / "contracts" / "examples" -POLICY_DIR = REPO_ROOT / "config" / "policies" - -FORBIDDEN_KEYS = { - "api_key", - "chain_of_thought", - "password", - "private_reasoning", - "reasoning_trace", - "secret", - "token", -} - - -def reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: - result: dict[str, Any] = {} - for key, value in pairs: - if key in result: - raise ValueError(f"duplicate JSON object key: {key}") - result[key] = value - return result - - -def load_json(path: Path) -> Any: - with path.open("r", encoding="utf-8") as handle: - return json.load(handle, object_pairs_hook=reject_duplicate_keys) - - -def load_policy_registry() -> dict[str, dict[str, Any]]: - registry: dict[str, dict[str, Any]] = {} - for path in sorted(POLICY_DIR.glob("*.json")): - if path.name == "policy-schema.json": - continue - policy = load_json(path) - registry[policy["policy_id"]] = { - "version": policy["version"], - "policy_digest": digest_payload("policy", policy), - } - return registry - - -def parse_timestamp(value: str) -> datetime: - return datetime.fromisoformat(value.replace("Z", "+00:00")) - - -def iter_keys(value: Any) -> Iterable[str]: - if isinstance(value, dict): - for key, nested in value.items(): - yield key - yield from iter_keys(nested) - elif isinstance(value, list): - for nested in value: - yield from iter_keys(nested) - - -def validate_reference( - reference: dict[str, str], - expected: dict[str, Any], - label: str, - errors: list[str], -) -> None: - if reference["record_id"] != expected["record_id"]: - errors.append(f"{label}.record_id does not match {expected['record_type']}") - if reference["record_digest"] != expected["record_digest"]: - errors.append(f"{label}.record_digest does not match {expected['record_type']}") - - -def validate_record_set( - document: dict[str, Any], - validator: Draft202012Validator, -) -> list[str]: - errors: list[str] = [] - records = document.get("records") - if document.get("contract_set_version") != "1.0.0": - errors.append("contract_set_version must be 1.0.0") - if not isinstance(records, list): - return errors + ["records must be an array"] - - for index, record in enumerate(records): - for error in validator.iter_errors(record): - location = ".".join(str(part) for part in error.absolute_path) or "" - errors.append(f"record[{index}] {location}: {error.message}") - - if errors: - return errors - - counts = Counter(record["record_type"] for record in records) - for record_type in ("request", "context", "plan"): - if counts[record_type] != 1: - errors.append(f"workflow must contain exactly one {record_type} record") - if counts["proposed_action"] < 1: - errors.append("workflow must contain at least one proposed_action record") - if errors: - return errors - - ids = [record["record_id"] for record in records] - if len(ids) != len(set(ids)): - errors.append("record_id values must be unique") - - workflow_ids = {record["workflow_id"] for record in records} - if len(workflow_ids) != 1: - errors.append("all records must share one workflow_id") - - sequences = [record["sequence"] for record in records] - if sequences != list(range(1, len(records) + 1)): - errors.append("sequence values must be contiguous, ordered, and start at 1") - - timestamps = [parse_timestamp(record["occurred_at"]) for record in records] - if timestamps != sorted(timestamps): - errors.append("occurred_at values must be monotonically nondecreasing") - - by_id = {record["record_id"]: record for record in records} - index_by_id = {record["record_id"]: index for index, record in enumerate(records)} - - for index, record in enumerate(records): - for key in ( - "request_ref", - "context_ref", - "plan_ref", - "action_ref", - "approval_ref", - "decision_ref", - "object_ref", - ): - reference = record.get(key) - if reference is None: - continue - referenced_id = reference["record_id"] - if referenced_id not in by_id: - errors.append(f"{record['record_id']}.{key} references an unknown record") - elif index_by_id[referenced_id] >= index: - errors.append(f"{record['record_id']}.{key} must reference an earlier record") - - previous_digest: str | None = None - for record in records: - if record["previous_record_digest"] != previous_digest: - errors.append(f"{record['record_id']}.previous_record_digest breaks the record chain") - expected_digest = calculate_record_digest(record) - if record["record_digest"] != expected_digest: - errors.append(f"{record['record_id']}.record_digest must be {expected_digest}") - previous_digest = record["record_digest"] - - forbidden = sorted({key for record in records for key in iter_keys(record)} & FORBIDDEN_KEYS) - if forbidden: - errors.append(f"forbidden sensitive/private-reasoning keys present: {forbidden}") - - request = next(record for record in records if record["record_type"] == "request") - context = next(record for record in records if record["record_type"] == "context") - plan = next(record for record in records if record["record_type"] == "plan") - actions = [record for record in records if record["record_type"] == "proposed_action"] - approvals = [record for record in records if record["record_type"] == "approval"] - decisions = [record for record in records if record["record_type"] == "policy_decision"] - results = [record for record in records if record["record_type"] == "execution_result"] - audits = [record for record in records if record["record_type"] == "audit_event"] - policy_registry = load_policy_registry() - - validate_reference(context["request_ref"], request, "context.request_ref", errors) - validate_reference(plan["request_ref"], request, "plan.request_ref", errors) - validate_reference(plan["context_ref"], context, "plan.context_ref", errors) - - if context["target"] != request["scope"]: - errors.append("context target must exactly equal request scope") - for evidence in context["evidence"]: - if parse_timestamp(evidence["retrieved_at"]) > parse_timestamp(context["occurred_at"]): - errors.append(f"{evidence['evidence_id']} is retrieved after the context record") - - request_limits = request["constraints"] - for field in ("max_steps", "max_actions", "max_duration_seconds"): - if plan["limits"][field] > request_limits[field]: - errors.append(f"plan.limits.{field} exceeds the request constraint") - if len(plan["steps"]) > plan["limits"]["max_steps"]: - errors.append("plan contains more steps than max_steps") - step_orders = [step["order"] for step in plan["steps"]] - if step_orders != list(range(1, len(plan["steps"]) + 1)): - errors.append("plan step order must be contiguous and start at 1") - step_id_list = [step["step_id"] for step in plan["steps"]] - if len(step_id_list) != len(set(step_id_list)): - errors.append("plan step_id values must be unique") - step_ids = set(step_id_list) - - if len(actions) > min(request_limits["max_actions"], plan["limits"]["max_actions"]): - errors.append("proposed-action count exceeds the request or plan limit") - - def records_for_action(collection: list[dict[str, Any]], action_id: str) -> list[dict[str, Any]]: - return [record for record in collection if record["action_ref"]["record_id"] == action_id] - - for action in actions: - validate_reference(action["request_ref"], request, f"{action['record_id']}.request_ref", errors) - validate_reference(action["context_ref"], context, f"{action['record_id']}.context_ref", errors) - validate_reference(action["plan_ref"], plan, f"{action['record_id']}.plan_ref", errors) - if action["step_id"] not in step_ids: - errors.append(f"{action['record_id']}.step_id is not present in the plan") - if action["actor_id"] != context["actor"]["subject_id"]: - errors.append(f"{action['record_id']}.actor_id does not match the context actor") - if action["target"] != request["scope"]: - errors.append(f"{action['record_id']}.target must exactly equal request scope") - expected_action_digest = calculate_action_digest(action) - if action["action_digest"] != expected_action_digest: - errors.append(f"{action['record_id']}.action_digest must be {expected_action_digest}") - - action_approvals = records_for_action(approvals, action["record_id"]) - action_decisions = records_for_action(decisions, action["record_id"]) - action_results = records_for_action(results, action["record_id"]) - for label, related in ( - ("approval", action_approvals), - ("policy_decision", action_decisions), - ("execution_result", action_results), - ): - if len(related) != 1: - errors.append(f"{action['record_id']} must have exactly one {label} record") - if not all(len(related) == 1 for related in (action_approvals, action_decisions, action_results)): - continue - - approval, decision, result = action_approvals[0], action_decisions[0], action_results[0] - validate_reference(approval["request_ref"], request, f"{approval['record_id']}.request_ref", errors) - validate_reference(decision["request_ref"], request, f"{decision['record_id']}.request_ref", errors) - validate_reference(result["request_ref"], request, f"{result['record_id']}.request_ref", errors) - - for record in (approval, decision, result): - action_ref = record["action_ref"] - validate_reference(action_ref, action, f"{record['record_id']}.action_ref", errors) - if action_ref["action_digest"] != action["action_digest"]: - errors.append(f"{record['record_id']}.action_ref.action_digest does not match the action") - - validate_reference(decision["approval_ref"], approval, f"{decision['record_id']}.approval_ref", errors) - validate_reference(result["decision_ref"], decision, f"{result['record_id']}.decision_ref", errors) - - if approval["policy_ref"] != decision["policy_ref"]: - errors.append(f"{decision['record_id']}.policy_ref must exactly equal the approval policy_ref") - policy_ref = approval["policy_ref"] - registered_policy = policy_registry.get(policy_ref["policy_id"]) - if registered_policy is None: - errors.append(f"{approval['record_id']}.policy_ref references an unknown policy") - elif ( - policy_ref["version"] != registered_policy["version"] - or policy_ref["policy_digest"] != registered_policy["policy_digest"] - ): - errors.append(f"{approval['record_id']}.policy_ref does not match the repository policy") - if approval["issued_at"] != approval["occurred_at"]: - errors.append(f"{approval['record_id']}.issued_at must equal occurred_at") - - approval_time = parse_timestamp(approval["issued_at"]) - decision_time = parse_timestamp(decision["occurred_at"]) - if decision_time < approval_time: - errors.append(f"{decision['record_id']} occurs before its approval disposition") - - if approval["status"] == "approved": - expires_time = parse_timestamp(approval["expires_at"]) - if approval["approved_scope"] != action["target"]: - errors.append(f"{approval['record_id']}.approved_scope must exactly equal action target") - if expires_time <= approval_time: - errors.append(f"{approval['record_id']} validity window must have positive duration") - if decision_time >= expires_time: - errors.append(f"{decision['record_id']} occurs at or after approval expiry") - if action["requires_approval"] and approval["status"] != "approved": - errors.append(f"{action['record_id']} requires an approved approval record") - if approval["status"] == "not_required" and action["requires_approval"]: - errors.append(f"{approval['record_id']} cannot be not_required for this action") - - if decision["decision"] == "permit": - valid_basis = ( - approval["status"] == "approved" - or (approval["status"] == "not_required" and not action["requires_approval"]) - ) - if not valid_basis or decision["authorization_basis"] != approval["status"]: - errors.append(f"{decision['record_id']} permit lacks a valid approval basis") - elif result["status"] != "not_executed": - errors.append(f"{result['record_id']} must be not_executed after deny or referral") - - if request_limits["dry_run"] and result["status"] != "not_executed": - errors.append(f"{result['record_id']} must be not_executed for a dry-run request") - if result["status"] == "succeeded" and decision["decision"] != "permit": - errors.append(f"{result['record_id']} succeeded without a permit decision") - if result["output_digest"] != calculate_output_digest(result["output"]): - errors.append(f"{result['record_id']}.output_digest does not match output") - if decision["obligations"] != sorted(decision["obligations"]): - errors.append(f"{decision['record_id']}.obligations must be sorted") - - if result["started_at"] is not None: - start = parse_timestamp(result["started_at"]) - complete = parse_timestamp(result["completed_at"]) - if start < decision_time: - errors.append(f"{result['record_id']} starts before the policy decision") - if approval["status"] == "approved" and start >= parse_timestamp(approval["expires_at"]): - errors.append(f"{result['record_id']} starts at or after approval expiry") - if complete < start: - errors.append(f"{result['record_id']} completes before it starts") - if parse_timestamp(result["occurred_at"]) < complete: - errors.append(f"{result['record_id']} is recorded before completion") - - for related in approvals + decisions + results: - if related["action_ref"]["record_id"] not in {action["record_id"] for action in actions}: - errors.append(f"{related['record_id']} references an unknown proposed action") - - outcome_for_status = { - "succeeded": "success", - "failed": "failure", - "not_executed": "not_executed", - } - for audit in audits: - validate_reference(audit["request_ref"], request, f"{audit['record_id']}.request_ref", errors) - object_id = audit["object_ref"]["record_id"] - if object_id not in by_id: - errors.append(f"{audit['record_id']}.object_ref references an unknown record") - continue - validate_reference(audit["object_ref"], by_id[object_id], f"{audit['record_id']}.object_ref", errors) - source_ids: set[str] = set() - source_sequences: list[int] = [] - for source in audit["source_records"]: - source_id = source["record_id"] - source_ids.add(source_id) - if source_id not in by_id: - errors.append(f"{audit['record_id']} references unknown source record {source_id}") - continue - expected = by_id[source_id] - source_sequences.append(expected["sequence"]) - if source["record_type"] != expected["record_type"] or source["record_digest"] != expected["record_digest"]: - errors.append(f"{audit['record_id']} source record {source_id} does not match") - if index_by_id[source_id] >= index_by_id[audit["record_id"]]: - errors.append(f"{audit['record_id']} source record {source_id} is not earlier") - if source_sequences != sorted(source_sequences): - errors.append(f"{audit['record_id']}.source_records must follow workflow sequence") - sourced_results = [result for result in results if result["record_id"] in source_ids] - if len(sourced_results) == 1: - result = sourced_results[0] - expected_outcome = outcome_for_status[result["status"]] - related_decision = by_id[result["decision_ref"]["record_id"]] - if result["status"] == "not_executed" and related_decision["decision"] == "deny": - expected_outcome = "denied" - if audit["outcome"] != expected_outcome: - errors.append(f"{audit['record_id']}.outcome is inconsistent with its execution result") - - scopes = [request["scope"], context["target"]] - scopes.extend(action["target"] for action in actions) - scopes.extend( - approval["approved_scope"] - for approval in approvals - if approval["approved_scope"] is not None - ) - for scope in scopes: - if scope["resource_ids"] != sorted(scope["resource_ids"]): - errors.append(f"{scope['scope_id']}.resource_ids must be sorted") - - return errors - - -def validate_all() -> list[str]: - schema = load_json(SCHEMA_PATH) - try: - Draft202012Validator.check_schema(schema) - except Exception as exc: # pragma: no cover - return [f"invalid schema: {exc}"] - - validator = Draft202012Validator(schema, format_checker=FormatChecker()) - paths = sorted(EXAMPLES_DIR.glob("*.json")) - if not paths: - return [f"no contract examples found in {EXAMPLES_DIR}"] - - errors: list[str] = [] - for path in paths: - try: - document = load_json(path) - errors.extend( - f"{path.relative_to(REPO_ROOT)}: {error}" - for error in validate_record_set(document, validator) - ) - except (OSError, json.JSONDecodeError, KeyError, TypeError, ValueError, rfc8785.CanonicalizationError) as exc: - errors.append(f"{path.relative_to(REPO_ROOT)}: {exc}") - return errors - - -def main() -> int: - errors = validate_all() - if errors: - print("F7-LAS contract validation FAILED:", file=sys.stderr) - for error in errors: - print(f" - {error}", file=sys.stderr) - return 1 - print("F7-LAS contract validation PASSED.") - return 0 +from src.canonical.validation import * # noqa: F403 +from src.canonical.validation import main if __name__ == "__main__": diff --git a/src/canonical/evidence.py b/src/canonical/evidence.py new file mode 100644 index 0000000..4a0d4f3 --- /dev/null +++ b/src/canonical/evidence.py @@ -0,0 +1,119 @@ +"""Integrity verification for canonical F7-LAS evidence sets.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +from typing import Any + +from jsonschema import Draft202012Validator, FormatChecker + +from .contracts import digest_payload +from .validation import REPO_ROOT, SCHEMA_PATH, load_json, validate_record_set + + +CANONICAL_RECORD_ORDER = ( + "request", + "context", + "plan", + "proposed_action", + "approval", + "policy_decision", + "execution_result", + "audit_event", +) + + +class EvidenceIntegrityError(ValueError): + """A canonical evidence set failed integrity or correlation checks.""" + + +def evidence_digest(document: dict[str, Any]) -> str: + """Return a domain-separated digest of the complete canonical evidence set.""" + + return digest_payload("evidence-set", document) + + +def _schema_validator() -> Draft202012Validator: + return Draft202012Validator( + load_json(SCHEMA_PATH), + format_checker=FormatChecker(), + ) + + +def verify_evidence(document: dict[str, Any]) -> str: + """Verify contracts plus the stricter single-action canonical evidence profile.""" + + if not isinstance(document, dict): + raise EvidenceIntegrityError("evidence document must be a JSON object") + if set(document) != {"contract_set_version", "records"}: + raise EvidenceIntegrityError("evidence document fields do not match the canonical profile") + + errors = validate_record_set(document, _schema_validator()) + if errors: + raise EvidenceIntegrityError("; ".join(errors)) + + records = document["records"] + record_types = tuple(record["record_type"] for record in records) + if record_types != CANONICAL_RECORD_ORDER: + raise EvidenceIntegrityError( + "canonical evidence must contain exactly one record of each type in workflow order" + ) + + audit = records[-1] + expected_sources = [ + { + "record_type": record["record_type"], + "record_id": record["record_id"], + "record_digest": record["record_digest"], + } + for record in records[:-1] + ] + if audit["source_records"] != expected_sources: + raise EvidenceIntegrityError( + "final audit_event.source_records must bind every preceding record in order" + ) + + action = records[3] + decision = records[5] + result = records[6] + expected_object = { + "record_id": action["record_id"], + "record_digest": action["record_digest"], + } + if audit["object_ref"] != expected_object: + raise EvidenceIntegrityError("final audit event must identify the canonical proposed action") + if audit["event_type"] != "workflow-completed": + raise EvidenceIntegrityError("final audit event must be workflow-completed") + if audit["details"].get("execution_status") != result["status"]: + raise EvidenceIntegrityError("audit execution_status does not match execution_result") + if audit["details"].get("policy_reason_code") != decision["reason_code"]: + raise EvidenceIntegrityError("audit policy_reason_code does not match policy_decision") + + return evidence_digest(document) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--evidence", required=True, type=Path) + args = parser.parse_args() + + try: + document = load_json(args.evidence) + digest = verify_evidence(document) + except (OSError, json.JSONDecodeError, KeyError, TypeError, ValueError) as exc: + print(f"F7-LAS evidence verification FAILED: {exc}") + return 4 + + relative = args.evidence + try: + relative = args.evidence.resolve().relative_to(REPO_ROOT) + except ValueError: + pass + print(f"F7-LAS evidence verification PASSED: file={relative} digest={digest}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/canonical/replay.py b/src/canonical/replay.py new file mode 100644 index 0000000..1a56d4f --- /dev/null +++ b/src/canonical/replay.py @@ -0,0 +1,84 @@ +"""Deterministically replay and compare a canonical F7-LAS evidence set.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +from typing import Any + +import rfc8785 + +from .evidence import EvidenceIntegrityError, evidence_digest, verify_evidence +from .validation import load_json +from .workflow import CanonicalWorkflow, WorkflowError + + +class ReplayMismatchError(ValueError): + """A valid evidence set was not reproduced exactly by replay.""" + + +def replay_evidence( + workflow_input: dict[str, Any], + expected: dict[str, Any], + opa_binary: str = "opa", +) -> dict[str, Any]: + """Verify prior evidence, rerun the workflow, and require canonical equality.""" + + expected_digest = verify_evidence(expected) + replayed = CanonicalWorkflow(opa_binary=opa_binary).run(workflow_input) + replayed_digest = verify_evidence(replayed) + if rfc8785.dumps(replayed) != rfc8785.dumps(expected): + expected_records = expected["records"] + replayed_records = replayed["records"] + mismatch = "document structure" + for old, new in zip(expected_records, replayed_records): + if old != new: + mismatch = old.get("record_type", "unknown record") + break + raise ReplayMismatchError( + "deterministic replay differs at " + f"{mismatch}: expected={expected_digest} replayed={replayed_digest}" + ) + return replayed + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--input", required=True, type=Path) + parser.add_argument("--evidence", required=True, type=Path) + parser.add_argument("--output", type=Path) + parser.add_argument("--opa-binary", default="opa") + args = parser.parse_args() + + if args.output is not None and args.output.resolve() in { + args.input.resolve(), + args.evidence.resolve(), + }: + print("F7-LAS replay FAILED: output must not overwrite input or reviewed evidence") + return 4 + + try: + workflow_input = load_json(args.input) + expected = load_json(args.evidence) + replayed = replay_evidence(workflow_input, expected, args.opa_binary) + if args.output is not None: + args.output.write_text(json.dumps(replayed, indent=2) + "\n", encoding="utf-8") + except ( + OSError, + json.JSONDecodeError, + KeyError, + TypeError, + ValueError, + WorkflowError, + EvidenceIntegrityError, + ) as exc: + print(f"F7-LAS replay FAILED: {exc}") + return 4 + + print(f"F7-LAS replay PASSED: digest={evidence_digest(replayed)}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/canonical/validation.py b/src/canonical/validation.py new file mode 100644 index 0000000..4a1d170 --- /dev/null +++ b/src/canonical/validation.py @@ -0,0 +1,412 @@ +#!/usr/bin/env python3 +"""Validate F7-LAS canonical records and their cross-record relationships.""" + +from __future__ import annotations + +import json +import sys +from collections import Counter +from datetime import datetime +from pathlib import Path +from typing import Any, Iterable + +import rfc8785 +from jsonschema import Draft202012Validator, FormatChecker + +REPO_ROOT = Path(__file__).resolve().parents[2] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from src.canonical.contracts import ( + calculate_action_digest, + calculate_output_digest, + calculate_record_digest, + digest_payload, +) + + +SCHEMA_PATH = REPO_ROOT / "schemas" / "contracts" / "f7las-records-v1.schema.json" +EXAMPLES_DIR = REPO_ROOT / "schemas" / "contracts" / "examples" +POLICY_DIR = REPO_ROOT / "config" / "policies" + +FORBIDDEN_KEYS = { + "api_key", + "chain_of_thought", + "password", + "private_reasoning", + "reasoning_trace", + "secret", + "token", +} + + +def reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise ValueError(f"duplicate JSON object key: {key}") + result[key] = value + return result + + +def load_json(path: Path) -> Any: + with path.open("r", encoding="utf-8") as handle: + return json.load(handle, object_pairs_hook=reject_duplicate_keys) + + +def load_policy_registry() -> dict[str, dict[str, Any]]: + registry: dict[str, dict[str, Any]] = {} + for path in sorted(POLICY_DIR.glob("*.json")): + if path.name == "policy-schema.json": + continue + policy = load_json(path) + registry[policy["policy_id"]] = { + "version": policy["version"], + "policy_digest": digest_payload("policy", policy), + } + return registry + + +def parse_timestamp(value: str) -> datetime: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + + +def iter_keys(value: Any) -> Iterable[str]: + if isinstance(value, dict): + for key, nested in value.items(): + yield key + yield from iter_keys(nested) + elif isinstance(value, list): + for nested in value: + yield from iter_keys(nested) + + +def validate_reference( + reference: dict[str, str], + expected: dict[str, Any], + label: str, + errors: list[str], +) -> None: + if reference["record_id"] != expected["record_id"]: + errors.append(f"{label}.record_id does not match {expected['record_type']}") + if reference["record_digest"] != expected["record_digest"]: + errors.append(f"{label}.record_digest does not match {expected['record_type']}") + + +def validate_record_set( + document: dict[str, Any], + validator: Draft202012Validator, +) -> list[str]: + errors: list[str] = [] + records = document.get("records") + if document.get("contract_set_version") != "1.0.0": + errors.append("contract_set_version must be 1.0.0") + if not isinstance(records, list): + return errors + ["records must be an array"] + + for index, record in enumerate(records): + for error in validator.iter_errors(record): + location = ".".join(str(part) for part in error.absolute_path) or "" + errors.append(f"record[{index}] {location}: {error.message}") + + if errors: + return errors + + counts = Counter(record["record_type"] for record in records) + for record_type in ("request", "context", "plan"): + if counts[record_type] != 1: + errors.append(f"workflow must contain exactly one {record_type} record") + if counts["proposed_action"] < 1: + errors.append("workflow must contain at least one proposed_action record") + if errors: + return errors + + ids = [record["record_id"] for record in records] + if len(ids) != len(set(ids)): + errors.append("record_id values must be unique") + + workflow_ids = {record["workflow_id"] for record in records} + if len(workflow_ids) != 1: + errors.append("all records must share one workflow_id") + + sequences = [record["sequence"] for record in records] + if sequences != list(range(1, len(records) + 1)): + errors.append("sequence values must be contiguous, ordered, and start at 1") + + timestamps = [parse_timestamp(record["occurred_at"]) for record in records] + if timestamps != sorted(timestamps): + errors.append("occurred_at values must be monotonically nondecreasing") + + by_id = {record["record_id"]: record for record in records} + index_by_id = {record["record_id"]: index for index, record in enumerate(records)} + + for index, record in enumerate(records): + for key in ( + "request_ref", + "context_ref", + "plan_ref", + "action_ref", + "approval_ref", + "decision_ref", + "object_ref", + ): + reference = record.get(key) + if reference is None: + continue + referenced_id = reference["record_id"] + if referenced_id not in by_id: + errors.append(f"{record['record_id']}.{key} references an unknown record") + elif index_by_id[referenced_id] >= index: + errors.append(f"{record['record_id']}.{key} must reference an earlier record") + + previous_digest: str | None = None + for record in records: + if record["previous_record_digest"] != previous_digest: + errors.append(f"{record['record_id']}.previous_record_digest breaks the record chain") + expected_digest = calculate_record_digest(record) + if record["record_digest"] != expected_digest: + errors.append(f"{record['record_id']}.record_digest must be {expected_digest}") + previous_digest = record["record_digest"] + + forbidden = sorted({key for record in records for key in iter_keys(record)} & FORBIDDEN_KEYS) + if forbidden: + errors.append(f"forbidden sensitive/private-reasoning keys present: {forbidden}") + + request = next(record for record in records if record["record_type"] == "request") + context = next(record for record in records if record["record_type"] == "context") + plan = next(record for record in records if record["record_type"] == "plan") + actions = [record for record in records if record["record_type"] == "proposed_action"] + approvals = [record for record in records if record["record_type"] == "approval"] + decisions = [record for record in records if record["record_type"] == "policy_decision"] + results = [record for record in records if record["record_type"] == "execution_result"] + audits = [record for record in records if record["record_type"] == "audit_event"] + policy_registry = load_policy_registry() + + validate_reference(context["request_ref"], request, "context.request_ref", errors) + validate_reference(plan["request_ref"], request, "plan.request_ref", errors) + validate_reference(plan["context_ref"], context, "plan.context_ref", errors) + + if context["target"] != request["scope"]: + errors.append("context target must exactly equal request scope") + for evidence in context["evidence"]: + if parse_timestamp(evidence["retrieved_at"]) > parse_timestamp(context["occurred_at"]): + errors.append(f"{evidence['evidence_id']} is retrieved after the context record") + + request_limits = request["constraints"] + for field in ("max_steps", "max_actions", "max_duration_seconds"): + if plan["limits"][field] > request_limits[field]: + errors.append(f"plan.limits.{field} exceeds the request constraint") + if len(plan["steps"]) > plan["limits"]["max_steps"]: + errors.append("plan contains more steps than max_steps") + step_orders = [step["order"] for step in plan["steps"]] + if step_orders != list(range(1, len(plan["steps"]) + 1)): + errors.append("plan step order must be contiguous and start at 1") + step_id_list = [step["step_id"] for step in plan["steps"]] + if len(step_id_list) != len(set(step_id_list)): + errors.append("plan step_id values must be unique") + step_ids = set(step_id_list) + + if len(actions) > min(request_limits["max_actions"], plan["limits"]["max_actions"]): + errors.append("proposed-action count exceeds the request or plan limit") + + def records_for_action(collection: list[dict[str, Any]], action_id: str) -> list[dict[str, Any]]: + return [record for record in collection if record["action_ref"]["record_id"] == action_id] + + for action in actions: + validate_reference(action["request_ref"], request, f"{action['record_id']}.request_ref", errors) + validate_reference(action["context_ref"], context, f"{action['record_id']}.context_ref", errors) + validate_reference(action["plan_ref"], plan, f"{action['record_id']}.plan_ref", errors) + if action["step_id"] not in step_ids: + errors.append(f"{action['record_id']}.step_id is not present in the plan") + if action["actor_id"] != context["actor"]["subject_id"]: + errors.append(f"{action['record_id']}.actor_id does not match the context actor") + if action["target"] != request["scope"]: + errors.append(f"{action['record_id']}.target must exactly equal request scope") + expected_action_digest = calculate_action_digest(action) + if action["action_digest"] != expected_action_digest: + errors.append(f"{action['record_id']}.action_digest must be {expected_action_digest}") + + action_approvals = records_for_action(approvals, action["record_id"]) + action_decisions = records_for_action(decisions, action["record_id"]) + action_results = records_for_action(results, action["record_id"]) + for label, related in ( + ("approval", action_approvals), + ("policy_decision", action_decisions), + ("execution_result", action_results), + ): + if len(related) != 1: + errors.append(f"{action['record_id']} must have exactly one {label} record") + if not all(len(related) == 1 for related in (action_approvals, action_decisions, action_results)): + continue + + approval, decision, result = action_approvals[0], action_decisions[0], action_results[0] + validate_reference(approval["request_ref"], request, f"{approval['record_id']}.request_ref", errors) + validate_reference(decision["request_ref"], request, f"{decision['record_id']}.request_ref", errors) + validate_reference(result["request_ref"], request, f"{result['record_id']}.request_ref", errors) + + for record in (approval, decision, result): + action_ref = record["action_ref"] + validate_reference(action_ref, action, f"{record['record_id']}.action_ref", errors) + if action_ref["action_digest"] != action["action_digest"]: + errors.append(f"{record['record_id']}.action_ref.action_digest does not match the action") + + validate_reference(decision["approval_ref"], approval, f"{decision['record_id']}.approval_ref", errors) + validate_reference(result["decision_ref"], decision, f"{result['record_id']}.decision_ref", errors) + + if approval["policy_ref"] != decision["policy_ref"]: + errors.append(f"{decision['record_id']}.policy_ref must exactly equal the approval policy_ref") + policy_ref = approval["policy_ref"] + registered_policy = policy_registry.get(policy_ref["policy_id"]) + if registered_policy is None: + errors.append(f"{approval['record_id']}.policy_ref references an unknown policy") + elif ( + policy_ref["version"] != registered_policy["version"] + or policy_ref["policy_digest"] != registered_policy["policy_digest"] + ): + errors.append(f"{approval['record_id']}.policy_ref does not match the repository policy") + if approval["issued_at"] != approval["occurred_at"]: + errors.append(f"{approval['record_id']}.issued_at must equal occurred_at") + + approval_time = parse_timestamp(approval["issued_at"]) + decision_time = parse_timestamp(decision["occurred_at"]) + if decision_time < approval_time: + errors.append(f"{decision['record_id']} occurs before its approval disposition") + + if approval["status"] == "approved": + expires_time = parse_timestamp(approval["expires_at"]) + if approval["approved_scope"] != action["target"]: + errors.append(f"{approval['record_id']}.approved_scope must exactly equal action target") + if expires_time <= approval_time: + errors.append(f"{approval['record_id']} validity window must have positive duration") + if decision_time >= expires_time: + errors.append(f"{decision['record_id']} occurs at or after approval expiry") + if action["requires_approval"] and approval["status"] != "approved": + errors.append(f"{action['record_id']} requires an approved approval record") + if approval["status"] == "not_required" and action["requires_approval"]: + errors.append(f"{approval['record_id']} cannot be not_required for this action") + + if decision["decision"] == "permit": + valid_basis = ( + approval["status"] == "approved" + or (approval["status"] == "not_required" and not action["requires_approval"]) + ) + if not valid_basis or decision["authorization_basis"] != approval["status"]: + errors.append(f"{decision['record_id']} permit lacks a valid approval basis") + elif result["status"] != "not_executed": + errors.append(f"{result['record_id']} must be not_executed after deny or referral") + + if request_limits["dry_run"] and result["status"] != "not_executed": + errors.append(f"{result['record_id']} must be not_executed for a dry-run request") + if result["status"] == "succeeded" and decision["decision"] != "permit": + errors.append(f"{result['record_id']} succeeded without a permit decision") + if result["output_digest"] != calculate_output_digest(result["output"]): + errors.append(f"{result['record_id']}.output_digest does not match output") + if decision["obligations"] != sorted(decision["obligations"]): + errors.append(f"{decision['record_id']}.obligations must be sorted") + + if result["started_at"] is not None: + start = parse_timestamp(result["started_at"]) + complete = parse_timestamp(result["completed_at"]) + if start < decision_time: + errors.append(f"{result['record_id']} starts before the policy decision") + if approval["status"] == "approved" and start >= parse_timestamp(approval["expires_at"]): + errors.append(f"{result['record_id']} starts at or after approval expiry") + if complete < start: + errors.append(f"{result['record_id']} completes before it starts") + if parse_timestamp(result["occurred_at"]) < complete: + errors.append(f"{result['record_id']} is recorded before completion") + + for related in approvals + decisions + results: + if related["action_ref"]["record_id"] not in {action["record_id"] for action in actions}: + errors.append(f"{related['record_id']} references an unknown proposed action") + + outcome_for_status = { + "succeeded": "success", + "failed": "failure", + "not_executed": "not_executed", + } + for audit in audits: + validate_reference(audit["request_ref"], request, f"{audit['record_id']}.request_ref", errors) + object_id = audit["object_ref"]["record_id"] + if object_id not in by_id: + errors.append(f"{audit['record_id']}.object_ref references an unknown record") + continue + validate_reference(audit["object_ref"], by_id[object_id], f"{audit['record_id']}.object_ref", errors) + source_ids: set[str] = set() + source_sequences: list[int] = [] + for source in audit["source_records"]: + source_id = source["record_id"] + source_ids.add(source_id) + if source_id not in by_id: + errors.append(f"{audit['record_id']} references unknown source record {source_id}") + continue + expected = by_id[source_id] + source_sequences.append(expected["sequence"]) + if source["record_type"] != expected["record_type"] or source["record_digest"] != expected["record_digest"]: + errors.append(f"{audit['record_id']} source record {source_id} does not match") + if index_by_id[source_id] >= index_by_id[audit["record_id"]]: + errors.append(f"{audit['record_id']} source record {source_id} is not earlier") + if source_sequences != sorted(source_sequences): + errors.append(f"{audit['record_id']}.source_records must follow workflow sequence") + sourced_results = [result for result in results if result["record_id"] in source_ids] + if len(sourced_results) == 1: + result = sourced_results[0] + expected_outcome = outcome_for_status[result["status"]] + related_decision = by_id[result["decision_ref"]["record_id"]] + if result["status"] == "not_executed" and related_decision["decision"] == "deny": + expected_outcome = "denied" + if audit["outcome"] != expected_outcome: + errors.append(f"{audit['record_id']}.outcome is inconsistent with its execution result") + + scopes = [request["scope"], context["target"]] + scopes.extend(action["target"] for action in actions) + scopes.extend( + approval["approved_scope"] + for approval in approvals + if approval["approved_scope"] is not None + ) + for scope in scopes: + if scope["resource_ids"] != sorted(scope["resource_ids"]): + errors.append(f"{scope['scope_id']}.resource_ids must be sorted") + + return errors + + +def validate_all() -> list[str]: + schema = load_json(SCHEMA_PATH) + try: + Draft202012Validator.check_schema(schema) + except Exception as exc: # pragma: no cover + return [f"invalid schema: {exc}"] + + validator = Draft202012Validator(schema, format_checker=FormatChecker()) + paths = sorted(EXAMPLES_DIR.glob("*.json")) + if not paths: + return [f"no contract examples found in {EXAMPLES_DIR}"] + + errors: list[str] = [] + for path in paths: + try: + document = load_json(path) + errors.extend( + f"{path.relative_to(REPO_ROOT)}: {error}" + for error in validate_record_set(document, validator) + ) + except (OSError, json.JSONDecodeError, KeyError, TypeError, ValueError, rfc8785.CanonicalizationError) as exc: + errors.append(f"{path.relative_to(REPO_ROOT)}: {exc}") + return errors + + +def main() -> int: + errors = validate_all() + if errors: + print("F7-LAS contract validation FAILED:", file=sys.stderr) + for error in errors: + print(f" - {error}", file=sys.stderr) + return 1 + print("F7-LAS contract validation PASSED.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_evidence_replay.py b/tests/test_evidence_replay.py new file mode 100644 index 0000000..316e678 --- /dev/null +++ b/tests/test_evidence_replay.py @@ -0,0 +1,181 @@ +import copy +import json +import os +import shutil +from pathlib import Path + +import pytest + +from src.canonical import evidence, replay +from src.canonical.contracts import calculate_record_digest +from src.canonical.evidence import EvidenceIntegrityError, evidence_digest, verify_evidence +from src.canonical.replay import ReplayMismatchError, replay_evidence +from src.canonical.validation import load_json +from src.canonical.workflow import CanonicalWorkflow + + +INPUT_PATH = Path("examples/canonical-workflow/request.json") +OPA_BINARY = os.environ.get("OPA_BIN") or shutil.which("opa") + + +def workflow_input(): + return load_json(INPUT_PATH) + + +def denied_evidence(): + return CanonicalWorkflow(opa_binary="/does/not/exist/opa").run(workflow_input()) + + +def record(document, record_type): + return next(item for item in document["records"] if item["record_type"] == record_type) + + +def test_evidence_verifier_accepts_complete_denial_and_is_deterministic(): + document = denied_evidence() + + first_digest = verify_evidence(document) + second_digest = verify_evidence(copy.deepcopy(document)) + + assert first_digest == second_digest == evidence_digest(document) + assert first_digest.startswith("sha256:") + + +@pytest.mark.skipif(OPA_BINARY is None, reason="OPA CLI is not installed") +def test_evidence_verifier_accepts_complete_permit_path(): + document = CanonicalWorkflow(opa_binary=OPA_BINARY).run(workflow_input()) + + assert verify_evidence(document) == evidence_digest(document) + + +@pytest.mark.parametrize( + ("record_type", "field", "value"), + [ + ("request", "mission", "tampered mission"), + ("proposed_action", "action_digest", "sha256:" + "f" * 64), + ("execution_result", "output_digest", "sha256:" + "f" * 64), + ("audit_event", "outcome", "success"), + ], +) +def test_evidence_verifier_detects_record_tampering(record_type, field, value): + document = denied_evidence() + record(document, record_type)[field] = value + + with pytest.raises(EvidenceIntegrityError): + verify_evidence(document) + + +def test_evidence_verifier_requires_complete_ordered_audit_sources(): + document = denied_evidence() + audit = record(document, "audit_event") + audit["source_records"] = audit["source_records"][:-1] + audit["record_digest"] = calculate_record_digest(audit) + + with pytest.raises(EvidenceIntegrityError, match="bind every preceding record"): + verify_evidence(document) + + +def test_evidence_verifier_binds_audit_summary_to_decision_and_result(): + document = denied_evidence() + audit = record(document, "audit_event") + audit["details"]["policy_reason_code"] = "different-reason" + audit["record_digest"] = calculate_record_digest(audit) + + with pytest.raises(EvidenceIntegrityError, match="policy_reason_code"): + verify_evidence(document) + + +def test_evidence_verifier_rejects_extra_envelope_fields(): + document = denied_evidence() + document["unsigned_note"] = "not covered by record digests" + + with pytest.raises(EvidenceIntegrityError, match="document fields"): + verify_evidence(document) + + +def test_strict_evidence_loading_rejects_duplicate_keys(tmp_path): + path = tmp_path / "duplicate.json" + path.write_text('{"contract_set_version":"1.0.0","records":[],"records":[]}', encoding="utf-8") + + with pytest.raises(ValueError, match="duplicate JSON object key: records"): + load_json(path) + + +def test_denied_outcome_replays_exactly_and_is_a_successful_replay(): + expected = denied_evidence() + + replayed = replay_evidence( + workflow_input(), + expected, + opa_binary="/does/not/exist/opa", + ) + + assert replayed == expected + assert record(replayed, "policy_decision")["decision"] == "deny" + + +@pytest.mark.skipif(OPA_BINARY is None, reason="OPA CLI is not installed") +def test_permitted_outcome_replays_exactly_with_real_opa(): + expected = CanonicalWorkflow(opa_binary=OPA_BINARY).run(workflow_input()) + + assert replay_evidence(workflow_input(), expected, OPA_BINARY) == expected + + +def test_replay_detects_different_admitted_input(): + expected = denied_evidence() + changed_input = workflow_input() + changed_input["started_at"] = "2026-01-15T12:01:00Z" + + with pytest.raises(ReplayMismatchError, match="deterministic replay differs"): + replay_evidence(changed_input, expected, opa_binary="/does/not/exist/opa") + + +def test_evidence_cli_returns_nonzero_for_tampering(tmp_path, monkeypatch): + document = denied_evidence() + record(document, "request")["mission"] = "tampered" + path = tmp_path / "tampered.json" + path.write_text(json.dumps(document), encoding="utf-8") + monkeypatch.setattr("sys.argv", ["f7las-evidence", "--evidence", str(path)]) + + assert evidence.main() == 4 + + +def test_replay_cli_writes_exact_reproduction(tmp_path, monkeypatch): + expected_path = tmp_path / "expected.json" + output_path = tmp_path / "replayed.json" + expected_path.write_text(json.dumps(denied_evidence()), encoding="utf-8") + monkeypatch.setattr( + "sys.argv", + [ + "f7las-replay", + "--input", + str(INPUT_PATH), + "--evidence", + str(expected_path), + "--output", + str(output_path), + "--opa-binary", + "/does/not/exist/opa", + ], + ) + + assert replay.main() == 0 + assert load_json(output_path) == load_json(expected_path) + + +def test_replay_cli_does_not_overwrite_reviewed_evidence(tmp_path, monkeypatch): + expected_path = tmp_path / "expected.json" + expected_path.write_text(json.dumps(denied_evidence()), encoding="utf-8") + monkeypatch.setattr( + "sys.argv", + [ + "f7las-replay", + "--input", + str(INPUT_PATH), + "--evidence", + str(expected_path), + "--output", + str(expected_path), + ], + ) + + assert replay.main() == 4 From 4a9414cd55c8abdd3f66432ce8910e4d752b9a20 Mon Sep 17 00:00:00 2001 From: anthonyfuller <38588965+anthfuller@users.noreply.github.com> Date: Sun, 20 Sep 2026 16:51:11 -0400 Subject: [PATCH 2/2] fix: bind evidence to executable policy bundle --- config/policies/canonical-workflow.rego | 10 ++--- docs/F7-LAS-QA.md | 4 +- examples/canonical-workflow/README.md | 8 ++-- schemas/contracts/README.md | 8 +++- .../contracts/examples/approved-dry-run.json | 28 ++++++------- src/canonical/contracts.py | 12 ++++++ src/canonical/opa.py | 35 +++++++++++++++- src/canonical/replay.py | 20 ++++++---- src/canonical/validation.py | 11 ++++- src/canonical/workflow.py | 15 ++----- tests/test_canonical_workflow.py | 29 ++++++++++++-- tests/test_evidence_replay.py | 40 ++++++++++++++++++- 12 files changed, 168 insertions(+), 52 deletions(-) diff --git a/config/policies/canonical-workflow.rego b/config/policies/canonical-workflow.rego index 924fb91..94470c8 100644 --- a/config/policies/canonical-workflow.rego +++ b/config/policies/canonical-workflow.rego @@ -6,18 +6,14 @@ default result := { "obligations": ["audit-required"], } -expected_policy_ref := { - "policy_id": "constraints-default-v1", - "version": "v1.0", - "policy_digest": "sha256:091de3f0a96ec85a610f42456aaba98c8d04e148b9f910c570f96af37795b44d", -} - result := { "decision": "permit", "reason_code": "approved-synthetic-read", "obligations": ["audit-required", "offline-runtime-required"], } if { - input.policy_ref == expected_policy_ref + input.policy_ref.policy_id == "constraints-default-v1" + input.policy_ref.version == "v1.0" + regex.match("^sha256:[0-9a-f]{64}$", input.policy_ref.policy_digest) input.request.dry_run == false input.request.scope.scope_id == "lab-boundary-0001" input.request.scope.environment == "lab" diff --git a/docs/F7-LAS-QA.md b/docs/F7-LAS-QA.md index ce80109..25f94e6 100644 --- a/docs/F7-LAS-QA.md +++ b/docs/F7-LAS-QA.md @@ -53,7 +53,9 @@ enforced network-isolation boundary. ## What do evidence verification and replay prove? The canonical verifier independently checks the complete record chain, -cross-record bindings, policy and action digests, and final audit correlation. +cross-record bindings, the policy-bundle digest covering metadata and exact +executable Rego bytes, action digests, and final audit correlation. The OPA +adapter verifies that bundle reference before executing the policy. Deterministic replay reruns the same admitted input through the synthetic workflow and requires the complete canonical evidence document to match. These checks detect inconsistent mutation, and a previously trusted evidence-set diff --git a/examples/canonical-workflow/README.md b/examples/canonical-workflow/README.md index d76de82..299c432 100644 --- a/examples/canonical-workflow/README.md +++ b/examples/canonical-workflow/README.md @@ -12,9 +12,11 @@ correlated canonical audit records (Layer 7). Layer 6 here is not an OS or container sandbox and does not enforce a network-isolation boundary. The synthetic approval is bound to the exact request and action references and -digests, complete scope, complete policy reference, approving authority, issue -time, and expiry. OPA validates the binding before permitting, and the -in-process executor independently revalidates it at execution time. This is +digests, complete scope, approving authority, issue time, expiry, and a +versioned policy-bundle digest covering both policy metadata and the exact Rego +bytes executed by OPA. The adapter verifies that bundle before invoking OPA, +and the in-process executor independently revalidates the reference at execution +time. This is deterministic approval evidence for the reference workflow, not an interactive human-approval service or identity proofing system. diff --git a/schemas/contracts/README.md b/schemas/contracts/README.md index e52e146..c108aeb 100644 --- a/schemas/contracts/README.md +++ b/schemas/contracts/README.md @@ -58,7 +58,7 @@ SHA-256 inputs are domain-separated: record: SHA-256("F7-LAS:record::1.0.0\n" || JCS(record without record_digest)) action: SHA-256("F7-LAS:action:1.0.0\n" || JCS(security-relevant action projection)) output: SHA-256("F7-LAS:output:1.0.0\n" || JCS(output)) -policy: SHA-256("F7-LAS:policy:1.0.0\n" || JCS(policy document)) +policy-bundle: SHA-256("F7-LAS:policy-bundle:1.0.0\n" || JCS({metadata, SHA-256(exact Rego bytes)})) evidence-set: SHA-256("F7-LAS:evidence-set:1.0.0\n" || JCS(complete evidence document)) ``` @@ -70,6 +70,12 @@ defaults, and self-inclusion of a digest field are not permitted. both `record_id` and `record_digest`; action-specific records additionally bind the exact `action_digest`. +The canonical `policy_ref.policy_digest` is the policy-bundle digest. It binds +the versioned JSON metadata and exact bytes of the Rego module executed +by OPA. The adapter recomputes and verifies this reference before invoking OPA; +the approval, decision, evidence verifier, and replay path require the same +bundle reference. + Milestone 2 defined these contracts. The canonical workflow emits and validates them around a real offline OPA decision and binds its deterministic synthetic approval through PDP and PEP enforcement. The canonical evidence verifier adds diff --git a/schemas/contracts/examples/approved-dry-run.json b/schemas/contracts/examples/approved-dry-run.json index e72cfed..5f1509d 100644 --- a/schemas/contracts/examples/approved-dry-run.json +++ b/schemas/contracts/examples/approved-dry-run.json @@ -159,7 +159,7 @@ "policy_ref": { "policy_id": "constraints-default-v1", "version": "v1.0", - "policy_digest": "sha256:091de3f0a96ec85a610f42456aaba98c8d04e148b9f910c570f96af37795b44d" + "policy_digest": "sha256:124eb62fbbf458a189184eafd09156be8bd764f26d65a51b7373db2cc61ce5b8" }, "issued_at": "2026-01-15T12:00:04Z", "expires_at": "2026-01-15T12:05:04Z", @@ -168,7 +168,7 @@ "environment": "lab", "resource_ids": ["workspace-0001"] }, - "record_digest": "sha256:7b3d00486cac26d09ad41d2f6cfeb1c11ef495bc31ddd6461570ef4aa64347e9" + "record_digest": "sha256:d4d2019534c49063811ae731618f6bd0134b1dcfa2c951f09226d0817886a68f" }, { "schema_version": "1.0.0", @@ -177,7 +177,7 @@ "workflow_id": "workflow-0001", "sequence": 6, "occurred_at": "2026-01-15T12:00:05Z", - "previous_record_digest": "sha256:7b3d00486cac26d09ad41d2f6cfeb1c11ef495bc31ddd6461570ef4aa64347e9", + "previous_record_digest": "sha256:d4d2019534c49063811ae731618f6bd0134b1dcfa2c951f09226d0817886a68f", "request_ref": { "record_id": "request-0001", "record_digest": "sha256:0fdf37fabbfd4b3364bf8377a83086afb92777126ef193330c6387125b1c1f58" @@ -189,7 +189,7 @@ }, "approval_ref": { "record_id": "approval-0001", - "record_digest": "sha256:7b3d00486cac26d09ad41d2f6cfeb1c11ef495bc31ddd6461570ef4aa64347e9" + "record_digest": "sha256:d4d2019534c49063811ae731618f6bd0134b1dcfa2c951f09226d0817886a68f" }, "pdp_id": "opa-pdp-0001", "decision": "permit", @@ -198,10 +198,10 @@ "policy_ref": { "policy_id": "constraints-default-v1", "version": "v1.0", - "policy_digest": "sha256:091de3f0a96ec85a610f42456aaba98c8d04e148b9f910c570f96af37795b44d" + "policy_digest": "sha256:124eb62fbbf458a189184eafd09156be8bd764f26d65a51b7373db2cc61ce5b8" }, "obligations": ["audit-required", "sandbox-required"], - "record_digest": "sha256:c3d41fd9f3c92bda6f178481428fae51cf1f92704889ce886602e12afe4947ed" + "record_digest": "sha256:0002be93d338e5013290940ac94ba246c2f83ca23200010f10b4efdff9a2d301" }, { "schema_version": "1.0.0", @@ -210,7 +210,7 @@ "workflow_id": "workflow-0001", "sequence": 7, "occurred_at": "2026-01-15T12:00:06Z", - "previous_record_digest": "sha256:c3d41fd9f3c92bda6f178481428fae51cf1f92704889ce886602e12afe4947ed", + "previous_record_digest": "sha256:0002be93d338e5013290940ac94ba246c2f83ca23200010f10b4efdff9a2d301", "request_ref": { "record_id": "request-0001", "record_digest": "sha256:0fdf37fabbfd4b3364bf8377a83086afb92777126ef193330c6387125b1c1f58" @@ -222,7 +222,7 @@ }, "decision_ref": { "record_id": "decision-0001", - "record_digest": "sha256:c3d41fd9f3c92bda6f178481428fae51cf1f92704889ce886602e12afe4947ed" + "record_digest": "sha256:0002be93d338e5013290940ac94ba246c2f83ca23200010f10b4efdff9a2d301" }, "execution_environment": { "sandbox_id": "sandbox-0001", @@ -239,7 +239,7 @@ "output_digest": "sha256:8cd39e51bc8da562bf5a85cb483365d3f9b4b0e8232f5be1c6cbe7b675527ed5", "error_code": null, "side_effects": [], - "record_digest": "sha256:acc4aa7d3b41cffc442915e5d6e82c00f5e928872015048192136340cd078b23" + "record_digest": "sha256:396b5feaaa48d651efc0214168fe90801d8ed410f25e104b26f3ce5c5b5aa5bc" }, { "schema_version": "1.0.0", @@ -248,7 +248,7 @@ "workflow_id": "workflow-0001", "sequence": 8, "occurred_at": "2026-01-15T12:00:07Z", - "previous_record_digest": "sha256:acc4aa7d3b41cffc442915e5d6e82c00f5e928872015048192136340cd078b23", + "previous_record_digest": "sha256:396b5feaaa48d651efc0214168fe90801d8ed410f25e104b26f3ce5c5b5aa5bc", "request_ref": { "record_id": "request-0001", "record_digest": "sha256:0fdf37fabbfd4b3364bf8377a83086afb92777126ef193330c6387125b1c1f58" @@ -288,23 +288,23 @@ { "record_type": "approval", "record_id": "approval-0001", - "record_digest": "sha256:7b3d00486cac26d09ad41d2f6cfeb1c11ef495bc31ddd6461570ef4aa64347e9" + "record_digest": "sha256:d4d2019534c49063811ae731618f6bd0134b1dcfa2c951f09226d0817886a68f" }, { "record_type": "policy_decision", "record_id": "decision-0001", - "record_digest": "sha256:c3d41fd9f3c92bda6f178481428fae51cf1f92704889ce886602e12afe4947ed" + "record_digest": "sha256:0002be93d338e5013290940ac94ba246c2f83ca23200010f10b4efdff9a2d301" }, { "record_type": "execution_result", "record_id": "result-0001", - "record_digest": "sha256:acc4aa7d3b41cffc442915e5d6e82c00f5e928872015048192136340cd078b23" + "record_digest": "sha256:396b5feaaa48d651efc0214168fe90801d8ed410f25e104b26f3ce5c5b5aa5bc" } ], "details": { "message": "Deterministic dry-run example; no external tool was invoked." }, - "record_digest": "sha256:df1997656225af8f250fdd5cee5370ac15750f2c8be9864de8bbbf48c4aab0d6" + "record_digest": "sha256:015bd9f3e0011894ef0eaf48b4944f3ca2d971103166aa45dee1f0030199e681" } ] } diff --git a/src/canonical/contracts.py b/src/canonical/contracts.py index 114fcd9..0a7c5f3 100644 --- a/src/canonical/contracts.py +++ b/src/canonical/contracts.py @@ -43,3 +43,15 @@ def calculate_action_digest(action: dict[str, Any]) -> str: def calculate_output_digest(output: dict[str, Any]) -> str: return digest_payload("output", output) + + +def calculate_policy_bundle_digest(metadata: dict[str, Any], rego_source: bytes) -> str: + """Bind versioned policy metadata to the exact executable Rego bytes.""" + + return digest_payload( + "policy-bundle", + { + "metadata": metadata, + "rego_sha256": f"sha256:{hashlib.sha256(rego_source).hexdigest()}", + }, + ) diff --git a/src/canonical/opa.py b/src/canonical/opa.py index f84e950..692a845 100644 --- a/src/canonical/opa.py +++ b/src/canonical/opa.py @@ -8,9 +8,13 @@ from pathlib import Path from typing import Any +from .contracts import calculate_policy_bundle_digest + QUERY = "data.f7las.canonical.result" IDENTIFIER = re.compile(r"^[a-z][a-z0-9]*(?:[-_][a-z0-9]+)*$") +REPO_ROOT = Path(__file__).resolve().parents[2] +DEFAULT_POLICY_METADATA_PATH = REPO_ROOT / "config" / "policies" / "policy-constraints-default.json" def _is_identifier(value: Any) -> bool: @@ -24,10 +28,17 @@ def _is_identifier(value: Any) -> bool: class OfflineOPA: """Evaluate one Rego decision without starting a network service.""" - def __init__(self, binary: str, policy_path: Path, timeout_seconds: float = 5.0) -> None: + def __init__( + self, + binary: str, + policy_path: Path, + timeout_seconds: float = 5.0, + policy_metadata_path: Path = DEFAULT_POLICY_METADATA_PATH, + ) -> None: self.binary = binary self.policy_path = policy_path self.timeout_seconds = timeout_seconds + self.policy_metadata_path = policy_metadata_path @staticmethod def _deny(reason_code: str) -> dict[str, Any]: @@ -37,7 +48,29 @@ def _deny(reason_code: str) -> dict[str, Any]: "obligations": ["audit-required"], } + def policy_ref(self) -> dict[str, str]: + """Describe the exact metadata and Rego bytes used by this adapter.""" + + with self.policy_metadata_path.open("r", encoding="utf-8") as handle: + metadata = json.load(handle) + rego_source = self.policy_path.read_bytes() + return { + "policy_id": metadata["policy_id"], + "version": metadata["version"], + "policy_digest": calculate_policy_bundle_digest(metadata, rego_source), + } + def evaluate(self, policy_input: dict[str, Any]) -> dict[str, Any]: + try: + expected_policy_ref = self.policy_ref() + except (OSError, UnicodeError, json.JSONDecodeError, KeyError, TypeError, ValueError): + return self._deny("pdp-policy-bundle-invalid") + if ( + policy_input.get("policy_ref") != expected_policy_ref + or policy_input.get("approval", {}).get("policy_ref") != expected_policy_ref + ): + return self._deny("pdp-policy-bundle-mismatch") + command = [ self.binary, "eval", diff --git a/src/canonical/replay.py b/src/canonical/replay.py index 1a56d4f..b12aad6 100644 --- a/src/canonical/replay.py +++ b/src/canonical/replay.py @@ -4,6 +4,7 @@ import argparse import json +import os from pathlib import Path from typing import Any @@ -18,6 +19,12 @@ class ReplayMismatchError(ValueError): """A valid evidence set was not reproduced exactly by replay.""" +def _same_file(first: Path, second: Path) -> bool: + if first.resolve() == second.resolve(): + return True + return first.exists() and second.exists() and os.path.samefile(first, second) + + def replay_evidence( workflow_input: dict[str, Any], expected: dict[str, Any], @@ -51,14 +58,13 @@ def main() -> int: parser.add_argument("--opa-binary", default="opa") args = parser.parse_args() - if args.output is not None and args.output.resolve() in { - args.input.resolve(), - args.evidence.resolve(), - }: - print("F7-LAS replay FAILED: output must not overwrite input or reviewed evidence") - return 4 - try: + if args.output is not None and ( + _same_file(args.output, args.input) or _same_file(args.output, args.evidence) + ): + raise ReplayMismatchError( + "output must not overwrite input or reviewed evidence" + ) workflow_input = load_json(args.input) expected = load_json(args.evidence) replayed = replay_evidence(workflow_input, expected, args.opa_binary) diff --git a/src/canonical/validation.py b/src/canonical/validation.py index 4a1d170..77ad27a 100644 --- a/src/canonical/validation.py +++ b/src/canonical/validation.py @@ -20,6 +20,7 @@ from src.canonical.contracts import ( calculate_action_digest, calculate_output_digest, + calculate_policy_bundle_digest, calculate_record_digest, digest_payload, ) @@ -28,6 +29,8 @@ SCHEMA_PATH = REPO_ROOT / "schemas" / "contracts" / "f7las-records-v1.schema.json" EXAMPLES_DIR = REPO_ROOT / "schemas" / "contracts" / "examples" POLICY_DIR = REPO_ROOT / "config" / "policies" +CANONICAL_POLICY_METADATA_PATH = POLICY_DIR / "policy-constraints-default.json" +CANONICAL_REGO_PATH = POLICY_DIR / "canonical-workflow.rego" FORBIDDEN_KEYS = { "api_key", @@ -60,9 +63,15 @@ def load_policy_registry() -> dict[str, dict[str, Any]]: if path.name == "policy-schema.json": continue policy = load_json(path) + policy_digest = digest_payload("policy", policy) + if path == CANONICAL_POLICY_METADATA_PATH: + policy_digest = calculate_policy_bundle_digest( + policy, + CANONICAL_REGO_PATH.read_bytes(), + ) registry[policy["policy_id"]] = { "version": policy["version"], - "policy_digest": digest_payload("policy", policy), + "policy_digest": policy_digest, } return registry diff --git a/src/canonical/workflow.py b/src/canonical/workflow.py index ba4fe26..0be0233 100644 --- a/src/canonical/workflow.py +++ b/src/canonical/workflow.py @@ -2,7 +2,6 @@ from __future__ import annotations -import json from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any @@ -19,7 +18,6 @@ REPO_ROOT = Path(__file__).resolve().parents[2] DEFAULT_POLICY_PATH = REPO_ROOT / "config" / "policies" / "canonical-workflow.rego" -POLICY_DOCUMENT_PATH = REPO_ROOT / "config" / "policies" / "policy-constraints-default.json" EXPECTED_MISSION = "Evaluate synthetic lab workspace health." EXPECTED_REQUESTER = { "subject_id": "operator-0001", @@ -94,15 +92,8 @@ def _validate_input(workflow_input: dict[str, Any]) -> datetime: raise WorkflowError("started_at must be whole-second RFC 3339 UTC") from exc return started_at.replace(tzinfo=timezone.utc) - @staticmethod - def _policy_ref() -> dict[str, str]: - with POLICY_DOCUMENT_PATH.open("r", encoding="utf-8") as handle: - document = json.load(handle) - return { - "policy_id": document["policy_id"], - "version": document["version"], - "policy_digest": digest_payload("policy", document), - } + def _policy_ref(self) -> dict[str, str]: + return self.opa.policy_ref() @staticmethod def _finalize( @@ -154,7 +145,7 @@ def _execute( or approval["issued_at"] != approval["occurred_at"] or approval["authority"] != EXPECTED_APPROVER or approval["approved_scope"] != action["target"] - or approval["policy_ref"] != CanonicalWorkflow._policy_ref() + or approval["policy_ref"] != OfflineOPA("opa", DEFAULT_POLICY_PATH).policy_ref() or approval["policy_ref"] != decision["policy_ref"] or decision["request_ref"] != _reference(request) or decision["approval_ref"] != _reference(approval) diff --git a/tests/test_canonical_workflow.py b/tests/test_canonical_workflow.py index f8d5ba1..a0fc10f 100644 --- a/tests/test_canonical_workflow.py +++ b/tests/test_canonical_workflow.py @@ -140,7 +140,30 @@ def test_real_opa_policy_denies_invalid_approval_binding(path, value): result = OfflineOPA(OPA_BINARY, DEFAULT_POLICY_PATH).evaluate(policy_input) assert result["decision"] == "deny" - assert result["reason_code"] == "policy-denied" + expected_reason = ( + "pdp-policy-bundle-mismatch" + if path == ("approval", "policy_ref", "version") + else "policy-denied" + ) + assert result["reason_code"] == expected_reason + + +def test_opa_rejects_modified_rego_before_subprocess(tmp_path, monkeypatch): + modified_policy = tmp_path / "modified.rego" + modified_policy.write_bytes(DEFAULT_POLICY_PATH.read_bytes() + b"\n# substituted policy\n") + monkeypatch.setattr( + subprocess, + "run", + lambda *args, **kwargs: pytest.fail("OPA executed before bundle verification"), + ) + + result = OfflineOPA("opa", modified_policy).evaluate(capture_policy_input()) + + assert result == { + "decision": "deny", + "reason_code": "pdp-policy-bundle-mismatch", + "obligations": ["audit-required"], + } @pytest.mark.skipif(OPA_BINARY is None, reason="OPA CLI is not installed") @@ -348,7 +371,7 @@ def test_malformed_opa_response_fails_closed(monkeypatch): completed = subprocess.CompletedProcess(args=["opa"], returncode=0, stdout="{}", stderr="") monkeypatch.setattr(subprocess, "run", lambda *args, **kwargs: completed) - result = OfflineOPA("opa", DEFAULT_POLICY_PATH).evaluate({}) + result = OfflineOPA("opa", DEFAULT_POLICY_PATH).evaluate(capture_policy_input()) assert result == { "decision": "deny", @@ -361,7 +384,7 @@ def test_opa_evaluation_error_fails_closed(monkeypatch): completed = subprocess.CompletedProcess(args=["opa"], returncode=2, stdout="", stderr="bad policy") monkeypatch.setattr(subprocess, "run", lambda *args, **kwargs: completed) - result = OfflineOPA("opa", DEFAULT_POLICY_PATH).evaluate({}) + result = OfflineOPA("opa", DEFAULT_POLICY_PATH).evaluate(capture_policy_input()) assert result["decision"] == "deny" assert result["reason_code"] == "pdp-evaluation-failed" diff --git a/tests/test_evidence_replay.py b/tests/test_evidence_replay.py index 316e678..287cc7d 100644 --- a/tests/test_evidence_replay.py +++ b/tests/test_evidence_replay.py @@ -6,12 +6,12 @@ import pytest -from src.canonical import evidence, replay +from src.canonical import evidence, replay, validation from src.canonical.contracts import calculate_record_digest from src.canonical.evidence import EvidenceIntegrityError, evidence_digest, verify_evidence from src.canonical.replay import ReplayMismatchError, replay_evidence from src.canonical.validation import load_json -from src.canonical.workflow import CanonicalWorkflow +from src.canonical.workflow import CanonicalWorkflow, DEFAULT_POLICY_PATH INPUT_PATH = Path("examples/canonical-workflow/request.json") @@ -129,6 +129,19 @@ def test_replay_detects_different_admitted_input(): replay_evidence(changed_input, expected, opa_binary="/does/not/exist/opa") +@pytest.mark.skipif(OPA_BINARY is None, reason="OPA CLI is not installed") +def test_modified_rego_fails_evidence_verification_and_replay(tmp_path, monkeypatch): + expected = CanonicalWorkflow(opa_binary=OPA_BINARY).run(workflow_input()) + modified_policy = tmp_path / "modified.rego" + modified_policy.write_bytes(DEFAULT_POLICY_PATH.read_bytes() + b"\n# substituted policy\n") + monkeypatch.setattr(validation, "CANONICAL_REGO_PATH", modified_policy) + + with pytest.raises(EvidenceIntegrityError, match="does not match the repository policy"): + verify_evidence(expected) + with pytest.raises(EvidenceIntegrityError, match="does not match the repository policy"): + replay_evidence(workflow_input(), expected, OPA_BINARY) + + def test_evidence_cli_returns_nonzero_for_tampering(tmp_path, monkeypatch): document = denied_evidence() record(document, "request")["mission"] = "tampered" @@ -179,3 +192,26 @@ def test_replay_cli_does_not_overwrite_reviewed_evidence(tmp_path, monkeypatch): ) assert replay.main() == 4 + + +def test_replay_cli_does_not_overwrite_hard_linked_evidence(tmp_path, monkeypatch): + expected_path = tmp_path / "expected.json" + output_path = tmp_path / "hard-linked-output.json" + original = json.dumps(denied_evidence()) + expected_path.write_text(original, encoding="utf-8") + os.link(expected_path, output_path) + monkeypatch.setattr( + "sys.argv", + [ + "f7las-replay", + "--input", + str(INPUT_PATH), + "--evidence", + str(expected_path), + "--output", + str(output_path), + ], + ) + + assert replay.main() == 4 + assert expected_path.read_text(encoding="utf-8") == original