from __future__ import annotations import csv import hashlib import importlib.util import json import os import shutil from copy import deepcopy from pathlib import Path import pytest import app.core.governance.domain_replication_contract_registry as contract_registry from app.core.governance.domain_replication import ( REQUIRED_STAGES, DomainReplicationError, validate_contract_reference, validate_execution_evidence, ) from app.core.governance.domain_replication_contract_registry import registry_snapshot ROOT = Path(__file__).resolve().parents[2] PACKAGE_DIR = ROOT / "docs/phase3/p3-wp08-third-domain-template" P2_PACKAGE_DIR = ROOT / "docs/phase2/p2-wp12-spare-parts-package" VALIDATOR_PATH = ROOT / "scripts/validate_domain_replication.py" VALIDATOR_SPEC = importlib.util.spec_from_file_location("wp08_validator", VALIDATOR_PATH) assert VALIDATOR_SPEC and VALIDATOR_SPEC.loader validator = importlib.util.module_from_spec(VALIDATOR_SPEC) VALIDATOR_SPEC.loader.exec_module(validator) def _package_json(name: str) -> dict: return json.loads((PACKAGE_DIR / name).read_text(encoding="utf-8")) def _package_rows(name: str) -> list[dict[str, str]]: with (PACKAGE_DIR / "data" / name).open(encoding="utf-8", newline="") as handle: return list(csv.DictReader(handle)) def _package_copy(tmp_path: Path) -> Path: package = tmp_path / "package" shutil.copytree(PACKAGE_DIR, package) return package def _execution_evidence() -> dict: return { "schema_version": 1, "stages": [ { "stage": stage, "actions": [ { "action": "dry_run", "status": "passed", "write_count": 0, "evidence_ref": f"dry-run://third_domain_template/{stage}", "audit_ref": f"audit://third_domain_template/{stage}/dry-run", }, { "action": "execute", "status": "passed", "write_count": 1, "evidence_ref": f"execution://third_domain_template/{stage}", "audit_ref": f"audit://third_domain_template/{stage}/execute", }, { "action": "diff", "status": "passed", "write_count": 0, "evidence_ref": f"diff://third_domain_template/{stage}", "audit_ref": f"audit://third_domain_template/{stage}/diff", }, ], } for stage in REQUIRED_STAGES ], } def _strict_execution_evidence() -> dict: """V2 execution receipts bind rollback to its executed state transition.""" evidence = _execution_evidence() evidence["schema_version"] = 2 replay_receipts: list[dict[str, object]] = [] for index, receipt in enumerate(evidence["stages"]): stage = receipt["stage"] execute = receipt["actions"][1] execute.update( { "operation_uid": f"execute-{index:02d}", "idempotency_key": f"execute-{index:02d}", "request_digest": f"{index + 1:064x}", "attempt": 1, "lease_fence": index + 1, "pre_state_digest": f"{index + 101:064x}", "post_state_digest": f"{index + 201:064x}", "result_digest": f"{index + 301:064x}", } ) if stage == "remediation": rollback = receipt["actions"][2] rollback.update( { "action": "rollback", "operation": "restore", "operation_uid": execute["operation_uid"], "idempotency_key": execute["idempotency_key"], "request_digest": execute["request_digest"], "attempt": execute["attempt"], "lease_fence": execute["lease_fence"], "from_digest": execute["post_state_digest"], "to_digest": execute["pre_state_digest"], "result_digest": f"{index + 401:064x}", "status": "passed", } ) replay_receipts.append( { "stage": stage, "operation_uid": execute["operation_uid"], "idempotency_key": execute["idempotency_key"], "request_digest": execute["request_digest"], "attempt": execute["attempt"], "lease_fence": execute["lease_fence"], "from_digest": execute["post_state_digest"], "to_digest": execute["pre_state_digest"], "result_digest": rollback["result_digest"], "status": "exact_replay", } ) evidence["replay_receipts"] = replay_receipts return evidence def test_v2_rollback_requires_matching_execute_state_and_independent_replay_receipt(): execution = _strict_execution_evidence() assert validate_execution_evidence(execution)["remediation"]["recovery_strategy"] == "rollback" delayed = deepcopy(execution) delayed["stages"][6]["actions"][2]["lease_fence"] = 6 with pytest.raises(DomainReplicationError, match="lease_fence"): validate_execution_evidence(delayed) changed = deepcopy(execution) changed["replay_receipts"][0]["to_digest"] = "a" * 64 with pytest.raises(DomainReplicationError, match="replay receipt"): validate_execution_evidence(changed) forged = deepcopy(execution) forged["stages"][6]["actions"][2]["from_digest"] = "b" * 64 with pytest.raises(DomainReplicationError, match="execute post_state_digest"): validate_execution_evidence(forged) malformed = deepcopy(execution) malformed["stages"][6]["actions"][2]["result_digest"] = "not-a-digest" malformed["replay_receipts"][0]["result_digest"] = "not-a-digest" with pytest.raises(DomainReplicationError, match="remediation.rollback.result_digest"): validate_execution_evidence(malformed) def test_execution_evidence_requires_ordered_dry_run_execute_and_diff(): summary = validate_execution_evidence(_execution_evidence()) assert list(summary) == list(REQUIRED_STAGES) assert all(item["actions"] == ["dry_run", "execute", "diff"] for item in summary.values()) assert all(item["dry_run_write_count"] == 0 for item in summary.values()) assert all(item["recovery_strategy"] == "diff" for item in summary.values()) assert all(len(item["evidence_digest"]) == 64 for item in summary.values()) @pytest.mark.parametrize( ("mutate", "message"), [ ( lambda evidence: evidence["stages"][0]["actions"].__setitem__( 0, evidence["stages"][0]["actions"][1] ), "ordered", ), ( lambda evidence: evidence["stages"][0]["actions"][0].update( {"write_count": 1} ), "dry_run", ), ( lambda evidence: evidence["stages"][0]["actions"][2].update( {"action": "verify"} ), "diff or rollback", ), ( lambda evidence: evidence["stages"][0]["actions"][1].update( {"source_url": "https://enterprise.invalid/source"} ), "URL", ), ( lambda evidence: evidence["stages"][0]["actions"][1].update( {"raw_rows": [{"id": "sample"}]} ), "raw source rows", ), ], ) def test_execution_evidence_fails_closed_for_unsafe_or_incomplete_receipts( mutate, message: str ): evidence = _execution_evidence() mutate(evidence) with pytest.raises(DomainReplicationError, match=message): validate_execution_evidence(evidence) def test_execution_evidence_rejects_cross_domain_receipt_reference(): evidence = _execution_evidence() evidence["stages"][1]["actions"][1]["evidence_ref"] = ( "execution://other_domain/catalog" ) with pytest.raises(DomainReplicationError, match="cross-domain"): validate_execution_evidence(evidence) def test_execution_evidence_is_deterministic_across_a_replay(): evidence = _execution_evidence() assert validate_execution_evidence(evidence) == validate_execution_evidence( deepcopy(evidence) ) def test_third_domain_template_is_hash_bound_and_uat_blocked(): manifest = _package_json("manifest.json") evidence = _package_json("evidence.json") execution_evidence = _package_json("execution-evidence.json") report = __import__( "app.core.governance.domain_replication", fromlist=["evaluate_replication_package"] ).evaluate_replication_package( manifest, evidence, template=_package_json("domain-template.json"), snapshot_rows=_package_rows("governance_object_snapshot.csv"), delta_rows=_package_rows("governance_object_delta.csv"), execution_evidence=execution_evidence, ) assert report["schema_version"] == 2 assert report["status"] == "passed" assert report["enterprise_uat"] == { "status": "blocked_external", "unbound_requirements": [ "data_steward", "domain_owner", "readonly_source", "uat_users", ], } assert set(report["execution_evidence"]) == set(REQUIRED_STAGES) assert report["device_specific_core_changes"] == 0 assert "spare_parts" not in json.dumps(report, ensure_ascii=False) assert "/device-" not in json.dumps(report, ensure_ascii=False) assert "sample-001" not in json.dumps(report, ensure_ascii=False) for binding in manifest["files"].values(): path = PACKAGE_DIR / binding["path"] assert path.is_file() assert hashlib.sha256(path.read_bytes()).hexdigest() == binding["sha256"] assert json.loads((PACKAGE_DIR / "acceptance-report.json").read_text(encoding="utf-8")) == report @pytest.mark.parametrize("path_name", ["governance_object_snapshot.csv", "manifest.json"]) def test_package_reader_rejects_hardlinked_inputs(tmp_path: Path, path_name: str): package = _package_copy(tmp_path) target = next(path for path in package.rglob(path_name)) os.link(target, package / f"second-{path_name}") with pytest.raises(DomainReplicationError, match="hardlink"): validator.validate_package(package) def test_package_reader_rejects_symlinked_input(tmp_path: Path): package = _package_copy(tmp_path) snapshot = package / "data/governance_object_snapshot.csv" replacement = package / "snapshot-safe-copy.csv" snapshot.rename(replacement) snapshot.symlink_to(replacement.name) with pytest.raises(DomainReplicationError, match="symlink"): validator.validate_package(package) def test_package_reader_fails_closed_when_open_target_is_replaced( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): package = _package_copy(tmp_path) original_open = os.open swapped = False def race_open(path, flags, *args, **kwargs): nonlocal swapped if path == "governance_object_snapshot.csv" and not swapped: swapped = True snapshot = package / "data/governance_object_snapshot.csv" replacement = package / "replacement.csv" snapshot.rename(replacement) snapshot.symlink_to(replacement.name) return original_open(path, flags, *args, **kwargs) monkeypatch.setattr(validator.os, "open", race_open) with pytest.raises(DomainReplicationError, match="symlink|changed"): validator.validate_package(package) @pytest.mark.parametrize( ("target", "key", "value", "message"), [ ("manifest", "unknown_field", True, "unsupported fields"), ("manifest", "tоken", "x", "sensitive"), ("manifest", "rаw_rows", [{"id": "sample"}], "raw source"), ("template", "records", [{"id": "sample"}], "unsupported fields"), ("evidence", "raw_rows", [{"id": "sample"}], "raw source"), ("evidence", "source_url", "https://enterprise.invalid", "URL"), ], ) def test_closed_package_schemas_reject_unknown_confusable_and_unsafe_values( target: str, key: str, value: object, message: str ): manifest = _package_json("manifest.json") evidence = _package_json("evidence.json") template = _package_json("domain-template.json") {"manifest": manifest, "evidence": evidence, "template": template}[target][key] = value with pytest.raises(DomainReplicationError, match=message): __import__( "app.core.governance.domain_replication", fromlist=["evaluate_replication_package"] ).evaluate_replication_package( manifest, evidence, template=template, snapshot_rows=_package_rows("governance_object_snapshot.csv"), delta_rows=_package_rows("governance_object_delta.csv"), execution_evidence=_package_json("execution-evidence.json"), ) def test_terminal_evidence_requires_allowlisted_api_and_typed_connector_contract(): evidence = _package_json("evidence.json") incremental = next(item for item in evidence["receipts"] if item["stage"] == "incremental_ingestion") incremental["api_refs"] = ["not-a-contract"] with pytest.raises(DomainReplicationError, match="allowlisted API"): _accept_third_domain(evidence=evidence) evidence = _package_json("evidence.json") incremental = next(item for item in evidence["receipts"] if item["stage"] == "incremental_ingestion") incremental["contract_refs"][0]["run_id"] = "not-a-contract" with pytest.raises(DomainReplicationError, match="trusted registry"): _accept_third_domain(evidence=evidence) def test_closed_schema_rejects_terminal_unknown_field_and_depth_or_size_attack(): evidence = _package_json("evidence.json") evidence["receipts"][0]["unexpected"] = True with pytest.raises(DomainReplicationError, match="unsupported fields"): _accept_third_domain(evidence=evidence) manifest = _package_json("manifest.json") nested: dict[str, object] = {"leaf": "safe"} for _ in range(13): nested = {"safe": nested} manifest["safe"] = nested with pytest.raises(DomainReplicationError, match="maximum package depth"): __import__( "app.core.governance.domain_replication", fromlist=["evaluate_replication_package"] ).evaluate_replication_package( manifest, _package_json("evidence.json"), template=_package_json("domain-template.json"), snapshot_rows=_package_rows("governance_object_snapshot.csv"), delta_rows=_package_rows("governance_object_delta.csv"), execution_evidence=_package_json("execution-evidence.json"), ) evidence = _package_json("evidence.json") evidence["safe"] = "x" * 4097 with pytest.raises(DomainReplicationError, match="maximum string length"): _accept_third_domain(evidence=evidence) def test_rollback_receipt_binds_replay_fence_and_digests(): execution = _execution_evidence() rollback = execution["stages"][0]["actions"][2] rollback.update( { "action": "rollback", "operation": "restore", "idempotency_key": "rollback-001", "request_digest": "a" * 64, "attempt": 1, "lease_fence": 4, "from_digest": "b" * 64, "to_digest": "c" * 64, "replay_status": "exact_replay", } ) assert validate_execution_evidence(execution)["template_initialization"]["recovery_strategy"] == "rollback" rollback["lease_fence"] = 0 with pytest.raises(DomainReplicationError, match="lease_fence"): validate_execution_evidence(execution) rollback["lease_fence"] = 4 rollback["replay_status"] = "changed_replay_conflict" with pytest.raises(DomainReplicationError, match="exact_replay"): validate_execution_evidence(execution) def test_v2_contract_reference_must_match_trusted_registry_run(): registry = registry_snapshot() contract = registry["contracts"][0] run = contract["runs"][0] reference = { "contract_id": contract["contract_id"], "contract_version": contract["contract_version"], "schema_digest": contract["schema_digest"], **run, } assert validate_contract_reference(reference, expected_contract_id=contract["contract_id"]) == reference for key, value in (("run_id", "invented-run"), ("operation", "unknown"), ("schema_digest", "f" * 64)): forged = deepcopy(reference) forged[key] = value with pytest.raises(DomainReplicationError, match="registry|contract"): validate_contract_reference(forged, expected_contract_id=contract["contract_id"]) def test_trusted_registry_rejects_tamper_expiry_and_contract_version_mismatch(monkeypatch): original_registry = registry_snapshot() original_digest = contract_registry.REGISTRY_SHA256 contract = registry_snapshot()["contracts"][0] reference = { "contract_id": contract["contract_id"], "contract_version": contract["contract_version"], "schema_digest": contract["schema_digest"], **contract["runs"][0], } tampered = registry_snapshot() tampered["contracts"][0]["allowed_operations"] = ["catalog"] monkeypatch.setattr(contract_registry, "CONTRACT_REGISTRY", tampered) with pytest.raises(DomainReplicationError, match="integrity"): validate_contract_reference(reference, expected_contract_id=contract["contract_id"]) expired = registry_snapshot() expired["expires_on"] = "2000-01-01" monkeypatch.setattr(contract_registry, "CONTRACT_REGISTRY", expired) monkeypatch.setattr( contract_registry, "REGISTRY_SHA256", hashlib.sha256(contract_registry._canonical(expired)).hexdigest(), ) with pytest.raises(DomainReplicationError, match="expired"): validate_contract_reference(reference, expected_contract_id=contract["contract_id"]) monkeypatch.setattr(contract_registry, "CONTRACT_REGISTRY", original_registry) monkeypatch.setattr(contract_registry, "REGISTRY_SHA256", original_digest) mismatch = deepcopy(reference) mismatch["contract_version"] = "999" with pytest.raises(DomainReplicationError, match="version"): validate_contract_reference(mismatch, expected_contract_id=contract["contract_id"]) def test_v1_p2_golden_package_is_accepted_without_v2_contract_refs(tmp_path: Path): package = tmp_path / "p2-v1-copy" shutil.copytree(P2_PACKAGE_DIR, package) report = validator.validate_package(package) assert report["schema_version"] == 1 assert all("contract_refs" not in stage for stage in report["stages"].values()) def test_v2_seed_items_reject_unknown_nested_object_field(): template = _package_json("domain-template.json") template["seed_data"][0]["items"][0]["unexpected"] = True with pytest.raises(DomainReplicationError, match="unsupported fields"): _accept_third_domain(evidence=_package_json("evidence.json"), template=template) def _accept_third_domain(*, evidence: dict, template: dict | None = None): return __import__( "app.core.governance.domain_replication", fromlist=["evaluate_replication_package"] ).evaluate_replication_package( _package_json("manifest.json"), evidence, template=template or _package_json("domain-template.json"), snapshot_rows=_package_rows("governance_object_snapshot.csv"), delta_rows=_package_rows("governance_object_delta.csv"), execution_evidence=_package_json("execution-evidence.json"), )