Files

414 lines
23 KiB
Python

#!/usr/bin/env python3
"""Synthetic evidence controls. No application, Docker, or workload execution."""
import json
import os
from pathlib import Path
import stat
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch
from uuid import UUID
import xml.etree.ElementTree as ET
sys.dont_write_bytecode = True
import analyze
import bundle
import launch
REVISION = "a" * 40
IMAGE = "sha256:" + "b" * 64
CONTEXT = {"name": "local-control", "endpoint": "unix:///tmp/control.sock"}
RUN_ID = "c" * 32
def canonical_rows(period=4, rejected=True):
yield {"kind": "start", "mode": analyze.MODE, "requestedSeconds": 7200, "minimumFullSeconds": 7200,
"runtime": analyze.RUNTIME, "sdk": analyze.SDK, "architecture": "Arm64", "processId": 123,
"startupSeconds": 1, "build": REVISION, "acceptance": False, "limits": analyze.BUDGETS.copy()}
cycles = 7200 // period
baseline = None
for cycle in range(1, cycles + 1):
elapsed = cycle * period - 1
value = dict.fromkeys(analyze.SAMPLE_FIELDS, 0)
value.update(Cycle=cycle, Seconds=elapsed, Variables=19, Folders=4, Conditions=1,
CachedCertificateStores=2, Descriptors=190, ManagedBytes=20_000_000 + elapsed * 32,
RssBytes=200_000_000 + elapsed * 128, GcCommittedBytes=48_000_000, JitMethods=76000,
JitIlBytes=11500000, ActiveTimers=12, Threads=24)
if cycle == 3:
baseline = value.copy()
if rejected and cycle == 4:
limit = baseline["RssBytes"] + analyze.BUDGETS["rssGrowthBytes"]
value["RssBytes"] = limit + 1
yield {"kind": "active", "cycle": cycle, "sessions": 2, "subscriptions": 1, "items": 3,
"browsePoints": 1, "historyPoints": 6}
yield {"kind": "network-stall-recovered", "cycle": cycle, "status": "BadRequestInterrupted", "seconds": 1}
yield {"kind": "resource-observation", "resources": value.copy()}
if rejected and cycle == 4:
yield {"kind": "budget-rejected", "cycle": cycle, "seconds": elapsed + 0.1,
"resources": value.copy(), "baseline": baseline,
"failure": (f"Shouldly.ShouldAssertException: {limit + 1}L\r\n should be less than or equal to"
f"\r\n{limit}L\r\n but was not\n at synthetic fixture")}
yield {"kind": "quiescent-budget-rejected" if rejected and cycle >= 4 else "quiescent",
"cycleSeconds": period - 1, "resources": value.copy()}
if cycle % 4 == 0:
yield {"kind": "restart", "cycles": cycle, "restarts": cycle // 4, "seconds": 1, "boundedRestartSeconds": 0.9}
yield {"kind": "late-slope-cycle", "cycle": cycle, "restarts": cycle // 4,
"elapsedSeconds": cycle * period, "cycleSeconds": period}
yield {"kind": "late-slope-result", "mode": analyze.MODE, "cycles": cycles, "restarts": cycles // 4,
"cycleSeconds": cycles * period, "shutdownSeconds": 2, "durationCompleted": True,
"shutdownVerified": True, "fatalFailure": False, "failureTypes": [], "coldBudgetRejected": rejected, "acceptance": False}
def write_json(path, value):
path.write_text(json.dumps(value) + "\n", encoding="utf-8")
def write_trx(path, names, failure=None):
root = ET.Element("TestRun", xmlns=analyze.TRX["t"])
results = ET.SubElement(root, "Results")
for name in sorted(names):
result = ET.SubElement(results, "UnitTestResult", testName=name, outcome="Failed" if failure else "Passed")
if failure:
output = ET.SubElement(result, "Output")
ET.SubElement(ET.SubElement(output, "ErrorInfo"), "Message").text = failure
total = len(names)
summary = ET.SubElement(root, "ResultSummary")
ET.SubElement(summary, "Counters", total=str(total), executed=str(total), passed=str(total - bool(failure)),
failed=str(int(bool(failure))), notExecuted="0", error="0", timeout="0", aborted="0")
ET.ElementTree(root).write(path, encoding="utf-8")
class RowControls(unittest.TestCase):
def test_fixed_windows_and_known_linear_slopes(self):
_, samples, completions, rejection, _ = analyze.validate_rows(canonical_rows(), REVISION)
report = analyze.describe(samples, completions)
self.assertEqual(rejection["cycle"], 4)
self.assertEqual(report["allQuiescent"]["sampleCount"], 1800)
self.assertEqual(report["allQuiescent"]["RssBytes"]["max"], rejection["resources"]["RssBytes"])
self.assertEqual([row["windowSeconds"] for row in report["windows"]], [[1800, 2400], [3600, 4200], [6600, 7200]])
self.assertEqual([row["sampleCount"] for row in report["windows"]], [150, 150, 150])
self.assertAlmostEqual(report["leastSquares"]["rss"], 128 * 3600)
self.assertAlmostEqual(report["leastSquares"]["retainedManaged"], 32 * 3600)
self.assertEqual(report["leastSquares"]["sampleCount"], 1350)
self.assertTrue(report["workCompletedBy1800Seconds"]["earlierSaturationWorkloadReached"])
def test_slopes_remain_descriptive_below_earlier_workload(self):
_, samples, completions, rejection, _ = analyze.validate_rows(canonical_rows(25, rejected=False), REVISION)
report = analyze.describe(samples, completions)
self.assertIsNone(rejection)
work = report["workCompletedBy1800Seconds"]
self.assertEqual((work["cycles"], work["restarts"]), (72, 18))
self.assertFalse(work["earlierSaturationWorkloadReached"])
self.assertIn("attribution withheld", work["interpretation"])
self.assertAlmostEqual(report["leastSquares"]["rss"], 128 * 3600)
def test_wrong_mode_duration_runtime_or_startup_are_rejected(self):
for field, value in (("mode", "full"), ("requestedSeconds", 90), ("runtime", "10.0.3"),
("architecture", "X64"), ("sdk", "2.0.0"), ("startupSeconds", 121), ("acceptance", True)):
with self.subTest(field=field):
rows = list(canonical_rows(25))
rows[0][field] = value
with self.assertRaises(ValueError):
analyze.validate_rows(rows, REVISION)
def test_wrong_final_state_never_yields_a_report(self):
for field, value in (("kind", "complete"), ("cycleSeconds", 7199.9), ("cycleSeconds", 7231),
("shutdownVerified", False), ("durationCompleted", False), ("fatalFailure", True),
("shutdownSeconds", 61), ("coldBudgetRejected", False), ("acceptance", True), ("restarts", 1)):
with self.subTest(field=field, value=value):
rows = list(canonical_rows(25))
rows[-1][field] = value
with self.assertRaises(ValueError):
analyze.validate_rows(rows, REVISION)
def test_duplicate_reordered_missing_or_malformed_samples(self):
def change(rows, position, field, value):
rows[position]["resources"][field] = value
edits = (
lambda rows: rows.insert(4, rows[3]),
lambda rows: rows.pop(3),
lambda rows: change(rows, 8, "Cycle", 1),
lambda rows: change(rows, 8, "Seconds", rows[3]["resources"]["Seconds"]),
lambda rows: change(rows, 3, "RssBytes", -1),
lambda rows: change(rows, 3, "ManagedBytes", float("inf")),
lambda rows: change(rows, 3, "PkiDescriptors", 1),
lambda rows: rows.append({"kind": "complete"}),
)
for edit in edits:
rows = list(canonical_rows(25))
edit(rows)
with self.assertRaises(ValueError):
analyze.validate_rows(rows, REVISION)
def test_first_rejection_cannot_disappear_or_change_baseline(self):
rows = list(canonical_rows(25))
first = next(i for i, row in enumerate(rows) if row["kind"] == "budget-rejected")
rows[first]["baseline"]["RssBytes"] -= 1
with self.assertRaisesRegex(ValueError, "baseline"):
analyze.validate_rows(rows, REVISION)
rows = list(canonical_rows(25))
later = next(row for row in rows if row["kind"] == "quiescent-budget-rejected" and row["resources"]["Cycle"] == 5)
later["kind"] = "quiescent"
with self.assertRaisesRegex(ValueError, "quiescent-budget-rejected"):
analyze.validate_rows(rows, REVISION)
rows = list(canonical_rows(25))
rows.pop(first)
with self.assertRaisesRegex(ValueError, "budget-rejected"):
analyze.validate_rows(rows, REVISION)
def test_bounded_restart_and_complete_cycle_durations(self):
for kind, field, value, expected in (
("restart", "boundedRestartSeconds", 25.1, "bounded restart duration"),
("restart", "seconds", 31, "combined restart duration"),
("late-slope-cycle", "cycleSeconds", 31, "complete cycle duration"),
):
with self.subTest(field=field):
rows = list(canonical_rows(25))
row = next(row for row in rows if row["kind"] == kind)
row[field] = value
with self.assertRaisesRegex(ValueError, expected):
analyze.validate_rows(rows, REVISION)
def test_missing_windows_edges_and_internal_or_between_window_gaps(self):
_, samples, completions, _, _ = analyze.validate_rows(canonical_rows(25), REVISION)
for begin, end in ((3600, 4200), (1800, 1880), (7110, 7200), (1900, 1980), (2900, 2980)):
with self.subTest(begin=begin, end=end):
incomplete = [sample for sample in samples if not begin <= sample["Seconds"] <= end]
with self.assertRaises(ValueError):
analyze.describe(incomplete, completions)
def test_jsonl_rejects_duplicate_keys_nonfinite_literals_and_truncation(self):
with tempfile.TemporaryDirectory() as temporary:
path = Path(temporary) / "samples.jsonl"
for content in ('{"kind":"start","kind":"complete"}\n', '{"seconds":NaN}\n', '{"kind":"start"}'):
path.write_text(content, encoding="utf-8")
with self.assertRaises(ValueError):
list(analyze.read_rows(path))
class ReceiptControls(unittest.TestCase):
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.root = Path(temporary.name).resolve()
self.run = self.root / "run"
self.bundle_receipt = self.root / "bundle.json"
(self.run / "results").mkdir(parents=True)
self.build_evidence()
def build_evidence(self, rejected=True):
self.bundle = {**analyze.SETTINGS, "kind": "bundle", "sourceRevision": REVISION, "image": IMAGE,
"dockerContext": CONTEXT, "files": {"app/libcoreclr.so": "d" * 64}}
write_json(self.bundle_receipt, self.bundle)
self.bundle_hash = analyze.digest(self.bundle_receipt)
self.provenance = {"kind": "launch", "protocol": analyze.PROTOCOL, "acceptance": False,
"sourceRevision": REVISION, "image": IMAGE, "dockerContext": CONTEXT, "runId": RUN_ID,
"bundleReceiptSha256": self.bundle_hash, "coreClrSha256": "d" * 64,
"containerTimeoutSeconds": 7440, "bundlePath": str(self.root / "bundle"), "outputPath": str(self.run)}
self.provenance["command"] = analyze.container_command(self.root / "bundle", self.run, self.bundle, RUN_ID)
write_json(self.run / "launch.json", self.provenance)
rows = list(canonical_rows(25, rejected))
(self.run / "resources.jsonl").write_text("".join(json.dumps(row) + "\n" for row in rows), encoding="utf-8")
status = int(rejected)
for filename in ("test-exit-status.txt", "container-exit-status.txt"):
(self.run / filename).write_text(f"{status}\n", encoding="ascii")
(self.run / "target-runtime.txt").write_text(
f"targetPid=123\nactualExecutable=/app/{analyze.APP}\nmappedCoreClr=/app/libcoreclr.so\ncoreClrSha256={'d' * 64}\n",
encoding="ascii")
for filename in ("container.log", "controller-runtime.txt", "descriptor-controls.log", "test.log"):
(self.run / filename).write_text("Synthetic control fixture\n", encoding="utf-8")
write_trx(self.run / "results/descriptor-controls.trx", analyze.DESCRIPTOR_TESTS)
failure = next((row["failure"] for row in rows if row["kind"] == "budget-rejected"), None)
message = analyze.assertion_message(failure).replace("Exception:", "Exception :", 1) if failure else None
write_trx(self.run / "results/late-slope.trx", {analyze.TEST_NAME}, message)
actual = analyze.inspect_run(self.run, self.provenance, 7205)
self.receipt = {**self.provenance, "kind": "run", "diagnosticCompleted": True, "containerRemoved": True,
"wallSeconds": 7205, "launcherExitStatus": status, **{key: actual[key] for key in (
"testExitStatus", "containerExitStatus", "descriptorTests", "experimentTests", "diagnosticOutcome")}}
self.refresh_receipt()
def refresh_receipt(self):
self.receipt["evidenceSha256"] = analyze.evidence_hashes(self.run)
write_json(self.run / "run-receipt.json", self.receipt)
def report(self):
return analyze.analyze_run(self.run, self.bundle_receipt, self.bundle_hash)
def test_cold_failure_and_zero_exit_are_separate_diagnostic_results(self):
cold = self.report()
self.assertEqual(cold["testExitStatus"], 1)
self.assertFalse(cold["acceptance"])
self.assertEqual(cold["firstColdRejection"]["exceededFields"], ["RssBytes"])
self.build_evidence(rejected=False)
within_cap = self.report()
self.assertEqual(within_cap["testExitStatus"], 0)
self.assertFalse(within_cap["acceptance"])
self.assertIsNone(within_cap["firstColdRejection"])
def test_changed_hash_or_bad_run_receipt_rejected(self):
(self.run / "test.log").write_text("Changed evidence\n", encoding="utf-8")
with self.assertRaisesRegex(ValueError, "evidence hashes"):
self.report()
for key, value in (("launcherExitStatus", 0), ("containerRemoved", False), ("diagnosticCompleted", False),
("wallSeconds", 7441), ("bundleReceiptSha256", "f" * 64)):
with self.subTest(key=key):
self.build_evidence()
self.receipt[key] = value
self.refresh_receipt()
with self.assertRaises(ValueError):
self.report()
def test_rehashed_exit_runtime_and_trx_cannot_mask_incomplete_evidence(self):
for filename, contents, pattern in (
("test-exit-status.txt", "0\n", "test/container exit"),
("target-runtime.txt", "targetPid=122\n", "mapped CLR"),
):
with self.subTest(filename=filename):
self.build_evidence()
(self.run / filename).write_text(contents, encoding="ascii")
self.refresh_receipt()
with self.assertRaisesRegex(ValueError, pattern):
self.report()
self.build_evidence()
write_trx(self.run / "results/late-slope.trx", {analyze.TEST_NAME}, "Shouldly.ShouldAssertException: wrong assertion")
self.refresh_receipt()
with self.assertRaisesRegex(ValueError, "first cold-cap assertion"):
self.report()
write_trx(self.run / "results/descriptor-controls.trx", {"SkippedOrWrongControl"})
self.refresh_receipt()
with self.assertRaisesRegex(ValueError, "descriptor-controls.trx"):
self.report()
def test_recorded_command_must_use_the_same_local_isolation(self):
private_mount = f"{self.run / 'tmp'}:/tmp:rw"
self.assertIn(private_mount, self.provenance["command"])
for isolation in ("network", "read-only rootfs", "missing /tmp bind", "foreign /tmp bind"):
with self.subTest(isolation=isolation):
self.build_evidence()
invocation = self.provenance["command"].copy()
if isolation == "network":
invocation[invocation.index("--network") + 1] = "host"
elif isolation == "read-only rootfs":
invocation.remove("--read-only")
elif isolation == "missing /tmp bind":
index = invocation.index(private_mount)
del invocation[index - 1:index + 1]
else:
invocation[invocation.index(private_mount)] = f"{self.root / 'foreign'}:/tmp:rw"
self.provenance["command"] = self.receipt["command"] = invocation
write_json(self.run / "launch.json", self.provenance)
self.refresh_receipt()
with self.assertRaisesRegex(ValueError, "container invocation"):
self.report()
class BundleControls(unittest.TestCase):
def test_task_output_and_tmp_are_fresh_private_directories(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary).resolve()
output = launch.create_task_output(root / "run", root / "bundle")
for directory in (output, output / "tmp"):
mode = directory.lstat().st_mode
self.assertTrue(stat.S_ISDIR(mode))
self.assertEqual(stat.S_IMODE(mode), 0o700)
with self.assertRaises(FileExistsError):
launch.create_task_output(output, root / "bundle")
file = root / "existing-file"
file.touch()
with self.assertRaises(FileExistsError):
launch.create_task_output(file, root / "bundle")
link = root / "linked-output"
destination = root / "missing-target"
link.symlink_to(destination, target_is_directory=True)
with self.assertRaisesRegex(ValueError, "symbolic link"):
launch.create_task_output(link, root / "bundle")
self.assertFalse(destination.exists())
def test_mapped_runtime_requires_the_mounted_inode_and_live_file(self):
script = Path(__file__).with_name("container.sh")
with tempfile.TemporaryDirectory() as temporary:
maps = Path(temporary) / "maps"
live = "1000-2000 r-xp 00000000 00:01 42 /app/libcoreclr.so"
for line, expected in ((live, 0), (live + " (deleted)", 1), (live.replace(" 42 ", " 43 "), 1),
(live + "\n" + live.replace("/app/", "/different/"), 1)):
with self.subTest(line=line):
maps.write_text(line + "\n", encoding="ascii")
result = subprocess.run(["bash", "-c", 'source "$1"\nverify_mapped_coreclr "$2" "$3"',
"map-control", str(script), str(maps), "42 /app/libcoreclr.so"],
capture_output=True, text=True, timeout=5)
self.assertEqual(result.returncode, expected, result.stderr)
def test_remote_override_fails_before_asking_docker(self):
with patch.dict(os.environ, {"DOCKER_HOST": "tcp://example.invalid:2375"}), patch("bundle.subprocess.check_output") as docker:
with self.assertRaisesRegex(ValueError, "override"):
bundle.context_identity("local-control")
docker.assert_not_called()
def test_bundle_file_set_runtime_and_package_graph(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary).resolve()
app = root / "app"
app.mkdir()
elf = b"\x7fELF\x02\x01" + b"\x00" * 12 + b"\xb7\x00"
(app / analyze.APP).write_bytes(elf)
(app / analyze.APP).chmod(0o700)
(app / "libcoreclr.so").write_bytes(elf)
runtime = app / f"{analyze.APP}.runtimeconfig.json"
write_json(runtime, {"runtimeOptions": {"tfm": "net10.0", "includedFrameworks": [
{"name": "Microsoft.NETCore.App", "version": analyze.RUNTIME}]}})
dependencies = app / f"{analyze.APP}.deps.json"
write_json(dependencies, {"runtimeTarget": {"name": ".NETCoreApp,Version=v10.0/linux-arm64"},
"libraries": {key: {} for key in bundle.SDK_PACKAGES | bundle.GATEWAY_PACKAGES}})
modules = []
for index, assembly in enumerate((analyze.APP, "Opc.Ua.Server", "System.Private.CoreLib",
"ZB.MOM.WW.HistorianGateway.Client", "ZB.MOM.WW.HistorianGateway.Contracts"), 1):
filename = f"{assembly}.dll"
(app / filename).write_text("Synthetic metadata fixture\n", encoding="utf-8")
modules.append({"file": f"app/{filename}", "assembly": assembly, "version": "1.0.0.0", "mvid": str(UUID(int=index))})
tools = root / "tools"
tools.mkdir()
for name in bundle.TOOLS:
(tools / name).write_text("Synthetic tool fixture\n", encoding="utf-8")
receipt = {**analyze.SETTINGS, "kind": "bundle", "sourceRevision": REVISION, "image": IMAGE,
"imageIdentity": {"id": IMAGE, "os": "linux", "architecture": "arm64"}, "dockerContext": CONTEXT,
"sdkLibraries": sorted(bundle.SDK_PACKAGES), "gatewayLibraries": sorted(bundle.GATEWAY_PACKAGES),
"managedModules": modules, "sourceFiles": {**dict.fromkeys(bundle.CANONICAL_SOURCES, "e" * 64), **{
f"scripts/validation/opcua-lifecycle-memory/late-slope/{name}": analyze.digest(tools / name)
for name in bundle.TOOLS}}, "files": bundle.manifest(root)}
path = root / "receipt.json"
write_json(path, receipt)
approved = analyze.digest(path)
bundle.verify_bundle(root, approved)
extra = app / "extra"
extra.touch()
with self.assertRaisesRegex(ValueError, "file set"):
bundle.verify_bundle(root, approved)
extra.unlink()
extra.symlink_to(app / "libcoreclr.so")
with self.assertRaisesRegex(ValueError, "artifact type"):
bundle.verify_bundle(root, approved)
extra.unlink()
write_json(runtime, {"runtimeOptions": {"tfm": "net10.0", "includedFrameworks": [
{"name": "Microsoft.NETCore.App", "version": "10.0.3"}]}})
receipt["files"] = bundle.manifest(root)
write_json(path, receipt)
with self.assertRaisesRegex(ValueError, "self-contained application runtime"):
bundle.verify_bundle(root, analyze.digest(path))
write_json(runtime, {"runtimeOptions": {"tfm": "net10.0", "includedFrameworks": [
{"name": "Microsoft.NETCore.App", "version": analyze.RUNTIME}]}})
write_json(dependencies, {"runtimeTarget": {"name": ".NETCoreApp,Version=v10.0/linux-arm64"}, "libraries": {}})
receipt["files"] = bundle.manifest(root)
write_json(path, receipt)
with self.assertRaisesRegex(ValueError, "dependency|graph"):
bundle.verify_bundle(root, analyze.digest(path))
if __name__ == "__main__":
unittest.main(verbosity=2)