lmxopcua/scripts/validation/opcua-lifecycle-memory/run.py

#!/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())