"""Tests for the bounded, metadata-only G7 topology sampler.""" from __future__ import annotations import importlib.util import hashlib import json import sys import unittest from pathlib import Path from typing import Sequence RUNNER_PATH = Path(__file__).with_name("capture-g7-topology-evidence.py") REPO_ROOT = RUNNER_PATH.resolve().parents[1] def file_sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def load_runner(): spec = importlib.util.spec_from_file_location("g7_topology_evidence", RUNNER_PATH) if spec is None or spec.loader is None: raise RuntimeError("G7 topology evidence sampler could not be loaded") module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module API_ID = "a" * 64 CADDY_ID = "c" * 64 API_DIGEST = "sha256:" + "1" * 64 CADDY_DIGEST = "sha256:" + "2" * 64 API_EXE_DIGEST = "3" * 64 CLOUDFLARED_EXE_DIGEST = "4" * 64 GIT_SHA = "5" * 40 GIT_TREE_SHA = "8" * 40 PSUTIL_VERSION = "6.1.1" SCRIPT_DIGESTS = { "runner": file_sha256(REPO_ROOT / "scripts/run-g7-external-proof-window.py"), "collector": file_sha256(REPO_ROOT / "scripts/capture-g7-topology-evidence.py"), "checker": file_sha256(REPO_ROOT / "scripts/check-g7-external-proof.py"), } def inspect_item( *, container_id: str, image_digest: str, service: str, pid: int, restart_count: int = 0, started_at: str = "2026-08-07T00:00:00Z", ) -> dict[str, object]: return { "Id": container_id, "Name": f"/vignette-{service}-1", "Image": image_digest, "RestartCount": restart_count, "Config": { "Labels": { "com.docker.compose.project": "vignette-preview-20260807", "com.docker.compose.service": service, } }, "State": {"Running": True, "Pid": pid, "StartedAt": started_at}, } def stable_inspect() -> str: return json.dumps( [ inspect_item( container_id=API_ID, image_digest=API_DIGEST, service="api", pid=101, ), inspect_item( container_id=CADDY_ID, image_digest=CADDY_DIGEST, service="caddy", pid=202, ), ] ) def stats_output() -> str: return "\n".join( [ json.dumps( { "ID": API_ID[:12], "CPUPerc": "1.25%", "MemUsage": "128MiB / 2GiB", "MemPerc": "6.25%", "NetIO": "1.5MB / 2MB", "BlockIO": "4KiB / 8KiB", "PIDs": "9", } ), json.dumps( { "ID": CADDY_ID[:12], "CPUPerc": "0.50%", "MemUsage": "32MiB / 1GiB", "MemPerc": "3.12%", "NetIO": "3MB / 4MB", "BlockIO": "0B / 1KiB", "PIDs": "4", } ), ] ) SS_OUTPUT = """State Recv-Q Send-Q Local Address:Port Peer Address:Port Process ESTAB 3 5 10.0.0.1:443 10.0.0.2:50000 cubic wscale:7,7 rto:204 retrans:1/4 cwnd:10 LISTEN 0 128 0.0.0.0:8080 0.0.0.0:* cubic cwnd:10 """ class FixtureRunner: def __init__( self, *, inspect_outputs: list[str] | None = None, git_outputs: list[str] | None = None, git_branch: str = "HEAD\n", git_status: str = "", git_tree_sha: str = GIT_TREE_SHA, unavailable: set[str] | None = None, ) -> None: self.inspect_outputs = list(inspect_outputs or [stable_inspect()]) self.git_outputs = list(git_outputs or [GIT_SHA + "\n"]) self.git_branch = git_branch self.git_status = git_status self.git_tree_sha = git_tree_sha self.unavailable = unavailable or set() self.calls: list[tuple[str, ...]] = [] def available(self, command: str) -> bool: return command not in self.unavailable def run(self, args: Sequence[str]) -> str: command = tuple(args) self.calls.append(command) if command[:3] == ("docker", "inspect", "--type"): if len(self.inspect_outputs) > 1: return self.inspect_outputs.pop(0) return self.inspect_outputs[0] if command[:3] == ("docker", "stats", "--no-stream"): return stats_output() if command == ("ss", "-tinm"): return SS_OUTPUT if command and command[0] == "git": operation = command[3:] if operation == ("rev-parse", "--abbrev-ref", "HEAD"): return self.git_branch if operation == ( "status", "--porcelain=v1", "--untracked-files=no", ): return self.git_status if operation == ("rev-parse", "--verify", "HEAD^{tree}"): return self.git_tree_sha + "\n" if len(self.git_outputs) > 1: return self.git_outputs.pop(0) return self.git_outputs[0] raise AssertionError(f"unexpected command: {command!r}") class FixtureSource: def __init__(self) -> None: self.files = { "/proc/101/cgroup": "0::/docker/api.scope\n", "/proc/202/cgroup": "0::/docker/caddy.scope\n", "/proc/101/status": "VmRSS:\t100 kB\nVmHWM:\t120 kB\nThreads:\t3\n", "/proc/202/status": "VmRSS:\t50 kB\nVmHWM:\t70 kB\nThreads:\t2\n", "/sys/fs/cgroup/docker/api.scope/memory.current": "104857600\n", "/sys/fs/cgroup/docker/api.scope/memory.peak": "125829120\n", "/sys/fs/cgroup/docker/api.scope/cpu.stat": ( "usage_usec 1000\nuser_usec 700\nsystem_usec 300\n" "nr_periods 10\nnr_throttled 1\nthrottled_usec 20\n" ), "/sys/fs/cgroup/docker/api.scope/pids.current": "9\n", "/sys/fs/cgroup/docker/caddy.scope/memory.current": "33554432\n", "/sys/fs/cgroup/docker/caddy.scope/memory.peak": "41943040\n", "/sys/fs/cgroup/docker/caddy.scope/cpu.stat": ( "usage_usec 500\nuser_usec 300\nsystem_usec 200\n" ), "/sys/fs/cgroup/docker/caddy.scope/pids.current": "4\n", } self.directories = { "/proc/101/fd": ["0", "1", "2", "3"], "/proc/202/fd": ["0", "1", "2"], } def read_text(self, path: str) -> str: return self.files[path] def list_names(self, path: str) -> list[str]: return list(self.directories[path]) class FixtureWindowsSource: def __init__( self, module, *, identity_overrides: dict[str, list[dict[str, object]]] | None = None, tcp_overrides: dict[str, object] | None = None, psutil_version: str = PSUTIL_VERSION, ) -> None: self.module = module self.identity_overrides = identity_overrides or {} self.tcp_overrides = tcp_overrides or {} self.psutil_version = psutil_version self.metric_calls = {"api": 0, "cloudflared": 0} def dependency_version(self, name: str) -> str: if name != "psutil": raise AssertionError(f"unexpected dependency: {name}") return self.psutil_version def _base_identity(self, role: str, git_sha: str) -> dict[str, object]: if role == "api": return { "role": role, "pid": 301, "started_at": "2026-08-07T00:00:00Z", "executable_name": "python.exe", "executable_sha256": API_EXE_DIGEST, "command_line_sha256": "6" * 64, "git_sha": git_sha, "cwd": r"D:\workspace\vignette\apps\api", } return { "role": role, "pid": 302, "started_at": "2026-08-07T00:00:00Z", "executable_name": "cloudflared.exe", "executable_sha256": CLOUDFLARED_EXE_DIGEST, "command_line_sha256": "7" * 64, "git_sha": git_sha, "cwd": r"D:\workspace\vignette", } def read_identity(self, *, pid: int, role: str, git_sha: str): values = self._base_identity(role, git_sha) overrides = self.identity_overrides.get(role, []) if overrides: values.update(overrides[0]) if len(overrides) > 1: overrides.pop(0) self.assert_pid(role, pid) return self.module.WindowsProcessIdentity(**values) def assert_pid(self, role: str, pid: int) -> None: expected = 301 if role == "api" else 302 if pid != expected: raise AssertionError(f"unexpected {role} pid: {pid}") def read_metrics(self, *, pid: int, role: str) -> dict[str, int | float]: self.assert_pid(role, pid) self.metric_calls[role] += 1 factor = 2 if role == "api" else 1 return { "rss_bytes": 100_000 * factor, "peak_rss_bytes": 120_000 * factor, "cpu_time_seconds": float(self.metric_calls[role] * factor), "cpu_percent": 1.5 * factor, "handles": 20 * factor, "threads": 4 * factor, } def read_tcp_metrics( self, *, api_pid: int, cloudflared_pid: int, api_listen_port: int, ) -> dict[str, object]: self.assert_pid("api", api_pid) self.assert_pid("cloudflared", cloudflared_pid) payload: dict[str, object] = { "processes": { "api": {"connections": 2, "established": 1, "listeners": 1}, "cloudflared": { "connections": 4, "established": 4, "listeners": 0, }, }, "host_tcp": {"connections": 20}, "api_listener": { "port": api_listen_port, "owned_listener_count": 1, "conflicting_listener_count": 0, }, } payload.update(self.tcp_overrides) return payload class G7TopologyEvidenceTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.runner_module = load_runner() def config(self, **overrides): values = { "compose_project": "vignette-preview-20260807", "public_host": "api.example.test", "api_container": "vignette-api-1", "api_service": "api", "api_image_digest": API_DIGEST, "caddy_container": "vignette-caddy-1", "caddy_service": "caddy", "caddy_image_digest": CADDY_DIGEST, "samples": 1, "interval_seconds": 1.0, } values.update(overrides) return self.runner_module.CaptureConfig(**values) def windows_config(self, **overrides): values = { "public_host": "api.example.test", "repo_root": r"D:\workspace\vignette", "git_sha": GIT_SHA, "git_tree_sha": GIT_TREE_SHA, "runner_script_sha256": SCRIPT_DIGESTS["runner"], "collector_script_sha256": SCRIPT_DIGESTS["collector"], "checker_script_sha256": SCRIPT_DIGESTS["checker"], "psutil_version": PSUTIL_VERSION, "api_pid": 301, "api_executable_name": "python.exe", "api_executable_sha256": API_EXE_DIGEST, "api_cwd": r"D:\workspace\vignette\apps\api", "api_listen_port": 8001, "cloudflared_pid": 302, "cloudflared_executable_name": "cloudflared.exe", "cloudflared_executable_sha256": CLOUDFLARED_EXE_DIGEST, "cloudflared_cwd": r"D:\workspace\vignette", "samples": 2, "interval_seconds": 1.0, } values.update(overrides) return self.runner_module.WindowsHostConfig(**values) def test_parsers_emit_numeric_metadata_without_endpoints(self) -> None: source = FixtureSource() targets = self.runner_module.pin_targets(self.config(), FixtureRunner(), source) stats = self.runner_module.parse_docker_stats(stats_output(), targets) tcp = self.runner_module.parse_ss_tcp(SS_OUTPUT) proc = self.runner_module.read_proc_metrics( source, proc_root="/proc", target=targets["api"], ) cgroup = self.runner_module.read_cgroup_metrics( source, cgroup_root="/sys/fs/cgroup", target=targets["api"], ) self.assertEqual(134_217_728, stats["api"]["memory_usage_bytes"]) self.assertEqual(1_500_000, stats["api"]["network_rx_bytes"]) self.assertEqual(2, tcp["connections"]) self.assertEqual(3, tcp["recv_q_bytes_total"]) self.assertEqual(133, tcp["send_q_bytes_total"]) self.assertEqual(1, tcp["retransmit_current_total"]) self.assertEqual(4, tcp["retransmit_cumulative_total"]) self.assertEqual(102_400, proc["vm_rss_bytes"]) self.assertEqual(4, proc["fd_count"]) self.assertEqual(1_000, cgroup["cpu_stat"]["usage_usec"]) serialized = json.dumps({"stats": stats, "tcp": tcp}) self.assertNotIn("10.0.0.1", serialized) self.assertNotIn("10.0.0.2", serialized) def test_capture_pins_targets_and_builds_high_water_summary(self) -> None: runner = FixtureRunner() evidence = self.runner_module.capture_evidence( self.config(), runner, FixtureSource(), sleep=lambda _seconds: self.fail("single sample must not sleep"), ) self.assertEqual("passed", evidence["status"]) self.assertEqual("linux_compose", evidence["topology_mode"]) self.assertEqual(1, evidence["samples_completed"]) self.assertEqual(API_ID, evidence["targets"]["api"]["container_id"]) self.assertEqual(API_DIGEST, evidence["targets"]["api"]["image_digest"]) self.assertEqual("api.example.test", evidence["requested"]["public_host"]) self.assertEqual( 125_829_120, evidence["summary"]["containers"]["api"]["cgroup_memory_peak_bytes_max"], ) scope = evidence["scope"] self.assertTrue(scope["metadata_only"]) self.assertFalse(scope["socket_endpoints_retained"]) self.assertFalse(scope["cloudflare_edge"]["internal_queue_measured"]) self.assertEqual( "separate_external_artifact_required", scope["cloudflare_edge"]["evidence_boundary"], ) inspect_calls = [call for call in runner.calls if call[1] == "inspect"] self.assertEqual(3, len(inspect_calls)) def test_pin_validation_fails_on_pid_restart_and_image_drift(self) -> None: cases = { "init_pid_drift:api": inspect_item( container_id=API_ID, image_digest=API_DIGEST, service="api", pid=999, ), "container_restart_drift:api": inspect_item( container_id=API_ID, image_digest=API_DIGEST, service="api", pid=101, restart_count=1, started_at="2026-08-07T00:01:00Z", ), "image_digest_drift:api": inspect_item( container_id=API_ID, image_digest="sha256:" + "9" * 64, service="api", pid=101, ), } for expected_failure, changed_api in cases.items(): with self.subTest(expected_failure=expected_failure): drifted = json.dumps( [ changed_api, inspect_item( container_id=CADDY_ID, image_digest=CADDY_DIGEST, service="caddy", pid=202, ), ] ) runner = FixtureRunner(inspect_outputs=[stable_inspect(), drifted]) with self.assertRaises(self.runner_module.EvidenceFailure) as caught: self.runner_module.capture_evidence( self.config(), runner, FixtureSource(), sleep=lambda _: None ) self.assertEqual(expected_failure, str(caught.exception)) def test_missing_cgroup_source_fails_closed(self) -> None: source = FixtureSource() del source.files["/sys/fs/cgroup/docker/api.scope/memory.peak"] with self.assertRaises(self.runner_module.EvidenceFailure) as caught: self.runner_module.capture_evidence( self.config(), FixtureRunner(), source, sleep=lambda _: None ) self.assertEqual( "cgroup_source_missing:api:memory.peak", str(caught.exception), ) def test_missing_command_fails_before_any_runtime_read(self) -> None: runner = FixtureRunner(unavailable={"ss"}) with self.assertRaises(self.runner_module.EvidenceFailure) as caught: self.runner_module.capture_evidence( self.config(), runner, FixtureSource(), sleep=lambda _: None ) self.assertEqual("command_unavailable:ss", str(caught.exception)) self.assertEqual([], runner.calls) def test_capture_bounds_are_fail_closed(self) -> None: for config, failure in ( (self.config(samples=0), "sample_count_out_of_bounds"), (self.config(interval_seconds=0), "sample_interval_out_of_bounds"), ( self.config(samples=7_200, interval_seconds=2), "capture_window_out_of_bounds", ), ): with self.subTest(failure=failure): with self.assertRaises(self.runner_module.EvidenceFailure) as caught: self.runner_module.validate_config(config) self.assertEqual(failure, str(caught.exception)) def test_windows_capture_pins_processes_listener_and_high_water(self) -> None: source = FixtureWindowsSource(self.runner_module) evidence = self.runner_module.capture_windows_evidence( self.windows_config(), FixtureRunner(), source, sleep=lambda _seconds: None, ) self.assertEqual("passed", evidence["status"]) self.assertEqual("windows_host", evidence["topology_mode"]) self.assertEqual({"api", "cloudflared"}, set(evidence["targets"])) self.assertEqual( { "detached_head": True, "tracked_clean": True, "git_sha": GIT_SHA, "git_tree_sha": GIT_TREE_SHA, "script_sha256": SCRIPT_DIGESTS, "runtime_dependencies": {"psutil": PSUTIL_VERSION}, }, evidence["source_provenance"], ) self.assertEqual(GIT_SHA, evidence["targets"]["api"]["git_sha"]) self.assertEqual( API_EXE_DIGEST, evidence["targets"]["api"]["executable_sha256"], ) self.assertEqual( r"D:\workspace\vignette\apps\api", evidence["targets"]["api"]["cwd"], ) summary = evidence["summary"] self.assertEqual(4.0, summary["processes"]["api"]["cpu_time_seconds_max"]) self.assertEqual(1, summary["api_listener"]["owned_listener_count_min"]) self.assertEqual(4, summary["processes"]["cloudflared"]["tcp_established_max"]) serialized = json.dumps(evidence) self.assertNotIn("127.0.0.1", serialized) def test_windows_identity_and_git_drift_fail_closed(self) -> None: cases = ( ( FixtureRunner(), FixtureWindowsSource( self.runner_module, identity_overrides={ "api": [{"started_at": "2999-01-01T00:00:00Z"}] }, ), "process_started_at_future:api", ), ( FixtureRunner(), FixtureWindowsSource( self.runner_module, identity_overrides={ "api": [ {}, {"command_line_sha256": "8" * 64}, ] }, ), "process_command_line_drift:api", ), ( FixtureRunner(git_outputs=[GIT_SHA + "\n", "9" * 40 + "\n"]), FixtureWindowsSource(self.runner_module), "git_sha_drift", ), ) for runner, source, failure in cases: with self.subTest(failure=failure): with self.assertRaises(self.runner_module.EvidenceFailure) as caught: self.runner_module.capture_windows_evidence( self.windows_config(samples=1), runner, source, sleep=lambda _seconds: None, ) self.assertEqual(failure, str(caught.exception)) def test_windows_source_and_toolchain_provenance_fail_closed(self) -> None: cases = ( ( FixtureRunner(git_branch="master\n"), FixtureWindowsSource(self.runner_module), self.windows_config(samples=1), "git_head_not_detached", ), ( FixtureRunner(git_status=" M scripts/check-g7-external-proof.py\n"), FixtureWindowsSource(self.runner_module), self.windows_config(samples=1), "git_tracked_worktree_dirty", ), ( FixtureRunner(git_tree_sha="9" * 40), FixtureWindowsSource(self.runner_module), self.windows_config(samples=1), "git_tree_sha_drift", ), ( FixtureRunner(), FixtureWindowsSource(self.runner_module), self.windows_config(samples=1, checker_script_sha256="9" * 64), "evidence_script_sha256_drift:checker", ), ( FixtureRunner(), FixtureWindowsSource(self.runner_module, psutil_version="6.1.0"), self.windows_config(samples=1), "psutil_version_drift", ), ) for runner, source, config, failure in cases: with self.subTest(failure=failure): with self.assertRaises(self.runner_module.EvidenceFailure) as caught: self.runner_module.capture_windows_evidence( config, runner, source, sleep=lambda _seconds: None, ) self.assertEqual(failure, str(caught.exception)) def test_windows_listener_and_tunnel_connection_are_required(self) -> None: cases = ( ( { "api_listener": { "port": 8001, "owned_listener_count": 0, "conflicting_listener_count": 0, } }, "api_listener_not_owned", ), ( { "processes": { "api": { "connections": 2, "established": 1, "listeners": 1, }, "cloudflared": { "connections": 0, "established": 0, "listeners": 0, }, } }, "cloudflared_tunnel_connection_missing", ), ) for overrides, failure in cases: with self.subTest(failure=failure): source = FixtureWindowsSource( self.runner_module, tcp_overrides=overrides, ) with self.assertRaises(self.runner_module.EvidenceFailure) as caught: self.runner_module.capture_windows_evidence( self.windows_config(samples=1), FixtureRunner(), source, sleep=lambda _seconds: None, ) self.assertEqual(failure, str(caught.exception)) def test_windows_config_rejects_shared_pid_and_outside_cwd(self) -> None: cases = ( ( self.windows_config(cloudflared_pid=301), "process_targets_not_distinct", ), ( self.windows_config(api_cwd=r"C:\elsewhere"), "api_cwd_outside_repo", ), ( self.windows_config(git_tree_sha="not-a-tree"), "git_tree_sha_invalid", ), ( self.windows_config(psutil_version="6.1.0"), "psutil_version_invalid", ), ) for config, failure in cases: with self.subTest(failure=failure): with self.assertRaises(self.runner_module.EvidenceFailure) as caught: self.runner_module.validate_windows_config(config) self.assertEqual(failure, str(caught.exception)) if __name__ == "__main__": unittest.main()