diff --git a/cmd/gvisor-start-probe/ids_linux.go b/cmd/gvisor-start-probe/ids_linux.go new file mode 100644 index 0000000000..cfb4cbe2cc --- /dev/null +++ b/cmd/gvisor-start-probe/ids_linux.go @@ -0,0 +1,31 @@ +//go:build linux + +package main + +import ( + "errors" + "fmt" + "strings" +) + +// containerIDs is an exact allowlist for controlled trials, not a runtime +// inventory. Never accept prefixes or derive this list from receiver output. +type containerIDs map[string]struct{} + +func (ids *containerIDs) Set(id string) error { + if id == "" || len(id) > 256 || strings.ContainsAny(id, " \t\r\n") { + return errors.New("container-id must be a nonempty exact runtime ID of at most 256 bytes without whitespace") + } + if *ids == nil { + *ids = make(containerIDs) + } + (*ids)[id] = struct{}{} + return nil +} + +func (ids *containerIDs) String() string { return fmt.Sprintf("%d exact runtime IDs", len(*ids)) } + +func (ids containerIDs) resolves(id string) bool { + _, ok := ids[id] + return ok +} diff --git a/cmd/gvisor-start-probe/ids_linux_test.go b/cmd/gvisor-start-probe/ids_linux_test.go new file mode 100644 index 0000000000..7dacf5b403 --- /dev/null +++ b/cmd/gvisor-start-probe/ids_linux_test.go @@ -0,0 +1,46 @@ +//go:build linux + +package main + +import ( + "flag" + "strings" + "testing" +) + +func TestContainerIDsExactMatching(t *testing.T) { + var ids containerIDs + flags := flag.NewFlagSet("trial", flag.ContinueOnError) + flags.Var(&ids, "container-id", "runtime identity") + if err := flags.Parse([]string{"--container-id", "sandbox-a", "--container-id", "sandbox-b", "--container-id", "sandbox-a"}); err != nil { + t.Fatal(err) + } + if len(ids) != 2 { + t.Fatalf("got %d identities, want 2", len(ids)) + } + for _, id := range []string{"sandbox-a", "sandbox-b"} { + if !ids.resolves(id) { + t.Fatal("known ID rejected") + } + } + for _, id := range []string{"", "sandbox", "sandbox-a-suffix", "unknown"} { + if ids.resolves(id) { + t.Fatal("unverified ID accepted") + } + } +} + +func TestContainerIDsRejectInvalidValues(t *testing.T) { + var ids containerIDs + for _, id := range []string{"", "with space", "with\nnewline", strings.Repeat("a", 257)} { + if err := ids.Set(id); err == nil { + t.Fatal("invalid ID accepted") + } + } + if len(ids) != 0 { + t.Fatal("invalid IDs changed allowlist") + } + if err := ids.Set(strings.Repeat("a", 256)); err != nil { + t.Fatal(err) + } +} diff --git a/cmd/gvisor-start-probe/main_linux.go b/cmd/gvisor-start-probe/main_linux.go index 348e688420..5a90c231f4 100644 --- a/cmd/gvisor-start-probe/main_linux.go +++ b/cmd/gvisor-start-probe/main_linux.go @@ -22,9 +22,10 @@ func main() { func run() (exitCode int) { socket := flag.String("socket", "/run/kubescape/gvisor-events.sock", "private Unix socket for the SecCheck remote sink") - containerID := flag.String("container-id", "", "exact container ID obtained independently from the local runtime before start") + var ids containerIDs + flag.Var(&ids, "container-id", "exact container ID obtained independently from the local runtime before start; repeat for multiple containers") flag.Parse() - if *containerID == "" { + if len(ids) == 0 { fmt.Fprintln(os.Stderr, "container-id is required") return 2 } @@ -46,9 +47,7 @@ func run() (exitCode int) { outputErrors := make(chan error, 1) receiver := &gvisor.Receiver{ SocketPath: *socket, - Resolve: func(id string) bool { - return id == *containerID - }, + Resolve: ids.resolves, OnStart: func(ctx context.Context, start gvisor.Start) { if err := output.Encode(ctx, start); err != nil { if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || diff --git a/docs/gvisor-start-probe.md b/docs/gvisor-start-probe.md index eaf4a60c39..0d90c2732b 100644 --- a/docs/gvisor-start-probe.md +++ b/docs/gvisor-start-probe.md @@ -94,3 +94,126 @@ This document records the procedure, not results. Live `runsc` and node-agent host eBPF observations must be added after the Linux trial; the existing [gVisor proof of concept](https://github.com/yellow-forrest/gvisor-visibility-poc) is useful prior work but is not validation of this receiver. + +## Repeated and concurrent starts + +The probe accepts repeated `--container-id` flags. Each value is matched in +full, never as a prefix. The original single-ID invocation still works. +Obtain all IDs from the local runtime before starting the selected containers. +Reuse the receiver-owned `/run/kubescape` directory (mode `0700`) and the +trace configuration from the controlled trial above. The sink endpoint, probe +and readiness check all use `/run/kubescape/gvisor-events.sock`; `TRIAL_DIR` +stores evidence only. If you choose a different socket, replace that literal +with the same absolute path in all three places. JSON does not expand shell +variables, so the trace configuration must contain the actual pathname. + +Create a private evidence directory before redirecting output: + +```sh +umask 077 +TRIAL_DIR="$(mktemp -d)" +./gvisor-start-probe \ + --socket /run/kubescape/gvisor-events.sock \ + --container-id "$CONTAINER_A" \ + --container-id "$CONTAINER_B" \ + > "$TRIAL_DIR/starts.jsonl" 2> "$TRIAL_DIR/probe.stderr" +``` + +Run the following readiness check in a second terminal while the probe runs. +Socket existence alone is not readiness: the receiver binds the path +before calling `listen`. Before starting either container, require a successful +connection and version-1 handshake with a bounded retry, for example: + +```sh +python3 - /run/kubescape/gvisor-events.sock <<'PYREADY' +import socket +import sys +import time + +deadline = time.monotonic() + 5 +while time.monotonic() < deadline: + with socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) as connection: + connection.settimeout(0.2) + try: + connection.connect(sys.argv[1]) + connection.sendall(b"\x08\x01") # Version-1 protobuf handshake. + if connection.recv(32) == b"\x08\x01": + break + except OSError: + pass + time.sleep(0.05) +else: + raise SystemExit("receiver readiness handshake timed out") +PYREADY +``` + +The check closes its connection without sending a start and produces no trial +event. Abort the trial if the check fails. It only establishes that the receiver +answered; keep checking that the probe is running while the workload starts. +Stop it with SIGTERM after the workload and wait for a zero exit status before +checking artifacts. A nonzero exit invalidates that run. + +For twenty repeated starts, either collect each independently created ID in +advance and repeat the flag for every ID, or run separate probe invocations +and combine only the sanitized start JSONL files. Keep a runtime-derived +expected-count manifest for exactly those invocations. For the concurrent +case use one receiver with two IDs, overlapping sandbox lifetimes, and runtime +records that demonstrate the overlap. Two sequential starts are not evidence +of concurrency. The ID flag remains a manual trial allowlist; it does not +implement the production runtime inventory join. + +## Check retained evidence + +Use Python 3.9 or newer. Create `expected.json` from independently recorded +runtime starts, not from the probe output. Its keys must be exact runtime IDs +and its values the expected number of starts. For example, replace these +placeholders with the two real IDs: + +```json +{"exact-runtime-id-a": 1, "exact-runtime-id-b": 1} +``` + +Use a harmless marker matching `[A-Za-z0-9_-]{16,}`, such as +`GVS_TRIAL_CANARY_2026`, in the root and child container argv and cwd fixtures. +Only ASCII letters, digits, underscores and hyphens are accepted, so ordinary +JSON string encoding leaves the marker literal. Quotes, backslashes, whitespace +and non-ASCII markers are rejected before reading evidence. Do not use real +credentials. After the probe has stopped, check each run: + +```sh +python3 scripts/check-gvisor-start-trial.py \ + --expected "$TRIAL_DIR/expected.json" \ + --events "$TRIAL_DIR/starts.jsonl" \ + --canary GVS_TRIAL_CANARY_2026 \ + --artifact "$TRIAL_DIR/probe.stderr" +``` + +Repeat `--artifact` for additional retained logs or exports. The checker +rejects missing or excess start counts, unknown identities, unexpected event +fields, invalid +source/timestamps, blank JSONL records, nonzero sender-reported drop counts, +and the canary in the specified artifacts. Each input file is limited to +16 MiB. It prints neither +input records nor parser error payloads. It does not inspect unspecified +files, prove that the canary was actually used, verify the manifest's origin, +measure local queue drops, or prove runtime identity from a JSON file. JSON +record counts cannot prove timing, concurrency, or per-connection isolation. +If a manifest expects two starts for one ID, two identical records satisfy that +count; the checker does not authenticate events or distinguish replayed copies +from distinct starts. Independent runtime evidence is still required. +A passing check is an artifact consistency check, not a lifecycle integration +gate or proof that application code executed. + +Fill in [the trial record](gvisor-start-trial-record.md) with version details, +runtime ground truth, privacy checks, host eBPF observations, and limitations. +Keep the synthetic fixture separate from sanitized evidence. Debug/strace +output can itself include argv or cwd; inspect it privately and sanitize it +before retaining or sharing. The checker can detect a known marker, not every +possible secret or encoding. A failed privacy check means the artifacts must +not be published. + +The checker can be tested without `runsc`: + +```sh +python3 -m unittest discover -s scripts -p test_check_gvisor_start_trial.py +``` diff --git a/docs/gvisor-start-trial-record.md b/docs/gvisor-start-trial-record.md new file mode 100644 index 0000000000..12b529d7bd --- /dev/null +++ b/docs/gvisor-start-trial-record.md @@ -0,0 +1,71 @@ +# gVisor start trial record + +**Status: NOT RUN.** This is a template for the live experiment from +[design #23](https://github.com/kubescape/designs-and-proposals/pull/23). +Neither the merged receiver nor the offline checker establishes live coverage. +Copy this record for a real trial and replace pending entries with measured +results and sanitized evidence references. Do not mark a gate passed from a +synthetic sender test or an offline checker result alone. + +## Environment and ownership + +| Item | Recorded value | +|---|---| +| Trial date and operator | Pending | +| Linux distribution, kernel and architecture | Pending | +| Exact runsc version/commit and binary checksum | Pending | +| containerd and CRI versions | Pending | +| node-agent version/commit and configuration | Pending | +| Probe version/commit | Pending | +| Runtime root and namespace | Pending | +| Sandbox/root/child ID relationship from runtime inventory | Pending | +| Selected trace metadata and initialization configuration | Pending | +| Receiver owner, socket permissions and runtime access | Pending | +| Owner of Default trace session, conflict behavior | Pending | + +Record actual commands against the pinned runtime and any deviations from +[the procedure](gvisor-start-probe.md). Configure tracing before startup and +never overwrite another monitor's session with `--force`. Trace configuration +is host setup, not something this branch installs. + +## Trial gates + +| Scenario | Expected evidence | Result | +|---|---|---| +| Ordinary container baseline | Runtime IDs/times and host process/network observations | Pending | +| One gVisor container | Independently recorded runtime ID/times, trace start and host comparison | Pending | +| Twenty repeated starts | Expected vs observed per-ID counts, 20/20 starts without unexplained misses | Pending | +| Two concurrent sandboxes | Overlapping runtime lifetimes, distinct IDs/connections, no identity crossover | Pending | +| Root and child privacy fixtures | Canary used in both argv and cwd; absent from retained/forwarded output and errors | Pending | +| Receiver absent and slow | Startup impact, retry delay, sender/local loss, bounded resources | Pending | +| Receiver restart | Trace-session ownership, recovery behavior and any missed early starts | Pending | +| Normal exit and forced termination | Runtime state vs trace transport; no stop inferred from disconnect alone | Pending | +| Controlled outbound, refused connection and DNS | Compare runtime/trace/host evidence and name only the observed behavior | Pending | + +## Sanitized evidence and interpretation + +- Runtime-derived expected manifest and provenance: pending. +- Probe exit status, checker invocation and result: pending. +- Sanitized observation files, including stderr and privacy review: pending. +- Versioned workload and harmless canary fixture: pending. +- Expected/observed starts and timing comparison: pending. +- Sender-reported loss and separately measured local loss: pending. +- Host eBPF comparison, including what could not be observed: pending. +- Startup latency and CPU/memory measurements with method: pending. +- Failures, missing events and unexplained differences: pending. + +The start hook precedes application execution. Receiver timestamps are +observation times, not runtime start timestamps. Correlating exact IDs does +not establish actor/session identity or Kubernetes attribution. The current +probe uses a manually supplied allowlist, not node-agent's runtime inventory. + +## Decision before integration + +State which gates passed, failed or remain untested, with supporting evidence. +Unexplained missed starts, identity ambiguity or privacy failures block the +verified-start integration. Source-health gaps must be recorded and addressed +before claiming reliable collection. Network and stop integration require +separate source validation. Managed GKE remains unverified unless its runtime +configuration and socket access are independently demonstrated. + +**Current decision: pending live Linux trial.** diff --git a/scripts/check-gvisor-start-trial.py b/scripts/check-gvisor-start-trial.py new file mode 100644 index 0000000000..876d28d039 --- /dev/null +++ b/scripts/check-gvisor-start-trial.py @@ -0,0 +1,109 @@ +#!/usr/bin/env python3 +"""Check sanitized probe records against independently recorded runtime starts. + +This does not authenticate the runtime manifest or prove any live-trial gate. +Never include input records, IDs, canaries, or parser errors in diagnostics. +""" +import argparse +from collections import Counter +from datetime import datetime +import json +from pathlib import Path +import re +import sys + +MAX_BYTES = 16 * 1024 * 1024 +FIELDS = {"Source", "ContainerID", "ObservedAt", "Dropped"} + + +class InvalidEvidence(ValueError): + pass + + +def unique_object(pairs): + result = {} + for key, value in pairs: + if key in result: + raise InvalidEvidence("duplicate JSON field") + result[key] = value + return result + + +def load_json(data): + return json.loads(data, object_pairs_hook=unique_object) + + +def read_artifact(path, canary): + with Path(path).open("rb") as artifact: + data = artifact.read(MAX_BYTES + 1) + if len(data) > MAX_BYTES: + raise InvalidEvidence("artifact exceeds checker size limit") + if canary.encode() in data: + raise InvalidEvidence("privacy canary found in retained artifact") + return data + + +def check(expected_path, events_path, canary, artifacts=()): + if re.fullmatch(r"[A-Za-z0-9_-]{16,}", canary) is None: + raise InvalidEvidence("use a synthetic canary of at least 16 ASCII letters, digits, underscores or hyphens") + expected = load_json(read_artifact(expected_path, canary)) + if not isinstance(expected, dict) or not expected: + raise InvalidEvidence("expected starts must be a nonempty ID-to-count object") + if any(not key or type(count) is not int or count < 1 + for key, count in expected.items()): + raise InvalidEvidence("expected start counts must be positive integers") + if canary in json.dumps(expected): + raise InvalidEvidence("privacy canary found in decoded manifest") + observed = Counter() + for line in read_artifact(events_path, canary).splitlines(): + if not line.strip(): + raise InvalidEvidence("blank JSONL record") + event = load_json(line) + if not isinstance(event, dict) or set(event) != FIELDS: + raise InvalidEvidence("start record has missing or unexpected fields") + if canary in json.dumps(event): + raise InvalidEvidence("privacy canary found in decoded start record") + if event["Source"] != "gvisor_trace": + raise InvalidEvidence("unexpected event source") + if not isinstance(event["ContainerID"], str) or event["ContainerID"] not in expected: + raise InvalidEvidence("start record has an unexpected identity") + if type(event["Dropped"]) is not int or event["Dropped"] != 0: + raise InvalidEvidence("start record reports invalid or nonzero sender drops") + if not isinstance(event["ObservedAt"], str): + raise InvalidEvidence("invalid observation timestamp") + match = re.fullmatch(r"(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(\.\d{1,9})?(Z|[+-](?:[01][0-9]|2[0-3]):[0-5][0-9])", event["ObservedAt"]) + if match is None: + raise InvalidEvidence("observation timestamp must be RFC3339 with a timezone") + # Python 3.9 accepts microseconds; Go emits up to nine fractional digits. + fraction = (match[2] or "")[:7] + datetime.fromisoformat(match[1] + fraction + match[3].replace("Z", "+00:00")) + observed[event["ContainerID"]] += 1 + for artifact in artifacts: + read_artifact(artifact, canary) + if observed != Counter(expected): + raise InvalidEvidence("observed start counts do not match runtime manifest") + return sum(observed.values()) + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--expected", required=True, help="runtime-derived JSON ID-to-count manifest") + parser.add_argument("--events", required=True, help="probe JSONL stdout") + parser.add_argument("--canary", required=True, help="synthetic marker matching [A-Za-z0-9_-]{16,}, never a real secret") + parser.add_argument("--artifact", action="append", default=[], help="additional retained file to scan for the canary; repeatable") + args = parser.parse_args(argv) + try: + count = check(args.expected, args.events, args.canary, args.artifact) + except InvalidEvidence as error: + print(f"FAIL: {error}", file=sys.stderr) + return 1 + except (OSError, ValueError, TypeError, UnicodeError, OverflowError, RecursionError): + print("FAIL: unreadable or malformed evidence (input omitted)", file=sys.stderr) + return 1 + print(f"PASS: {count} retained start records match expected counts with zero reported sender drops.") + print("Live runtime provenance, concurrency, timing, local queue loss, and deployment gates still require review.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/test_check_gvisor_start_trial.py b/scripts/test_check_gvisor_start_trial.py new file mode 100644 index 0000000000..72f2c5e7aa --- /dev/null +++ b/scripts/test_check_gvisor_start_trial.py @@ -0,0 +1,145 @@ +import contextlib +import importlib.util +import io +import json +from pathlib import Path +import tempfile +import unittest + +spec = importlib.util.spec_from_file_location("trial", Path(__file__).with_name("check-gvisor-start-trial.py")) +trial = importlib.util.module_from_spec(spec) +spec.loader.exec_module(trial) +CANARY = "GVS_TRIAL_CANARY_2026" + + +class TrialCheckTests(unittest.TestCase): + def setUp(self): + self.directory = tempfile.TemporaryDirectory() + self.addCleanup(self.directory.cleanup) + self.root = Path(self.directory.name) + self.expected = self.root / "expected.json" + self.events = self.root / "events.jsonl" + self.expected.write_text(json.dumps({"first": 1, "second": 1})) + self.records = [dict(Source="gvisor_trace", ContainerID=id, + ObservedAt="2026-10-22T10:00:00.123456789Z", Dropped=0) + for id in ["first", "second"]] + + def run_check(self, records=None, artifacts=()): + self.events.write_text("\n".join(json.dumps(record) for record in + (self.records if records is None else records))) + return trial.check(self.expected, self.events, CANARY, artifacts) + + def test_two_identities_and_twenty_starts(self): + self.assertEqual(self.run_check(), 2) + self.expected.write_text(json.dumps({str(i): 1 for i in range(20)})) + self.assertEqual(self.run_check([dict(self.records[0], ContainerID=str(i)) for i in range(20)]), 20) + + def test_missing_duplicate_and_unknown_starts(self): + for records in [[], self.records[:1], self.records + self.records[:1], + [dict(self.records[0], ContainerID="unknown")]]: + with self.subTest(records=len(records)), self.assertRaises(trial.InvalidEvidence): + self.run_check(records) + + def test_source_loss_timestamp_and_unexpected_fields(self): + for update in [dict(Source="ebpf"), dict(Dropped=1), dict(Dropped=True), + dict(ObservedAt="2026-10-22T10:00:00"), dict(Args=["harmless"]), + dict(ContainerID=["first"])]: + with self.subTest(update=update), self.assertRaises(trial.InvalidEvidence): + self.run_check([dict(self.records[0], **update), self.records[1]]) + + def test_privacy_marker_in_additional_artifact(self): + artifact = self.root / "stderr.txt" + artifact.write_text(CANARY) + with self.assertRaises(trial.InvalidEvidence): + self.run_check(artifacts=[artifact]) + + def test_no_payload_in_failure_diagnostics(self): + for payload in [CANARY, '{"Args":"' + CANARY + '"}', + '{"Source":"gvisor_trace","Source":"ebpf"}', + json.dumps(dict(self.records[0], ContainerID=CANARY))]: + self.events.write_text(payload) + output = io.StringIO() + with contextlib.redirect_stderr(output): + self.assertEqual(trial.main(["--expected", str(self.expected), "--events", str(self.events), "--canary", CANARY]), 1) + self.assertNotIn(CANARY, output.getvalue()) + self.assertNotIn(payload, output.getvalue()) + + def test_canary_alphabet_rejects_json_escaped_markers(self): + artifact = self.root / "retained.json" + self.run_check() + for suffix in ['"', "\\", "\n", "\t", " ", "é"]: + canary = "A" * 15 + suffix + artifact.write_text(json.dumps({"argv": [canary]})) + with self.subTest(suffix=suffix), self.assertRaises(trial.InvalidEvidence): + trial.check(self.expected, self.events, canary, [artifact]) + + def test_invalid_canary_in_matching_escaped_identities(self): + for suffix in ['"', "\\"]: + canary = "A" * 15 + suffix + self.expected.write_text(json.dumps({canary: 1})) + self.events.write_text(json.dumps(dict(self.records[0], ContainerID=canary))) + with self.subTest(suffix=suffix), self.assertRaisesRegex(trial.InvalidEvidence, "use a synthetic canary"): + trial.check(self.expected, self.events, canary) + + def test_supported_canary_alphabet(self): + self.run_check() + for canary in ["aZ09_-" * 3, "A" * 16]: + with self.subTest(canary=canary): + self.assertEqual(trial.check(self.expected, self.events, canary), 2) + with self.assertRaises(trial.InvalidEvidence): + trial.check(self.expected, self.events, "A" * 15) + + def test_literal_canary_in_json_artifact_is_rejected(self): + artifact = self.root / "retained.json" + artifact.write_text(json.dumps({"argv": [CANARY]})) + with self.assertRaises(trial.InvalidEvidence): + self.run_check(artifacts=[artifact]) + + def test_escaped_identity_canary_is_rejected(self): + # Exercise decoded checks even when every marker byte is JSON-escaped. + escaped = "".join("\\u%04x" % ord(char) for char in CANARY) + self.expected.write_text(json.dumps({CANARY: 1}).replace(CANARY, escaped)) + self.events.write_text(json.dumps(dict(self.records[0], ContainerID=CANARY)).replace(CANARY, escaped)) + with self.assertRaises(trial.InvalidEvidence): + trial.check(self.expected, self.events, CANARY) + # A marker in an event also fails if it is absent from the manifest. + self.expected.write_text(json.dumps({"first": 1})) + with self.assertRaisesRegex(trial.InvalidEvidence, "privacy canary"): + trial.check(self.expected, self.events, CANARY) + + def test_canary_validation_does_not_echo_marker(self): + canary = "A" * 15 + '"' + output = io.StringIO() + with contextlib.redirect_stderr(output): + status = trial.main(["--expected", str(self.expected), "--events", str(self.events), "--canary", canary]) + self.assertEqual(status, 1) + self.assertNotIn(canary, output.getvalue()) + self.assertIn("canary", output.getvalue()) + + def test_blank_records_fail_but_final_newline_is_valid(self): + self.run_check() + records = self.events.read_text() + for contents in ["\n" + records, records.replace("\n", "\n\n"), records + "\n \t\n"]: + self.events.write_text(contents) + with self.subTest(contents=contents), self.assertRaises(trial.InvalidEvidence): + trial.check(self.expected, self.events, CANARY) + self.events.write_text(records + "\n") + self.assertEqual(trial.check(self.expected, self.events, CANARY), 2) + + def test_timestamp_offsets(self): + for offset in ["+00:60", "-00:60", "+24:00", "+12:99"]: + with self.subTest(offset=offset), self.assertRaises(ValueError): + self.run_check([dict(self.records[0], ObservedAt="2026-10-22T10:00:00" + offset), self.records[1]]) + for offset in ["Z", "+00:00", "-05:30", "+23:59"]: + with self.subTest(offset=offset): + self.assertEqual(self.run_check([dict(self.records[0], ObservedAt="2026-10-22T10:00:00.123456789" + offset), self.records[1]]), 2) + + def test_invalid_expected_counts(self): + for expected in [{}, {"first": 0}, {"first": True}, {"first": -1}]: + self.expected.write_text(json.dumps(expected)) + with self.subTest(expected=expected), self.assertRaises(trial.InvalidEvidence): + self.run_check() + + +if __name__ == "__main__": + unittest.main()