feat(evidence): add safe retention and materialized reads

This commit is contained in:
2026-07-12 05:02:47 +02:00
parent b6a52995ae
commit f3b49f41c8
11 changed files with 310 additions and 14 deletions
+41
View File
@@ -1,4 +1,5 @@
import pytest
import os
from tht.corpus.models import CorpusManifest
from tht.corpus.store import CorpusStore, UnsafeCorpusPath
@@ -56,3 +57,43 @@ def test_publish_restores_previous_active_when_directory_fsync_fails_after_repla
with pytest.raises(OSError, match="post replace"):
store.publish(second)
assert store.active_generation() == first
def test_read_document_rejects_symlink_hardlink_and_hash_mismatch(tmp_path):
from tht.corpus.models import CanonicalDocument
content = "trusted"
digest = "sha256:" + __import__("hashlib").sha256(content.encode()).hexdigest()
document = CanonicalDocument(
document_id="doc:" + "a" * 64, source_id="fs:one", source_uri="file:///one",
source_fingerprint="sha256:" + "b" * 64, content_hash=digest, content=content,
pipeline_version="evidence-v1",
)
store = CorpusStore(tmp_path / "corpus")
generation = store.stage(CorpusManifest(documents=(document,)), {document.document_id: content})
path = store.resolve_document(document.document_id, generation)
assert store.read_document(document.document_id, generation) == content
path.unlink()
path.symlink_to(tmp_path / "outside")
(tmp_path / "outside").write_text(content)
with pytest.raises(UnsafeCorpusPath):
store.read_document(document.document_id, generation)
path.unlink()
os.link(tmp_path / "outside", path)
with pytest.raises(UnsafeCorpusPath):
store.read_document(document.document_id, generation)
path.unlink()
path.write_text("tampered")
with pytest.raises(UnsafeCorpusPath):
store.read_document(document.document_id, generation)
def test_generation_inventory_is_validated_and_sorted(tmp_path):
store = CorpusStore(tmp_path / "corpus")
first = store.stage(CorpusManifest(), {}, generation="gen:" + "1" * 32)
second = store.stage(CorpusManifest(), {}, generation="gen:" + "2" * 32)
(store.root / "unrelated").mkdir()
assert store.list_generations() == [first, second]