lmxopcua/scripts/validation/opcua-lifecycle-memory/late-slope/bundle.py

"""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