From 613369fc54fbbdd08bcf77feb7d1256a4e1c4462 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Tue, 10 Nov 2026 17:59:15 -0500 Subject: [PATCH] test: add immutable Linux late-slope diagnostic tooling --- .../late-slope/README.md | 71 +++ .../late-slope/analyze.py | 423 ++++++++++++++++++ .../late-slope/bundle.py | 129 ++++++ .../late-slope/container.sh | 70 +++ .../late-slope/launch.py | 130 ++++++ .../late-slope/prepare.py | 97 ++++ .../late-slope/test_offline.py | 368 +++++++++++++++ 7 files changed, 1288 insertions(+) create mode 100644 scripts/validation/opcua-lifecycle-memory/late-slope/README.md create mode 100644 scripts/validation/opcua-lifecycle-memory/late-slope/analyze.py create mode 100644 scripts/validation/opcua-lifecycle-memory/late-slope/bundle.py create mode 100644 scripts/validation/opcua-lifecycle-memory/late-slope/container.sh create mode 100644 scripts/validation/opcua-lifecycle-memory/late-slope/launch.py create mode 100644 scripts/validation/opcua-lifecycle-memory/late-slope/prepare.py create mode 100644 scripts/validation/opcua-lifecycle-memory/late-slope/test_offline.py diff --git a/scripts/validation/opcua-lifecycle-memory/late-slope/README.md b/scripts/validation/opcua-lifecycle-memory/late-slope/README.md new file mode 100644 index 00000000..0dedca9d --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/late-slope/README.md @@ -0,0 +1,71 @@ +# Linux late-slope diagnostic + +The [protocol](../LATE_SLOPE.md) fixes 7,200 seconds of canonical cycling from +startup completion, with the existing active restart every fourth cycle. The +cycle-three baseline and resource caps remain in force. A latched rejection +keeps the original failing test and container exit. The result records +`acceptance=false` even when the diagnostic completes its duration and shutdown. + +The analyzer requires an ordered active, recovered-stall, resource-observation, +quiescent and complete-cycle ledger. It checks the 25-second restart helper +bound, 30-second complete cycle bound and 60-second shutdown bound. It +recomputes the first cold rejection and verifies its assertion against the +actual TRX. It checks each fixed time window and the entire 1,800–7,200 second +range before reporting RSS and retained-managed slopes in bytes per hour. +The report includes sample counts, quiescent peaks, final values and coverage. +It identifies whether the earlier saturation workload had completed by 1,800 +seconds, which bounds any interpretation relative to the previously observed +restart step. The raw JSONL keeps every sample and the first assertion stack. + +Certificate allocation stack tracking remains enabled and contributes to the +measured process. The slope report is descriptive and carries no resource +pass threshold. Quiescent peaks refer to the sampled points. + +## Offline controls and artifact preparation + +The synthetic controls exercise the parser, receipts and bundle manifest with +temporary files. Run them with Python 3.11 or later. + +```bash +python3 scripts/validation/opcua-lifecycle-memory/late-slope/test_offline.py +``` + +Prepare from the reviewed clean commit containing the canonical test and tools. +The image argument is the immutable ID of a locally cached Linux ARM64 SDK +image. The explicitly selected Docker context must use a local Unix socket. +The preparation command publishes the self-contained .NET 10.0.14 app and +records the exact SDK 2.0.1 and HistorianGateway 1.0.0 graph, source hashes, file +hashes and managed module MVIDs. It uses the repository's read-only PE metadata +inspector. Preparation completes before any descriptor or lifecycle test runs. + +```bash +python3 scripts/validation/opcua-lifecycle-memory/late-slope/prepare.py \ + --source-revision "$reviewedRevision" --docker-context "$localContext" \ + --image "$cachedImageId" --output "$bundle" +python3 "$bundle/tools/launch.py" --bundle "$bundle" \ + --receipt-sha256 "$reviewedBundleReceiptHash" --verify-only +``` + +## Reviewed run and analysis + +Root owns launch after review of the exact bundle receipt and controls. The +bundled launcher uses the pinned Unix endpoint, immutable image, network-none +container, read-only root filesystem and read-only application and tools mounts. +A fresh private task directory holds results and fixture files. The two +descriptor controls must execute successfully before the canonical test starts. The launcher verifies +the actual test process's apphost, mapped CLR and CLR hash. Its outer deadline +is 7,440 seconds, with cleanup limited to the uniquely labeled container. + +```bash +python3 "$bundle/tools/launch.py" --bundle "$bundle" \ + --receipt-sha256 "$reviewedBundleReceiptHash" --output "$run" +python3 "$bundle/tools/analyze.py" --run "$run" \ + --bundle-receipt "$bundle/receipt.json" --bundle-receipt-sha256 "$reviewedBundleReceiptHash" +``` + +A completed diagnostic with a cold-cap rejection has a nonzero launcher exit. +Its receipt records diagnostic completion separately from the actual test and +container status. The offline analyzer verifies the reviewed bundle receipt, +raw evidence hashes, test identities and outcomes, fixed duration and verified +shutdown before emitting a report. It can run from copies of the small bundle +receipt and run evidence on a machine without Docker. diff --git a/scripts/validation/opcua-lifecycle-memory/late-slope/analyze.py b/scripts/validation/opcua-lifecycle-memory/late-slope/analyze.py new file mode 100644 index 00000000..b621058f --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/late-slope/analyze.py @@ -0,0 +1,423 @@ +#!/usr/bin/env python3 +"""Validate a recorded Linux late-slope diagnostic and describe its quiescent samples.""" + +import argparse +import hashlib +import json +import math +from pathlib import Path +import re +from statistics import median +import sys +import xml.etree.ElementTree as ET + + +PROTOCOL = "sdk2-late-slope-v1" +DOCKER_LABEL = "org.otopcua.late-slope-run" +MODE = "late-slope" +SECONDS = 7200 +MAX_CYCLE_SECONDS = 30 +RESTART_SECONDS = 25 +SHUTDOWN_SECONDS = 60 +CONTAINER_SECONDS = 7440 +RUNTIME = "10.0.14" +SDK = "2.0.1+07e62f72500cc0923a2bab940653c8c1116fa286" +APP = "ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests" +TEST_CLASS = APP + ".Sdk2LifecycleSoakTests." +TEST_NAME = TEST_CLASS + "Repeated_lifecycles_return_resources_to_a_steady_baseline" +DESCRIPTOR_TESTS = { + TEST_CLASS + "Descriptor_probe_observes_an_open_PKI_file_through_a_symlink_alias_and_its_close", + TEST_CLASS + "Descriptor_probe_times_out_and_reaps_a_process_that_keeps_both_streams_open", +} +STALL_STATUSES = {"OperationCanceled", "BadTimeout", "BadRequestCancelledByClient", "BadRequestTimeout", "BadRequestInterrupted"} +WINDOWS = ((1800, 2400), (3600, 4200), (6600, 7200)) +BUDGETS = {"fdGrowth": 32, "retainedManagedGrowthBytes": 64 << 20, "rssGrowthBytes": 128 << 20} +SETTINGS = { + "protocol": PROTOCOL, "mode": MODE, "requestedSeconds": SECONDS, "rid": "linux-arm64", + "runtimeVersion": RUNTIME, "selfContained": True, "acceptance": False, + "startupTimeoutSeconds": 120, "cycleTimeoutSeconds": MAX_CYCLE_SECONDS, + "cycleDeadlineSeconds": SECONDS + MAX_CYCLE_SECONDS, "restartTimeoutSeconds": RESTART_SECONDS, + "ownerTimeoutSeconds": 20, "shutdownTimeoutSeconds": SHUTDOWN_SECONDS, + "containerTimeoutSeconds": CONTAINER_SECONDS, "certificateLeakTracking": True, + "windowsSeconds": [list(window) for window in WINDOWS], "limits": BUDGETS, +} +CAP_FIELDS = (("Descriptors", "fdGrowth"), ("ManagedBytes", "retainedManagedGrowthBytes"), ("RssBytes", "rssGrowthBytes")) +ZERO_FIELDS = ( + "Sessions", "Subscriptions", "Items", "BrowsePoints", "HistoryPoints", + "LiveDataItems", "LiveEventItems", "MonitoredNodes", "PkiDescriptors", +) +STEADY_FIELDS = ("Variables", "Folders", "Conditions", "CachedCertificateStores") +MEASUREMENTS = ( + "Descriptors", "ManagedBytes", "RssBytes", "AllocatedBytes", "Gen0", "Gen1", "Gen2", + "GcCommittedBytes", "GcHeapBytes", "GcFragmentedBytes", "PinnedObjects", "PendingFinalizers", + "JitMethods", "JitIlBytes", "ActiveTimers", "Threads", +) +SAMPLE_FIELDS = {"Cycle", "Seconds", *ZERO_FIELDS, *STEADY_FIELDS, *MEASUREMENTS} +EVIDENCE_FILES = ( + "launch.json", "container.log", "container-exit-status.txt", "controller-runtime.txt", + "descriptor-controls.log", "test.log", "target-runtime.txt", "resources.jsonl", + "test-exit-status.txt", "results/descriptor-controls.trx", "results/late-slope.trx", +) +TRX = {"t": "http://microsoft.com/schemas/VisualStudio/TeamTest/2010"} + + +def require(condition, message): + if not condition: + raise ValueError(message) + + +def local_context(context): + return (type(context) is dict and set(context) == {"name", "endpoint"} + and isinstance(context["name"], str) and re.fullmatch(r"[A-Za-z0-9_.-]+", context["name"]) + and isinstance(context["endpoint"], str) and context["endpoint"].startswith("unix:///") + and not any(character in context["endpoint"] for character in "\r\n\0")) + + +def container_command(root, output, bundle, run_id): + require(all(path.is_absolute() and ":" not in str(path) for path in (root, output)), "Invalid Docker bind paths") + require(local_context(bundle["dockerContext"]), "Expected the reviewed local Unix Docker context") + return ["docker", "--host", bundle["dockerContext"]["endpoint"], "run", "--rm", "--init", "--pull", "never", + "--network", "none", "--platform", "linux/arm64", "--read-only", "--workdir", "/task", + "--name", f"opcua-late-slope-{run_id}", + "--label", f"{DOCKER_LABEL}={run_id}", "-e", "OTOPCUA_SOAK_BUILD_SHA=" + bundle["sourceRevision"], + "-v", f"{root / 'app'}:/app:ro", "-v", f"{root / 'tools'}:/tools:ro", "-v", f"{output}:/task", + bundle["image"], "bash", "/tools/container.sh"] + + +def digest(path): + with path.open("rb") as stream: + return hashlib.file_digest(stream, "sha256").hexdigest() + + +def unique_object(pairs): + result = {} + for key, value in pairs: + require(key not in result, f"Duplicate JSON key {key}") + result[key] = value + return result + + +def reject_constant(value): + raise ValueError(f"Invalid JSON constant {value}") + + +def json_object(text): + value = json.loads(text, object_pairs_hook=unique_object, parse_constant=reject_constant) + require(type(value) is dict, "Expected a JSON object") + return value + + +def read_object(path): + return json_object(path.read_text(encoding="utf-8")) + + +def read_rows(path): + with path.open(encoding="utf-8") as stream: + for line_number, line in enumerate(stream, 1): + require(line.endswith("\n"), f"Truncated JSONL row {line_number}") + try: + yield json_object(line) + except ValueError as error: + raise ValueError(f"Invalid JSONL row {line_number}: {error}") from error + + +def count(value, label, minimum=0): + require(type(value) is int and minimum <= value <= (1 << 63) - 1, f"Invalid {label}") + return value + + +def seconds(value, label, minimum=0, maximum=CONTAINER_SECONDS): + require(type(value) in (int, float) and minimum <= value <= maximum, f"Invalid {label}") + return value + + +def assertion_message(text): + require(isinstance(text, str), "Missing original resource-cap assertion") + message = text.replace("\r", "").split("\n at ", 1)[0].strip() + normalized = re.sub(r"^Shouldly\.ShouldAssertException\s*:\s*", "Shouldly.ShouldAssertException: ", message) + require(normalized.startswith("Shouldly.ShouldAssertException: ") and len(normalized) > 32, + "The recorded failure is not a resource-cap assertion") + return normalized + + +def sample(row, cycle): + value = row["resources"] + require(type(value) is dict and set(value) == SAMPLE_FIELDS, f"Invalid sample fields at cycle {cycle}") + require(count(value["Cycle"], "sample cycle", 1) == cycle, f"Out-of-order sample cycle {cycle}") + seconds(value["Seconds"], "sample time", maximum=SECONDS + MAX_CYCLE_SECONDS) + for field in SAMPLE_FIELDS - {"Cycle", "Seconds"}: + count(value[field], field) + require(all(value[field] == 0 for field in ZERO_FIELDS), f"Quiescent ownership failure at cycle {cycle}") + return value + + +def validate_rows(rows, revision): + """Validate the canonical row order, original budget and fixed-duration outcome.""" + rows = iter(rows) + + def take(kind): + row = next(rows, {}) + require(row.get("kind") == kind, f"Expected {kind}, found {row.get('kind')!r}") + require(row.get("acceptance", False) is False and "satisfiesTwoHourDuration" not in row, + "An ordinary acceptance result is forbidden") + return row + + start = take("start") + require(start.get("mode") == MODE and start.get("requestedSeconds") == SECONDS + and type(start["requestedSeconds"]) is int and start.get("minimumFullSeconds") == SECONDS + and start.get("acceptance") is False, "Wrong fixed-duration diagnostic mode") + require(start.get("build") == revision and re.fullmatch(r"[0-9a-f]{40}", revision), "Wrong source revision") + require(start.get("runtime") == RUNTIME and start.get("architecture") == "Arm64" and start.get("sdk") == SDK, + "Wrong Linux ARM64 runtime or SDK version") + count(start.get("processId"), "test process ID", 1) + seconds(start.get("startupSeconds"), "startup duration", maximum=SETTINGS["startupTimeoutSeconds"]) + limits = start.get("limits") + require(type(limits) is dict and limits == BUDGETS and all(type(value) is int for value in limits.values()), + "Original cold resource caps changed") + + samples = [] + first_rejection = None + completions = [] + # Buffer only a single row so the actual record order cannot be reconstructed by sorting. + pending = next(rows, {}) + while pending.get("kind") == "active": + cycle = len(samples) + 1 + require(count(pending.get("cycle"), "active cycle", 1) == cycle, "Missing or duplicate active cycle") + active_counts = tuple(count(pending.get(field), field) for field in ("sessions", "subscriptions", "items")) + require(active_counts == (2, 1, 3), f"Incomplete active session, subscription or item exercise at cycle {cycle}") + for field in ("browsePoints", "historyPoints"): + count(pending.get(field), field, 1) + stall = take("network-stall-recovered") + require(count(stall.get("cycle"), "recovered stall cycle", 1) == cycle, "Missing or duplicate recovered stall") + seconds(stall.get("seconds"), "network stall time", maximum=MAX_CYCLE_SECONDS) + require(isinstance(stall.get("status"), str) and stall["status"] in STALL_STATUSES, + "Unexpected recovered network stall status") + value = sample(take("resource-observation"), cycle) + require(not samples or value["Seconds"] > samples[-1]["Seconds"], "Nonmonotonic or duplicate sample time") + require(not completions or value["Seconds"] > completions[-1]["elapsedSeconds"], + "Sample precedes completion of the previous cycle") + samples.append(value) + + if cycle >= 3: + baseline = samples[2] + require(all(value[field] == baseline[field] for field in STEADY_FIELDS), + f"Quiescent model or certificate-store inventory changed at cycle {cycle}") + exceeded = [field for field, cap in CAP_FIELDS if value[field] > baseline[field] + BUDGETS[cap]] + if exceeded and first_rejection is None: + rejection = take("budget-rejected") + require(count(rejection.get("cycle"), "rejection cycle", 1) == cycle + and sample(rejection, cycle) == value + and sample({"resources": rejection.get("baseline")}, 3) == baseline, + "First rejection does not match the failing sample and original cycle-three baseline") + seconds(rejection.get("seconds"), "rejection time", minimum=value["Seconds"], + maximum=SECONDS + MAX_CYCLE_SECONDS) + first_rejection = {**rejection, "exceededFields": exceeded, + "assertionMessage": assertion_message(rejection.get("failure"))} + if cycle >= 26: + retained = [row["ManagedBytes"] for row in samples[-24:]] + require(not (all(a < b for a, b in zip(retained, retained[1:])) + and retained[-1] - retained[0] > (16 << 20)), + f"Managed growth guard failed at cycle {cycle}") + + quiescent = take("quiescent-budget-rejected" if first_rejection else "quiescent") + require(sample(quiescent, cycle) == value, "Quiescent sample differs from its pre-assertion observation") + quiescent_seconds = seconds(quiescent.get("cycleSeconds"), "quiescent cycle duration", maximum=MAX_CYCLE_SECONDS) + restart_seconds = 0 + if cycle % 4 == 0: + restart = take("restart") + require(count(restart.get("cycles"), "restart cycle", 1) == cycle + and count(restart.get("restarts"), "restart count", 1) == cycle // 4, "Missing or duplicate active restart") + bounded = seconds(restart.get("boundedRestartSeconds"), "bounded restart duration", maximum=RESTART_SECONDS) + restart_seconds = seconds(restart.get("seconds"), "combined restart duration", minimum=bounded, maximum=MAX_CYCLE_SECONDS) + completion = take("late-slope-cycle") + require(count(completion.get("cycle"), "completed cycle", 1) == cycle + and count(completion.get("restarts"), "completed restarts") == cycle // 4, "Invalid completed cycle ledger") + elapsed = seconds(completion.get("elapsedSeconds"), "cycle completion time", minimum=value["Seconds"], + maximum=SECONDS + MAX_CYCLE_SECONDS) + seconds(completion.get("cycleSeconds"), "complete cycle duration", minimum=quiescent_seconds + restart_seconds, + maximum=MAX_CYCLE_SECONDS) + if first_rejection and first_rejection["cycle"] == cycle: + require(first_rejection["seconds"] <= elapsed, "Rejection follows its completed cycle") + require(not completions or completions[-1]["elapsedSeconds"] < SECONDS, + "Another cycle began after the fixed duration") + completions.append(completion) + pending = next(rows, {}) + + result = pending + require(result.get("kind") == "late-slope-result" and next(rows, None) is None, + "Expected a unique final diagnostic result with no ordinary completion or trailing failure") + require(result.get("mode") == MODE and result.get("acceptance") is False + and "satisfiesTwoHourDuration" not in result and result.get("durationCompleted") is True + and result.get("shutdownVerified") is True and result.get("fatalFailure") is False and result.get("failureTypes") == [], + "Duration, shutdown or diagnostic outcome was incomplete") + cycles = count(result.get("cycles"), "result cycles", 4) + require(cycles == len(samples) and count(result.get("restarts"), "result restarts", 1) == cycles // 4, + "Final cycle or restart count differs from the ledger") + phase = seconds(result.get("cycleSeconds"), "canonical cycle phase", minimum=SECONDS, maximum=SECONDS + MAX_CYCLE_SECONDS) + require(phase >= completions[-1]["elapsedSeconds"] and completions[-1]["elapsedSeconds"] >= SECONDS, + "The final cycle does not cover the fixed duration") + seconds(result.get("shutdownSeconds"), "shutdown duration", maximum=SHUTDOWN_SECONDS) + require(result.get("coldBudgetRejected") is (first_rejection is not None), "The first cold-cap rejection was not preserved") + return start, samples, completions, first_rejection, result + + +def sample_summary(samples, fields): + times = [value["Seconds"] for value in samples] + summary = {"sampleCount": len(samples), "firstSeconds": times[0], "lastSeconds": times[-1], + "spanSeconds": times[-1] - times[0], "maximumGapSeconds": max((b - a for a, b in zip(times, times[1:])), default=0)} + for field in fields: + values = [value[field] for value in samples] + summary[field] = {"first": values[0], "last": values[-1], "min": min(values), "max": max(values), "median": median(values)} + return summary + + +def slope(samples, field): + times = [value["Seconds"] for value in samples] + mean = math.fsum(times) / len(times) + mean_value = math.fsum(value[field] for value in samples) / len(samples) + variance = math.fsum((time - mean) ** 2 for time in times) + return 3600 * math.fsum((time - mean) * (value[field] - mean_value) + for time, value in zip(times, samples)) / variance + + +def describe(samples, completions): + windows = [] + for begin, end in WINDOWS: + subset = [value for value in samples if begin <= value["Seconds"] <= end] + require(len(subset) >= 10, f"Insufficient observations in {begin}–{end} seconds") + summary = sample_summary(subset, MEASUREMENTS) + require(summary["firstSeconds"] <= begin + 60 and summary["lastSeconds"] >= end - 60 + and summary["spanSeconds"] >= 480 and summary["maximumGapSeconds"] <= 60, + f"Incomplete coverage in {begin}–{end} seconds") + windows.append({"windowSeconds": [begin, end], **summary}) + late = [value for value in samples if 1800 <= value["Seconds"] <= SECONDS] + coverage = sample_summary(late, ()) + require(coverage["maximumGapSeconds"] <= 60, "An observation gap exceeds 60 seconds across the late phase") + previous = [row for row in completions if row["elapsedSeconds"] <= 1800] + cycles = len(previous) + restarts = previous[-1]["restarts"] if previous else 0 + work = {"cycles": cycles, "restarts": restarts, "earlierSaturationWorkloadReached": cycles >= 385 and restarts >= 96} + if not work["earlierSaturationWorkloadReached"]: + work["interpretation"] = "Post-restart-step attribution withheld because the earlier workload counts fell short by 1800 seconds." + return { + "allQuiescent": sample_summary(samples, (*ZERO_FIELDS, *STEADY_FIELDS, *MEASUREMENTS)), + "finalQuiescentSample": samples[-1], + "windows": windows, + "leastSquares": {**coverage, "rangeSeconds": [1800, SECONDS], "units": "bytes/hour", + "rss": slope(late, "RssBytes"), "retainedManaged": slope(late, "ManagedBytes")}, + "workCompletedBy1800Seconds": work, + } + + +def check_trx(path, expected_names, failure=None): + document = ET.parse(path).getroot() + counters = document.findall("./t:ResultSummary/t:Counters", TRX) + require(len(counters) == 1, f"Missing unique counters in {path.name}") + counters = counters[0].attrib + total = len(expected_names) + failed = 1 if failure else 0 + expected = {"total": total, "executed": total, "passed": total - failed, "failed": failed, + "error": 0, "timeout": 0, "aborted": 0, "notExecuted": 0} + for key, value in expected.items(): + require(counters.get(key) == str(value), f"Unexpected {key} count in {path.name}") + for key in ("inconclusive", "passedButRunAborted", "notRunnable", "disconnected", "inProgress", "pending"): + require(counters.get(key, "0") == "0", f"Unexpected {key} count in {path.name}") + results = document.findall("./t:Results/t:UnitTestResult", TRX) + require(len(results) == total and {row.get("testName") for row in results} == expected_names, + f"Wrong test identities in {path.name}") + require(all(row.get("outcome") == ("Failed" if failure else "Passed") for row in results), + f"Unexpected test outcome in {path.name}") + if failure: + actual = results[0].findtext("./t:Output/t:ErrorInfo/t:Message", namespaces=TRX) + require(assertion_message(actual) == failure, "TRX failure differs from the first cold-cap assertion") + return {"executed": total, "passed": total - failed, "failed": failed, "skipped": 0} + + +def exit_status(path): + value = path.read_text(encoding="ascii").strip() + require(re.fullmatch(r"0|[1-9][0-9]{0,2}", value) and int(value) <= 255, f"Invalid {path.name}") + return int(value) + + +def inspect_run(path, provenance, wall_seconds): + """Check actual run evidence. This function neither launches Docker nor changes the evidence.""" + start, samples, completions, rejection, result = validate_rows(read_rows(path / "resources.jsonl"), provenance["sourceRevision"]) + wall = seconds(wall_seconds, "container wall time", minimum=SECONDS) + require(start["startupSeconds"] + result["cycleSeconds"] + result["shutdownSeconds"] <= wall, + "Startup, cycle and shutdown phases exceed measured wall time") + test_exit = exit_status(path / "test-exit-status.txt") + container_exit = exit_status(path / "container-exit-status.txt") + require(test_exit == container_exit and ((test_exit != 0) == bool(rejection)), "Original test/container exit was not preserved") + mapping = (path / "target-runtime.txt").read_text(encoding="ascii").splitlines() + require(mapping == [f"targetPid={start['processId']}", f"actualExecutable=/app/{APP}", + "mappedCoreClr=/app/libcoreclr.so", f"coreClrSha256={provenance['coreClrSha256']}"], + "Published Linux apphost or mapped CLR was not verified for the recorded PID") + controls = check_trx(path / "results/descriptor-controls.trx", DESCRIPTOR_TESTS) + workload = check_trx(path / "results/late-slope.trx", {TEST_NAME}, rejection and rejection["assertionMessage"]) + return {"diagnosticOutcome": result, "firstColdRejection": rejection, "descriptorTests": controls, + "experimentTests": workload, "testExitStatus": test_exit, "containerExitStatus": container_exit, + **describe(samples, completions)} + + +def evidence_hashes(path): + result = {} + for name in EVIDENCE_FILES: + file = path / name + require(file.is_file() and file.resolve() == file, f"Missing or linked evidence file {name}") + result[name] = digest(file) + return result + + +def analyze_run(path, bundle_receipt_path, bundle_hash): + require(re.fullmatch(r"[0-9a-f]{64}", bundle_hash) and digest(bundle_receipt_path) == bundle_hash, + "Reviewed bundle receipt hash differs") + bundle = read_object(bundle_receipt_path) + require(bundle.get("kind") == "bundle" and all(type(bundle.get(key)) is type(value) and bundle[key] == value + for key, value in SETTINGS.items()), + "Invalid immutable bundle protocol") + receipt = read_object(path / "run-receipt.json") + provenance = read_object(path / "launch.json") + require(receipt.get("kind") == "run" and provenance.get("kind") == "launch", "Wrong receipt types") + for record in (receipt, provenance): + require(record.get("protocol") == PROTOCOL and record.get("acceptance") is False + and record.get("bundleReceiptSha256") == bundle_hash, "Wrong diagnostic provenance") + for field in ("runId", "sourceRevision", "image", "coreClrSha256", "dockerContext", "command", "containerTimeoutSeconds"): + require(receipt[field] == provenance[field], f"Launch and run receipts disagree on {field}") + require(receipt["sourceRevision"] == bundle["sourceRevision"] and receipt["image"] == bundle["image"] + and receipt["coreClrSha256"] == bundle["files"]["app/libcoreclr.so"] + and receipt["dockerContext"] == bundle["dockerContext"], + "Recorded source, image or runtime differs from the reviewed bundle") + require(re.fullmatch(r"[0-9a-f]{32}", receipt["runId"]) + and re.fullmatch(r"sha256:[0-9a-f]{64}", receipt["image"]) + and re.fullmatch(r"[0-9a-f]{64}", receipt["coreClrSha256"]), "Invalid run identity or immutable provenance") + require(receipt.get("diagnosticCompleted") is True and receipt.get("containerRemoved") is True, + "The launcher did not verify duration, shutdown and container cleanup") + require("validationErrors" not in receipt and provenance["containerTimeoutSeconds"] == CONTAINER_SECONDS + and provenance["command"] == container_command(Path(provenance["bundlePath"]), Path(provenance["outputPath"]), + bundle, receipt["runId"]), "The recorded container invocation differs") + require(receipt.get("evidenceSha256") == evidence_hashes(path), "Recorded evidence hashes differ") + report = inspect_run(path, provenance, receipt["wallSeconds"]) + for field in ("diagnosticOutcome", "testExitStatus", "containerExitStatus", "descriptorTests", "experimentTests"): + require(receipt[field] == report[field], f"Run receipt differs from the raw {field}") + require(receipt.get("launcherExitStatus") == report["testExitStatus"], "The launcher did not preserve the original exit") + return {"protocol": PROTOCOL, "acceptance": False, "bundleReceiptSha256": bundle_hash, + "sourceRevision": receipt["sourceRevision"], "evidenceSha256": receipt["evidenceSha256"], **report} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--run", type=Path, required=True) + parser.add_argument("--bundle-receipt", type=Path, required=True) + parser.add_argument("--bundle-receipt-sha256", required=True) + options = parser.parse_args() + try: + print(json.dumps(analyze_run(options.run.resolve(strict=True), options.bundle_receipt, + options.bundle_receipt_sha256), indent=2, allow_nan=False)) + except (OSError, ValueError, KeyError, TypeError, ET.ParseError) as error: + print(f"Incomplete late-slope evidence: {error}", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/validation/opcua-lifecycle-memory/late-slope/bundle.py b/scripts/validation/opcua-lifecycle-memory/late-slope/bundle.py new file mode 100644 index 00000000..892b1117 --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/late-slope/bundle.py @@ -0,0 +1,129 @@ +"""Checks shared by immutable artifact preparation and launch.""" + +import json +import os +from pathlib import Path +import re +import stat +import subprocess +import sys +from uuid import UUID + +sys.dont_write_bytecode = True + +from analyze import APP, RUNTIME, SETTINGS, digest, local_context, read_object, require + + +TOOLS = ("analyze.py", "bundle.py", "launch.py", "container.sh", "README.md") +TEST_DIRECTORY = f"tests/Server/{APP}" +CANONICAL_SOURCES = tuple(f"{TEST_DIRECTORY}/{name}" for name in ( + "Sdk2LifecycleSoakTests.cs", "Sdk2LifecycleLateSlopeTests.cs", f"{APP}.csproj")) +SDK_PACKAGES = { + f"OPCFoundation.NetStandard.Opc.Ua.{component}/2.0.1" + for component in ("Client", "Configuration", "Core", "Core.Schema", "Core.Types", + "Security.Certificates", "Server", "Types") +} +GATEWAY_PACKAGES = {f"ZB.MOM.WW.HistorianGateway.{component}/1.0.0" for component in ("Client", "Contracts")} + + +def context_identity(name): + require(re.fullmatch(r"[A-Za-z0-9_.-]+", name), "Invalid local Docker context name") + require(not any(os.environ.get(key) for key in ("DOCKER_HOST", "DOCKER_TLS", "DOCKER_TLS_VERIFY", "DOCKER_CERT_PATH")) + and os.environ.get("DOCKER_CONTEXT", name) in ("", name), "Docker endpoint or TLS override is set") + result = subprocess.check_output(["docker", "context", "inspect", name], text=True, timeout=15) + contexts = json.loads(result) + require(len(contexts) == 1 and contexts[0]["Name"] == name, "Expected one explicit Docker context") + identity = {"name": name, "endpoint": contexts[0]["Endpoints"]["docker"]["Host"]} + require(local_context(identity) and Path(identity["endpoint"][len("unix://"):]).is_socket(), + "The reviewed Docker context must use a local Unix socket") + return identity + + +def image_identity(image, context): + require(re.fullmatch(r"sha256:[0-9a-f]{64}", image), "Use a locally cached immutable image ID") + result = subprocess.check_output(["docker", "--host", context["endpoint"], "image", "inspect", image], text=True, timeout=15) + images = json.loads(result) + require(len(images) == 1, "Expected one cached image") + value = images[0] + require((value["Id"], value["Os"], value["Architecture"]) == (image, "linux", "arm64"), + "The immutable image must be Linux ARM64") + return {"id": image, "os": "linux", "architecture": "arm64"} + + +def manifest(root): + """Hash the exact regular-file tree and exclude only the receipt that contains it.""" + result = {} + for path in sorted(root.rglob("*")): + kind = path.lstat().st_mode + require(stat.S_ISDIR(kind) or stat.S_ISREG(kind), f"Unsupported artifact type {path.relative_to(root)}") + if stat.S_ISREG(kind) and path != root / "receipt.json": + result[path.relative_to(root).as_posix()] = digest(path) + return result + + +def check_elf(path): + with path.open("rb") as stream: + header = stream.read(20) + require(header[:6] == b"\x7fELF\x02\x01" and header[18:20] == b"\xb7\x00", + f"{path.name} is not ELF64 AArch64") + + +def check_application(app): + check_elf(app / APP) + require(os.access(app / APP, os.X_OK), "The Linux apphost is not executable") + check_elf(app / "libcoreclr.so") + options = read_object(app / f"{APP}.runtimeconfig.json")["runtimeOptions"] + frameworks = options["includedFrameworks"] + require(options["tfm"] == "net10.0" and "framework" not in options and "frameworks" not in options + and isinstance(frameworks, list) and any(value["name"] == "Microsoft.NETCore.App" for value in frameworks) + and all(value["version"] == RUNTIME for value in frameworks), "Wrong self-contained application runtime") + dependencies = read_object(app / f"{APP}.deps.json") + require(dependencies["runtimeTarget"]["name"] == ".NETCoreApp,Version=v10.0/linux-arm64", "Wrong application RID") + libraries = dependencies["libraries"] + sdk = {key for key in libraries if key.startswith("OPCFoundation.")} + gateway = {key for key in libraries if key.startswith("ZB.MOM.WW.HistorianGateway.")} + require(sdk == SDK_PACKAGES and gateway == GATEWAY_PACKAGES, "Expected exact SDK 2.0.1 and HistorianGateway 1.0.0 graph") + return sorted(sdk), sorted(gateway) + + +def check_modules(modules, files): + require(isinstance(modules, list), "Missing managed module identities") + seen = set() + for module in modules: + name = module["file"] + require(name in files and name.startswith("app/") and name.endswith(".dll") and name not in seen, + "Unexpected or duplicate managed module file") + require(module["assembly"] and re.fullmatch(r"\d+(?:\.\d+){3}", module["version"]) + and UUID(module["mvid"]).int != 0, "Invalid assembly version or MVID") + seen.add(name) + required = {f"app/{APP}.dll", "app/Opc.Ua.Server.dll", "app/System.Private.CoreLib.dll", + "app/ZB.MOM.WW.HistorianGateway.Client.dll", "app/ZB.MOM.WW.HistorianGateway.Contracts.dll"} + require(required <= seen, "Missing application, runtime or dependency module identity") + + +def verify_bundle(root, expected_hash): + path = root / "receipt.json" + require(path.is_file() and not path.is_symlink() and re.fullmatch(r"[0-9a-f]{64}", expected_hash) + and digest(path) == expected_hash, "Reviewed bundle receipt hash differs") + receipt = read_object(path) + require(receipt.get("kind") == "bundle" and all(type(receipt.get(key)) is type(value) and receipt[key] == value + for key, value in SETTINGS.items()), "Invalid fixed diagnostic settings") + require(re.fullmatch(r"[0-9a-f]{40}", receipt["sourceRevision"]), "Invalid committed source revision") + files = manifest(root) + require(receipt["files"] == files, "The immutable bundle file set or a file hash changed") + require({name for name in files if name.startswith("tools/")} == {f"tools/{name}" for name in TOOLS}, + "Unexpected bundled tools") + sdk, gateway = check_application(root / "app") + require(receipt["sdkLibraries"] == sdk and receipt["gatewayLibraries"] == gateway, + "Dependency receipt differs from the published application") + check_modules(receipt["managedModules"], files) + require(receipt["imageIdentity"] == {"id": receipt["image"], "os": "linux", "architecture": "arm64"} + and re.fullmatch(r"sha256:[0-9a-f]{64}", receipt["image"]), "Invalid immutable image identity") + require(local_context(receipt["dockerContext"]), "Invalid reviewed local Docker context") + require(receipt["sourceFiles"] and all(re.fullmatch(r"[0-9a-f]{64}", value) for value in receipt["sourceFiles"].values()), + "Missing committed source hashes") + require(set(CANONICAL_SOURCES) <= receipt["sourceFiles"].keys(), "Missing canonical test or project hash") + for name in TOOLS: + source = "scripts/validation/opcua-lifecycle-memory/late-slope/" + name + require(receipt["sourceFiles"].get(source) == files[f"tools/{name}"], f"Bundled {name} differs from source") + return receipt diff --git a/scripts/validation/opcua-lifecycle-memory/late-slope/container.sh b/scripts/validation/opcua-lifecycle-memory/late-slope/container.sh new file mode 100644 index 00000000..8cb84478 --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/late-slope/container.sh @@ -0,0 +1,70 @@ +#!/usr/bin/env bash +# Execute the immutable self-contained apphost through the cached Linux VSTest controller. +set -euo pipefail +umask 077 +app=ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests +test_class="$app.Sdk2LifecycleSoakTests" +mkdir -p /task/tmp /task/data /task/cli /task/results +export TMPDIR=/task/tmp XDG_DATA_HOME=/task/data DOTNET_CLI_HOME=/task/cli +export OTOPCUA_SDK2_SOAK=1 OTOPCUA_SDK2_SOAK_MODE=late-slope OTOPCUA_SDK2_SOAK_SECONDS=7200 +export OTOPCUA_SDK2_SOAK_OUTPUT=/task/resources.jsonl OPCUA_CERTIFICATE_LEAK_TRACKING=1 +export DOTNET_CLI_TELEMETRY_OPTOUT=1 DOTNET_NOLOGO=1 +[[ $(uname -s) == Linux && $(uname -m) == aarch64 && -d /proc/self/fd ]] +[[ ${OTOPCUA_SOAK_BUILD_SHA:-} =~ ^[0-9a-f]{40}$ ]] +dotnet --info > /task/controller-runtime.txt + +dotnet vstest "/app/$app.dll" \ + --TestCaseFilter:"FullyQualifiedName~$test_class.Descriptor_probe_" \ + --ResultsDirectory:/task/results '--Logger:trx;LogFileName=descriptor-controls.trx' > /task/descriptor-controls.log 2>&1 + +# Verify actual execution before starting the long test. The host parses the complete TRX independently. +controls=/task/results/descriptor-controls.trx +counters=$(grep -E '^[[:space:]]*/dev/null || true; fi +} +trap cleanup EXIT +dotnet vstest "/app/$app.dll" \ + --TestCaseFilter:"FullyQualifiedName=$test_class.Repeated_lifecycles_return_resources_to_a_steady_baseline" \ + --ResultsDirectory:/task/results '--Logger:trx;LogFileName=late-slope.trx' > /task/test.log 2>&1 & +vstest_pid=$! + +# The canonical startup deadline is 120 seconds. Allow controller startup before observing its first complete row. +proof_deadline=$((SECONDS + 150)) +observed=0 +while ((SECONDS < proof_deadline)) && kill -0 "$vstest_pid" 2>/dev/null; do + if [[ -s /task/resources.jsonl ]] && IFS= read -r header < /task/resources.jsonl; then + [[ $header == *'"kind":"start"'* && $header == *'"mode":"late-slope"'* ]] + pid=$(sed -n 's/.*"processId":\([1-9][0-9]*\),.*/\1/p' <<< "$header") + [[ $pid =~ ^[1-9][0-9]*$ ]] + executable=$(readlink "/proc/$pid/exe") + [[ $executable == "/app/$app" ]] + runtime=$(awk '$6 ~ /\/libcoreclr.so$/ {print $6}' "/proc/$pid/maps" | sort -u) + [[ $runtime == /app/libcoreclr.so ]] + runtime_hash=$(sha256sum "$runtime") + printf 'targetPid=%s\nactualExecutable=%s\nmappedCoreClr=%s\ncoreClrSha256=%s\n' \ + "$pid" "$executable" "$runtime" "${runtime_hash%% *}" > /task/target-runtime.txt + observed=1 + break + fi + sleep 0.1 +done +if [[ $observed != 1 ]]; then kill "$vstest_pid" 2>/dev/null || true; fi +test_status=0 +wait "$vstest_pid" || test_status=$? +vstest_pid= +printf '%s\n' "$test_status" > /task/test-exit-status.txt +if [[ $test_status != 0 ]]; then exit "$test_status"; fi +[[ $observed == 1 ]] diff --git a/scripts/validation/opcua-lifecycle-memory/late-slope/launch.py b/scripts/validation/opcua-lifecycle-memory/late-slope/launch.py new file mode 100644 index 00000000..5328303e --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/late-slope/launch.py @@ -0,0 +1,130 @@ +#!/usr/bin/env python3 +"""Verify an immutable bundle or launch its fixed-duration Linux diagnostic.""" + +import argparse +import json +from pathlib import Path +import re +import subprocess +import sys +import time +import uuid +import xml.etree.ElementTree as ET + +sys.dont_write_bytecode = True + +from analyze import CONTAINER_SECONDS, DOCKER_LABEL, PROTOCOL, container_command, evidence_hashes, inspect_run, require +from bundle import context_identity, image_identity, verify_bundle + + +def remove_container(context, run_id): + docker = ["docker", "--host", context["endpoint"]] + + def containers(): + result = subprocess.check_output([*docker, "ps", "--all", "--quiet", "--no-trunc", + "--filter", f"label={DOCKER_LABEL}={run_id}"], text=True, timeout=10) + ids = result.split() + require(all(re.fullmatch(r"[0-9a-f]{64}", value) for value in ids), "Unexpected container ID") + return ids + + ids = containers() + if ids: + subprocess.run([*docker, "rm", "--force", *ids], stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, timeout=15) + require(not containers(), "The task-owned container is still present") + + +def write(path, value): + with path.open("x", encoding="utf-8") as stream: + json.dump(value, stream, indent=2, allow_nan=False) + stream.write("\n") + + +def verify_local_image(bundle): + context = bundle["dockerContext"] + require(context_identity(context["name"]) == context, "Reviewed Docker context endpoint changed") + require(image_identity(bundle["image"], context) == bundle["imageIdentity"], "Wrong cached image identity") + + +def launch(root, bundle, bundle_hash, output): + require(Path(__file__).resolve() == root / "tools/launch.py", "Invoke the reviewed bundled launcher") + require(not output.is_relative_to(root), "Run output must be outside the immutable bundle") + run_id = uuid.uuid4().hex + invocation = container_command(root, output, bundle, run_id) + output.mkdir(mode=0o700, parents=True, exist_ok=False) + provenance = {"kind": "launch", "protocol": PROTOCOL, "acceptance": False, "runId": run_id, + "bundleReceiptSha256": bundle_hash, "sourceRevision": bundle["sourceRevision"], "image": bundle["image"], + "coreClrSha256": bundle["files"]["app/libcoreclr.so"], "command": invocation, + "dockerContext": bundle["dockerContext"], "bundlePath": str(root), "outputPath": str(output), + "containerTimeoutSeconds": CONTAINER_SECONDS} + write(output / "launch.json", provenance) + errors = [] + removed = False + started = time.monotonic() + try: + with (output / "container.log").open("x", encoding="utf-8") as log: + process = subprocess.run(invocation, stdout=log, stderr=subprocess.STDOUT, timeout=CONTAINER_SECONDS) + status = process.returncode + except subprocess.TimeoutExpired: + status = 124 + errors.append("The container reached the fixed outer deadline") + except KeyboardInterrupt: + status = 130 + errors.append("Container execution was interrupted") + except OSError as error: + status = 125 + errors.append(str(error)) + finally: + wall = time.monotonic() - started + try: + remove_container(bundle["dockerContext"], run_id) + removed = True + except (OSError, subprocess.SubprocessError, ValueError) as error: + errors.append(f"Container cleanup failed: {error}") + (output / "container-exit-status.txt").write_text(f"{status}\n", encoding="ascii") + receipt = {**provenance, "kind": "run", "containerExitStatus": status, "wallSeconds": wall, + "containerRemoved": removed, "diagnosticCompleted": False} + try: + verify_bundle(root, bundle_hash) + verify_local_image(bundle) + receipt["evidenceSha256"] = evidence_hashes(output) + inspected = inspect_run(output, provenance, wall) + for field in ("diagnosticOutcome", "testExitStatus", "descriptorTests", "experimentTests"): + receipt[field] = inspected[field] + except (OSError, KeyError, TypeError, ValueError, ET.ParseError, subprocess.SubprocessError) as error: + errors.append(f"Evidence validation failed: {error}") + receipt["diagnosticCompleted"] = not errors + if errors: + receipt["validationErrors"] = errors + # Return the original container status when the test rethrows its first resource cap. + receipt["launcherExitStatus"] = (128 - status if status < 0 else status) or int(bool(errors)) + write(output / "run-receipt.json", receipt) + print(json.dumps(receipt, indent=2, allow_nan=False)) + return receipt["launcherExitStatus"] + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--bundle", type=Path, required=True) + parser.add_argument("--receipt-sha256", required=True) + action = parser.add_mutually_exclusive_group(required=True) + action.add_argument("--verify-only", action="store_true") + action.add_argument("--output", type=Path) + options = parser.parse_args() + try: + root = options.bundle.resolve(strict=True) + bundle = verify_bundle(root, options.receipt_sha256) + verify_local_image(bundle) + if options.verify_only: + print(json.dumps({"protocol": PROTOCOL, "sourceRevision": bundle["sourceRevision"], + "receiptSha256": options.receipt_sha256, "verified": True, + "acceptance": False, "workloadLaunched": False}, indent=2)) + return 0 + return launch(root, bundle, options.receipt_sha256, options.output.resolve()) + except (OSError, KeyError, TypeError, ValueError, subprocess.SubprocessError) as error: + print(f"Late-slope launcher rejected the request: {error}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/validation/opcua-lifecycle-memory/late-slope/prepare.py b/scripts/validation/opcua-lifecycle-memory/late-slope/prepare.py new file mode 100644 index 00000000..4f20eaf5 --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/late-slope/prepare.py @@ -0,0 +1,97 @@ +#!/usr/bin/env python3 +"""Publish and record a fresh immutable Linux ARM64 bundle without running the workload.""" + +import argparse +from datetime import datetime, timezone +import json +from pathlib import Path +import shutil +import subprocess +import sys +import tempfile +import xml.etree.ElementTree as ET + +sys.dont_write_bytecode = True + +from analyze import APP, RUNTIME, SETTINGS, digest, require +from bundle import CANONICAL_SOURCES, TOOLS, check_application, context_identity, image_identity, manifest, verify_bundle + + +HERE = Path(__file__).resolve().parent +REPO = HERE.parents[3] +PROJECT = REPO / "tests/Server" / APP / (APP + ".csproj") +SUPPORT = PROJECT.with_name("Sdk2LifecycleLateSlopeTests.cs") +MODULE_SOURCE = HERE.parent / "ArtifactModuleInfo.cs" +DESIGN = HERE.parent / "LATE_SLOPE.md" + + +def check_source(revision): + current = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=REPO, text=True).strip() + require(revision == current, "Publish the specified committed source revision") + require(not subprocess.check_output(["git", "status", "--porcelain"], cwd=REPO, text=True).strip(), + "The publication checkout must be clean") + require('"late-slope-result"' in SUPPORT.read_text(encoding="utf-8"), "The committed canonical test lacks late-slope mode") + sources = [*(REPO / name for name in CANONICAL_SOURCES), MODULE_SOURCE, DESIGN, + HERE / "prepare.py", HERE / "test_offline.py", *(HERE / name for name in TOOLS)] + return {path.relative_to(REPO).as_posix(): digest(path) for path in sources} + + +def managed_modules(build, root): + project = ET.Element("Project", Sdk="Microsoft.NET.Sdk") + settings = ET.SubElement(project, "PropertyGroup") + for key, value in {"OutputType": "Exe", "TargetFramework": "net10.0", "ImplicitUsings": "enable", + "EnableDefaultCompileItems": "false", "TreatWarningsAsErrors": "true"}.items(): + ET.SubElement(settings, key).text = value + ET.SubElement(ET.SubElement(project, "ItemGroup"), "Compile", Include=str(MODULE_SOURCE)) + path = build / "module-info.csproj" + ET.ElementTree(project).write(path, encoding="unicode") + with (root / "module-info-build.log").open("w", encoding="utf-8") as log: + subprocess.run(["dotnet", "build", str(path), "-c", "Release", "-m:2", "--nologo"], + stdout=log, stderr=subprocess.STDOUT, check=True, timeout=300) + command = ["dotnet", str(build / "bin/Release/net10.0/module-info.dll"), str(root), str(root / "app")] + return json.loads(subprocess.check_output(command, text=True, timeout=60)) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--source-revision", required=True) + parser.add_argument("--image", required=True) + parser.add_argument("--docker-context", required=True) + options = parser.parse_args() + sources = check_source(options.source_revision) + context = context_identity(options.docker_context) + image = image_identity(options.image, context) + root = options.output.resolve() + root.mkdir(mode=0o700, parents=True, exist_ok=False) + with tempfile.TemporaryDirectory(prefix="opcua-late-slope-publish-") as scratch: + build = Path(scratch) + command = ["dotnet", "publish", str(PROJECT), "-c", "Release", "-r", "linux-arm64", + "--self-contained", "true", f"-p:RuntimeFrameworkVersion={RUNTIME}", + "-p:PublishSingleFile=false", "-p:PublishTrimmed=false", "-p:UseAppHost=true", + "--artifacts-path", str(build / "artifacts"), "-o", str(root / "app"), "-m:2", "--nologo"] + with (root / "publish.log").open("w", encoding="utf-8") as log: + subprocess.run(command, cwd=REPO, stdout=log, stderr=subprocess.STDOUT, check=True, timeout=900) + sdk, gateway = check_application(root / "app") + modules = managed_modules(build, root) + tools = root / "tools" + tools.mkdir() + for name in TOOLS: + shutil.copy2(HERE / name, tools / name) + require(check_source(options.source_revision) == sources, "Committed inputs changed during publication") + require(context_identity(options.docker_context) == context and image_identity(options.image, context) == image, + "The local Docker context or image changed during publication") + receipt = {**SETTINGS, "kind": "bundle", "sourceRevision": options.source_revision, "sourceFiles": sources, "dockerContext": context, + "image": options.image, "imageIdentity": image, "preparedUtc": datetime.now(timezone.utc).isoformat(), + "publishCommand": command, "sdkLibraries": sdk, "gatewayLibraries": gateway, + "managedModules": modules, "files": manifest(root)} + path = root / "receipt.json" + path.write_text(json.dumps(receipt, indent=2, allow_nan=False) + "\n", encoding="utf-8") + sha = digest(path) + verify_bundle(root, sha) + print(json.dumps({"bundle": str(root), "sourceRevision": options.source_revision, + "receiptSha256": sha, "acceptance": False, "workloadLaunched": False}, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/scripts/validation/opcua-lifecycle-memory/late-slope/test_offline.py b/scripts/validation/opcua-lifecycle-memory/late-slope/test_offline.py new file mode 100644 index 00000000..f8d270e3 --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/late-slope/test_offline.py @@ -0,0 +1,368 @@ +#!/usr/bin/env python3 +"""Synthetic evidence controls. No application, Docker, or workload execution.""" + +import json +import os +from pathlib import Path +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 + + +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): + for isolation in ("network", "read-only rootfs"): + with self.subTest(isolation=isolation): + self.build_evidence() + invocation = self.provenance["command"].copy() + if isolation == "network": + invocation[invocation.index("--network") + 1] = "host" + else: + invocation.remove("--read-only") + 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_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)