#!/usr/bin/env python3 """Build opt-in controls against existing published DLLs; never rebuild product code.""" import argparse import hashlib import json import os from pathlib import Path import secrets import subprocess import sys import time import xml.etree.ElementTree as ET def digest(path): return hashlib.sha256(path.read_bytes()).hexdigest() def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("kind", choices=["lifecycle", "certificate", "telemetry"]) parser.add_argument("--references", type=Path, action="append", required=True, help="Existing published DLL directory; first directory wins duplicate names") parser.add_argument("--output", type=Path, required=True, help="New local artifact directory") parser.add_argument("--sdk2", action="store_true") parser.add_argument("--seconds", type=int, default=600) parser.add_argument("--mode", help="Certificate: public/pfx/shared-copy/sdk-copy; telemetry: dispose/undisposed") parser.add_argument("--build-only", action="store_true") parser.add_argument("--certificate-tracking", action="store_true", help="SDK2 opt-in allocation-stack diagnostics") options = parser.parse_args() if options.kind != "lifecycle" and not options.sdk2: parser.error("certificate and telemetry controls require --sdk2") if not 1 <= options.seconds <= 1800: parser.error("Diagnostic duration must be 1..1800 seconds; this is never a full soak") options.output = options.output.resolve() options.output.mkdir(parents=True, exist_ok=False) names = {"lifecycle": "LifecycleControl.cs", "certificate": "CertificateControl.cs", "telemetry": "TelemetryControl.cs"} source = Path(__file__).resolve().parent / names[options.kind] references = {} for directory in options.references: directory = directory.resolve(strict=True) for dll in sorted(directory.glob("*.dll")): references.setdefault(dll.name, dll) if "Opc.Ua.Core.dll" not in references: parser.error("References must contain actual OPC UA managed assemblies") project = ET.Element("Project", Sdk="Microsoft.NET.Sdk") properties = ET.SubElement(project, "PropertyGroup") for key, value in {"OutputType": "Exe", "TargetFramework": "net10.0", "ImplicitUsings": "enable", "Nullable": "enable", "EnableDefaultCompileItems": "false", "DefineConstants": "SDK2" if options.sdk2 else ""}.items(): ET.SubElement(properties, key).text = value items = ET.SubElement(project, "ItemGroup") ET.SubElement(items, "FrameworkReference", Include="Microsoft.AspNetCore.App") ET.SubElement(items, "Compile", Include=str(source)) for name, path in references.items(): reference = ET.SubElement(items, "Reference", Include=path.stem) ET.SubElement(reference, "HintPath").text = str(path) project_path = options.output / "control.csproj" ET.indent(project) ET.ElementTree(project).write(project_path, encoding="unicode") manifest = {"kind": options.kind, "sdk2": options.sdk2, "seconds": options.seconds, "fullSoakGate": False, "source": {source.name: digest(source)}, "referencePrecedence": "first directory wins duplicate names", "references": {name: digest(path) for name, path in references.items()}} (options.output / "provenance.json").write_text(json.dumps(manifest, indent=2) + "\n") with (options.output / "build.log").open("w") as log: result = subprocess.run(["dotnet", "build", str(project_path), "-c", "Release", "-m:2"], stdout=log, stderr=subprocess.STDOUT) if result.returncode: return result.returncode if options.build_only: return 0 executable = options.output / "bin/Release/net10.0/control.dll" environment = os.environ.copy() environment["OPCUA_MEMORY_PASSWORD"] = secrets.token_urlsafe(32) if options.certificate_tracking: environment["OPCUA_CERTIFICATE_LEAK_TRACKING"] = "1" if options.kind == "lifecycle": command = ["dotnet", str(executable), str(options.output / "private-fixture"), str(options.seconds)] with (options.output / "samples.jsonl").open("w") as log: result = subprocess.run(command, env=environment, stdout=log, stderr=subprocess.STDOUT, timeout=options.seconds + 180) status = result.returncode elif options.kind == "telemetry": if options.mode not in ["dispose", "undisposed"]: parser.error("Telemetry mode must be dispose or undisposed") with (options.output / "samples.jsonl").open("w") as log: result = subprocess.run(["dotnet", str(executable), options.mode], env=environment, stdout=log, stderr=subprocess.STDOUT, timeout=180) status = result.returncode else: if sys.platform != "darwin": parser.error("Native certificate control currently requires macOS heap/vmmap") if options.mode not in ["public", "pfx", "shared-copy", "sdk-copy"]: parser.error("Certificate mode must be public, pfx, shared-copy or sdk-copy") status = certificate_control(executable, options.output, options.mode, environment) (options.output / "exit-status.txt").write_text(str(status) + "\n") return status def certificate_control(executable, output, mode, environment): # Native snapshots contain process metadata; retain locally, do not publish them unredacted. with (output / "samples.jsonl").open("w") as log: process = subprocess.Popen(["dotnet", str(executable), mode, str(output)], env=environment, stdout=log, stderr=subprocess.STDOUT) try: deadline = time.monotonic() + 180 while not (output / "samples.jsonl").stat().st_size: if process.poll() is not None or time.monotonic() > deadline: raise RuntimeError("Certificate control failed before ready") time.sleep(.05) for label in ["before", "after"]: if label == "after": (output / "go").touch() while True: rows = (output / "samples.jsonl").read_text().splitlines() if rows and json.loads(rows[-1]).get("kind") == "done": break if process.poll() is not None or time.monotonic() > deadline: raise RuntimeError("Certificate control did not finish") time.sleep(.1) with (output / ("heap-" + label + ".txt")).open("w") as log_heap: subprocess.run(["/usr/bin/heap", "--noContent", "-s", str(process.pid)], stdout=log_heap, stderr=subprocess.STDOUT, timeout=20, check=True) (output / "stop").touch() return process.wait(timeout=10) finally: if process.poll() is None: process.terminate() process.wait(timeout=10) if __name__ == "__main__": raise SystemExit(main())