weichen/Agent-System-Dev
1
0
Fork 0
Code Issues 1 Pull Requests 3 Actions 4 Packages Projects Releases Wiki Activity

Harden DHF catalog provenance and review checks

Browse Source 
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
This commit is contained in:
weichen
2026-10-14 04:37:05 +00:00
parent 3262459886
commit 38ef548637
2 changed files with 207 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/dhf_catalog_guard.py
+80 -10
View File
@@ -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
tests/test_dhf_catalog_guard.py
+127 -48
View File
@@ -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()