From 38ef548637f426e2e36c33a50ae6b8af8c9da453 Mon Sep 17 00:00:00 2001 From: weichen Date: Wed, 14 Oct 2026 04:37:05 +0000 Subject: [PATCH] Harden DHF catalog provenance and review checks Why: Catalog identity and publication metadata need mechanical checks without implying full asset approval. Evidence: 16 focused tests and independent scoped review passed; authoritative RDM and human review remain pending. Impact: Reject symlinked source/package paths, identity drift, invalid reviewer metadata, and incomplete converter provenance. Refs: feature/rdm-two-trees-latest-templates W0 --- scripts/dhf_catalog_guard.py | 90 ++++++++++++++-- tests/test_dhf_catalog_guard.py | 175 +++++++++++++++++++++++--------- 2 files changed, 207 insertions(+), 58 deletions(-) diff --git a/scripts/dhf_catalog_guard.py b/scripts/dhf_catalog_guard.py index d6f1ec37..f7a69eaa 100644 --- a/scripts/dhf_catalog_guard.py +++ b/scripts/dhf_catalog_guard.py @@ -7,6 +7,7 @@ meaning or Word rendering fidelity; those require the per-asset review. import argparse from collections import Counter +from datetime import datetime import hashlib import json from pathlib import Path @@ -19,6 +20,7 @@ SOURCE_PREFIX = (".templates", "DHF 文件模板") STATES = {"needs-review", "blocked", "published", "retired"} SHA256 = re.compile(r"[0-9a-f]{64}\Z") COMMIT = re.compile(r"[0-9a-f]{40}\Z") +RFC3339 = re.compile(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})\Z") def digest(raw): @@ -30,6 +32,11 @@ def source_path(dev_root, source): if (path.is_absolute() or path.parts[:2] != SOURCE_PREFIX or ".." in path.parts or path.suffix.lower() not in {".doc", ".docx"}): raise ValueError(f"unsafe or non-Word source path: {source}") + segment = dev_root + for part in path.parts: + segment = segment / part + if segment.is_symlink(): + raise ValueError(f"source path symlink: {source}") absolute = dev_root / path if not absolute.resolve().is_relative_to(dev_root.resolve()) or absolute.is_symlink(): raise ValueError(f"source path leaves the Dev tree: {source}") @@ -44,6 +51,11 @@ def package_digest(root, file_ref): path = Path(relative) if path.is_absolute() or ".." in path.parts or path == Path("."): raise ValueError("published asset file reference is unsafe") + segment = root + for part in path.parts: + segment = segment / part + if segment.is_symlink(): + raise ValueError("published asset file symlink") absolute = root / path if not absolute.resolve().is_relative_to(root.resolve()) or absolute.is_symlink() or not absolute.is_file(): raise ValueError("published asset file missing or escapes catalog") @@ -52,7 +64,8 @@ def package_digest(root, file_ref): return expected -def audit(dev_root, inventory, catalog, expected_count, require_published=False, catalog_root=None): +def audit(dev_root, inventory, catalog, expected_count, require_published=False, + catalog_root=None, identity_baseline=None): records = inventory.get("templates") assets = catalog.get("assets") if not isinstance(records, list) or not isinstance(assets, list): @@ -71,6 +84,21 @@ def audit(dev_root, inventory, catalog, expected_count, require_published=False, raise ValueError("duplicate or missing logical template ID") by_path[path] = asset ids.add(template_id) + if identity_baseline is not None: + prior = identity_baseline.get("assets") + if not isinstance(prior, list) or len(prior) != expected_count: + raise ValueError("identity baseline is incomplete") + prior_ids = {} + for asset in prior: + if not isinstance(asset, dict): + raise ValueError("identity baseline has invalid asset") + path = (asset.get("sourceProvenance") or {}).get("path") + if path in prior_ids or not isinstance(path, str): + raise ValueError("identity baseline has duplicate source") + prior_ids[path] = asset.get("templateId") + for path, asset in by_path.items(): + if prior_ids.get(path) != asset["templateId"]: + raise ValueError(f"template identity drift: {path}") seen = set() states = Counter() hashes = set() @@ -116,23 +144,51 @@ def audit(dev_root, inventory, catalog, expected_count, require_published=False, raise ValueError(f"invalid asset state: {relative}") if state == "published": review = asset.get("review") or {} - if (not catalog_types or not applicability.get("confirmed") or review.get("decision") != "approved" + normalized = asset.get("normalizedDocx") or {} + condition = applicability.get("condition") + if (not normalized or not catalog_types or applicability.get("confirmed") is not True + or not isinstance(condition, str) or not condition.strip() + or review.get("decision") != "approved" or review.get("sourceSha256") != expected or not asset.get("structure") or not asset.get("render")): raise ValueError(f"published asset lacks approved binding: {relative}") + if (normalized.get("format") != "docx" or asset["structure"].get("format") != "json" + or asset["render"].get("format") != "docx"): + raise ValueError(f"published asset has invalid derived formats: {relative}") + reviewer = review.get("reviewer") + evidence = review.get("evidence") + reviewed_at = review.get("reviewedAt") + if (not isinstance(reviewer, str) or not reviewer.strip() + or not isinstance(evidence, str) or not evidence.strip() + or not isinstance(reviewed_at, str)): + raise ValueError(f"published asset lacks reviewer evidence: {relative}") + if not RFC3339.fullmatch(reviewed_at): + raise ValueError(f"published asset has invalid review time: {relative}") + try: + at = datetime.fromisoformat(reviewed_at.replace("Z", "+00:00")) + except ValueError as exc: + raise ValueError(f"published asset has invalid review time: {relative}") from exc + if at.tzinfo is None: + raise ValueError(f"published asset has invalid review time: {relative}") + converter = asset.get("converter") or {} + if (asset["source"].get("format") == "doc" or + asset["source"].get("sha256") != normalized.get("sha256")): + if (converter.get("inputSha256") != expected or + converter.get("outputSha256") != normalized.get("sha256") or + any(not isinstance(converter.get(field), str) or not converter[field].strip() + for field in ("name", "version", "configuration"))): + raise ValueError(f"published asset lacks conversion provenance: {relative}") if any(issue.get("status") != "resolved" for issue in asset.get("issues", [])): raise ValueError(f"published asset has unresolved issues: {relative}") if (review.get("structureSha256") != asset["structure"].get("sha256") or review.get("renderSha256") != asset["render"].get("sha256") - or (asset.get("normalizedDocx") and - review.get("normalizedSha256") != asset["normalizedDocx"].get("sha256"))): + or review.get("normalizedSha256") != normalized.get("sha256")): raise ValueError(f"published asset review hash mismatch: {relative}") if catalog_root is not None: package_digest(catalog_root, asset["source"]) + package_digest(catalog_root, normalized) package_digest(catalog_root, asset["structure"]) package_digest(catalog_root, asset["render"]) - if asset.get("normalizedDocx"): - package_digest(catalog_root, asset["normalizedDocx"]) states[state] += 1 hashes.add(expected) if require_published and states.get("published", 0) != expected_count: @@ -145,7 +201,9 @@ def audit(dev_root, inventory, catalog, expected_count, require_published=False, "published": states.get("published", 0), "states": dict(sorted(states.items())), "candidateTypeDifferences": candidate_differences, - "limits": "This does not verify content classification, applicability decisions, Word fidelity or real product behavior.", + "identityBaselineVerified": identity_baseline is not None, + "publicationValidation": "NOT_RUN: RDM LoadDHFAssetCatalog and human content/Word review are authoritative", + "limits": "This checks provenance and obvious metadata errors; state counts do not prove published bundles are valid.", } @@ -155,14 +213,26 @@ def main(): parser.add_argument("--inventory", type=Path, required=True) parser.add_argument("--catalog", type=Path, required=True) parser.add_argument("--expected-count", type=int, default=44) + parser.add_argument("--identity-baseline", type=Path) parser.add_argument("--require-published", action="store_true") args = parser.parse_args() try: if args.expected_count < 1: raise ValueError("expected-count must be positive") - result = audit(args.dev_root, json.loads(args.inventory.read_text()), - json.loads(args.catalog.read_text()), args.expected_count, - args.require_published, args.catalog.parent) + inventory_raw = args.inventory.read_bytes() + catalog_raw = args.catalog.read_bytes() + baseline_raw = args.identity_baseline.read_bytes() if args.identity_baseline else None + result = audit(args.dev_root, json.loads(inventory_raw), json.loads(catalog_raw), + args.expected_count, args.require_published, args.catalog.parent, + json.loads(baseline_raw) if baseline_raw is not None else None) + if args.catalog.read_bytes() != catalog_raw: + raise ValueError("catalog changed during check") + result["catalogSha256"] = digest(catalog_raw) + result["inventorySha256"] = digest(inventory_raw) + if baseline_raw is not None: + result["identityBaselineSha256"] = digest(baseline_raw) + if args.require_published: + raise ValueError("RDM publication validator required for full bundle safety and human fidelity") except (OSError, ValueError, json.JSONDecodeError) as exc: print(f"dhf catalog guard: {exc}", file=sys.stderr) return 1 diff --git a/tests/test_dhf_catalog_guard.py b/tests/test_dhf_catalog_guard.py index d7139c72..31bf34b9 100644 --- a/tests/test_dhf_catalog_guard.py +++ b/tests/test_dhf_catalog_guard.py @@ -5,6 +5,7 @@ import subprocess import sys import tempfile import unittest +from copy import deepcopy ROOT = Path(__file__).resolve().parents[1] @@ -65,6 +66,32 @@ class DHFCatalogGuardTest(unittest.TestCase): "--expected-count", "2", *extra, ], capture_output=True, text=True) + def publish_fixture(self, index=0, *, structure=b'{"blocks":[]}', render=b"rendered-docx"): + asset = self.assets[index] + number = index + 1 + source = (self.dev / self.items[index]["source"]).read_bytes() + names = {"source": f"source{number}.docx", "structure": f"structure{number}.json", + "render": f"render{number}.docx"} + for label, raw in [("source", source), ("structure", structure), ("render", render)]: + (self.root / names[label]).write_bytes(raw) + asset["state"] = "published" + asset["applicability"].update({"confirmed": True, "condition": "test scope"}) + asset["source"]["path"] = names["source"] + asset["normalizedDocx"] = {"path": names["source"], "format": "docx", + "sha256": hashlib.sha256(source).hexdigest()} + asset["structure"] = {"path": names["structure"], "format": "json", + "sha256": hashlib.sha256(structure).hexdigest()} + asset["render"] = {"path": names["render"], "format": "docx", + "sha256": hashlib.sha256(render).hexdigest()} + asset["review"] = { + "decision": "approved", "reviewer": "reviewer", "reviewedAt": "2026-10-14T00:00:00Z", + "evidence": "review-record", "sourceSha256": self.items[index]["sourceSha256"], + "normalizedSha256": asset["normalizedDocx"]["sha256"], + "structureSha256": asset["structure"]["sha256"], + "renderSha256": asset["render"]["sha256"], + } + return asset + def test_accepts_matching_immutable_sources_but_does_not_call_them_published(self): result = self.run_guard() self.assertEqual(result.returncode, 0, result.stdout + result.stderr) @@ -79,6 +106,16 @@ class DHFCatalogGuardTest(unittest.TestCase): self.assertNotEqual(result.returncode, 0) self.assertIn("checkout hash mismatch", result.stderr) + def test_rejects_original_in_symlinked_parent_directory(self): + source = self.dev / self.items[0]["source"] + original_parent = source.parent + moved_parent = original_parent.parent / "alias-target" + original_parent.rename(moved_parent) + original_parent.symlink_to(moved_parent, target_is_directory=True) + result = self.run_guard() + self.assertNotEqual(result.returncode, 0) + self.assertIn("source path symlink", result.stderr) + def test_rejects_false_publication(self): self.assets[0]["state"] = "published" self.save() @@ -106,72 +143,114 @@ class DHFCatalogGuardTest(unittest.TestCase): self.assertIn("not published", result.stderr) def test_rejects_published_asset_whose_review_does_not_bind_bundle_bytes(self): - structure = b'{"blocks":[]}' - render = b"rendered-docx" - (self.root / "structure.json").write_bytes(structure) - (self.root / "render.docx").write_bytes(render) - asset = self.assets[0] - asset["state"] = "published" - asset["applicability"]["confirmed"] = True - asset["structure"] = {"path": "structure.json", "sha256": hashlib.sha256(structure).hexdigest()} - asset["render"] = {"path": "render.docx", "sha256": hashlib.sha256(render).hexdigest()} - asset["review"] = { - "decision": "approved", "sourceSha256": self.items[0]["sourceSha256"], - "structureSha256": "0" * 64, - "renderSha256": hashlib.sha256(render).hexdigest(), - } + asset = self.publish_fixture() + asset["review"]["structureSha256"] = "0" * 64 self.save() result = self.run_guard() self.assertNotEqual(result.returncode, 0) self.assertIn("published asset review hash mismatch", result.stderr) def test_rejects_bundle_file_changed_after_approval(self): - structure = b'{"blocks":[]}' - render = b"rendered-docx" - original = (self.dev / self.items[0]["source"]).read_bytes() - (self.root / "source.docx").write_bytes(original) - (self.root / "structure.json").write_bytes(structure) - (self.root / "render.docx").write_bytes(render) - asset = self.assets[0] - asset["state"] = "published" - asset["applicability"]["confirmed"] = True - asset["source"]["path"] = "source.docx" - asset["structure"] = {"path": "structure.json", "sha256": hashlib.sha256(structure).hexdigest()} - asset["render"] = {"path": "render.docx", "sha256": hashlib.sha256(render).hexdigest()} - asset["review"] = { - "decision": "approved", "sourceSha256": self.items[0]["sourceSha256"], - "structureSha256": asset["structure"]["sha256"], - "renderSha256": asset["render"]["sha256"], - } + asset = self.publish_fixture() self.save() - (self.root / "render.docx").write_bytes(b"modified-after-review") + (self.root / asset["render"]["path"]).write_bytes(b"modified-after-review") result = self.run_guard() self.assertNotEqual(result.returncode, 0) self.assertIn("published asset file hash mismatch", result.stderr) def test_rejects_published_asset_with_unresolved_fidelity_issue(self): - structure = b'{"blocks":[]}' - render = b"rendered-docx" - (self.root / "source.docx").write_bytes((self.dev / self.items[0]["source"]).read_bytes()) - (self.root / "structure.json").write_bytes(structure) - (self.root / "render.docx").write_bytes(render) - asset = self.assets[0] - asset["state"] = "published" - asset["applicability"]["confirmed"] = True - asset["source"]["path"] = "source.docx" - asset["structure"] = {"path": "structure.json", "sha256": hashlib.sha256(structure).hexdigest()} - asset["render"] = {"path": "render.docx", "sha256": hashlib.sha256(render).hexdigest()} - asset["review"] = { - "decision": "approved", "sourceSha256": self.items[0]["sourceSha256"], - "structureSha256": asset["structure"]["sha256"], - "renderSha256": asset["render"]["sha256"], - } + asset = self.publish_fixture() asset["issues"] = [{"id": "fidelity", "status": "open"}] self.save() result = self.run_guard() self.assertNotEqual(result.returncode, 0) self.assertIn("published asset has unresolved issues", result.stderr) + def test_publication_flag_never_claims_authoritative_rdm_validation(self): + for index in range(2): + self.publish_fixture(index, structure=b"not-a-valid-structure", render=b"not-a-valid-docx") + self.save() + result = self.run_guard("--require-published") + self.assertNotEqual(result.returncode, 0) + self.assertIn("RDM publication validator required", result.stderr) + + def test_rejects_swapped_logical_ids_against_fixed_identity_baseline(self): + baseline = self.root / "identity-baseline.json" + baseline.write_text(json.dumps({"assets": deepcopy(self.assets)})) + self.assets[0]["templateId"], self.assets[1]["templateId"] = ( + self.assets[1]["templateId"], self.assets[0]["templateId"]) + self.save() + result = self.run_guard("--identity-baseline", str(baseline)) + self.assertNotEqual(result.returncode, 0) + self.assertIn("template identity drift", result.stderr) + + def test_report_binds_exact_catalog_bytes(self): + result = self.run_guard() + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(json.loads(result.stdout)["catalogSha256"], + hashlib.sha256(self.catalog.read_bytes()).hexdigest()) + + def test_rejects_published_package_through_symlink_directory(self): + asset = self.publish_fixture() + real = self.root / "real" + real.mkdir() + for key in ["source", "normalizedDocx", "structure", "render"]: + p = self.root / asset[key]["path"] + (real / p.name).write_bytes(p.read_bytes()) + asset[key]["path"] = "alias/" + p.name + (self.root / "alias").symlink_to(real, target_is_directory=True) + self.save() + result = self.run_guard() + self.assertNotEqual(result.returncode, 0) + self.assertIn("symlink", result.stderr) + + def test_rejects_string_false_and_null_applicability(self): + asset = self.publish_fixture() + asset["applicability"]["confirmed"] = "false" + self.save() + result = self.run_guard() + self.assertNotEqual(result.returncode, 0) + self.assertIn("published asset lacks approved binding", result.stderr) + + asset["applicability"]["confirmed"] = True + asset["applicability"]["condition"] = None + self.save() + result = self.run_guard() + self.assertNotEqual(result.returncode, 0) + self.assertIn("published asset lacks approved binding", result.stderr) + + def test_rejects_null_reviewer_and_non_rfc3339_timestamp(self): + asset = self.publish_fixture() + asset["review"]["reviewer"] = None + self.save() + result = self.run_guard() + self.assertNotEqual(result.returncode, 0) + self.assertIn("published asset lacks reviewer evidence", result.stderr) + + asset["review"]["reviewer"] = "reviewer" + asset["review"]["reviewedAt"] = "2026-10-14 00:00:00Z" + self.save() + result = self.run_guard() + self.assertNotEqual(result.returncode, 0) + self.assertIn("published asset has invalid review time", result.stderr) + + def test_rejects_changed_normalized_file_without_named_converter(self): + asset = self.publish_fixture() + normalized = b"changed-normalized-docx" + (self.root / "normalized.docx").write_bytes(normalized) + asset["normalizedDocx"] = { + "path": "normalized.docx", "format": "docx", "sha256": hashlib.sha256(normalized).hexdigest(), + } + asset["review"]["normalizedSha256"] = asset["normalizedDocx"]["sha256"] + asset["converter"] = { + "inputSha256": self.items[0]["sourceSha256"], + "outputSha256": asset["normalizedDocx"]["sha256"], + } + self.save() + result = self.run_guard() + self.assertNotEqual(result.returncode, 0) + self.assertIn("published asset lacks conversion provenance", result.stderr) + if __name__ == "__main__": unittest.main()