693 lines
25 KiB
Python
693 lines
25 KiB
Python
"""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()
|