245 lines
12 KiB
Python
Executable File
245 lines
12 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Compare the DHF inventory, immutable originals and one RDM asset catalog.
|
|
|
|
This checks provenance and publication metadata. It cannot decide document
|
|
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
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
|
|
|
|
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):
|
|
return hashlib.sha256(raw).hexdigest()
|
|
|
|
|
|
def source_path(dev_root, source):
|
|
path = Path(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}")
|
|
return absolute
|
|
|
|
|
|
def package_digest(root, file_ref):
|
|
relative = file_ref.get("path")
|
|
expected = file_ref.get("sha256")
|
|
if not isinstance(relative, str) or not isinstance(expected, str) or not SHA256.fullmatch(expected):
|
|
raise ValueError("published asset file reference missing path or hash")
|
|
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")
|
|
if digest(absolute.read_bytes()) != expected:
|
|
raise ValueError("published asset file hash mismatch")
|
|
return expected
|
|
|
|
|
|
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):
|
|
raise ValueError("inventory/templates or catalog/assets is missing")
|
|
if len(records) != expected_count or len(assets) != expected_count:
|
|
raise ValueError(f"expected {expected_count} logical sources and assets")
|
|
by_path = {}
|
|
ids = set()
|
|
for asset in assets:
|
|
provenance = asset.get("sourceProvenance") or {}
|
|
path = provenance.get("path")
|
|
template_id = asset.get("templateId")
|
|
if path in by_path or not isinstance(path, str):
|
|
raise ValueError("duplicate or missing provenance path")
|
|
if template_id in ids or not isinstance(template_id, str) or not template_id:
|
|
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()
|
|
candidate_differences = []
|
|
for record in records:
|
|
relative = record.get("source")
|
|
if not isinstance(relative, str) or relative in seen or relative not in by_path:
|
|
raise ValueError(f"inventory source missing, duplicate or unmapped: {relative}")
|
|
seen.add(relative)
|
|
asset = by_path[relative]
|
|
expected = record.get("sourceSha256")
|
|
found = (asset.get("source") or {}).get("sha256")
|
|
if not isinstance(expected, str) or not SHA256.fullmatch(expected) or found != expected:
|
|
raise ValueError(f"catalog source hash mismatch: {relative}")
|
|
path = source_path(dev_root, relative)
|
|
if not path.is_file() or digest(path.read_bytes()) != expected:
|
|
raise ValueError(f"checkout hash mismatch: {relative}")
|
|
provenance = asset["sourceProvenance"]
|
|
commit = provenance.get("commit")
|
|
if provenance.get("repository") != "Agent-System-Dev" or not isinstance(commit, str) or not COMMIT.fullmatch(commit):
|
|
raise ValueError(f"untrusted source provenance: {relative}")
|
|
historical = subprocess.run(
|
|
["git", "-C", str(dev_root), "show", f"{commit}:{relative}"],
|
|
capture_output=True, check=False,
|
|
)
|
|
if historical.returncode or digest(historical.stdout) != expected:
|
|
raise ValueError(f"committed source hash mismatch: {relative}")
|
|
if (asset.get("source") or {}).get("format") != path.suffix.lower()[1:]:
|
|
raise ValueError(f"catalog format mismatch: {relative}")
|
|
applicability = asset.get("applicability") or {}
|
|
catalog_types = applicability.get("docTypes", [])
|
|
if not isinstance(catalog_types, list):
|
|
raise ValueError(f"invalid catalog docTypes: {relative}")
|
|
if record.get("proposedDocType") not in catalog_types:
|
|
candidate_differences.append({
|
|
"path": relative,
|
|
"inventoryCandidate": record.get("proposedDocType"),
|
|
"catalogCandidates": catalog_types,
|
|
"confirmed": bool(applicability.get("confirmed")),
|
|
})
|
|
state = asset.get("state")
|
|
if state not in STATES:
|
|
raise ValueError(f"invalid asset state: {relative}")
|
|
if state == "published":
|
|
review = asset.get("review") or {}
|
|
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 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"])
|
|
states[state] += 1
|
|
hashes.add(expected)
|
|
if require_published and states.get("published", 0) != expected_count:
|
|
raise ValueError(f"{expected_count - states.get('published', 0)} assets not published")
|
|
return {
|
|
"source": "SELF mechanical provenance check",
|
|
"matched": len(seen),
|
|
"logicalTemplateIds": len(ids),
|
|
"uniqueSourceBytes": len(hashes),
|
|
"published": states.get("published", 0),
|
|
"states": dict(sorted(states.items())),
|
|
"candidateTypeDifferences": candidate_differences,
|
|
"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.",
|
|
}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--dev-root", type=Path, required=True)
|
|
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")
|
|
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
|
|
print(json.dumps(result, ensure_ascii=False, sort_keys=True))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|