Reject history overwrite and unreviewed publication routing changes
All checks were successful
Build and publish policy-nexus image / build-and-push (push) Successful in 1m25s

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
This commit is contained in:
tegwick 2026-10-11 02:31:41 +02:00
parent bb193b2e87
commit 65862b1e0c
4 changed files with 58 additions and 5 deletions

View file

@ -47,3 +47,22 @@ class TransitionTests(unittest.TestCase):
self.old.write_text(json.dumps({'documents':[self.doc]}))
self.audit.write_text(json.dumps({'immutable_results':[{'id':'a','revision_path':'../outside','http_status':200,'served_html_sha256':'a'*64}]}))
with self.assertRaisesRegex(ValueError,'unsafe'): self.verify()
def test_address_and_owner_routing_changes_require_review(self):
for field in ('canonical_path','revision_path','source_repo','source_path','owner'):
with self.subTest(field=field):
self.write_candidate(self.doc | {field:'changed'})
with self.assertRaisesRegex(ValueError,'unreviewed'):
self.verify()
def test_symlink_parent_cannot_supply_history_from_outside_candidate(self):
import shutil
outside = self.root.parent / (self.root.name + '-outside')
outside.mkdir()
self.addCleanup(lambda: shutil.rmtree(outside))
(outside/'revisions/1').mkdir(parents=True)
(outside/'revisions/1/index.html').write_bytes(b'original')
shutil.rmtree(self.root/'a')
(self.root/'a').symlink_to(outside, target_is_directory=True)
with self.assertRaisesRegex(ValueError,'absent or unsafe'):
self.verify()

View file

@ -64,3 +64,26 @@ class PublishedHistoryTests(unittest.TestCase):
def test_changed_source_under_same_revision_refuses(self):
self.source.write_text(self.source.read_text().replace('No position.','Changed position.'))
with self.assertRaisesRegex(RuntimeError,'Publish a new revision'):self.clean_build()
def test_canonical_cannot_overwrite_archived_revision(self):
manifest=json.loads(self.manifest.read_text())
manifest['documents'][0]['canonical_path']=self.doc['revision_path']
self.manifest.write_text(json.dumps(manifest))
# Bump current revision so the ordinary manifest path check cannot catch it.
self.source.write_text(self.source.read_text().replace('draft-1','draft-2'))
with self.assertRaisesRegex(ValueError,'archived immutable'):
self.clean_build()
def test_legacy_redirect_cannot_overwrite_archived_revision(self):
manifest=json.loads(self.manifest.read_text())
manifest['documents'][0]['legacy_paths']=[self.doc['revision_path']]
self.manifest.write_text(json.dumps(manifest))
self.source.write_text(self.source.read_text().replace('draft-1','draft-2'))
with self.assertRaisesRegex(ValueError,'archived immutable'):
self.clean_build()
def test_delivered_commit_field_must_match_captured_manifest(self):
self.receipt['documents'][0]['delivered_manifest_source_revision']='c'*40
self.configure()
with self.assertRaisesRegex(RuntimeError,'delivered source revision'):
self.clean_build()

View file

@ -162,7 +162,7 @@ def _restore_historical_revisions(
target.write_bytes(page_bytes)
def _restore_published_history(manifest_path: Path, manifest: dict[str, Any], output: Path) -> None:
def _restore_published_history(manifest_path: Path, manifest: dict[str, Any], output: Path) -> set[PurePosixPath]:
"""Restore publisher-held observed bytes, never re-render old owner decisions."""
seen: set[PurePosixPath] = set()
root = manifest_path.parent.resolve()
@ -198,6 +198,9 @@ def _restore_published_history(manifest_path: Path, manifest: dict[str, Any], ou
for field in ("revision", "revision_path", "source_repo", "source_path", "source_digest"):
if record[field] != original[field]:
raise RuntimeError("published history differs from delivered manifest")
delivered_revision = record.get("delivered_manifest_source_revision", record["source_revision"])
if delivered_revision != original["source_revision"]:
raise RuntimeError("published history delivered source revision mismatch")
revision_path = _safe_path(record["revision_path"])
if "/revisions/" not in revision_path.as_posix() or revision_path in seen:
raise ValueError("published history requires unique immutable revision paths")
@ -225,6 +228,7 @@ def _restore_published_history(manifest_path: Path, manifest: dict[str, Any], ou
raise RuntimeError("published history immutable path conflicts with existing bytes")
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(page_bytes)
return seen
def _redirect(target: str, title: str) -> str:
@ -299,7 +303,11 @@ def build(
records: list[dict[str, str]] = []
try:
_restore_published_history(manifest_path, manifest, temporary)
archived_paths = _restore_published_history(manifest_path, manifest, temporary)
for document in manifest["documents"]:
for raw in (document["canonical_path"], *document.get("legacy_paths", [])):
if _safe_path(raw) in archived_paths:
raise ValueError("current publication write conflicts with archived immutable path")
for document in manifest["documents"]:
repo = resolved_repositories[document["source_repo"]]
source = (repo / document["source_path"]).resolve()

View file

@ -5,7 +5,8 @@ import hashlib
import json
from pathlib import Path, PurePosixPath
SEMANTIC_FIELDS = ('source_digest', 'revision', 'status', 'lifecycle', 'currency')
SEMANTIC_FIELDS = ('source_digest', 'revision', 'status', 'lifecycle', 'currency',
'canonical_path', 'revision_path', 'source_repo', 'source_path', 'owner')
def safe_path(value):
p = PurePosixPath(value)
@ -14,7 +15,7 @@ def safe_path(value):
return p
def verify(candidate, baseline_manifest, immutable_audit, allowed_changed_ids):
candidate = Path(candidate)
candidate = Path(candidate).resolve()
old = json.loads(Path(baseline_manifest).read_text())['documents']
new = json.loads((candidate / 'publication-manifest.json').read_text())['documents']
old_by = {d['id']: d for d in old}
@ -43,7 +44,9 @@ def verify(candidate, baseline_manifest, immutable_audit, allowed_changed_ids):
errors.append(f'{identity}: captured path differs from baseline manifest')
continue
path = candidate / safe_path(record['revision_path'])
if not path.is_file() or path.is_symlink():
if (not path.is_file() or path.is_symlink()
or candidate not in path.resolve().parents
or any(parent.is_symlink() for parent in path.parents if parent != candidate and candidate in parent.parents)) :
errors.append(f'{identity}: old immutable revision absent or unsafe')
elif hashlib.sha256(path.read_bytes()).hexdigest() != record['served_html_sha256']:
errors.append(f'{identity}: old immutable bytes changed')