205 lines
8.6 KiB
Python
205 lines
8.6 KiB
Python
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()
|