mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-11-02 19:22:22 +01:00
2623 lines
120 KiB
Python
2623 lines
120 KiB
Python
"""Deterministic tests for audit_c5_benchmark_evidence.py.
|
|
|
|
All tests use temporary synthetic fixtures - no network, no runtime, no model.
|
|
Tests are self-contained and safe to run repeatedly without changing source data.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
# Ensure project root is on path for import without conftest.
|
|
_ROOT = Path(__file__).resolve().parents[1]
|
|
if str(_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(_ROOT))
|
|
|
|
import scripts.audit_c5_benchmark_evidence as auditor
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _sha256(data: bytes) -> str:
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def _write(path: Path, content: bytes) -> str:
|
|
"""Write bytes to path and return its sha256."""
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(content)
|
|
return _sha256(content)
|
|
|
|
|
|
def _manifest(artifacts: list, campaign_id: str = "TEST-CAMPAIGN-01", expected_sets: dict | None = None) -> dict:
|
|
d = {
|
|
"schema_version": 1,
|
|
"campaign_id": campaign_id,
|
|
"description": "Synthetic test manifest",
|
|
"artifacts": artifacts,
|
|
}
|
|
if expected_sets:
|
|
d["expected_sets"] = expected_sets
|
|
return d
|
|
|
|
|
|
def _run_audit(
|
|
tmp: Path,
|
|
manifest_dict: dict,
|
|
deterministic: bool = False,
|
|
trusted_sources: list[str] | None = None,
|
|
trusted_expected_sets: dict | None = None,
|
|
) -> dict:
|
|
"""Write manifest to tmp, run audit, return parsed result."""
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(manifest_dict))
|
|
return auditor.audit(
|
|
str(mpath),
|
|
str(tmp),
|
|
deterministic=deterministic,
|
|
trusted_sources=trusted_sources,
|
|
trusted_expected_sets=trusted_expected_sets,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Baseline Integrity and Safety Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestValidMatchingHash(unittest.TestCase):
|
|
"""VERIFIED: artifact present, observed hash matches expected hash."""
|
|
|
|
def test_verified_when_observed_matches_expected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
content = b"hello world"
|
|
h = _write(tmp / "file.txt", content)
|
|
result = _run_audit(tmp, _manifest([{
|
|
"path": "file.txt",
|
|
"type": "result_record",
|
|
"expected_sha256": h,
|
|
}]))
|
|
art = result["artifacts"][0]
|
|
self.assertEqual(art["status"], "VERIFIED")
|
|
self.assertTrue(art["present"])
|
|
self.assertEqual(art["observed_sha256"], h)
|
|
self.assertEqual(art["expected_sha256"], h)
|
|
self.assertTrue(art["hash_match"])
|
|
self.assertEqual(result["totals"]["VERIFIED"], 1)
|
|
|
|
|
|
class TestMismatchedHash(unittest.TestCase):
|
|
"""MISMATCH: artifact present, but observed hash differs from expected."""
|
|
|
|
def test_mismatch_when_hash_differs(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
content = b"actual content"
|
|
_write(tmp / "data.json", content)
|
|
wrong_hash = _sha256(b"different content")
|
|
result = _run_audit(tmp, _manifest([{
|
|
"path": "data.json",
|
|
"type": "campaign_manifest",
|
|
"expected_sha256": wrong_hash,
|
|
}]))
|
|
art = result["artifacts"][0]
|
|
self.assertEqual(art["status"], "MISMATCH")
|
|
self.assertTrue(art["present"])
|
|
self.assertNotEqual(art["observed_sha256"], art["expected_sha256"])
|
|
self.assertFalse(art["hash_match"])
|
|
self.assertEqual(result["totals"]["MISMATCH"], 1)
|
|
|
|
|
|
class TestMissingArtifact(unittest.TestCase):
|
|
"""MISSING: expected artifact is absent from the filesystem."""
|
|
|
|
def test_missing_when_file_absent(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
result = _run_audit(tmp, _manifest([{
|
|
"path": "nonexistent/file.json",
|
|
"type": "trajectory",
|
|
"expected_sha256": "a" * 64,
|
|
}]))
|
|
art = result["artifacts"][0]
|
|
self.assertEqual(art["status"], "MISSING")
|
|
self.assertFalse(art["present"])
|
|
self.assertIsNone(art["observed_sha256"])
|
|
self.assertIsNone(art["size_bytes"])
|
|
self.assertEqual(result["totals"]["MISSING"], 1)
|
|
|
|
|
|
class TestObservedOnlyNoExpectedHash(unittest.TestCase):
|
|
"""OBSERVED_ONLY: artifact exists but manifest has no expected hash."""
|
|
|
|
def test_observed_only_when_no_expected_hash(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
_write(tmp / "report.md", b"# Report\n")
|
|
result = _run_audit(tmp, _manifest([{
|
|
"path": "report.md",
|
|
"type": "result_record",
|
|
}]))
|
|
art = result["artifacts"][0]
|
|
self.assertEqual(art["status"], "OBSERVED_ONLY")
|
|
self.assertTrue(art["present"])
|
|
self.assertIsNotNone(art["observed_sha256"])
|
|
self.assertIsNone(art["expected_sha256"])
|
|
self.assertFalse(art["hash_match"])
|
|
self.assertEqual(result["totals"]["OBSERVED_ONLY"], 1)
|
|
|
|
def test_observed_only_when_expected_hash_is_null(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
_write(tmp / "report.md", b"# Report\n")
|
|
result = _run_audit(tmp, _manifest([{
|
|
"path": "report.md",
|
|
"type": "result_record",
|
|
"expected_sha256": None,
|
|
}]))
|
|
art = result["artifacts"][0]
|
|
self.assertEqual(art["status"], "OBSERVED_ONLY")
|
|
|
|
|
|
class TestInvalidManifest(unittest.TestCase):
|
|
"""INVALID: manifest fails validation."""
|
|
|
|
def test_invalid_missing_schema_version(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps({"campaign_id": "X", "artifacts": []}))
|
|
result = auditor.audit(str(mpath), str(tmp))
|
|
self.assertFalse(result["manifest_valid"])
|
|
self.assertGreater(len(result["errors"]), 0)
|
|
|
|
def test_invalid_wrong_schema_version(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
result = _run_audit(tmp, {"schema_version": 99, "campaign_id": "X", "artifacts": []})
|
|
self.assertFalse(result["manifest_valid"])
|
|
|
|
def test_invalid_artifacts_not_a_list(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
result = _run_audit(tmp, {"schema_version": 1, "campaign_id": "X", "artifacts": "oops"})
|
|
self.assertFalse(result["manifest_valid"])
|
|
|
|
def test_invalid_malformed_json(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text("{not valid json}")
|
|
result = auditor.audit(str(mpath), str(tmp))
|
|
self.assertFalse(result["manifest_valid"])
|
|
self.assertGreater(len(result["errors"]), 0)
|
|
|
|
def test_invalid_missing_manifest_file(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "does_not_exist.json"
|
|
result = auditor.audit(str(mpath), str(tmp))
|
|
self.assertFalse(result["manifest_valid"])
|
|
self.assertGreater(len(result["errors"]), 0)
|
|
|
|
|
|
class TestPathTraversal(unittest.TestCase):
|
|
"""INVALID: path traversal attempts must be rejected."""
|
|
|
|
def test_reject_dotdot_path(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
result = _run_audit(tmp, _manifest([{
|
|
"path": "../escape.txt",
|
|
"type": "result_record",
|
|
}]))
|
|
art = result["artifacts"][0]
|
|
self.assertEqual(art["status"], "INVALID")
|
|
|
|
def test_reject_absolute_path(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
result = _run_audit(tmp, _manifest([{
|
|
"path": "/etc/passwd",
|
|
"type": "result_record",
|
|
}]))
|
|
art = result["artifacts"][0]
|
|
self.assertEqual(art["status"], "INVALID")
|
|
|
|
def test_reject_nested_dotdot(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
result = _run_audit(tmp, _manifest([{
|
|
"path": "subdir/../../etc/passwd",
|
|
"type": "result_record",
|
|
}]))
|
|
art = result["artifacts"][0]
|
|
self.assertEqual(art["status"], "INVALID")
|
|
|
|
|
|
class TestSymlinkEscape(unittest.TestCase):
|
|
"""UNVERIFIABLE: symlinks that escape the designated root."""
|
|
|
|
def test_symlink_escape_is_unverifiable(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
link = tmp / "escape_link.txt"
|
|
link.symlink_to("/etc/hostname")
|
|
result = _run_audit(tmp, _manifest([{
|
|
"path": "escape_link.txt",
|
|
"type": "result_record",
|
|
}]))
|
|
art = result["artifacts"][0]
|
|
self.assertIn(art["status"], ("UNVERIFIABLE", "INVALID"))
|
|
|
|
def test_internal_symlink_is_handled(self):
|
|
"""A symlink that stays within root is rejected as UNVERIFIABLE under O_NOFOLLOW boundary."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
target = tmp / "real.txt"
|
|
target.write_bytes(b"data")
|
|
link = tmp / "link.txt"
|
|
link.symlink_to(target)
|
|
result = _run_audit(tmp, _manifest([{
|
|
"path": "link.txt",
|
|
"type": "result_record",
|
|
}]))
|
|
art = result["artifacts"][0]
|
|
self.assertNotEqual(art["status"], "INVALID")
|
|
self.assertEqual(art["status"], "UNVERIFIABLE")
|
|
|
|
|
|
class TestStableRepeatedOutput(unittest.TestCase):
|
|
"""Stable: running the audit twice must produce identical output."""
|
|
|
|
def test_identical_output_on_repeated_runs(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
content = b"stable content"
|
|
h = _write(tmp / "a.json", content)
|
|
_write(tmp / "b.json", b"other")
|
|
manifest = _manifest([
|
|
{"path": "a.json", "type": "result_record", "expected_sha256": h},
|
|
{"path": "b.json", "type": "campaign_manifest"},
|
|
{"path": "missing.json", "type": "trajectory", "expected_sha256": "b" * 64},
|
|
])
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(manifest))
|
|
|
|
r1 = auditor.audit(str(mpath), str(tmp), deterministic=True)
|
|
r2 = auditor.audit(str(mpath), str(tmp), deterministic=True)
|
|
|
|
self.assertEqual(
|
|
json.dumps(r1, sort_keys=True),
|
|
json.dumps(r2, sort_keys=True),
|
|
"Repeated runs produced different output",
|
|
)
|
|
|
|
def test_artifacts_sorted_by_path(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
_write(tmp / "zzz.txt", b"z")
|
|
_write(tmp / "aaa.txt", b"a")
|
|
result = _run_audit(tmp, _manifest([
|
|
{"path": "zzz.txt", "type": "result_record"},
|
|
{"path": "aaa.txt", "type": "result_record"},
|
|
]))
|
|
paths = [a["path"] for a in result["artifacts"]]
|
|
self.assertEqual(paths, sorted(paths))
|
|
|
|
|
|
class TestEmptyManifest(unittest.TestCase):
|
|
"""Empty artifact list is valid but produces zero results."""
|
|
|
|
def test_empty_artifacts_list(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
result = _run_audit(tmp, _manifest([]))
|
|
self.assertTrue(result["manifest_valid"])
|
|
self.assertEqual(result["artifacts"], [])
|
|
self.assertEqual(result["totals"]["total"], 0)
|
|
|
|
|
|
class TestMixedOutcomes(unittest.TestCase):
|
|
"""Multiple artifacts with mixed verification outcomes."""
|
|
|
|
def test_mixed_verified_missing_mismatch_observed(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
good_content = b"good"
|
|
good_h = _write(tmp / "good.json", good_content)
|
|
bad_content = b"actual"
|
|
_write(tmp / "bad.json", bad_content)
|
|
_write(tmp / "noexpect.json", b"whatever")
|
|
wrong_h = _sha256(b"expected but wrong")
|
|
|
|
result = _run_audit(tmp, _manifest([
|
|
{"path": "good.json", "type": "result_record", "expected_sha256": good_h},
|
|
{"path": "bad.json", "type": "result_record", "expected_sha256": wrong_h},
|
|
{"path": "missing.json", "type": "trajectory", "expected_sha256": "c" * 64},
|
|
{"path": "noexpect.json", "type": "campaign_manifest"},
|
|
]))
|
|
|
|
statuses = {a["path"]: a["status"] for a in result["artifacts"]}
|
|
self.assertEqual(statuses["good.json"], "VERIFIED")
|
|
self.assertEqual(statuses["bad.json"], "MISMATCH")
|
|
self.assertEqual(statuses["missing.json"], "MISSING")
|
|
self.assertEqual(statuses["noexpect.json"], "OBSERVED_ONLY")
|
|
|
|
t = result["totals"]
|
|
self.assertEqual(t["VERIFIED"], 1)
|
|
self.assertEqual(t["MISMATCH"], 1)
|
|
self.assertEqual(t["MISSING"], 1)
|
|
self.assertEqual(t["OBSERVED_ONLY"], 1)
|
|
self.assertEqual(t["total"], 4)
|
|
|
|
|
|
class TestNeverVerifiedFromOwnHash(unittest.TestCase):
|
|
"""The utility must not auto-verify by computing and comparing its own hash."""
|
|
|
|
def test_no_expected_hash_means_not_verified(self):
|
|
"""An artifact with no expected_sha256 must never be VERIFIED."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
_write(tmp / "auto.txt", b"some data")
|
|
result = _run_audit(tmp, _manifest([{
|
|
"path": "auto.txt",
|
|
"type": "model_artifact",
|
|
}]))
|
|
art = result["artifacts"][0]
|
|
self.assertNotEqual(art["status"], "VERIFIED",
|
|
"Artifact must not be VERIFIED without an independently provided expected hash")
|
|
self.assertEqual(art["status"], "OBSERVED_ONLY")
|
|
|
|
|
|
class TestReproducibilityPrerequisites(unittest.TestCase):
|
|
"""Reproducibility prerequisites section is present and populated."""
|
|
|
|
def test_reproducibility_keys_present(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
result = _run_audit(tmp, _manifest([]))
|
|
prereqs = result.get("reproducibility_prerequisites", {})
|
|
expected_keys = {
|
|
"model_artifact_hash",
|
|
"source_checkpoint_revision",
|
|
"quantization_info",
|
|
"backend_executable_identity",
|
|
"chat_template_identity",
|
|
"context_and_decoding_config",
|
|
"task_fixture_identities",
|
|
"expected_outcomes_and_oracles",
|
|
"raw_trajectories",
|
|
"result_and_latency_records",
|
|
}
|
|
self.assertEqual(set(prereqs.keys()), expected_keys)
|
|
|
|
def test_model_artifact_hash_unknown_without_artifact(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
result = _run_audit(tmp, _manifest([]))
|
|
self.assertEqual(
|
|
result["reproducibility_prerequisites"]["model_artifact_hash"],
|
|
"UNKNOWN",
|
|
)
|
|
|
|
def test_model_artifact_hash_observed_when_type_present(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
_write(tmp / "model.gguf.txt", b"placeholder")
|
|
result = _run_audit(tmp, _manifest([{
|
|
"path": "model.gguf.txt",
|
|
"type": "model_artifact",
|
|
}]))
|
|
val = result["reproducibility_prerequisites"]["model_artifact_hash"]
|
|
self.assertNotEqual(val, "UNKNOWN",
|
|
"model_artifact_hash should not be UNKNOWN when a model_artifact is present")
|
|
|
|
|
|
class TestOutputStructure(unittest.TestCase):
|
|
"""Output JSON has required top-level fields."""
|
|
|
|
def test_required_top_level_fields(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
result = _run_audit(tmp, _manifest([]))
|
|
for key in ("schema_version", "audited_at", "campaign_id", "root",
|
|
"manifest_path", "totals", "artifacts",
|
|
"reproducibility_prerequisites", "reproducibility_details",
|
|
"errors", "manifest_valid"):
|
|
self.assertIn(key, result, f"Missing top-level key: {key}")
|
|
|
|
def test_schema_version_is_one(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
result = _run_audit(tmp, _manifest([]))
|
|
self.assertEqual(result["schema_version"], 1)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Finding 1: Output Destination Overwrite & TOCTOU Protection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestFinding1OutputProtection(unittest.TestCase):
|
|
"""Regression tests for Finding 1 (P1): Output publication remains vulnerable to TOCTOU."""
|
|
|
|
def test_output_targeting_manifest_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
content = json.dumps(_manifest([])).encode()
|
|
mpath.write_bytes(content)
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--output", str(mpath)])
|
|
self.assertEqual(code, 2)
|
|
self.assertEqual(mpath.read_bytes(), content, "Manifest was modified by output destination")
|
|
|
|
def test_summary_targeting_audited_artifact_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
art_file = tmp / "evidence.dat"
|
|
original_bytes = b"precious_historical_evidence"
|
|
art_file.write_bytes(original_bytes)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([{"path": "evidence.dat", "type": "result_record"}])))
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--summary", str(art_file)])
|
|
self.assertEqual(code, 2)
|
|
self.assertEqual(art_file.read_bytes(), original_bytes, "Audited artifact was overwritten")
|
|
|
|
def test_output_and_summary_same_destination_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
dest = tmp / "collision.txt"
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--output", str(dest), "--summary", str(dest)])
|
|
self.assertEqual(code, 2)
|
|
self.assertFalse(dest.exists(), "File should not be created on destination collision")
|
|
|
|
def test_existing_symlink_alias_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
original = json.dumps(_manifest([])).encode()
|
|
mpath.write_bytes(original)
|
|
link_out = tmp / "link_manifest.json"
|
|
link_out.symlink_to(mpath)
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--output", str(link_out)])
|
|
self.assertEqual(code, 2)
|
|
self.assertEqual(mpath.read_bytes(), original)
|
|
|
|
def test_existing_hardlink_alias_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
art_file = tmp / "original.txt"
|
|
original_bytes = b"immutable_artifact"
|
|
art_file.write_bytes(original_bytes)
|
|
hardlink_file = tmp / "hardlink.txt"
|
|
os.link(art_file, hardlink_file)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([{"path": "original.txt", "type": "result_record"}])))
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--summary", str(hardlink_file)])
|
|
self.assertEqual(code, 2)
|
|
self.assertEqual(art_file.read_bytes(), original_bytes)
|
|
|
|
def test_injected_destination_replacement_after_preflight_blocks_overwrite(self):
|
|
"""Adversarial TOCTOU test: destination is substituted with symlink after preflight."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
art_file = tmp / "protected.dat"
|
|
original_bytes = b"IMMUTABLE_HISTORICAL_DATA"
|
|
art_file.write_bytes(original_bytes)
|
|
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([{"path": "protected.dat", "type": "result_record"}])))
|
|
|
|
target_out = tmp / "safe_destination.json"
|
|
|
|
# Inject symlink substitution after preflight succeeds
|
|
orig_validate = auditor._validate_output_destinations
|
|
|
|
def injected_validation(*args, **kwargs):
|
|
errs = orig_validate(*args, **kwargs)
|
|
# Inject malicious symlink pointing to protected artifact after preflight check
|
|
if target_out.exists():
|
|
target_out.unlink()
|
|
target_out.symlink_to(art_file)
|
|
return errs
|
|
|
|
auditor._validate_output_destinations = injected_validation
|
|
try:
|
|
code = auditor.main([str(mpath), str(tmp), "--output", str(target_out)])
|
|
self.assertEqual(code, 2, "CLI should reject publication when destination points to protected artifact")
|
|
self.assertEqual(art_file.read_bytes(), original_bytes, "Protected artifact was overwritten!")
|
|
finally:
|
|
auditor._validate_output_destinations = orig_validate
|
|
|
|
def test_nested_output_directory_symlink_replacement_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
|
|
outside = tmp / "outside"
|
|
outside.mkdir()
|
|
sub = tmp / "nested_dir"
|
|
sub.symlink_to(outside)
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--output", str(sub / "report.json")])
|
|
self.assertEqual(code, 2)
|
|
self.assertFalse((outside / "report.json").exists())
|
|
|
|
def test_failed_publication_preserves_existing_protected_input_bytes(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
art = tmp / "artifact.dat"
|
|
sentinel = b"SENTINEL_BYTES_PRESERVED"
|
|
art.write_bytes(sentinel)
|
|
mpath = tmp / "manifest.json"
|
|
m_bytes = json.dumps(_manifest([{"path": "artifact.dat", "type": "result_record"}])).encode()
|
|
mpath.write_bytes(m_bytes)
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--output", str(art)])
|
|
self.assertEqual(code, 2)
|
|
self.assertEqual(art.read_bytes(), sentinel)
|
|
self.assertEqual(mpath.read_bytes(), m_bytes)
|
|
|
|
def test_temporary_files_are_cleaned_up_after_failure(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
art = tmp / "artifact.dat"
|
|
art.write_bytes(b"data")
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([{"path": "artifact.dat", "type": "result_record"}])))
|
|
|
|
out_dir = tmp / "out_dir"
|
|
out_dir.mkdir()
|
|
out_file = out_dir / "target.json"
|
|
out_file.symlink_to(art)
|
|
|
|
auditor.main([str(mpath), str(tmp), "--output", str(out_file)])
|
|
# Verify no .tmp_* files remain in out_dir
|
|
leftover_tmps = [p for p in out_dir.iterdir() if p.name.startswith(".tmp_")]
|
|
self.assertEqual(leftover_tmps, [], "Temporary files leaked in output directory")
|
|
|
|
def test_valid_distinct_json_output_destination_works(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
out = tmp / "out.json"
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--output", str(out)])
|
|
self.assertEqual(code, 0)
|
|
self.assertTrue(out.is_file())
|
|
parsed = json.loads(out.read_text())
|
|
self.assertTrue(parsed["manifest_valid"])
|
|
|
|
def test_valid_distinct_markdown_summary_destination_works(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
summ = tmp / "summary.md"
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--summary", str(summ)])
|
|
self.assertEqual(code, 0)
|
|
self.assertTrue(summ.is_file())
|
|
self.assertIn("Benchmark Evidence Audit Summary", summ.read_text())
|
|
|
|
def test_repeated_successful_publication_produces_complete_valid_files(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
out = tmp / "out.json"
|
|
|
|
code1 = auditor.main([str(mpath), str(tmp), "--output", str(out), "--deterministic"])
|
|
bytes1 = out.read_bytes()
|
|
code2 = auditor.main([str(mpath), str(tmp), "--output", str(out), "--deterministic"])
|
|
bytes2 = out.read_bytes()
|
|
|
|
self.assertEqual(code1, 0)
|
|
self.assertEqual(code2, 0)
|
|
self.assertEqual(bytes1, bytes2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Finding 2: Conservative Prerequisite Completeness Model
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestFinding2PrerequisiteCompleteness(unittest.TestCase):
|
|
"""Mandatory regressions for Finding 2 (P2): Campaign prerequisite completeness."""
|
|
|
|
def test_1_one_verified_trajectory_without_completeness_evidence(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
h = _write(tmp / "t1.json", b"single_trajectory")
|
|
res = _run_audit(tmp, _manifest([{"path": "t1.json", "type": "trajectory", "expected_sha256": h}]))
|
|
|
|
# Prerequisite completeness is NOT verified when expected-set completeness is unknown
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], "PARTIAL")
|
|
det = res["reproducibility_details"]["raw_trajectories"]
|
|
self.assertEqual(det["artifact_integrity"], "VERIFIED")
|
|
self.assertEqual(det["expected_set_status"], "UNKNOWN")
|
|
self.assertFalse(det["prerequisite_complete"])
|
|
|
|
def test_2_two_verified_trajectories_with_unknown_expected_set_completeness(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
h1 = _write(tmp / "t1.json", b"traj1")
|
|
h2 = _write(tmp / "t2.json", b"traj2")
|
|
res = _run_audit(tmp, _manifest([
|
|
{"path": "t1.json", "type": "trajectory", "expected_sha256": h1},
|
|
{"path": "t2.json", "type": "trajectory", "expected_sha256": h2},
|
|
]))
|
|
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], "PARTIAL")
|
|
det = res["reproducibility_details"]["raw_trajectories"]
|
|
self.assertEqual(det["artifact_integrity"], "VERIFIED")
|
|
self.assertEqual(det["expected_set_status"], "UNKNOWN")
|
|
self.assertFalse(det["prerequisite_complete"])
|
|
|
|
def test_3_complete_independently_established_expected_set_all_verify(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
h1 = _write(tmp / "t1.json", b"traj1")
|
|
h2 = _write(tmp / "t2.json", b"traj2")
|
|
trusted_src = "https://specs.example.org/qwen-e00-trajectories.json"
|
|
expected_sets = {
|
|
"raw_trajectories": {
|
|
"expected_paths": ["t1.json", "t2.json"],
|
|
"provenance": trusted_src,
|
|
"independent_oracle": True,
|
|
}
|
|
}
|
|
res = _run_audit(
|
|
tmp,
|
|
_manifest([
|
|
{"path": "t1.json", "type": "trajectory", "expected_sha256": h1},
|
|
{"path": "t2.json", "type": "trajectory", "expected_sha256": h2},
|
|
], expected_sets=expected_sets),
|
|
trusted_sources=[trusted_src],
|
|
)
|
|
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], "VERIFIED")
|
|
det = res["reproducibility_details"]["raw_trajectories"]
|
|
self.assertEqual(det["artifact_integrity"], "VERIFIED")
|
|
self.assertEqual(det["expected_set_status"], "COMPLETE")
|
|
self.assertTrue(det["prerequisite_complete"])
|
|
self.assertEqual(det["missing_or_unverified"], [])
|
|
|
|
def test_4_established_expected_set_with_one_missing_artifact(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
h1 = _write(tmp / "t1.json", b"traj1")
|
|
trusted_src = "https://specs.example.org/qwen-e00-trajectories.json"
|
|
expected_sets = {
|
|
"raw_trajectories": {
|
|
"expected_paths": ["t1.json", "t2_missing.json"],
|
|
"provenance": trusted_src,
|
|
"independent_oracle": True,
|
|
}
|
|
}
|
|
res = _run_audit(
|
|
tmp,
|
|
_manifest([
|
|
{"path": "t1.json", "type": "trajectory", "expected_sha256": h1},
|
|
{"path": "t2_missing.json", "type": "trajectory", "expected_sha256": "0" * 64},
|
|
], expected_sets=expected_sets),
|
|
trusted_sources=[trusted_src],
|
|
)
|
|
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], "INCOMPLETE")
|
|
det = res["reproducibility_details"]["raw_trajectories"]
|
|
self.assertFalse(det["prerequisite_complete"])
|
|
self.assertIn("t2_missing.json", det["missing_or_unverified"])
|
|
|
|
def test_5_established_expected_set_with_one_mismatch_artifact(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
h1 = _write(tmp / "t1.json", b"traj1")
|
|
_write(tmp / "t2.json", b"actual_content")
|
|
trusted_src = "https://specs.example.org/qwen-e00-trajectories.json"
|
|
expected_sets = {
|
|
"raw_trajectories": {
|
|
"expected_paths": ["t1.json", "t2.json"],
|
|
"provenance": trusted_src,
|
|
"independent_oracle": True,
|
|
}
|
|
}
|
|
res = _run_audit(
|
|
tmp,
|
|
_manifest([
|
|
{"path": "t1.json", "type": "trajectory", "expected_sha256": h1},
|
|
{"path": "t2.json", "type": "trajectory", "expected_sha256": "f" * 64},
|
|
], expected_sets=expected_sets),
|
|
trusted_sources=[trusted_src],
|
|
)
|
|
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], "MISMATCH")
|
|
det = res["reproducibility_details"]["raw_trajectories"]
|
|
self.assertFalse(det["prerequisite_complete"])
|
|
self.assertIn("t2.json", det["missing_or_unverified"])
|
|
|
|
def test_6_verified_plus_observed_only(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
h1 = _write(tmp / "t1.json", b"traj1")
|
|
_write(tmp / "t2.json", b"traj2")
|
|
res = _run_audit(tmp, _manifest([
|
|
{"path": "t1.json", "type": "trajectory", "expected_sha256": h1},
|
|
{"path": "t2.json", "type": "trajectory"},
|
|
]))
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], "PARTIAL")
|
|
|
|
def test_7_optional_verified_artifacts_with_missing_mandatory_artifact(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
h_opt = _write(tmp / "optional.json", b"opt")
|
|
res = _run_audit(tmp, _manifest([
|
|
{"path": "mandatory.json", "type": "trajectory", "expected_sha256": "1" * 64, "mandatory": True},
|
|
{"path": "optional.json", "type": "trajectory", "expected_sha256": h_opt, "mandatory": False},
|
|
]))
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], "INCOMPLETE")
|
|
|
|
def test_8_empty_declared_artifact_set(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
res = _run_audit(tmp, _manifest([]))
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], "UNKNOWN")
|
|
|
|
def test_9_incomplete_archive_with_self_declared_completeness_flag(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
h1 = _write(tmp / "t1.json", b"traj1")
|
|
expected_sets = {
|
|
"raw_trajectories": {
|
|
"expected_paths": ["t1.json"],
|
|
"complete": True,
|
|
"provenance": "internal-collector-self-claim",
|
|
"independent_oracle": False, # Not independently verified!
|
|
}
|
|
}
|
|
res = _run_audit(tmp, _manifest([
|
|
{"path": "t1.json", "type": "trajectory", "expected_sha256": h1},
|
|
], expected_sets=expected_sets))
|
|
|
|
# Unverified self-claim must not be accepted as verified completeness
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], "PARTIAL")
|
|
det = res["reproducibility_details"]["raw_trajectories"]
|
|
self.assertEqual(det["expected_set_status"], "UNVERIFIED_CLAIM")
|
|
self.assertFalse(det["prerequisite_complete"])
|
|
|
|
def test_10_checksum_matched_summary_document_does_not_prove_task_level_oracles(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
content = b'{"task_specifications": "summary_only"}'
|
|
h = _write(tmp / "freeze-manifest.json", content)
|
|
res = _run_audit(tmp, _manifest([
|
|
{"path": "freeze-manifest.json", "type": "oracle_definition", "expected_sha256": h, "is_summary": True},
|
|
]))
|
|
|
|
art = res["artifacts"][0]
|
|
self.assertEqual(art["status"], "VERIFIED", "File checksum integrity verified")
|
|
|
|
# Prerequisite completeness is NOT verified: summary does not prove task-level oracles
|
|
self.assertEqual(res["reproducibility_prerequisites"]["expected_outcomes_and_oracles"], "OBSERVED_ONLY")
|
|
det = res["reproducibility_details"]["expected_outcomes_and_oracles"]
|
|
self.assertEqual(det["expected_set_status"], "SUMMARY_ONLY")
|
|
self.assertFalse(det["prerequisite_complete"])
|
|
self.assertIn("summary document does not prove", det["notes"])
|
|
|
|
def test_11_checksum_matched_aggregate_report_does_not_prove_complete_latency_records(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
content = b"# Aggregate Benchmark Report\nOverall latency: 120ms\n"
|
|
h = _write(tmp / "campaign-report.md", content)
|
|
res = _run_audit(tmp, _manifest([
|
|
{"path": "campaign-report.md", "type": "result_record", "expected_sha256": h, "is_summary": True},
|
|
]))
|
|
|
|
art = res["artifacts"][0]
|
|
self.assertEqual(art["status"], "VERIFIED", "Report file checksum integrity verified")
|
|
|
|
# Prerequisite completeness is NOT verified: aggregate statistics do not prove per-turn latency
|
|
self.assertEqual(res["reproducibility_prerequisites"]["result_and_latency_records"], "OBSERVED_ONLY")
|
|
det = res["reproducibility_details"]["result_and_latency_records"]
|
|
self.assertEqual(det["expected_set_status"], "AGGREGATE_ONLY")
|
|
self.assertFalse(det["prerequisite_complete"])
|
|
self.assertIn("aggregate statistics do not prove", det["notes"])
|
|
|
|
def test_12_stable_output_regardless_of_manifest_ordering(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
h1 = _write(tmp / "t1.json", b"traj1")
|
|
_write(tmp / "t2.json", b"traj2")
|
|
e1 = {"path": "t1.json", "type": "trajectory", "expected_sha256": h1}
|
|
e2 = {"path": "t2.json", "type": "trajectory", "expected_sha256": "0" * 64}
|
|
|
|
res1 = _run_audit(tmp, _manifest([e1, e2], campaign_id="STABLE"))
|
|
res2 = _run_audit(tmp, _manifest([e2, e1], campaign_id="STABLE"))
|
|
|
|
self.assertEqual(
|
|
res1["reproducibility_prerequisites"],
|
|
res2["reproducibility_prerequisites"],
|
|
)
|
|
self.assertEqual(
|
|
res1["reproducibility_details"],
|
|
res2["reproducibility_details"],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Finding 3: Symlink TOCTOU and O_NOFOLLOW Enforcement
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestFinding3SymlinkTOCTOU(unittest.TestCase):
|
|
"""Regression tests for Finding 3 (P2): Symlink time-of-check/time-of-use vulnerability."""
|
|
|
|
def test_ordinary_regular_file_inside_root(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
content = b"regular_inside_root"
|
|
h = _write(tmp / "subdir" / "data.txt", content)
|
|
res = _run_audit(tmp, _manifest([{
|
|
"path": "subdir/data.txt",
|
|
"type": "result_record",
|
|
"expected_sha256": h,
|
|
}]))
|
|
art = res["artifacts"][0]
|
|
self.assertEqual(art["status"], "VERIFIED")
|
|
self.assertEqual(art["observed_sha256"], h)
|
|
|
|
def test_direct_symlink_escaping_root(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
link = tmp / "escape_passwd"
|
|
link.symlink_to("/etc/passwd")
|
|
res = _run_audit(tmp, _manifest([{"path": "escape_passwd", "type": "result_record"}]))
|
|
art = res["artifacts"][0]
|
|
self.assertEqual(art["status"], "UNVERIFIABLE")
|
|
self.assertIsNone(art["observed_sha256"])
|
|
|
|
def test_nested_symlink_escaping_root(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
sub = tmp / "a" / "b"
|
|
sub.mkdir(parents=True)
|
|
link = sub / "nested_escape"
|
|
link.symlink_to("/etc/passwd")
|
|
res = _run_audit(tmp, _manifest([{"path": "a/b/nested_escape", "type": "result_record"}]))
|
|
art = res["artifacts"][0]
|
|
self.assertEqual(art["status"], "UNVERIFIABLE")
|
|
self.assertIsNone(art["observed_sha256"])
|
|
|
|
def test_symlink_directory_component_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
outside = Path(tempfile.mkdtemp())
|
|
try:
|
|
(outside / "secret.txt").write_bytes(b"secret")
|
|
(tmp / "linkdir").symlink_to(outside)
|
|
res = _run_audit(tmp, _manifest([{"path": "linkdir/secret.txt", "type": "result_record"}]))
|
|
art = res["artifacts"][0]
|
|
self.assertEqual(art["status"], "UNVERIFIABLE")
|
|
self.assertIsNone(art["observed_sha256"])
|
|
finally:
|
|
import shutil
|
|
shutil.rmtree(outside)
|
|
|
|
def test_safe_handling_of_missing_or_replaced_files(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
res = _run_audit(tmp, _manifest([{"path": "missing/dir/file.txt", "type": "result_record"}]))
|
|
art = res["artifacts"][0]
|
|
self.assertEqual(art["status"], "MISSING")
|
|
|
|
def test_no_outside_root_file_access(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
secret_outside = Path(tempfile.mkdtemp())
|
|
try:
|
|
secret_file = secret_outside / "secret.dat"
|
|
secret_file.write_bytes(b"TOP_SECRET_DATA")
|
|
secret_hash = _sha256(b"TOP_SECRET_DATA")
|
|
|
|
link = tmp / "secret_link"
|
|
link.symlink_to(secret_file)
|
|
|
|
res = _run_audit(tmp, _manifest([{
|
|
"path": "secret_link",
|
|
"type": "result_record",
|
|
"expected_sha256": secret_hash,
|
|
}]))
|
|
art = res["artifacts"][0]
|
|
self.assertEqual(art["status"], "UNVERIFIABLE")
|
|
self.assertIsNone(art["observed_sha256"], "Outside-root content was read or hashed")
|
|
finally:
|
|
import shutil
|
|
shutil.rmtree(secret_outside)
|
|
|
|
def test_no_descriptor_leaks_after_failure(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
(tmp / "bad_link").symlink_to("/etc/passwd")
|
|
|
|
fd_dir = Path(f"/proc/{os.getpid()}/fd")
|
|
if not fd_dir.exists():
|
|
self.skipTest("/proc/self/fd not accessible on this platform")
|
|
|
|
# Warm-up run
|
|
_run_audit(tmp, _manifest([{"path": "bad_link", "type": "result_record"}]))
|
|
|
|
fds_before = len(list(fd_dir.iterdir()))
|
|
for _ in range(10):
|
|
_run_audit(tmp, _manifest([
|
|
{"path": "bad_link", "type": "result_record"},
|
|
{"path": "missing.txt", "type": "result_record"},
|
|
{"path": "../outside.txt", "type": "result_record"},
|
|
]))
|
|
fds_after = len(list(fd_dir.iterdir()))
|
|
self.assertEqual(fds_before, fds_after, "File descriptor leak detected")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Finding 4: Malformed Manifest Handling
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestFinding4MalformedManifest(unittest.TestCase):
|
|
"""Regression tests for Finding 4 (P2): Malformed manifest handling."""
|
|
|
|
def test_invalid_list_path(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
res = _run_audit(tmp, _manifest([{"path": ["invalid", "list"], "type": "result_record"}]))
|
|
self.assertEqual(res["totals"]["INVALID"], 1)
|
|
self.assertEqual(res["artifacts"][0]["status"], "INVALID")
|
|
|
|
def test_invalid_integer_path(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
res = _run_audit(tmp, _manifest([{"path": 12345, "type": "result_record"}]))
|
|
self.assertEqual(res["totals"]["INVALID"], 1)
|
|
self.assertEqual(res["artifacts"][0]["status"], "INVALID")
|
|
|
|
def test_invalid_null_path(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
res = _run_audit(tmp, _manifest([{"path": None, "type": "result_record"}]))
|
|
self.assertEqual(res["totals"]["INVALID"], 1)
|
|
self.assertEqual(res["artifacts"][0]["status"], "INVALID")
|
|
|
|
def test_invalid_boolean_expected_hash(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
_write(tmp / "file.txt", b"data")
|
|
res_false = _run_audit(tmp, _manifest([{"path": "file.txt", "type": "result_record", "expected_sha256": False}]))
|
|
self.assertEqual(res_false["totals"]["INVALID"], 1)
|
|
self.assertEqual(res_false["artifacts"][0]["status"], "INVALID")
|
|
|
|
res_true = _run_audit(tmp, _manifest([{"path": "file.txt", "type": "result_record", "expected_sha256": True}]))
|
|
self.assertEqual(res_true["totals"]["INVALID"], 1)
|
|
self.assertEqual(res_true["artifacts"][0]["status"], "INVALID")
|
|
|
|
def test_invalid_numeric_expected_hash(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
_write(tmp / "file.txt", b"data")
|
|
res = _run_audit(tmp, _manifest([{"path": "file.txt", "type": "result_record", "expected_sha256": 1234567890}]))
|
|
self.assertEqual(res["totals"]["INVALID"], 1)
|
|
self.assertEqual(res["artifacts"][0]["status"], "INVALID")
|
|
|
|
def test_invalid_malformed_hexadecimal_hash(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
_write(tmp / "file.txt", b"data")
|
|
res = _run_audit(tmp, _manifest([{"path": "file.txt", "type": "result_record", "expected_sha256": "g" * 64}]))
|
|
self.assertEqual(res["totals"]["INVALID"], 1)
|
|
self.assertEqual(res["artifacts"][0]["status"], "INVALID")
|
|
|
|
def test_invalid_wrong_length_hash(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
_write(tmp / "file.txt", b"data")
|
|
res = _run_audit(tmp, _manifest([{"path": "file.txt", "type": "result_record", "expected_sha256": "a" * 32}]))
|
|
self.assertEqual(res["totals"]["INVALID"], 1)
|
|
self.assertEqual(res["artifacts"][0]["status"], "INVALID")
|
|
|
|
def test_invalid_artifact_record_structure(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
res = _run_audit(tmp, _manifest(["not_a_dict", 42, None]))
|
|
self.assertEqual(res["totals"]["INVALID"], 3)
|
|
for art in res["artifacts"]:
|
|
self.assertEqual(art["status"], "INVALID")
|
|
|
|
def test_missing_required_path_field(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
res = _run_audit(tmp, _manifest([{"type": "result_record"}]))
|
|
self.assertEqual(res["totals"]["INVALID"], 1)
|
|
self.assertEqual(res["artifacts"][0]["status"], "INVALID")
|
|
|
|
def test_mixed_valid_and_invalid_entries(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
h = _write(tmp / "good.txt", b"content")
|
|
res = _run_audit(tmp, _manifest([
|
|
{"path": "good.txt", "type": "result_record", "expected_sha256": h},
|
|
{"path": ["bad", "list"], "type": "result_record"},
|
|
{"path": "bad_hash.txt", "type": "result_record", "expected_sha256": False},
|
|
]))
|
|
self.assertEqual(res["totals"]["VERIFIED"], 1)
|
|
self.assertEqual(res["totals"]["INVALID"], 2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Deterministic Output Mode
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDeterministicOutputMode(unittest.TestCase):
|
|
"""Test deterministic CLI mode generates identical output bytes."""
|
|
|
|
def test_deterministic_cli_flag_produces_identical_bytes(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
out1 = tmp / "out1.json"
|
|
out2 = tmp / "out2.json"
|
|
|
|
code1 = auditor.main([str(mpath), str(tmp), "--output", str(out1), "--deterministic"])
|
|
code2 = auditor.main([str(mpath), str(tmp), "--output", str(out2), "--deterministic"])
|
|
|
|
self.assertEqual(code1, 0)
|
|
self.assertEqual(code2, 0)
|
|
self.assertEqual(out1.read_bytes(), out2.read_bytes(), "Deterministic runs differed in output bytes")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Additional Regressions for Final Security & Evidence Findings
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestFinding1ParentDirectorySubstitution(unittest.TestCase):
|
|
"""P1: Output parent directory substitution protection."""
|
|
|
|
def test_intermediate_directory_symlink_substitution_blocked_and_target_preserved(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
# Create a protected target directory containing sensitive files
|
|
protected_dir = tmp / "protected"
|
|
protected_dir.mkdir()
|
|
protected_file = protected_dir / "secret_evidence.dat"
|
|
original_bytes = b"ORIGINAL PROTECTED EVIDENCE - DO NOT OVERWRITE"
|
|
protected_file.write_bytes(original_bytes)
|
|
|
|
# Create an intermediate directory structure
|
|
base_out = tmp / "output_root"
|
|
base_out.mkdir()
|
|
# Replace intermediate component with a symlink pointing to the protected directory
|
|
inter_symlink = base_out / "intermediate"
|
|
os.symlink(str(protected_dir), str(inter_symlink))
|
|
|
|
# Attempt publication targeting a path through the intermediate symlink
|
|
dest = inter_symlink / "secret_evidence.dat"
|
|
protected_paths, protected_inodes = auditor._collect_protected_targets(
|
|
str(tmp / "manifest.json"),
|
|
str(tmp),
|
|
[{"path": "secret_evidence.dat", "type": "result_record"}],
|
|
)
|
|
# Explicitly register protected file
|
|
protected_paths[protected_file.resolve()] = "protected file"
|
|
st = protected_file.stat()
|
|
protected_inodes[(st.st_dev, st.st_ino)] = "protected file"
|
|
|
|
with self.assertRaises(OSError) as ctx:
|
|
auditor._publish_file_securely(
|
|
str(dest),
|
|
"MALICIOUS OVERWRITE CONTENT",
|
|
protected_paths,
|
|
protected_inodes,
|
|
)
|
|
|
|
self.assertIn("symlink", str(ctx.exception).lower())
|
|
# Verify the protected file bytes are completely unchanged
|
|
self.assertEqual(protected_file.read_bytes(), original_bytes)
|
|
|
|
def test_cli_intermediate_directory_symlink_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
|
|
outside = tmp / "outside"
|
|
outside.mkdir()
|
|
canary = outside / "canary.txt"
|
|
canary.write_bytes(b"CANARY UNTOUCHED")
|
|
|
|
out_tree = tmp / "out_tree"
|
|
out_tree.mkdir()
|
|
sym_dir = out_tree / "sym_sub"
|
|
os.symlink(str(outside), str(sym_dir))
|
|
|
|
target_out = sym_dir / "out.json"
|
|
code = auditor.main([str(mpath), str(tmp), "--output", str(target_out)])
|
|
self.assertEqual(code, 2)
|
|
self.assertEqual(canary.read_bytes(), b"CANARY UNTOUCHED")
|
|
|
|
|
|
class TestFinding2ArtifactRootSubstitution(unittest.TestCase):
|
|
"""P2: Artifact root substitution and outside-root read protection."""
|
|
|
|
def test_root_directory_symlink_rejected_and_outside_file_not_verified(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
# Create outside secret directory
|
|
outside_dir = tmp / "outside_secret"
|
|
outside_dir.mkdir()
|
|
secret_file = outside_dir / "confidential.dat"
|
|
secret_bytes = b"CONFIDENTIAL SYSTEM DATA"
|
|
secret_file.write_bytes(secret_bytes)
|
|
expected_hash = _sha256(secret_bytes)
|
|
|
|
# Create symlink root pointing to outside directory
|
|
symlink_root = tmp / "symlink_root"
|
|
os.symlink(str(outside_dir), str(symlink_root))
|
|
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([{
|
|
"path": "confidential.dat",
|
|
"type": "result_record",
|
|
"expected_sha256": expected_hash,
|
|
}])))
|
|
|
|
# Audit against symlink root
|
|
res = auditor.audit(str(mpath), str(symlink_root))
|
|
|
|
art = res["artifacts"][0]
|
|
self.assertEqual(art["status"], "UNVERIFIABLE")
|
|
self.assertFalse(art["present"])
|
|
self.assertIsNone(art["observed_sha256"])
|
|
self.assertFalse(art["hash_match"])
|
|
self.assertIn("root directory is invalid or a symlink", art["notes"])
|
|
self.assertEqual(res["totals"]["VERIFIED"], 0)
|
|
self.assertEqual(res["totals"]["UNVERIFIABLE"], 1)
|
|
|
|
def test_root_replaced_with_symlink_during_audit(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
real_root = tmp / "root_dir"
|
|
real_root.mkdir()
|
|
outside = tmp / "outside"
|
|
outside.mkdir()
|
|
h = _write(outside / "data.txt", b"outside data")
|
|
|
|
# Replace real root with symlink
|
|
real_root.rmdir()
|
|
os.symlink(str(outside), str(real_root))
|
|
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([{
|
|
"path": "data.txt",
|
|
"type": "result_record",
|
|
"expected_sha256": h,
|
|
}])))
|
|
|
|
res = auditor.audit(str(mpath), str(real_root))
|
|
self.assertEqual(res["artifacts"][0]["status"], "UNVERIFIABLE")
|
|
self.assertEqual(res["totals"]["VERIFIED"], 0)
|
|
|
|
def test_no_descriptor_leaks_after_root_symlink_rejections(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
sym_root = tmp / "sym_root"
|
|
os.symlink(str(tmp), str(sym_root))
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([{"path": "file.txt", "type": "result_record"}])))
|
|
|
|
fd_dir = Path(f"/proc/{os.getpid()}/fd")
|
|
if not fd_dir.exists():
|
|
self.skipTest("/proc/self/fd not accessible on this platform")
|
|
|
|
# Warm-up run
|
|
auditor.audit(str(mpath), str(sym_root))
|
|
fds_before = len(list(fd_dir.iterdir()))
|
|
|
|
for _ in range(10):
|
|
auditor.audit(str(mpath), str(sym_root))
|
|
|
|
fds_after = len(list(fd_dir.iterdir()))
|
|
self.assertEqual(fds_before, fds_after, "Descriptor leak detected after root symlink rejections")
|
|
|
|
|
|
class TestFinding3ProvenanceTrust(unittest.TestCase):
|
|
"""P2: Self-attested independent provenance and trusted verification source."""
|
|
|
|
def test_fabricated_provenance_url_cannot_produce_verified(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
h1 = _write(tmp / "t1.json", b"traj1")
|
|
h2 = _write(tmp / "t2.json", b"traj2")
|
|
expected_sets = {
|
|
"raw_trajectories": {
|
|
"expected_paths": ["t1.json", "t2.json"],
|
|
"provenance": "https://fabricated-external-oracle.org/claims.json",
|
|
"independent_oracle": True,
|
|
}
|
|
}
|
|
# Run without trusted source
|
|
res = _run_audit(tmp, _manifest([
|
|
{"path": "t1.json", "type": "trajectory", "expected_sha256": h1},
|
|
{"path": "t2.json", "type": "trajectory", "expected_sha256": h2},
|
|
], expected_sets=expected_sets))
|
|
|
|
# Must NOT produce VERIFIED
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], "PARTIAL")
|
|
det = res["reproducibility_details"]["raw_trajectories"]
|
|
self.assertEqual(det["artifact_integrity"], "VERIFIED")
|
|
self.assertEqual(det["expected_set_status"], "UNVERIFIED")
|
|
self.assertEqual(det["expected_set_source"], "declared")
|
|
self.assertFalse(det["prerequisite_complete"])
|
|
self.assertIn("lacks separately trusted verification source", det["notes"])
|
|
|
|
def test_self_declared_completeness_flag_cannot_produce_verified(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
h1 = _write(tmp / "t1.json", b"traj1")
|
|
expected_sets = {
|
|
"raw_trajectories": {
|
|
"expected_paths": ["t1.json"],
|
|
"complete": True,
|
|
"provenance": "internal-self-assertion",
|
|
}
|
|
}
|
|
res = _run_audit(tmp, _manifest([
|
|
{"path": "t1.json", "type": "trajectory", "expected_sha256": h1},
|
|
], expected_sets=expected_sets))
|
|
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], "PARTIAL")
|
|
det = res["reproducibility_details"]["raw_trajectories"]
|
|
self.assertEqual(det["expected_set_status"], "UNVERIFIED_CLAIM")
|
|
self.assertFalse(det["prerequisite_complete"])
|
|
|
|
def test_partial_manifest_with_trusted_source_cannot_produce_verified(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
h1 = _write(tmp / "t1.json", b"traj1")
|
|
trusted_src = "https://official-specs.org/e00.json"
|
|
expected_sets = {
|
|
"raw_trajectories": {
|
|
"expected_paths": ["t1.json", "t2_missing.json"],
|
|
"provenance": trusted_src,
|
|
"independent_oracle": True,
|
|
}
|
|
}
|
|
res = _run_audit(
|
|
tmp,
|
|
_manifest([
|
|
{"path": "t1.json", "type": "trajectory", "expected_sha256": h1},
|
|
], expected_sets=expected_sets),
|
|
trusted_sources=[trusted_src],
|
|
)
|
|
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], "INCOMPLETE")
|
|
det = res["reproducibility_details"]["raw_trajectories"]
|
|
self.assertFalse(det["prerequisite_complete"])
|
|
self.assertIn("t2_missing.json", det["missing_or_unverified"])
|
|
|
|
def test_untrusted_provenance_mismatch_cannot_produce_verified(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
h1 = _write(tmp / "t1.json", b"traj1")
|
|
expected_sets = {
|
|
"raw_trajectories": {
|
|
"expected_paths": ["t1.json"],
|
|
"provenance": "https://unapproved-oracle.org",
|
|
"independent_oracle": True,
|
|
}
|
|
}
|
|
res = _run_audit(
|
|
tmp,
|
|
_manifest([
|
|
{"path": "t1.json", "type": "trajectory", "expected_sha256": h1},
|
|
], expected_sets=expected_sets),
|
|
trusted_sources=["https://approved-oracle.org"],
|
|
)
|
|
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], "PARTIAL")
|
|
det = res["reproducibility_details"]["raw_trajectories"]
|
|
self.assertEqual(det["expected_set_status"], "UNVERIFIED")
|
|
self.assertFalse(det["prerequisite_complete"])
|
|
|
|
def test_external_trusted_expected_sets_file_works(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
h1 = _write(tmp / "t1.json", b"traj1")
|
|
h2 = _write(tmp / "t2.json", b"traj2")
|
|
trusted_sets_file = tmp / "trusted_sets.json"
|
|
trusted_sets_file.write_text(json.dumps({
|
|
"raw_trajectories": {
|
|
"expected_paths": ["t1.json", "t2.json"],
|
|
"provenance": "official-e00-spec",
|
|
}
|
|
}))
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([
|
|
{"path": "t1.json", "type": "trajectory", "expected_sha256": h1},
|
|
{"path": "t2.json", "type": "trajectory", "expected_sha256": h2},
|
|
])))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(trusted_sets_file))
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], "VERIFIED")
|
|
det = res["reproducibility_details"]["raw_trajectories"]
|
|
self.assertEqual(det["expected_set_status"], "COMPLETE")
|
|
self.assertEqual(det["expected_set_source"], "externally_verified")
|
|
self.assertTrue(det["prerequisite_complete"])
|
|
|
|
|
|
class TestFinding4MalformedExpectedSets(unittest.TestCase):
|
|
"""P2: Malformed expected-set definitions and structural validation."""
|
|
|
|
def test_expected_sets_as_list_handled_without_attribute_error(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mdata = _manifest([])
|
|
mdata["expected_sets"] = [] # list instead of dict
|
|
mpath.write_text(json.dumps(mdata))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp))
|
|
self.assertFalse(res["manifest_valid"])
|
|
self.assertTrue(any("expected_sets must be a JSON object" in e for e in res["errors"]))
|
|
|
|
# Direct call to _build_prereqs with list must also not raise AttributeError
|
|
prereqs, details = auditor._build_prereqs([], [])
|
|
self.assertEqual(prereqs["raw_trajectories"], "INVALID")
|
|
self.assertEqual(details["raw_trajectories"]["status"], "INVALID")
|
|
|
|
def test_expected_paths_containing_dict_object_handled_without_type_error(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mdata = _manifest([])
|
|
mdata["expected_sets"] = {
|
|
"raw_trajectories": {
|
|
"expected_paths": [{"nested": "unhashable_dict"}],
|
|
}
|
|
}
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(mdata))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp))
|
|
self.assertFalse(res["manifest_valid"])
|
|
self.assertTrue(any("must be string" in e for e in res["errors"]))
|
|
|
|
# Direct call to _build_prereqs must also handle safely
|
|
prereqs, details = auditor._build_prereqs([], mdata["expected_sets"])
|
|
self.assertEqual(prereqs["raw_trajectories"], "INVALID")
|
|
self.assertEqual(details["raw_trajectories"]["expected_set_status"], "INVALID")
|
|
|
|
def test_empty_expected_paths_list_does_not_produce_verified(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
h = _write(tmp / "t1.json", b"content")
|
|
trusted_src = "https://specs.example.org/oracle.json"
|
|
mdata = _manifest([
|
|
{"path": "t1.json", "type": "trajectory", "expected_sha256": h},
|
|
])
|
|
mdata["expected_sets"] = {
|
|
"raw_trajectories": {
|
|
"expected_paths": [], # Empty list!
|
|
"provenance": trusted_src,
|
|
"independent_oracle": True,
|
|
}
|
|
}
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(mdata))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_sources=[trusted_src])
|
|
self.assertNotEqual(res["reproducibility_prerequisites"]["raw_trajectories"], "VERIFIED")
|
|
self.assertFalse(res["manifest_valid"])
|
|
self.assertTrue(any("cannot be empty" in e for e in res["errors"]))
|
|
|
|
# Direct call to _build_prereqs must also handle safely without producing VERIFIED
|
|
prereqs, details = auditor._build_prereqs(
|
|
[{"path": "t1.json", "type": "trajectory", "status": "VERIFIED"}],
|
|
mdata["expected_sets"],
|
|
trusted_sources=[trusted_src],
|
|
)
|
|
self.assertNotEqual(prereqs["raw_trajectories"], "VERIFIED")
|
|
self.assertEqual(prereqs["raw_trajectories"], "INVALID")
|
|
self.assertFalse(details["raw_trajectories"]["prerequisite_complete"])
|
|
|
|
def test_expected_paths_with_non_string_items_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mdata = _manifest([])
|
|
mdata["expected_sets"] = {
|
|
"raw_trajectories": {
|
|
"expected_paths": [123, True, None],
|
|
}
|
|
}
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(mdata))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp))
|
|
self.assertFalse(res["manifest_valid"])
|
|
|
|
def test_expected_paths_with_duplicate_paths_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mdata = _manifest([])
|
|
mdata["expected_sets"] = {
|
|
"raw_trajectories": {
|
|
"expected_paths": ["t1.json", "t1.json"],
|
|
}
|
|
}
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(mdata))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp))
|
|
self.assertFalse(res["manifest_valid"])
|
|
self.assertTrue(any("duplicate path" in e for e in res["errors"]))
|
|
|
|
def test_expected_count_invalid_type_or_negative_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mdata = _manifest([])
|
|
mdata["expected_sets"] = {
|
|
"raw_trajectories": {
|
|
"expected_count": -5,
|
|
}
|
|
}
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(mdata))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp))
|
|
self.assertFalse(res["manifest_valid"])
|
|
self.assertTrue(any("positive integer" in e for e in res["errors"]))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Finding 1: P1, Trusted Input Overwrite Protection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestFinding1TrustedInputProtection(unittest.TestCase):
|
|
"""Output publication must never overwrite any trusted input file."""
|
|
|
|
def test_output_targeting_trusted_expected_sets_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
tpath = tmp / "trusted_sets.json"
|
|
tbytes = json.dumps({"raw_trajectories": {"expected_count": 1}}).encode()
|
|
tpath.write_bytes(tbytes)
|
|
|
|
code = auditor.main([
|
|
str(mpath), str(tmp),
|
|
"--trusted-expected-sets", str(tpath),
|
|
"--output", str(tpath),
|
|
])
|
|
self.assertEqual(code, 2)
|
|
self.assertEqual(tpath.read_bytes(), tbytes, "Trusted expected sets file was overwritten")
|
|
|
|
def test_summary_targeting_trusted_expected_sets_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
tpath = tmp / "trusted_sets.json"
|
|
tbytes = json.dumps({"raw_trajectories": {"expected_count": 1}}).encode()
|
|
tpath.write_bytes(tbytes)
|
|
|
|
code = auditor.main([
|
|
str(mpath), str(tmp),
|
|
"--trusted-expected-sets", str(tpath),
|
|
"--summary", str(tpath),
|
|
])
|
|
self.assertEqual(code, 2)
|
|
self.assertEqual(tpath.read_bytes(), tbytes, "Trusted expected sets file was overwritten by summary")
|
|
|
|
def test_output_symlink_alias_targeting_trusted_expected_sets_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
tpath = tmp / "trusted_sets.json"
|
|
tbytes = json.dumps({"raw_trajectories": {"expected_count": 1}}).encode()
|
|
tpath.write_bytes(tbytes)
|
|
link_out = tmp / "link_trusted.json"
|
|
link_out.symlink_to(tpath)
|
|
|
|
code = auditor.main([
|
|
str(mpath), str(tmp),
|
|
"--trusted-expected-sets", str(tpath),
|
|
"--output", str(link_out),
|
|
])
|
|
self.assertEqual(code, 2)
|
|
self.assertEqual(tpath.read_bytes(), tbytes, "Trusted expected sets overwritten via symlink alias")
|
|
|
|
def test_output_hardlink_alias_targeting_trusted_expected_sets_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
tpath = tmp / "trusted_sets.json"
|
|
tbytes = json.dumps({"raw_trajectories": {"expected_count": 1}}).encode()
|
|
tpath.write_bytes(tbytes)
|
|
hlink_out = tmp / "hlink_trusted.json"
|
|
os.link(tpath, hlink_out)
|
|
|
|
code = auditor.main([
|
|
str(mpath), str(tmp),
|
|
"--trusted-expected-sets", str(tpath),
|
|
"--output", str(hlink_out),
|
|
])
|
|
self.assertEqual(code, 2)
|
|
self.assertEqual(tpath.read_bytes(), tbytes, "Trusted expected sets overwritten via hardlink alias")
|
|
|
|
def test_distinct_legitimate_destinations_with_trusted_sets_succeeds(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
content = b'{"task": "c5"}'
|
|
chash = hashlib.sha256(content).hexdigest()
|
|
(tmp / "t1.json").write_bytes(content)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([{
|
|
"path": "t1.json",
|
|
"type": "trajectory",
|
|
"expected_sha256": chash,
|
|
"size_bytes": len(content),
|
|
}])))
|
|
tpath = tmp / "trusted_sets.json"
|
|
tpath.write_text(json.dumps({"raw_trajectories": {"expected_paths": ["t1.json"]}}))
|
|
|
|
out = tmp / "out.json"
|
|
summ = tmp / "summary.md"
|
|
code = auditor.main([
|
|
str(mpath), str(tmp),
|
|
"--trusted-expected-sets", str(tpath),
|
|
"--output", str(out),
|
|
"--summary", str(summ),
|
|
"--deterministic",
|
|
])
|
|
self.assertEqual(code, 0)
|
|
self.assertTrue(out.is_file())
|
|
self.assertTrue(summ.is_file())
|
|
|
|
def test_adversarial_post_preflight_substitution_targeting_trusted_input_blocked(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
tpath = tmp / "trusted_sets.json"
|
|
original_bytes = b'{"immutable": "evidence"}'
|
|
tpath.write_bytes(original_bytes)
|
|
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
|
|
protected_paths, protected_inodes = auditor._collect_protected_targets(
|
|
str(mpath),
|
|
str(tmp),
|
|
[],
|
|
extra_input_paths=[tpath],
|
|
)
|
|
self.assertIn(tpath.resolve(), protected_paths)
|
|
self.assertIn((tpath.stat().st_dev, tpath.stat().st_ino), protected_inodes)
|
|
|
|
out_dest = tmp / "attack_out.json"
|
|
out_dest.symlink_to(tpath)
|
|
|
|
with self.assertRaises(OSError):
|
|
auditor._publish_file_securely(
|
|
str(out_dest),
|
|
"attacker payload",
|
|
protected_paths,
|
|
protected_inodes,
|
|
)
|
|
self.assertEqual(tpath.read_bytes(), original_bytes)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Finding 2: P2, Dot-Only Path Crash Rejection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestFinding2DotPathRejection(unittest.TestCase):
|
|
"""Dot-only path components must be rejected cleanly as INVALID without IndexError."""
|
|
|
|
def test_dot_path_in_manifest_classified_as_invalid(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
res = _run_audit(tmp, _manifest([{"path": ".", "type": "trajectory"}]))
|
|
self.assertEqual(len(res["artifacts"]), 1)
|
|
art = res["artifacts"][0]
|
|
self.assertEqual(art["status"], auditor.INVALID)
|
|
self.assertEqual(res["totals"][auditor.INVALID], 1)
|
|
|
|
def test_dot_slash_path_in_manifest_classified_as_invalid(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
res = _run_audit(tmp, _manifest([{"path": "./", "type": "trajectory"}]))
|
|
self.assertEqual(len(res["artifacts"]), 1)
|
|
art = res["artifacts"][0]
|
|
self.assertEqual(art["status"], auditor.INVALID)
|
|
|
|
def test_dot_slash_dot_path_classified_as_invalid(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
res = _run_audit(tmp, _manifest([{"path": "././", "type": "trajectory"}]))
|
|
self.assertEqual(len(res["artifacts"]), 1)
|
|
self.assertEqual(res["artifacts"][0]["status"], auditor.INVALID)
|
|
|
|
def test_multi_dot_slash_path_classified_as_invalid(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
res = _run_audit(tmp, _manifest([{"path": ".///.", "type": "trajectory"}]))
|
|
self.assertEqual(len(res["artifacts"]), 1)
|
|
self.assertEqual(res["artifacts"][0]["status"], auditor.INVALID)
|
|
|
|
def test_empty_path_classified_as_invalid(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
res = _run_audit(tmp, _manifest([{"path": "", "type": "trajectory"}]))
|
|
self.assertEqual(len(res["artifacts"]), 1)
|
|
self.assertEqual(res["artifacts"][0]["status"], auditor.INVALID)
|
|
|
|
def test_valid_nested_path_accepted(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
sub = tmp / "nested"
|
|
sub.mkdir()
|
|
content = b"trajectory_content"
|
|
(sub / "traj.json").write_bytes(content)
|
|
chash = hashlib.sha256(content).hexdigest()
|
|
res = _run_audit(tmp, _manifest([{
|
|
"path": "nested/traj.json",
|
|
"type": "trajectory",
|
|
"expected_sha256": chash,
|
|
"size_bytes": len(content),
|
|
}]))
|
|
self.assertEqual(len(res["artifacts"]), 1)
|
|
self.assertEqual(res["artifacts"][0]["status"], auditor.VERIFIED)
|
|
|
|
def test_mixed_manifest_with_dot_path_and_valid_path(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
content = b"valid_data"
|
|
(tmp / "valid.json").write_bytes(content)
|
|
chash = hashlib.sha256(content).hexdigest()
|
|
|
|
res = _run_audit(tmp, _manifest([
|
|
{"path": ".", "type": "trajectory"},
|
|
{"path": "valid.json", "type": "trajectory", "expected_sha256": chash, "size_bytes": len(content)},
|
|
]))
|
|
self.assertEqual(res["totals"][auditor.INVALID], 1)
|
|
self.assertEqual(res["totals"][auditor.VERIFIED], 1)
|
|
by_path = {a["path"]: a["status"] for a in res["artifacts"]}
|
|
self.assertEqual(by_path["."], auditor.INVALID)
|
|
self.assertEqual(by_path["valid.json"], auditor.VERIFIED)
|
|
|
|
def test_open_and_hash_artifact_direct_dot_calls_never_raise_index_error(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
for bad_path in [".", "./", "././", ".///.", "", " "]:
|
|
status, chash, size, note = auditor._open_and_hash_artifact(tmp, bad_path)
|
|
self.assertEqual(status, auditor.INVALID)
|
|
self.assertIsNone(chash)
|
|
self.assertIsNone(size)
|
|
self.assertIsNotNone(note)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Finding 3: P2, Unchecked Trusted-Expected-Sets Structure
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestFinding3MalformedTrustedExpectedSets(unittest.TestCase):
|
|
"""External trusted expected-sets configuration must be strictly validated."""
|
|
|
|
def test_trusted_sets_top_level_list_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text("[]")
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
self.assertFalse(res["configuration_valid"])
|
|
self.assertTrue(any("must be a JSON object" in e for e in res["errors"]))
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], auditor.INVALID)
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 2)
|
|
|
|
def test_trusted_sets_top_level_null_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text("null")
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
self.assertFalse(res["configuration_valid"])
|
|
self.assertTrue(any("cannot be null" in e for e in res["errors"]))
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 2)
|
|
|
|
def test_trusted_sets_top_level_empty_dict_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text("{}")
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
self.assertFalse(res["configuration_valid"])
|
|
self.assertTrue(any("cannot be an empty object" in e for e in res["errors"]))
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 2)
|
|
|
|
def test_trusted_sets_invalid_json_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text("THIS IS NOT JSON {{{{")
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
self.assertFalse(res["configuration_valid"])
|
|
self.assertTrue(any("cannot load trusted_expected_sets" in e for e in res["errors"]))
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 2)
|
|
|
|
def test_trusted_sets_unknown_category_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({"fabricated_unknown_category": {"expected_count": 1}}))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
self.assertFalse(res["configuration_valid"])
|
|
self.assertTrue(any("unrecognized or unsupported prerequisite category" in e for e in res["errors"]))
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 2)
|
|
|
|
def test_trusted_sets_empty_expected_paths_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({"raw_trajectories": {"expected_paths": []}}))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
self.assertFalse(res["configuration_valid"])
|
|
self.assertTrue(any("cannot be empty" in e for e in res["errors"]))
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 2)
|
|
|
|
def test_trusted_sets_non_string_paths_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({"raw_trajectories": {"expected_paths": [42, False, None]}}))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
self.assertFalse(res["configuration_valid"])
|
|
self.assertTrue(any("must be string" in e for e in res["errors"]))
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 2)
|
|
|
|
def test_trusted_sets_duplicate_paths_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({"raw_trajectories": {"expected_paths": ["a.json", "a.json"]}}))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
self.assertFalse(res["configuration_valid"])
|
|
self.assertTrue(any("duplicate path" in e for e in res["errors"]))
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 2)
|
|
|
|
def test_trusted_sets_unsafe_path_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({"raw_trajectories": {"expected_paths": ["../escape.json"]}}))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
self.assertFalse(res["configuration_valid"])
|
|
self.assertTrue(any("unsafe or invalid path" in e for e in res["errors"]))
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 2)
|
|
|
|
def test_trusted_sets_invalid_expected_count_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({"raw_trajectories": {"expected_count": -3}}))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
self.assertFalse(res["configuration_valid"])
|
|
self.assertTrue(any("positive integer" in e for e in res["errors"]))
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 2)
|
|
|
|
def test_trusted_sets_conflicting_paths_and_count_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({
|
|
"raw_trajectories": {
|
|
"expected_paths": ["a.json", "b.json"],
|
|
"expected_count": 5,
|
|
}
|
|
}))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
self.assertFalse(res["configuration_valid"])
|
|
self.assertTrue(any("conflicting definitions" in e for e in res["errors"]))
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 2)
|
|
|
|
def test_trusted_sets_invalid_provenance_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({"raw_trajectories": {"expected_count": 1, "provenance": " "}}))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
self.assertFalse(res["configuration_valid"])
|
|
self.assertTrue(any("invalid provenance" in e for e in res["errors"]))
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 2)
|
|
|
|
def test_trusted_sets_unsupported_fields_rejected(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({"raw_trajectories": {"expected_count": 1, "illegal_field": True}}))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
self.assertFalse(res["configuration_valid"])
|
|
self.assertTrue(any("unsupported field(s)" in e for e in res["errors"]))
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 2)
|
|
|
|
def test_valid_trusted_sets_with_complete_evidence_verified(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
content = b"evidence_payload"
|
|
chash = hashlib.sha256(content).hexdigest()
|
|
(tmp / "task1.json").write_bytes(content)
|
|
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([{
|
|
"path": "task1.json",
|
|
"type": "trajectory",
|
|
"expected_sha256": chash,
|
|
"size_bytes": len(content),
|
|
}])))
|
|
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({"raw_trajectories": {"expected_paths": ["task1.json"]}}))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
self.assertTrue(res["configuration_valid"])
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], auditor.VERIFIED)
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 0)
|
|
|
|
def test_valid_trusted_sets_with_incomplete_evidence_yields_code_1_not_2(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
content = b"partial_data"
|
|
chash = hashlib.sha256(content).hexdigest()
|
|
(tmp / "t1.json").write_bytes(content)
|
|
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([{
|
|
"path": "t1.json",
|
|
"type": "trajectory",
|
|
"expected_sha256": chash,
|
|
"size_bytes": len(content),
|
|
}])))
|
|
|
|
# Trusted expected sets requires 5 trajectories, but only 1 is present
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({"raw_trajectories": {"expected_count": 5}}))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
self.assertTrue(res["configuration_valid"])
|
|
self.assertEqual(len(res["errors"]), 0)
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], auditor.INCOMPLETE)
|
|
|
|
# Valid configuration with incomplete evidence must return exit code 1, NOT exit code 2
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 1)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# P2: Duplicate Evidence and Normalized Path Protection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestP2DuplicateEvidenceProtection(unittest.TestCase):
|
|
"""Campaign completeness must never be established by duplicate evidence."""
|
|
|
|
def test_1_physical_file_declared_twice_with_exact_same_path_cannot_satisfy_count(self):
|
|
"""Case A: One physical file declared twice cannot satisfy an expected count of two."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
content = b'{"trajectory_id": "traj_01"}'
|
|
chash = hashlib.sha256(content).hexdigest()
|
|
(tmp / "t1.json").write_bytes(content)
|
|
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([
|
|
{"path": "t1.json", "type": "trajectory", "expected_sha256": chash, "size_bytes": len(content)},
|
|
{"path": "t1.json", "type": "trajectory", "expected_sha256": chash, "size_bytes": len(content)},
|
|
])))
|
|
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({"raw_trajectories": {"expected_count": 2}}))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
det = res["reproducibility_details"]["raw_trajectories"]
|
|
self.assertFalse(det["prerequisite_complete"])
|
|
self.assertEqual(det["expected_set_status"], "INCOMPLETE")
|
|
self.assertEqual(det["distinct_verified_count"], 1)
|
|
self.assertTrue(len(det["duplicate_artifacts"]) > 0)
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], auditor.INCOMPLETE)
|
|
|
|
# CLI execution assertion
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 1, "Duplicate evidence should return exit code 1 (incomplete evidence)")
|
|
|
|
def test_2_file_declared_as_plain_and_dot_slash_path_cannot_be_counted_separately(self):
|
|
"""Case B: 'artifact.bin' and './artifact.bin' cannot be counted separately."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
content = b"artifact_binary_payload"
|
|
chash = hashlib.sha256(content).hexdigest()
|
|
(tmp / "artifact.bin").write_bytes(content)
|
|
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([
|
|
{"path": "artifact.bin", "type": "trajectory", "expected_sha256": chash, "size_bytes": len(content)},
|
|
{"path": "./artifact.bin", "type": "trajectory", "expected_sha256": chash, "size_bytes": len(content)},
|
|
])))
|
|
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({"raw_trajectories": {"expected_count": 2}}))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
det = res["reproducibility_details"]["raw_trajectories"]
|
|
self.assertFalse(det["prerequisite_complete"])
|
|
self.assertEqual(det["expected_set_status"], "INCOMPLETE")
|
|
self.assertEqual(det["distinct_verified_count"], 1)
|
|
self.assertTrue(any("normalized path alias" in d for d in det["duplicate_artifacts"]))
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], auditor.INCOMPLETE)
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 1)
|
|
|
|
def test_3_duplicate_nested_relative_paths_with_redundant_dot_slash(self):
|
|
"""Duplicate nested relative paths containing redundant dot components are detected."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
sub = tmp / "nested"
|
|
sub.mkdir()
|
|
content = b"nested_trajectory_data"
|
|
chash = hashlib.sha256(content).hexdigest()
|
|
(sub / "traj.json").write_bytes(content)
|
|
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([
|
|
{"path": "nested/traj.json", "type": "trajectory", "expected_sha256": chash, "size_bytes": len(content)},
|
|
{"path": "./nested/./traj.json", "type": "trajectory", "expected_sha256": chash, "size_bytes": len(content)},
|
|
])))
|
|
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({"raw_trajectories": {"expected_count": 2}}))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
det = res["reproducibility_details"]["raw_trajectories"]
|
|
self.assertFalse(det["prerequisite_complete"])
|
|
self.assertEqual(det["distinct_verified_count"], 1)
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], auditor.INCOMPLETE)
|
|
|
|
def test_4_expected_count_of_two_with_only_one_distinct_verified_artifact(self):
|
|
"""Expected count of two with only one distinct artifact reports incomplete."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
content = b"only_one_file"
|
|
chash = hashlib.sha256(content).hexdigest()
|
|
(tmp / "single.json").write_bytes(content)
|
|
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([
|
|
{"path": "single.json", "type": "trajectory", "expected_sha256": chash, "size_bytes": len(content)},
|
|
])))
|
|
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({"raw_trajectories": {"expected_count": 2}}))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
det = res["reproducibility_details"]["raw_trajectories"]
|
|
self.assertFalse(det["prerequisite_complete"])
|
|
self.assertEqual(det["expected_set_status"], "INCOMPLETE")
|
|
self.assertEqual(det["distinct_verified_count"], 1)
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], auditor.INCOMPLETE)
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 1)
|
|
|
|
def test_5_expected_set_containing_normalized_duplicate_paths_rejected(self):
|
|
"""Expected-set definition containing normalized duplicates is rejected as configuration error."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([])))
|
|
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({
|
|
"raw_trajectories": {
|
|
"expected_paths": ["sub/traj.json", "./sub/./traj.json"],
|
|
}
|
|
}))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
self.assertFalse(res["configuration_valid"])
|
|
self.assertTrue(any("duplicate path in expected_paths" in e for e in res["errors"]))
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], auditor.INVALID)
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 2, "Normalized duplicate in expected_paths must exit code 2")
|
|
|
|
def test_6_two_genuinely_distinct_checksum_verified_artifacts_satisfy_count(self):
|
|
"""Case C: Two genuinely distinct verified artifacts satisfy an expected count of two."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
c1 = b'{"task": "distinct_01"}'
|
|
c2 = b'{"task": "distinct_02"}'
|
|
h1 = hashlib.sha256(c1).hexdigest()
|
|
h2 = hashlib.sha256(c2).hexdigest()
|
|
(tmp / "t1.json").write_bytes(c1)
|
|
(tmp / "t2.json").write_bytes(c2)
|
|
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([
|
|
{"path": "t1.json", "type": "trajectory", "expected_sha256": h1, "size_bytes": len(c1)},
|
|
{"path": "t2.json", "type": "trajectory", "expected_sha256": h2, "size_bytes": len(c2)},
|
|
])))
|
|
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({"raw_trajectories": {"expected_count": 2}}))
|
|
|
|
out = tmp / "out.json"
|
|
code = auditor.main([
|
|
str(mpath), str(tmp),
|
|
"--trusted-expected-sets", str(tpath),
|
|
"--output", str(out),
|
|
])
|
|
self.assertEqual(code, 0)
|
|
res = json.loads(out.read_text())
|
|
det = res["reproducibility_details"]["raw_trajectories"]
|
|
self.assertTrue(det["prerequisite_complete"])
|
|
self.assertEqual(det["expected_set_status"], "COMPLETE")
|
|
self.assertEqual(det["distinct_verified_count"], 2)
|
|
self.assertEqual(det["duplicate_artifacts"], [])
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], auditor.VERIFIED)
|
|
|
|
def test_7_explicit_expected_paths_where_one_required_artifact_is_absent(self):
|
|
"""Explicit expected paths where one required artifact is absent remains incomplete."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
c1 = b"present_payload"
|
|
h1 = hashlib.sha256(c1).hexdigest()
|
|
(tmp / "t1.json").write_bytes(c1)
|
|
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([
|
|
{"path": "t1.json", "type": "trajectory", "expected_sha256": h1, "size_bytes": len(c1)},
|
|
])))
|
|
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({
|
|
"raw_trajectories": {
|
|
"expected_paths": ["t1.json", "t2.json"],
|
|
}
|
|
}))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
det = res["reproducibility_details"]["raw_trajectories"]
|
|
self.assertFalse(det["prerequisite_complete"])
|
|
self.assertEqual(det["expected_set_status"], "INCOMPLETE")
|
|
self.assertIn("t2.json", det["missing_or_unverified"])
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], auditor.INCOMPLETE)
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 1)
|
|
|
|
def test_8_explicit_expected_paths_where_every_required_artifact_verifies(self):
|
|
"""Explicit expected paths where every required distinct artifact verifies produces complete."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
c1 = b"payload_01"
|
|
c2 = b"payload_02"
|
|
h1 = hashlib.sha256(c1).hexdigest()
|
|
h2 = hashlib.sha256(c2).hexdigest()
|
|
(tmp / "t1.json").write_bytes(c1)
|
|
(tmp / "t2.json").write_bytes(c2)
|
|
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([
|
|
{"path": "./t1.json", "type": "trajectory", "expected_sha256": h1, "size_bytes": len(c1)},
|
|
{"path": "t2.json", "type": "trajectory", "expected_sha256": h2, "size_bytes": len(c2)},
|
|
])))
|
|
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({
|
|
"raw_trajectories": {
|
|
"expected_paths": ["t1.json", "t2.json"],
|
|
}
|
|
}))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
det = res["reproducibility_details"]["raw_trajectories"]
|
|
self.assertTrue(det["prerequisite_complete"])
|
|
self.assertEqual(det["expected_set_status"], "COMPLETE")
|
|
self.assertEqual(det["missing_or_unverified"], [])
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], auditor.VERIFIED)
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 0)
|
|
|
|
def test_9_manifest_entry_ordering_does_not_change_results(self):
|
|
"""Manifest entry ordering produces identical deterministic audit outcomes."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
c1 = b"first_file"
|
|
c2 = b"second_file"
|
|
h1 = hashlib.sha256(c1).hexdigest()
|
|
h2 = hashlib.sha256(c2).hexdigest()
|
|
(tmp / "a1.json").write_bytes(c1)
|
|
(tmp / "b2.json").write_bytes(c2)
|
|
|
|
e1 = {"path": "a1.json", "type": "trajectory", "expected_sha256": h1, "size_bytes": len(c1)}
|
|
e2 = {"path": "b2.json", "type": "trajectory", "expected_sha256": h2, "size_bytes": len(c2)}
|
|
|
|
mpath = tmp / "manifest.json"
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({"raw_trajectories": {"expected_count": 2}}))
|
|
|
|
out1 = tmp / "out1.json"
|
|
out2 = tmp / "out2.json"
|
|
|
|
mpath.write_text(json.dumps(_manifest([e1, e2])))
|
|
code1 = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath), "--output", str(out1), "--deterministic"])
|
|
|
|
mpath.write_text(json.dumps(_manifest([e2, e1])))
|
|
code2 = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath), "--output", str(out2), "--deterministic"])
|
|
|
|
self.assertEqual(code1, 0)
|
|
self.assertEqual(code2, 0)
|
|
self.assertEqual(out1.read_bytes(), out2.read_bytes())
|
|
|
|
def test_10_duplicate_evidence_never_produces_prerequisite_complete_or_verified(self):
|
|
"""Duplicate evidence never produces prerequisite_complete: true or VERIFIED status."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
content = b"sole_evidence_payload"
|
|
chash = hashlib.sha256(content).hexdigest()
|
|
(tmp / "sole.json").write_bytes(content)
|
|
|
|
# Even if expected_count is 1, a manifest with duplicate entries must not report prerequisite_complete
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([
|
|
{"path": "sole.json", "type": "trajectory", "expected_sha256": chash, "size_bytes": len(content)},
|
|
{"path": "sole.json", "type": "trajectory", "expected_sha256": chash, "size_bytes": len(content)},
|
|
])))
|
|
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({"raw_trajectories": {"expected_count": 1}}))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
det = res["reproducibility_details"]["raw_trajectories"]
|
|
self.assertFalse(det["prerequisite_complete"], "Duplicate evidence must not produce prerequisite_complete: true")
|
|
self.assertNotEqual(res["reproducibility_prerequisites"]["raw_trajectories"], auditor.VERIFIED)
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], auditor.INCOMPLETE)
|
|
|
|
def test_11_hardlink_alias_detected_and_cannot_satisfy_expected_count(self):
|
|
"""Filesystem hardlink alias to the same inode cannot satisfy an expected count of two."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
content = b"hardlink_target_bytes"
|
|
chash = hashlib.sha256(content).hexdigest()
|
|
(tmp / "orig.json").write_bytes(content)
|
|
os.link(tmp / "orig.json", tmp / "hardlink_alias.json")
|
|
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([
|
|
{"path": "orig.json", "type": "trajectory", "expected_sha256": chash, "size_bytes": len(content)},
|
|
{"path": "hardlink_alias.json", "type": "trajectory", "expected_sha256": chash, "size_bytes": len(content)},
|
|
])))
|
|
|
|
tpath = tmp / "trusted.json"
|
|
tpath.write_text(json.dumps({"raw_trajectories": {"expected_count": 2}}))
|
|
|
|
res = auditor.audit(str(mpath), str(tmp), trusted_expected_sets=str(tpath))
|
|
det = res["reproducibility_details"]["raw_trajectories"]
|
|
self.assertFalse(det["prerequisite_complete"])
|
|
self.assertEqual(det["expected_set_status"], "INCOMPLETE")
|
|
self.assertEqual(det["distinct_verified_count"], 1)
|
|
self.assertTrue(any("hardlinks" in d for d in det["duplicate_artifacts"]))
|
|
self.assertEqual(res["reproducibility_prerequisites"]["raw_trajectories"], auditor.INCOMPLETE)
|
|
|
|
code = auditor.main([str(mpath), str(tmp), "--trusted-expected-sets", str(tpath)])
|
|
self.assertEqual(code, 1)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# P2: Malformed Filename Rejection (Embedded NUL and Unpaired Unicode Surrogates)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestMalformedFilenameRejection(unittest.TestCase):
|
|
"""Malformed filesystem filenames must be rejected cleanly without traceback."""
|
|
|
|
def test_1_filename_containing_nul_classified_invalid(self):
|
|
"""Case 1: Filename containing embedded NUL is classified as INVALID without crash."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
bad_path = "artifact\x00.txt"
|
|
mpath.write_text(json.dumps(_manifest([
|
|
{"path": bad_path, "type": "result_record"}
|
|
])))
|
|
res = auditor.audit(str(mpath), str(tmp))
|
|
self.assertEqual(res["totals"][auditor.INVALID], 1)
|
|
self.assertEqual(len(res["artifacts"]), 1)
|
|
art = res["artifacts"][0]
|
|
self.assertEqual(art["status"], auditor.INVALID)
|
|
self.assertFalse(art["present"])
|
|
self.assertEqual(art["path"], bad_path)
|
|
|
|
def test_2_filename_containing_unpaired_high_surrogate_classified_invalid(self):
|
|
"""Case 2: Filename containing unpaired high surrogate U+D800 is classified as INVALID."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
bad_path = f"artifact{chr(0xd800)}.txt"
|
|
mpath.write_text(json.dumps(_manifest([
|
|
{"path": bad_path, "type": "result_record"}
|
|
])))
|
|
res = auditor.audit(str(mpath), str(tmp))
|
|
self.assertEqual(res["totals"][auditor.INVALID], 1)
|
|
art = res["artifacts"][0]
|
|
self.assertEqual(art["status"], auditor.INVALID)
|
|
self.assertFalse(art["present"])
|
|
|
|
def test_3_filename_containing_unpaired_low_surrogate_classified_invalid(self):
|
|
"""Case 3: Filename containing unpaired low surrogate (U+DC00 / U+DFFF) is classified as INVALID."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
for surrogate_cp in (0xdc00, 0xdc80, 0xdfff):
|
|
mpath = tmp / f"manifest_{surrogate_cp}.json"
|
|
bad_name = f"artifact{chr(surrogate_cp)}.txt"
|
|
mpath.write_text(json.dumps(_manifest([
|
|
{"path": bad_name, "type": "result_record"}
|
|
])))
|
|
res = auditor.audit(str(mpath), str(tmp))
|
|
self.assertEqual(res["totals"][auditor.INVALID], 1)
|
|
self.assertEqual(res["artifacts"][0]["status"], auditor.INVALID)
|
|
|
|
def test_4_nul_in_nested_relative_path_component(self):
|
|
"""Case 4: NUL in nested directory or filename relative path component is classified as INVALID."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
bad_path = "nested/dir\x00component/artifact.json"
|
|
mpath.write_text(json.dumps(_manifest([
|
|
{"path": bad_path, "type": "result_record"}
|
|
])))
|
|
res = auditor.audit(str(mpath), str(tmp))
|
|
self.assertEqual(res["totals"][auditor.INVALID], 1)
|
|
self.assertEqual(res["artifacts"][0]["status"], auditor.INVALID)
|
|
|
|
def test_5_valid_unicode_filename_with_non_ascii_characters(self):
|
|
"""Case 5: Valid Unicode filename with non-ASCII characters is accepted and verified."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
valid_name = "café_測試_файл_🚀.json"
|
|
content = b'{"status": "ok", "experiment": "valid_unicode"}'
|
|
(tmp / valid_name).write_bytes(content)
|
|
chash = hashlib.sha256(content).hexdigest()
|
|
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([
|
|
{"path": valid_name, "type": "result_record", "expected_sha256": chash, "size_bytes": len(content)}
|
|
])))
|
|
res = auditor.audit(str(mpath), str(tmp))
|
|
self.assertEqual(res["totals"][auditor.VERIFIED], 1)
|
|
self.assertEqual(res["totals"][auditor.INVALID], 0)
|
|
art = res["artifacts"][0]
|
|
self.assertEqual(art["status"], auditor.VERIFIED)
|
|
self.assertTrue(art["present"])
|
|
self.assertTrue(art["hash_match"])
|
|
self.assertEqual(art["observed_sha256"], chash)
|
|
|
|
def test_6_mixed_manifest_containing_one_valid_and_one_malformed_artifact(self):
|
|
"""Case 6: Mixed manifest continues processing other entries when one is invalid."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
valid_name = "valid_evidence.txt"
|
|
content = b"clean_verified_evidence_data"
|
|
(tmp / valid_name).write_bytes(content)
|
|
chash = hashlib.sha256(content).hexdigest()
|
|
|
|
bad_nul = "malformed\x00_entry.dat"
|
|
bad_surrogate = f"malformed{chr(0xd800)}_entry.dat"
|
|
|
|
mpath = tmp / "manifest.json"
|
|
mpath.write_text(json.dumps(_manifest([
|
|
{"path": valid_name, "type": "trajectory", "expected_sha256": chash, "size_bytes": len(content)},
|
|
{"path": bad_nul, "type": "result_record"},
|
|
{"path": bad_surrogate, "type": "result_record"},
|
|
])))
|
|
res = auditor.audit(str(mpath), str(tmp))
|
|
self.assertEqual(res["totals"][auditor.VERIFIED], 1)
|
|
self.assertEqual(res["totals"][auditor.INVALID], 2)
|
|
by_path = {a["path"]: a["status"] for a in res["artifacts"]}
|
|
self.assertEqual(by_path[valid_name], auditor.VERIFIED)
|
|
self.assertEqual(by_path[bad_nul], auditor.INVALID)
|
|
self.assertEqual(by_path[bad_surrogate], auditor.INVALID)
|
|
|
|
def test_7_invalid_paths_never_reach_filesystem_calls(self):
|
|
"""Case 7: Invalid paths with NUL or surrogates never reach filesystem stat/open calls."""
|
|
from unittest.mock import patch
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
mpath = tmp / "manifest.json"
|
|
bad_paths = ["file\x00.txt", f"file{chr(0xd800)}.txt", "nested\x00/file.txt"]
|
|
mpath.write_text(json.dumps(_manifest([
|
|
{"path": p, "type": "result_record"} for p in bad_paths
|
|
])))
|
|
|
|
with patch("os.lstat", wraps=os.lstat) as mock_lstat, patch("os.open", wraps=os.open) as mock_open:
|
|
res = auditor.audit(str(mpath), str(tmp))
|
|
for bad_p in bad_paths:
|
|
for c in mock_lstat.call_args_list:
|
|
self.assertNotIn(bad_p, str(c))
|
|
for c in mock_open.call_args_list:
|
|
self.assertNotIn(bad_p, str(c))
|
|
self.assertEqual(res["totals"][auditor.INVALID], len(bad_paths))
|
|
|
|
# Direct reader invocation also avoids filesystem calls
|
|
with patch("os.lstat", wraps=os.lstat) as mock_lstat, patch("os.open", wraps=os.open) as mock_open:
|
|
for bad_p in bad_paths:
|
|
status, chash, size, note = auditor._open_and_hash_artifact(tmp, bad_p)
|
|
self.assertEqual(status, auditor.INVALID)
|
|
for c in mock_lstat.call_args_list:
|
|
self.assertNotIn(bad_p, str(c))
|
|
for c in mock_open.call_args_list:
|
|
self.assertNotIn(bad_p, str(c))
|
|
|
|
def test_8_direct_invocation_of_path_validator_and_artifact_reader(self):
|
|
"""Case 8: Path validator and reader return safe results when invoked directly."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
bad_nul = "file\x00.txt"
|
|
bad_high = f"file{chr(0xd800)}.txt"
|
|
bad_low1 = f"file{chr(0xdc00)}.txt"
|
|
bad_low_esc = f"file{chr(0xdc80)}.txt"
|
|
bad_low2 = f"file{chr(0xdfff)}.txt"
|
|
bad_nested_nul = "dir/sub\x00/file.txt"
|
|
bad_nested_surr = f"dir/sub{chr(0xd800)}/file.txt"
|
|
|
|
# Validator returns False
|
|
self.assertFalse(auditor._is_safe_relative_path(bad_nul))
|
|
self.assertFalse(auditor._is_safe_relative_path(bad_high))
|
|
self.assertFalse(auditor._is_safe_relative_path(bad_low1))
|
|
self.assertFalse(auditor._is_safe_relative_path(bad_low_esc))
|
|
self.assertFalse(auditor._is_safe_relative_path(bad_low2))
|
|
self.assertFalse(auditor._is_safe_relative_path(bad_nested_nul))
|
|
self.assertFalse(auditor._is_safe_relative_path(bad_nested_surr))
|
|
|
|
# Normalizer returns None
|
|
self.assertIsNone(auditor._normalize_relative_path(bad_nul))
|
|
self.assertIsNone(auditor._normalize_relative_path(bad_high))
|
|
self.assertIsNone(auditor._normalize_relative_path(bad_low1))
|
|
self.assertIsNone(auditor._normalize_relative_path(bad_low_esc))
|
|
|
|
# Valid Unicode paths accepted
|
|
self.assertTrue(auditor._is_safe_relative_path("café_測試.json"))
|
|
self.assertEqual(auditor._normalize_relative_path("./café_測試.json"), "café_測試.json")
|
|
|
|
# Direct reader returns INVALID
|
|
for bad in [bad_nul, bad_high, bad_low1, bad_nested_nul]:
|
|
status, chash, size, note = auditor._open_and_hash_artifact(tmp, bad)
|
|
self.assertEqual(status, auditor.INVALID)
|
|
self.assertIsNone(chash)
|
|
self.assertIsNone(size)
|
|
self.assertIsNotNone(note)
|
|
|
|
def test_9_actual_cli_execution_with_malformed_paths(self):
|
|
"""Case 9: CLI execution returns exit code 2 without traceback for malformed paths."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
for bad_name in ["file\x00.txt", f"file{chr(0xd800)}.txt", "nested\x00/file.txt"]:
|
|
mpath = tmp / "manifest.json"
|
|
out_path = tmp / "out.json"
|
|
sum_path = tmp / "summary.md"
|
|
mpath.write_text(json.dumps(_manifest([
|
|
{"path": bad_name, "type": "result_record"}
|
|
])))
|
|
|
|
code = auditor.main([
|
|
str(mpath), str(tmp),
|
|
"--output", str(out_path),
|
|
"--summary", str(sum_path),
|
|
])
|
|
self.assertEqual(code, 2, f"Expected exit code 2 for malformed path: {bad_name!r}")
|
|
self.assertTrue(out_path.exists())
|
|
self.assertTrue(sum_path.exists())
|
|
out_path.unlink()
|
|
sum_path.unlink()
|
|
|
|
def test_10_no_traceback_correct_classification_and_expected_exit_code(self):
|
|
"""Case 10: Subprocess CLI execution produces no traceback, controlled INVALID, and exit code 2."""
|
|
import subprocess
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tmp = Path(td)
|
|
cases = [
|
|
("embedded_nul", "bad\x00filename.json"),
|
|
("unpaired_high_surrogate", f"bad{chr(0xd800)}filename.json"),
|
|
("unpaired_low_surrogate", f"bad{chr(0xdc00)}filename.json"),
|
|
]
|
|
for case_label, bad_name in cases:
|
|
mpath = tmp / f"manifest_{case_label}.json"
|
|
mpath.write_text(json.dumps(_manifest([
|
|
{"path": bad_name, "type": "result_record"}
|
|
])))
|
|
proc = subprocess.run(
|
|
[sys.executable, str(_ROOT / "scripts" / "audit_c5_benchmark_evidence.py"), str(mpath), str(tmp)],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
self.assertEqual(proc.returncode, 2, f"Case {case_label} must exit with code 2")
|
|
self.assertNotIn("Traceback", proc.stderr, f"Case {case_label} stderr must not contain traceback")
|
|
res = json.loads(proc.stdout)
|
|
self.assertEqual(res["totals"][auditor.INVALID], 1)
|
|
self.assertEqual(res["artifacts"][0]["status"], auditor.INVALID)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Security: command-line diagnostics must not disclose untrusted input values
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCliDiagnosticConfidentiality(unittest.TestCase):
|
|
"""Keep detailed diagnostics in the structured API, not in terminal logs."""
|
|
|
|
def test_invalid_manifest_category_does_not_leak_into_stderr(self):
|
|
import subprocess
|
|
marker = "SENSITIVE_MARKER_CATEGORY_0183"
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root = Path(td)
|
|
manifest = root / "manifest.json"
|
|
manifest.write_text(json.dumps(_manifest([], expected_sets={marker: {"expected_count": 1}})))
|
|
result = auditor.audit(str(manifest), str(root))
|
|
self.assertTrue(any(marker in err for err in result["errors"]))
|
|
proc = subprocess.run(
|
|
[sys.executable, str(_ROOT / "scripts" / "audit_c5_benchmark_evidence.py"), str(manifest), str(root)],
|
|
capture_output=True, text=True,
|
|
)
|
|
self.assertEqual(proc.returncode, 2)
|
|
self.assertIn("audit error: manifest is unreadable or invalid", proc.stderr)
|
|
self.assertNotIn(marker, proc.stderr)
|
|
self.assertNotIn("Traceback", proc.stderr)
|
|
|
|
def test_unreadable_trusted_config_path_does_not_leak_into_stderr(self):
|
|
import subprocess
|
|
marker = "SENSITIVE_MARKER_TRUSTED_PATH_0294"
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root = Path(td)
|
|
manifest = root / "manifest.json"
|
|
manifest.write_text(json.dumps(_manifest([])))
|
|
trusted_path = root / (marker + ".json")
|
|
proc = subprocess.run(
|
|
[sys.executable, str(_ROOT / "scripts" / "audit_c5_benchmark_evidence.py"), str(manifest), str(root),
|
|
"--trusted-expected-sets", str(trusted_path)],
|
|
capture_output=True, text=True,
|
|
)
|
|
self.assertEqual(proc.returncode, 2)
|
|
self.assertIn("audit error: invalid trusted expected-sets configuration", proc.stderr)
|
|
self.assertNotIn(marker, proc.stderr)
|
|
self.assertNotIn("Traceback", proc.stderr)
|
|
|
|
def test_preflight_conflict_does_not_leak_path_or_modify_manifest(self):
|
|
import subprocess
|
|
marker = "SENSITIVE_MARKER_DESTINATION_0312"
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root = Path(td)
|
|
manifest = root / (marker + ".json")
|
|
manifest.write_text(json.dumps(_manifest([])))
|
|
original = manifest.read_bytes()
|
|
proc = subprocess.run(
|
|
[sys.executable, str(_ROOT / "scripts" / "audit_c5_benchmark_evidence.py"), str(manifest), str(root),
|
|
"--output", str(manifest)],
|
|
capture_output=True, text=True,
|
|
)
|
|
self.assertEqual(proc.returncode, 2)
|
|
self.assertIn("preflight error: unsafe or conflicting output destination", proc.stderr)
|
|
self.assertNotIn(marker, proc.stderr)
|
|
self.assertNotIn("Traceback", proc.stderr)
|
|
self.assertEqual(manifest.read_bytes(), original)
|
|
|
|
def test_publication_error_does_not_leak_exception_text(self):
|
|
from contextlib import redirect_stderr
|
|
from io import StringIO
|
|
from unittest.mock import patch
|
|
marker = "SENSITIVE_MARKER_PUBLISH_0489"
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root = Path(td)
|
|
manifest = root / "manifest.json"
|
|
manifest.write_text(json.dumps(_manifest([])))
|
|
for option, expected in (
|
|
("--output", "publication error: unable to write output"),
|
|
("--summary", "publication error: unable to write summary"),
|
|
):
|
|
stderr = StringIO()
|
|
with patch.object(auditor, "_publish_file_securely", side_effect=OSError(marker)):
|
|
with redirect_stderr(stderr):
|
|
code = auditor.main([str(manifest), str(root), option, str(root / "output.txt")])
|
|
self.assertEqual(code, 2)
|
|
self.assertIn(expected, stderr.getvalue())
|
|
self.assertNotIn(marker, stderr.getvalue())
|
|
self.assertNotIn("Traceback", stderr.getvalue())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|