mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-09-10 18:45:53 +00:00
Richard Courtman
523 lines
24 KiB
Python
523 lines
24 KiB
Python
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
|
|
from secure_runtime_attestation import (
|
|
ARTIFACT_ARGUMENTS,
|
|
AttestationError,
|
|
DISPOSABLE_VM_GUARD_SHA256,
|
|
EXPECTED_ARTIFACT_PACKAGES,
|
|
REQUIRED_SCENARIOS,
|
|
SCENARIO_REQUIRED_CLAIMS,
|
|
SCENARIO_REQUIRED_OBSERVATIONS,
|
|
SOURCE_MANIFEST_ID,
|
|
SOURCE_MANIFEST_PATH,
|
|
SOURCE_MANIFEST_SCHEMA_VERSION,
|
|
create_attestation,
|
|
parse_utc_timestamp,
|
|
sha256_bytes,
|
|
verify_artifact_build_identity,
|
|
)
|
|
from repo_file_io import strip_local_git_env
|
|
|
|
|
|
class SecureRuntimeAttestationTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temporary.name)
|
|
self.repo = self.root / "repo"
|
|
self.repo.mkdir()
|
|
self.git("init", "-b", "main")
|
|
self.git("config", "user.name", "Pulse Test")
|
|
self.git("config", "user.email", "pulse-test@example.invalid")
|
|
manifest = {
|
|
"schema_version": SOURCE_MANIFEST_SCHEMA_VERSION,
|
|
"manifest_id": SOURCE_MANIFEST_ID,
|
|
"target_os": "linux",
|
|
"exact_paths": [
|
|
"scripts/install.sh",
|
|
"scripts/installtests/secure_runtime_systemd_lab_test.go",
|
|
"scripts/release_control/secure_runtime_attestation.py",
|
|
SOURCE_MANIFEST_PATH,
|
|
"scripts/release_ldflags.sh",
|
|
"scripts/release_update_key.go",
|
|
],
|
|
"recursive_roots": ["boundary", "internal/updatesignature"],
|
|
"include_suffixes": [".go"],
|
|
"exclude_suffixes": ["_test.go"],
|
|
}
|
|
manifest_raw = (json.dumps(manifest, indent=2) + "\n").encode()
|
|
source_values = {
|
|
"scripts/install.sh": b"fixture installer\n",
|
|
"scripts/installtests/secure_runtime_systemd_lab_test.go": b"fixture harness\n",
|
|
"scripts/release_control/secure_runtime_attestation.py": Path(__file__).with_name("secure_runtime_attestation.py").read_bytes(),
|
|
SOURCE_MANIFEST_PATH: manifest_raw,
|
|
"scripts/release_ldflags.sh": b"#!/bin/sh\n",
|
|
"scripts/release_update_key.go": b"package main\n",
|
|
"boundary/runtime.go": b"package boundary\n",
|
|
"internal/updatesignature/signature.go": b"package updatesignature\n",
|
|
}
|
|
self.source_hashes = {}
|
|
for relative, value in sorted(source_values.items()):
|
|
path = self.repo / relative
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(value)
|
|
self.source_hashes[relative] = sha256_bytes(value)
|
|
self.git("add", ".")
|
|
self.git("commit", "-m", "fixture")
|
|
self.commit = self.git("rev-parse", "HEAD")
|
|
self.git("branch", "origin/main")
|
|
self.git("checkout", "--detach", self.commit)
|
|
|
|
self.artifacts = {}
|
|
artifact_hashes = {}
|
|
for name in ARTIFACT_ARGUMENTS:
|
|
path = self.root / name
|
|
value = f"{name}\n".encode()
|
|
path.write_bytes(value)
|
|
self.artifacts[name] = path
|
|
artifact_hashes[name] = sha256_bytes(value)
|
|
self.receipt = self.root / "receipt.json"
|
|
self.transcript = self.root / "transcript.jsonl"
|
|
started_at = "2026-08-30T10:00:00+00:00"
|
|
scenarios = []
|
|
command_output = "fixture command output\n"
|
|
transcript_events = [
|
|
{
|
|
"sequence": 1,
|
|
"event_id": "event-0001",
|
|
"observed_at": started_at,
|
|
"kind": "command_output",
|
|
"operation": "fixture command",
|
|
"output": command_output,
|
|
"output_sha256": sha256_bytes(command_output.encode()),
|
|
}
|
|
]
|
|
previous = started_at
|
|
for index, name in enumerate(REQUIRED_SCENARIOS, 1):
|
|
completed = f"2026-08-30T10:00:{index:02d}+00:00"
|
|
event_sequence = index + 1
|
|
event_id = f"event-{event_sequence:04d}"
|
|
claims = sorted(SCENARIO_REQUIRED_CLAIMS[name])
|
|
observations = dict(SCENARIO_REQUIRED_OBSERVATIONS[name])
|
|
if name == "helper_update_authoritative_commit":
|
|
observations.update(
|
|
{
|
|
"candidate_sha256": artifact_hashes["collector_v3"],
|
|
"prior_sha256": artifact_hashes["collector_v2"],
|
|
"target_sha256": artifact_hashes["collector_v3"],
|
|
"last_known_good_sha256": artifact_hashes["collector_v2"],
|
|
"activator_pid": 4321,
|
|
"committer_pid": 4321,
|
|
}
|
|
)
|
|
elif name in (
|
|
"helper_update_watchdog_rollback",
|
|
"helper_update_interrupted_recovery",
|
|
):
|
|
observations.update(
|
|
{
|
|
"candidate_sha256": artifact_hashes["collector_v4"],
|
|
"prior_sha256": artifact_hashes["collector_v3"],
|
|
"target_sha256": artifact_hashes["collector_v3"],
|
|
"last_known_good_sha256": artifact_hashes["collector_v4"],
|
|
}
|
|
)
|
|
scenarios.append(
|
|
{
|
|
"sequence": index,
|
|
"name": name,
|
|
"passed": True,
|
|
"started_at": previous,
|
|
"completed_at": completed,
|
|
"evidence": {
|
|
"kind": "runtime-observation-v1",
|
|
"summary": f"observed {name}",
|
|
"claims": claims,
|
|
"observations": observations,
|
|
"transcript_event_ids": [event_id],
|
|
},
|
|
}
|
|
)
|
|
transcript_events.append(
|
|
{
|
|
"sequence": event_sequence,
|
|
"event_id": event_id,
|
|
"observed_at": completed,
|
|
"kind": "scenario_result",
|
|
"scenario": name,
|
|
"claims": claims,
|
|
"observations": observations,
|
|
"summary": f"observed {name}",
|
|
}
|
|
)
|
|
previous = completed
|
|
transcript_raw = b"".join(
|
|
(json.dumps(event, separators=(",", ":")) + "\n").encode()
|
|
for event in transcript_events
|
|
)
|
|
self.transcript.write_bytes(transcript_raw)
|
|
self.receipt.write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": 5,
|
|
"record_path": "records/secure-runtime-receipt-v5.json",
|
|
"started_at": started_at,
|
|
"completed_at": "2026-08-30T10:00:16+00:00",
|
|
"source_manifest": {
|
|
"schema_version": SOURCE_MANIFEST_SCHEMA_VERSION,
|
|
"manifest_id": SOURCE_MANIFEST_ID,
|
|
"path": SOURCE_MANIFEST_PATH,
|
|
"sha256": sha256_bytes(manifest_raw),
|
|
"target_os": "linux",
|
|
"target_arch": "arm64",
|
|
},
|
|
"source_hashes": self.source_hashes,
|
|
"artifact_hashes": artifact_hashes,
|
|
"artifact_versions": {
|
|
"collector_v1": "1.0.0",
|
|
"collector_v2": "1.1.0",
|
|
"collector_v3": "1.2.0",
|
|
"collector_v4": "1.3.0",
|
|
},
|
|
"disposable_vm_guard_sha256": DISPOSABLE_VM_GUARD_SHA256,
|
|
"os_release": 'PRETTY_NAME="Fixture Linux"',
|
|
"kernel": "Linux fixture",
|
|
"systemd_version": "systemd 255",
|
|
"architecture": "arm64",
|
|
"transcript": {
|
|
"format": "jsonl-v1",
|
|
"record_path": "records/secure-runtime-transcript-v5.jsonl",
|
|
"sha256": sha256_bytes(transcript_raw),
|
|
"event_count": len(transcript_events),
|
|
},
|
|
"collector_service_user": "pulse-agent",
|
|
"collector_process_uid": 1000,
|
|
"collector_authority": "monitoring-only",
|
|
"action_receipt_kind": "pulse.host_storage_cleanup_result",
|
|
"report_count": 3,
|
|
"first_report_at": "2026-08-30T10:00:01+00:00",
|
|
"last_report_at": "2026-08-30T10:00:15+00:00",
|
|
"scenarios": scenarios,
|
|
},
|
|
indent=2,
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
self.build_identity = mock.patch(
|
|
"secure_runtime_attestation.verify_artifact_build_identity",
|
|
return_value={
|
|
name: {"package": package, "vcs_revision": self.commit, "vcs_modified": "false"}
|
|
for name, package in EXPECTED_ARTIFACT_PACKAGES.items()
|
|
},
|
|
)
|
|
self.build_identity.start()
|
|
|
|
def tearDown(self) -> None:
|
|
self.build_identity.stop()
|
|
self.temporary.cleanup()
|
|
|
|
def git(self, *args: str) -> str:
|
|
return subprocess.run(
|
|
["git", *args],
|
|
cwd=self.repo,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
env=strip_local_git_env(os.environ.copy()),
|
|
).stdout.strip()
|
|
|
|
def attest(self, **overrides):
|
|
arguments = {
|
|
"checkout": self.repo,
|
|
"commit": self.commit,
|
|
"main_ref": "origin/main",
|
|
"receipt_path": self.receipt,
|
|
"receipt_record_path": "records/secure-runtime-receipt-v5.json",
|
|
"transcript_path": self.transcript,
|
|
"artifacts": self.artifacts,
|
|
"elapsed_seconds": 113.43,
|
|
}
|
|
arguments.update(overrides)
|
|
return create_attestation(**arguments)
|
|
|
|
def test_accepts_exact_committed_main_evidence(self) -> None:
|
|
result = self.attest()
|
|
self.assertEqual(result["proof_classification"], "committed-main-artifact-bound-self-attested-systemd")
|
|
self.assertEqual(result["qualified_commit"], self.commit)
|
|
self.assertTrue(result["source_hashes_match_commit"])
|
|
self.assertTrue(result["artifact_hashes_match_receipt"])
|
|
self.assertEqual(result["schema_version"], 5)
|
|
self.assertEqual(result["scenario_count"], 15)
|
|
self.assertTrue(result["receipt"]["path_bound_inside_receipt"])
|
|
self.assertEqual(result["transcript"]["event_count"], 16)
|
|
self.assertEqual(result["source_manifest"]["manifest_id"], SOURCE_MANIFEST_ID)
|
|
self.assertEqual(len(result["attestation_tool_sha256"]), 64)
|
|
|
|
def test_accepts_go_rfc3339nano_without_losing_fractional_order(self) -> None:
|
|
earlier = parse_utc_timestamp("2026-08-30T17:53:17.73472767Z", "earlier")
|
|
later = parse_utc_timestamp("2026-08-30T17:53:17.734727671Z", "later")
|
|
equivalent = parse_utc_timestamp("2026-08-30T17:53:17.734727670+00:00", "equivalent")
|
|
self.assertLess(earlier, later)
|
|
self.assertEqual(earlier, equivalent)
|
|
|
|
def test_rejects_non_utc_rfc3339nano(self) -> None:
|
|
with self.assertRaisesRegex(AttestationError, "must be UTC"):
|
|
parse_utc_timestamp("2026-08-30T18:53:17.73472767+01:00", "offset")
|
|
|
|
def test_accepts_exact_release_candidate_ref(self) -> None:
|
|
self.git("tag", "v9.0.0-rc.1", self.commit)
|
|
result = self.attest(release_candidate_ref="v9.0.0-rc.1")
|
|
self.assertEqual(result["proof_classification"], "release-candidate-artifact-bound-self-attested-systemd")
|
|
self.assertNotIn("exact-release-candidate", result["residual_proof"])
|
|
|
|
def test_rejects_attached_checkout(self) -> None:
|
|
self.git("checkout", "main")
|
|
with self.assertRaisesRegex(AttestationError, "must be detached"):
|
|
self.attest()
|
|
|
|
def test_rejects_tracked_checkout_change(self) -> None:
|
|
(self.repo / "scripts/install.sh").write_text("changed\n", encoding="utf-8")
|
|
with self.assertRaisesRegex(AttestationError, "tracked modifications"):
|
|
self.attest()
|
|
|
|
def test_rejects_untracked_source_outside_lab_artifacts(self) -> None:
|
|
(self.repo / "unexpected.go").write_text("package unexpected\n", encoding="utf-8")
|
|
with self.assertRaisesRegex(AttestationError, "unapproved untracked path"):
|
|
self.attest()
|
|
|
|
def test_rejects_source_digest_not_bound_to_commit(self) -> None:
|
|
receipt = json.loads(self.receipt.read_text(encoding="utf-8"))
|
|
receipt["source_hashes"]["scripts/install.sh"] = "0" * 64
|
|
self.receipt.write_text(json.dumps(receipt), encoding="utf-8")
|
|
with self.assertRaisesRegex(AttestationError, "source digest mismatch"):
|
|
self.attest()
|
|
|
|
def test_rejects_source_set_that_omits_manifest_expansion(self) -> None:
|
|
receipt = json.loads(self.receipt.read_text(encoding="utf-8"))
|
|
del receipt["source_hashes"]["boundary/runtime.go"]
|
|
self.receipt.write_text(json.dumps(receipt), encoding="utf-8")
|
|
with self.assertRaisesRegex(AttestationError, "exactly match the governed boundary manifest"):
|
|
self.attest()
|
|
|
|
def test_rejects_source_set_that_omits_update_signature_boundary(self) -> None:
|
|
receipt = json.loads(self.receipt.read_text(encoding="utf-8"))
|
|
del receipt["source_hashes"]["internal/updatesignature/signature.go"]
|
|
self.receipt.write_text(json.dumps(receipt), encoding="utf-8")
|
|
with self.assertRaisesRegex(AttestationError, "exactly match the governed boundary manifest"):
|
|
self.attest()
|
|
|
|
def test_rejects_artifact_digest_mismatch(self) -> None:
|
|
self.artifacts["runner"].write_bytes(b"tampered\n")
|
|
with self.assertRaisesRegex(AttestationError, "artifact digest mismatch"):
|
|
self.attest()
|
|
|
|
def test_rejects_wrong_go_command_package(self) -> None:
|
|
output = (
|
|
"fixture: go1.25\n"
|
|
"\tpath\tgithub.com/rcourtman/pulse-go-rewrite/cmd/not-pulse-agent\n"
|
|
f"\tbuild\tvcs=git\n\tbuild\tvcs.revision={self.commit}\n\tbuild\tvcs.modified=false\n"
|
|
)
|
|
with mock.patch("secure_runtime_attestation.subprocess.run", return_value=subprocess.CompletedProcess([], 0, output, "")):
|
|
with self.assertRaisesRegex(AttestationError, "expected Go command package"):
|
|
verify_artifact_build_identity(self.receipt_dict(), self.artifacts, self.commit)
|
|
|
|
def test_rejects_modified_go_build_identity(self) -> None:
|
|
def inspect(command, **_kwargs):
|
|
name = Path(command[-1]).name
|
|
output = (
|
|
"fixture: go1.25\n"
|
|
f"\tpath\t{EXPECTED_ARTIFACT_PACKAGES[name]}\n"
|
|
f"\tbuild\tvcs=git\n\tbuild\tvcs.revision={self.commit}\n\tbuild\tvcs.modified=true\n"
|
|
)
|
|
return subprocess.CompletedProcess(command, 0, output, "")
|
|
|
|
with mock.patch("secure_runtime_attestation.subprocess.run", side_effect=inspect):
|
|
with self.assertRaisesRegex(AttestationError, "modified checkout"):
|
|
verify_artifact_build_identity(self.receipt_dict(), self.artifacts, self.commit)
|
|
|
|
def test_rejects_failed_scenario(self) -> None:
|
|
receipt = json.loads(self.receipt.read_text(encoding="utf-8"))
|
|
receipt["scenarios"][-1]["passed"] = False
|
|
self.receipt.write_text(json.dumps(receipt), encoding="utf-8")
|
|
with self.assertRaisesRegex(AttestationError, "did not pass"):
|
|
self.attest()
|
|
|
|
def test_rejects_tampered_transcript(self) -> None:
|
|
self.transcript.write_bytes(self.transcript.read_bytes() + b"{}\n")
|
|
with self.assertRaisesRegex(AttestationError, "transcript digest mismatch"):
|
|
self.attest()
|
|
|
|
def test_rejects_scenario_chronology_drift(self) -> None:
|
|
receipt = json.loads(self.receipt.read_text(encoding="utf-8"))
|
|
receipt["scenarios"][2]["started_at"] = "2026-08-30T09:59:00+00:00"
|
|
self.receipt.write_text(json.dumps(receipt), encoding="utf-8")
|
|
with self.assertRaisesRegex(AttestationError, "chronology is invalid"):
|
|
self.attest()
|
|
|
|
def test_rejects_receipt_without_verified_mutation(self) -> None:
|
|
receipt = json.loads(self.receipt.read_text(encoding="utf-8"))
|
|
scenario = next(item for item in receipt["scenarios"] if item["name"] == "typed_action_receipt")
|
|
scenario["evidence"]["claims"].remove("typed_mutation_verified")
|
|
self.receipt.write_text(json.dumps(receipt), encoding="utf-8")
|
|
with self.assertRaisesRegex(AttestationError, "required causal claims"):
|
|
self.attest()
|
|
|
|
def test_rejects_scenario_without_typed_observation(self) -> None:
|
|
receipt = json.loads(self.receipt.read_text(encoding="utf-8"))
|
|
scenario = next(item for item in receipt["scenarios"] if item["name"] == "typed_action_receipt")
|
|
del scenario["evidence"]["observations"]["verification"]
|
|
self.receipt.write_text(json.dumps(receipt), encoding="utf-8")
|
|
with self.assertRaisesRegex(AttestationError, "required typed observations"):
|
|
self.attest()
|
|
|
|
def test_rejects_helper_update_without_accepted_report(self) -> None:
|
|
receipt = json.loads(self.receipt.read_text(encoding="utf-8"))
|
|
scenario = next(
|
|
item
|
|
for item in receipt["scenarios"]
|
|
if item["name"] == "helper_update_authoritative_commit"
|
|
)
|
|
scenario["evidence"]["observations"]["accepted_primary_report"] = False
|
|
self.receipt.write_text(json.dumps(receipt), encoding="utf-8")
|
|
with self.assertRaisesRegex(AttestationError, "required typed observations"):
|
|
self.attest()
|
|
|
|
def test_rejects_malformed_helper_update_digest(self) -> None:
|
|
self.rewrite_scenario_observations(
|
|
"helper_update_authoritative_commit",
|
|
lambda observations: observations.__setitem__("candidate_sha256", "not-a-digest"),
|
|
)
|
|
with self.assertRaisesRegex(AttestationError, "not a SHA-256 digest"):
|
|
self.attest()
|
|
|
|
def test_rejects_helper_update_commit_from_different_pid(self) -> None:
|
|
self.rewrite_scenario_observations(
|
|
"helper_update_authoritative_commit",
|
|
lambda observations: observations.__setitem__("committer_pid", 4322),
|
|
)
|
|
with self.assertRaisesRegex(AttestationError, "activated collector PID"):
|
|
self.attest()
|
|
|
|
def test_rejects_helper_update_commit_with_wrong_target_identity(self) -> None:
|
|
self.rewrite_scenario_observations(
|
|
"helper_update_authoritative_commit",
|
|
lambda observations: observations.__setitem__("target_sha256", "2" * 64),
|
|
)
|
|
with self.assertRaisesRegex(AttestationError, "target/LKG identities are inconsistent"):
|
|
self.attest()
|
|
|
|
def test_rejects_helper_update_identity_not_bound_to_artifact(self) -> None:
|
|
self.rewrite_scenario_observations(
|
|
"helper_update_authoritative_commit",
|
|
lambda observations: (
|
|
observations.__setitem__("prior_sha256", "f" * 64),
|
|
observations.__setitem__("last_known_good_sha256", "f" * 64),
|
|
),
|
|
)
|
|
with self.assertRaisesRegex(AttestationError, "not bound to the attested artifacts"):
|
|
self.attest()
|
|
|
|
def test_rejects_watchdog_rollback_with_wrong_lkg_identity(self) -> None:
|
|
self.rewrite_scenario_observations(
|
|
"helper_update_watchdog_rollback",
|
|
lambda observations: observations.__setitem__("last_known_good_sha256", "2" * 64),
|
|
)
|
|
with self.assertRaisesRegex(AttestationError, "restoration of the committed collector"):
|
|
self.attest()
|
|
|
|
def test_rejects_legacy_receipt_schema(self) -> None:
|
|
receipt = json.loads(self.receipt.read_text(encoding="utf-8"))
|
|
receipt["schema_version"] = 4
|
|
self.receipt.write_text(json.dumps(receipt), encoding="utf-8")
|
|
with self.assertRaisesRegex(AttestationError, "schema_version 5"):
|
|
self.attest()
|
|
|
|
def test_rejects_missing_scenario(self) -> None:
|
|
receipt = json.loads(self.receipt.read_text(encoding="utf-8"))
|
|
receipt["scenarios"].pop()
|
|
self.receipt.write_text(json.dumps(receipt), encoding="utf-8")
|
|
with self.assertRaisesRegex(AttestationError, "canonical 15-scenario"):
|
|
self.attest()
|
|
|
|
def test_rejects_mismatched_release_candidate_ref(self) -> None:
|
|
(self.repo / "later").write_text("later\n", encoding="utf-8")
|
|
self.git("add", "later")
|
|
self.git("commit", "-m", "later")
|
|
later = self.git("rev-parse", "HEAD")
|
|
self.git("tag", "v9.0.0-rc.2", later)
|
|
self.git("checkout", "--detach", self.commit)
|
|
with self.assertRaisesRegex(AttestationError, "does not resolve"):
|
|
self.attest(release_candidate_ref="v9.0.0-rc.2")
|
|
|
|
def test_rejects_commit_not_reachable_from_main(self) -> None:
|
|
self.git("checkout", "--orphan", "unreachable")
|
|
self.git("rm", "-rf", ".")
|
|
(self.repo / "orphan").write_text("orphan\n", encoding="utf-8")
|
|
self.git("add", "orphan")
|
|
self.git("commit", "-m", "orphan")
|
|
orphan = self.git("rev-parse", "HEAD")
|
|
self.git("checkout", "--detach", orphan)
|
|
with self.assertRaisesRegex(AttestationError, "not reachable"):
|
|
self.attest(commit=orphan)
|
|
|
|
def test_rejects_sensitive_receipt_keys(self) -> None:
|
|
receipt = json.loads(self.receipt.read_text(encoding="utf-8"))
|
|
receipt["secret"] = "not-allowed"
|
|
self.receipt.write_text(json.dumps(receipt), encoding="utf-8")
|
|
with self.assertRaisesRegex(AttestationError, "forbidden sensitive key"):
|
|
self.attest()
|
|
|
|
def test_rejects_noncanonical_receipt_record_path(self) -> None:
|
|
with self.assertRaisesRegex(AttestationError, "canonical repository-relative"):
|
|
self.attest(receipt_record_path="../receipt.json")
|
|
|
|
def test_rejects_receipt_record_path_not_bound_inside_blob(self) -> None:
|
|
with self.assertRaisesRegex(AttestationError, "not bound inside"):
|
|
self.attest(receipt_record_path="records/other.json")
|
|
|
|
def test_rejects_nonfinite_elapsed_time(self) -> None:
|
|
with self.assertRaisesRegex(AttestationError, "must be positive"):
|
|
self.attest(elapsed_seconds=float("nan"))
|
|
|
|
def receipt_dict(self):
|
|
return json.loads(self.receipt.read_text(encoding="utf-8"))
|
|
|
|
def rewrite_scenario_observations(self, scenario_name, mutate) -> None:
|
|
receipt = self.receipt_dict()
|
|
scenario = next(
|
|
item for item in receipt["scenarios"] if item["name"] == scenario_name
|
|
)
|
|
observations = scenario["evidence"]["observations"]
|
|
mutate(observations)
|
|
events = [
|
|
json.loads(line)
|
|
for line in self.transcript.read_text(encoding="utf-8").splitlines()
|
|
]
|
|
event = next(
|
|
item
|
|
for item in events
|
|
if item.get("kind") == "scenario_result"
|
|
and item.get("scenario") == scenario_name
|
|
)
|
|
event["observations"] = observations
|
|
transcript_raw = b"".join(
|
|
(json.dumps(item, separators=(",", ":")) + "\n").encode()
|
|
for item in events
|
|
)
|
|
self.transcript.write_bytes(transcript_raw)
|
|
receipt["transcript"]["sha256"] = sha256_bytes(transcript_raw)
|
|
self.receipt.write_text(json.dumps(receipt), encoding="utf-8")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|