from __future__ import annotations import datetime as dt import json import os import stat import tempfile import unittest from pathlib import Path from .evidence import ( EvidenceStore, EvidenceValidationError, create_summary, sha256_bytes, sha256_json, utc_timestamp, validate_summary, ) class EvidenceTests(unittest.TestCase): def test_documented_schema_matches_the_runtime_contract(self) -> None: root = Path(__file__).parents[3] schema = json.loads( (root / "config" / "nono" / "evidence.schema.json").read_text( encoding="utf-8" ) ) summary = self._summary() self.assertEqual(set(schema["required"]), set(summary)) self.assertFalse(schema["additionalProperties"]) self.assertEqual( set(schema["properties"]["suite"]["enum"]), {"hermetic", "host", "live"}, ) self.assertEqual( set(schema["properties"]["status"]["enum"]), {"passed", "failed", "blocked", "not-run"}, ) observation = summary["observations"][0] self.assertEqual( set(schema["$defs"]["observation"]["required"]), set(observation), ) self.assertFalse(schema["$defs"]["observation"]["additionalProperties"]) def _summary(self, **changes: object) -> dict[str, object]: observation = { "id": "negative-tmp", "capability": "filesystem", "expected": "denied", "actual": "denied", "status": "passed", "hash": sha256_json({"actual": "denied", "path": "tmp"}), } values: dict[str, object] = { "run_id": "run-001", "test_id": "negative-matrix", "suite": "hermetic", "repository_commit": "0" * 40, "policy_snapshot_digest": "1" * 64, "installed_snapshot_digest": "2" * 64, "nono_version": "0.74.0", "status": "passed", "started_at": "2026-08-27T04:00:00Z", "finished_at": "2026-08-27T04:00:01Z", "raw_evidence_sha256": "3" * 64, "raw_evidence_retained_until": "2026-09-26T04:00:00Z", "result_hashes": {"result": "4" * 64}, "observations": [observation], } values.update(changes) return create_summary(**values) # type: ignore[arg-type] def test_summary_contract_is_closed_and_validates_hashes_times_status(self) -> None: summary = self._summary() validate_summary(summary) for mutation in ( {"missing": "schemaVersion"}, {"extra": "not allowed"}, {"policySnapshotDigest": "not-a-digest"}, {"startedAt": "2026-08-27T04:00:00+00:00"}, {"status": "skipped"}, {"rawEvidence": {"sha256": "3" * 64, "retainedUntil": "bad"}}, ): candidate = dict(summary) if "missing" in mutation: candidate.pop(str(mutation["missing"])) elif "extra" in mutation: candidate["extra"] = mutation["extra"] else: candidate.update(mutation) with self.assertRaises(EvidenceValidationError, msg=str(mutation)): validate_summary(candidate) def test_raw_command_fields_and_live_approval_contract(self) -> None: with self.assertRaises(EvidenceValidationError): self._summary(result_hashes={"stdout": "4" * 64}) with self.assertRaises(EvidenceValidationError): self._summary(suite="live") live = self._summary(suite="live", approver="JB", status="blocked") validate_summary(live) with self.assertRaises(EvidenceValidationError): self._summary(observations=[{"stdout": "raw"}]) def test_raw_files_are_private_atomic_and_traceable(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) state = root / "state" summaries = root / "summaries" state.mkdir(mode=0o700) summaries.mkdir(mode=0o700) os.chmod(state, 0o700) os.chmod(summaries, 0o700) store = EvidenceStore(state, summaries) payload = { "runId": "run-001", "result": {"stdout": "raw fixture output", "stderr": "raw error"}, } reference = store.write_raw_evidence( "run-001", payload, now=dt.datetime(2026, 8, 27, tzinfo=dt.timezone.utc) ) self.assertEqual(stat.S_IMODE(reference.path.stat().st_mode), 0o600) self.assertEqual(stat.S_IMODE(reference.metadata_path.stat().st_mode), 0o600) self.assertEqual(stat.S_IMODE(reference.path.parent.stat().st_mode), 0o700) self.assertEqual(stat.S_IMODE((state / "pi-sbx").stat().st_mode), 0o700) self.assertEqual(stat.S_IMODE((state / "pi-sbx" / "evidence").stat().st_mode), 0o700) self.assertEqual( reference.sha256, sha256_bytes(reference.path.read_bytes()), ) restored, metadata = store.read_raw_evidence( "run-001", now=dt.datetime(2026, 8, 28, tzinfo=dt.timezone.utc), ) self.assertEqual(restored, payload) self.assertEqual(metadata["sha256"], reference.sha256) self.assertEqual(metadata["retainedUntil"], "2026-09-26T00:00:00Z") summary = self._summary(raw_evidence_sha256=reference.sha256) summary_path = store.write_summary( summary, repository_id="fixture-repo", policy_digest="1" * 64, ) self.assertEqual(stat.S_IMODE(summary_path.stat().st_mode), 0o600) self.assertEqual(stat.S_IMODE(summary_path.parent.stat().st_mode), 0o700) committed = json.loads(summary_path.read_text(encoding="utf-8")) self.assertNotIn("stdout", json.dumps(committed)) self.assertNotIn("stderr", json.dumps(committed)) self.assertEqual(committed["rawEvidence"]["sha256"], reference.sha256) self.assertEqual(list(summary_path.parent.glob(".*")), []) expired_at = dt.datetime(2026, 9, 26, tzinfo=dt.timezone.utc) with self.assertRaisesRegex(EvidenceValidationError, "expired"): store.read_raw_evidence("run-001", now=expired_at) self.assertEqual(store.purge_expired(now=expired_at), ["run-001"]) self.assertFalse(reference.path.parent.exists()) def test_raw_fifo_is_rejected_without_blocking(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) state = root / "state" summaries = root / "summaries" state.mkdir(mode=0o700) summaries.mkdir(mode=0o700) store = EvidenceStore(state, summaries) reference = store.write_raw_evidence("run-fifo", {"status": "blocked"}) reference.path.unlink() os.mkfifo(reference.path, mode=0o600) with self.assertRaisesRegex(EvidenceValidationError, "regular file"): store.read_raw_evidence("run-fifo") def test_interrupted_quarantine_is_recovered(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) state = root / "state" summaries = root / "summaries" state.mkdir(mode=0o700) summaries.mkdir(mode=0o700) store = EvidenceStore(state, summaries) reference = store.write_raw_evidence("run-interrupted", {"status": "blocked"}) quarantine = reference.path.parent.parent / (".purge-" + "a" * 32) reference.path.parent.rename(quarantine) self.assertEqual(store.purge_expired(), []) self.assertFalse(quarantine.exists()) def test_paths_cannot_escape_injected_roots(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) state = root / "state" summaries = root / "summaries" state.mkdir(mode=0o700) summaries.mkdir(mode=0o700) os.chmod(state, 0o700) os.chmod(summaries, 0o700) store = EvidenceStore(state, summaries) with self.assertRaises(EvidenceValidationError): store.read_raw_evidence("../outside") with self.assertRaises(EvidenceValidationError): store.summary_path("../outside", "1" * 64, "run-1") if __name__ == "__main__": unittest.main()