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

@@ -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
+                    or review.get("normalizedSha256") != normalized.get("sha256")):
-                        review.get("normalizedSha256") != asset["normalizedDocx"].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()),
+        inventory_raw = args.inventory.read_bytes()
-                       json.loads(args.catalog.read_text()), args.expected_count,
+        catalog_raw = args.catalog.read_bytes()
-                       args.require_published, args.catalog.parent)
+        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

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":[]}'
+        asset = self.publish_fixture()
-        render = b"rendered-docx"
+        asset["review"]["structureSha256"] = "0" * 64
-        (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(),
-        }
         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":[]}'
+        asset = self.publish_fixture()
-        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"],
-        }
         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":[]}'
+        asset = self.publish_fixture()
-        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["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()