NODEDC_DEVICE_CORE/infra/deploy-runner/nodedc-edge-deploy

#!/usr/bin/env python3
"""Canonical data-only deploy runner for the dedicated NODE.DC Device Edge."""

from __future__ import annotations

import hashlib
import json
import os
import re
import select
import shutil
import socket
import struct
import subprocess
import sys
import tarfile
import tempfile
import time
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath


RUNNER_PATH = Path("/usr/local/sbin/nodedc-edge-deploy")
LIVE_ROOT = Path("/home/ndcsudo/nodedc-device-edge/source")
INBOX_ROOT = Path("/home/ndcsudo/nodedc-device-edge/deploy/inbox")
STATE_ROOT = Path("/var/lib/nodedc-edge-deploy")
APPLIED_ROOT = STATE_ROOT / "applied"
FAILED_ROOT = STATE_ROOT / "failed"
BACKUP_ROOT = STATE_ROOT / "backups"
APPLIED_JOURNAL = STATE_ROOT / "state/applied.jsonl"
FAILED_JOURNAL = STATE_ROOT / "state/failed.jsonl"
DEPLOY_LOCK = STATE_ROOT / "state/deploy.lock"

DOCKER = "/usr/bin/docker"
COMPONENT = "device-edge"
ARTIFACT_TYPE = "app-overlay"
PATCH_ID_RE = re.compile(r"^[A-Za-z0-9._-]{1,96}$")
MAX_ARTIFACT_BYTES = 16 * 1024 * 1024

COMPOSE_PROJECT = "nodedc-device-edge"
BASE_COMPOSE = LIVE_ROOT / "docker-compose.device-edge.yml"
INGRESS_COMPOSE = LIVE_ROOT / "docker-compose.device-edge.ingress.yml"
RELAY_SERVICE = "device-edge-relay"
RELAY_CONTAINER = "nodedc-device-edge-device-edge-relay-1"
BACKHAUL_CONTAINER = "nodedc-device-edge-device-edge-backhaul-1"
TAILNET_CONTAINER = "nodedc-device-edge-tailnet-1"
RELAY_IMAGE = "nodedc/device-edge-relay:local"

INGRESS_PARENT = "enp1s0f0"
INGRESS_SUBNET = "192.168.68.0/22"
INGRESS_GATEWAY = "192.168.68.1"
INGRESS_IPV4 = "192.168.71.253"
INGRESS_PORT = 9921
INGRESS_NETWORK = "nodedc-device-edge-ingress"
INGRESS_IPV4_APPROVED = True
INGRESS_IPV4_APPROVAL = "approved-outside-dhcp-pool"

ENTRIES = (
    "docker-compose.device-edge.yml",
    "docker-compose.device-edge.ingress.yml",
    "services/device-edge-relay/Dockerfile",
    "services/device-edge-relay/src",
    "deployment/device-edge-admission-gate-v1.json",
)

PAYLOAD_FILE_SHA256 = {
    "docker-compose.device-edge.yml":
        "666945ffd9512355e610ecd36a9df96936477315150555def93e0243e8ff1e22",
    "docker-compose.device-edge.ingress.yml":
        "11bedfd7fdea749ca1bdb3b35b9c136c86b330f51a9001f0b38c4618f6f96108",
    "services/device-edge-relay/Dockerfile":
        "f2f15b7618ac2ab3a1d4dd40041d06d1e9a695edcfc168d8835f9560b4897e70",
    "services/device-edge-relay/src/runtime.mjs":
        "21e83678980aa61127bf9f3d77982dd485c4aaae208c43818db7bb1cc150b83a",
    "services/device-edge-relay/src/server.mjs":
        "1b99ec944f1d3fbadded045b159f08624e2829620cec39a97f6b4b8cdcd2be22",
    "deployment/device-edge-admission-gate-v1.json":
        "e6c1f21ff297b451c42b6746bc2063484874435dfa9f1614410a7cbe84f0ce6f",
}

PREDECESSOR_FILE_SHA256 = {
    "docker-compose.device-edge.yml":
        "7f13c11d6d4d541964053c0a8cf791e401947d34c42e0f7c26f9f9df26fa00b5",
    "docker-compose.device-edge.ingress.yml":
        "a4afd04755530fc3b9be64d1a65f0f7282a9539bcc1985bfd880aa904e1c4d8f",
    "services/device-edge-relay/Dockerfile":
        "f2f15b7618ac2ab3a1d4dd40041d06d1e9a695edcfc168d8835f9560b4897e70",
    "services/device-edge-relay/src/runtime.mjs":
        "ae8bf8b55603bab266b6fa6e9bc65c9f310a9d94a54db04e2130704e38622ffc",
    "services/device-edge-relay/src/server.mjs":
        "e4b051b74f934bd37322440e6a013fb6774a76607da08f9cc1e844fc109c83c1",
    "deployment/device-edge-ingress-ipvlan-v1.json":
        "b9ce402db0c059a76f07a8d4a34297aff2250fd1c0d1aed9970b3a88f4e75d7f",
}

PREDECESSOR_ABSENT = {
    "deployment/device-edge-admission-gate-v1.json",
}


class DeployError(RuntimeError):
    pass


def die(message: str) -> None:
    raise DeployError(message)


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def run(command, *, check=True, capture=True, cwd=None, timeout=180):
    result = subprocess.run(
        [str(value) for value in command],
        check=False,
        capture_output=capture,
        text=True,
        cwd=str(cwd) if cwd else None,
        timeout=timeout,
    )
    if check and result.returncode != 0:
        detail = (result.stderr or result.stdout or "command failed").strip()
        die(f"command failed: {command[0]}: {detail}")
    return result


def docker_json(*args):
    result = run([DOCKER, *args])
    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError as error:
        die(f"Docker JSON response invalid: {error}")


def expected_descriptor():
    return {
        "schemaVersion": "nodedc.device-edge.admission-gate.v1",
        "mode": "single-nic-ipvlan-b2-relay-only",
        "runtimeHost": "ndcmini12",
        "component": COMPONENT,
        "selectedServices": [RELAY_SERVICE],
        "preservedServices": ["device-edge-backhaul", "tailnet"],
        "composeProject": COMPOSE_PROJECT,
        "composeFiles": [
            "docker-compose.device-edge.yml",
            "docker-compose.device-edge.ingress.yml",
        ],
        "parentInterface": INGRESS_PARENT,
        "lanSubnet": INGRESS_SUBNET,
        "lanGateway": INGRESS_GATEWAY,
        "ingressIpv4": INGRESS_IPV4,
        "ingressIpv4Approval": INGRESS_IPV4_APPROVAL,
        "ingressNetwork": INGRESS_NETWORK,
        "deviceTcpListen": f"{INGRESS_IPV4}:{INGRESS_PORT}",
        "hostPortPublication": "disabled",
        "healthPublication": "disabled",
        "privateUpstream": "device-edge-backhaul:19921",
        "sourceAdmission": "public-ipv4-only",
        "maxTrackedSourceAddresses": 2048,
        "maxBytesPerDirection": 262144,
        "protocolInspection": "gateway-owned",
        "identityTrust": "claimed-not-ownership-proof",
        "discoveryLifecycle": "quarantine",
        "commandTransport": "disabled",
        "gelios": "untouched",
        "amneziaHostFullTunnel": "preserved",
        "routerNatFirewall": "separate-manual-gate",
        "rollback": "restore-reviewed-ipvlan-predecessor-without-network-or-router-mutation",
    }


def assert_root():
    if os.geteuid() != 0:
        die("nodedc-edge-deploy must run as root")


def assert_regular_nonsymlink(path: Path, label: str):
    if not path.exists() or path.is_symlink() or not path.is_file():
        die(f"{label} must be a regular non-symlink file")


def parse_manifest(raw: str):
    values = {}
    for line in raw.splitlines():
        if not line or "=" not in line:
            die("artifact manifest is malformed")
        key, value = line.split("=", 1)
        if key in values or key not in {"id", "component", "type"}:
            die("artifact manifest key set is invalid")
        values[key] = value
    if set(values) != {"id", "component", "type"}:
        die("artifact manifest key set is incomplete")
    if not PATCH_ID_RE.fullmatch(values["id"]):
        die("artifact patch id is invalid")
    if values["component"] != COMPONENT or values["type"] != ARTIFACT_TYPE:
        die("artifact component/type mismatch")
    return values


def safe_tar_member(member: tarfile.TarInfo):
    path = PurePosixPath(member.name)
    if path.is_absolute() or ".." in path.parts or not path.parts:
        die("artifact contains an unsafe path")
    if not (member.isfile() or member.isdir()):
        die("artifact contains a non-file/non-directory member")
    lowered = {part.lower() for part in path.parts}
    if any(
        part.startswith(".env")
        or part in {
            ".git",
            "node_modules",
            "secrets",
            "keys",
            "trust",
            "runtime",
            "logs",
            "uploads",
        }
        for part in lowered
    ):
        die("artifact contains a forbidden boundary")
    if any(part.startswith("._") for part in path.parts):
        die("artifact contains AppleDouble metadata")


def load_artifact(artifact: Path, extraction_root: Path):
    artifact = artifact.resolve(strict=True)
    if artifact.parent != INBOX_ROOT.resolve(strict=True):
        die("artifact must be an explicit file in the Device Edge inbox")
    assert_regular_nonsymlink(artifact, "artifact")
    if artifact.suffix != ".tgz" or artifact.stat().st_size > MAX_ARTIFACT_BYTES:
        die("artifact extension/size rejected")

    seen = set()
    with tarfile.open(artifact, "r:gz") as archive:
        for member in archive.getmembers():
            safe_tar_member(member)
            if member.name in seen:
                die("artifact contains duplicate members")
            seen.add(member.name)
        required = {"manifest.env", "files.txt", "payload"}
        if not required.issubset(seen):
            die("artifact top-level contract is incomplete")
        if any(name.split("/", 1)[0] not in required for name in seen):
            die("artifact contains an unexpected top-level member")
        archive.extractall(extraction_root, filter="data")

    manifest = parse_manifest(
        (extraction_root / "manifest.env").read_text(encoding="utf-8")
    )
    entries = tuple(
        line for line in
        (extraction_root / "files.txt").read_text(encoding="utf-8").splitlines()
        if line
    )
    if entries != ENTRIES or len(entries) != len(set(entries)):
        die("Device Edge artifact file selection mismatch")
    payload = extraction_root / "payload"
    validate_payload(payload)
    return manifest, entries, payload, sha256_file(artifact), artifact


def validate_payload(payload: Path):
    actual_files = {
        path.relative_to(payload).as_posix(): sha256_file(path)
        for path in payload.rglob("*")
        if path.is_file()
    }
    if actual_files != PAYLOAD_FILE_SHA256:
        die("Device Edge artifact payload digest set mismatch")
    descriptor = json.loads(
        (payload / "deployment/device-edge-admission-gate-v1.json")
        .read_text(encoding="utf-8")
    )
    if descriptor != expected_descriptor():
        die("Device Edge ingress descriptor mismatch")


def journal_records(path: Path):
    if not path.exists():
        return []
    records = []
    for line in path.read_text(encoding="utf-8").splitlines():
        if not line:
            continue
        try:
            records.append(json.loads(line))
        except json.JSONDecodeError:
            die(f"journal is malformed: {path}")
    return records


def assert_new_identity(patch_id: str, artifact_sha256: str):
    records = journal_records(APPLIED_JOURNAL) + journal_records(FAILED_JOURNAL)
    if any(record.get("patch") == patch_id for record in records):
        die("Device Edge patch id is terminally recorded")
    if any(record.get("sha256") == artifact_sha256 for record in records):
        die("Device Edge artifact digest is terminally recorded")


def current_source_state():
    state = {}
    for relative, expected in PREDECESSOR_FILE_SHA256.items():
        path = LIVE_ROOT / relative
        assert_regular_nonsymlink(path, f"predecessor {relative}")
        state[relative] = sha256_file(path)
        if state[relative] != expected:
            die(f"Device Edge predecessor drift: {relative}")
    for relative in PREDECESSOR_ABSENT:
        if (LIVE_ROOT / relative).exists():
            die(f"Device Edge predecessor unexpected path: {relative}")
    return state


def inspect_container(name: str):
    response = docker_json("inspect", name)
    if len(response) != 1:
        die(f"container inspect cardinality mismatch: {name}")
    return response[0]


def container_health(container):
    health = container.get("State", {}).get("Health")
    return health.get("Status") if health else None


def preserved_runtime_snapshot():
    snapshot = {}
    for name in (BACKHAUL_CONTAINER, TAILNET_CONTAINER):
        container = inspect_container(name)
        if container.get("State", {}).get("Status") != "running":
            die(f"preserved Device Edge service is not running: {name}")
        if name == BACKHAUL_CONTAINER and container_health(container) != "healthy":
            die("Device Edge backhaul is not healthy")
        snapshot[name] = {
            "Id": container.get("Id"),
            "Image": container.get("Image"),
            "StartedAt": container.get("State", {}).get("StartedAt"),
            "RestartCount": container.get("RestartCount"),
            "PortBindings": container.get("HostConfig", {}).get("PortBindings"),
        }
    return snapshot


def assert_preserved_runtime(snapshot):
    current_snapshot = preserved_runtime_snapshot()
    for name, expected in snapshot.items():
        current = current_snapshot[name]
        if current != expected:
            die(f"preserved Device Edge runtime changed: {name}")


def validate_predecessor_runtime():
    relay = inspect_container(RELAY_CONTAINER)
    if relay.get("State", {}).get("Status") != "running":
        die("Device Edge IPvlan predecessor relay is not running")
    if container_health(relay) != "healthy":
        die("Device Edge IPvlan predecessor relay is not healthy")
    environment = set(relay.get("Config", {}).get("Env") or [])
    required = {
        "DEVICE_EDGE_RELAY_HEALTH_HOST=127.0.0.1",
        "DEVICE_EDGE_RELAY_INGRESS_ENABLED=true",
        "DEVICE_EDGE_RELAY_TCP_HOST=0.0.0.0",
        "DEVICE_EDGE_RELAY_TCP_PORT=9921",
        "DEVICE_EDGE_RELAY_UPSTREAM_HOST=device-edge-backhaul",
        "DEVICE_EDGE_RELAY_UPSTREAM_PORT=19921",
    }
    if not required.issubset(environment):
        die("Device Edge IPvlan predecessor environment mismatch")
    bindings = relay.get("HostConfig", {}).get("PortBindings") or {}
    if bindings not in ({}, None):
        die("Device Edge IPvlan predecessor host publication mismatch")
    validate_network_runtime(relay)


def validate_host_network_boundary():
    if socket.gethostname() != "ndcmini12":
        die("Device Edge runtime host mismatch")
    route = run(["/usr/sbin/ip", "-4", "route", "show"]).stdout
    for line in (
        "0.0.0.0/1 dev amn0 metric 1",
        "128.0.0.0/1 dev amn0 metric 1",
        "default via 192.168.68.1 dev enp1s0f0",
        "192.168.68.0/22 dev enp1s0f0",
    ):
        if line not in route:
            die(f"Device Edge host route boundary mismatch: {line}")
    if run(["/usr/bin/systemctl", "is-active", "AmneziaVPN.service"]).stdout.strip() != "active":
        die("AmneziaVPN must remain active for this transition")
    interface = run([
        "/usr/sbin/ip", "-4", "-brief", "address", "show", "dev", INGRESS_PARENT,
    ]).stdout
    if "192.168.68.54/22" not in interface or "UP" not in interface:
        die("Device Edge physical interface boundary mismatch")


def arp_duplicate_detected(target_ip: str, interface: str, attempts=3):
    protocol = 0x0806
    raw = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(protocol))
    try:
        raw.bind((interface, 0))
        source_mac = raw.getsockname()[4]
        target = socket.inet_aton(target_ip)
        ethernet = b"\xff" * 6 + source_mac + struct.pack("!H", protocol)
        arp = struct.pack(
            "!HHBBH6s4s6s4s",
            1,
            0x0800,
            6,
            4,
            1,
            source_mac,
            b"\x00" * 4,
            b"\x00" * 6,
            target,
        )
        raw.setblocking(False)
        for _ in range(attempts):
            raw.send(ethernet + arp)
            deadline = time.monotonic() + 0.7
            while time.monotonic() < deadline:
                ready, _, _ = select.select([raw], [], [], deadline - time.monotonic())
                if not ready:
                    break
                packet = raw.recv(2048)
                if len(packet) < 42 or packet[12:14] != b"\x08\x06":
                    continue
                if packet[28:32] == target and packet[22:28] != source_mac:
                    return True
        return False
    finally:
        raw.close()


def preflight(manifest, artifact_sha256):
    if not INGRESS_IPV4_APPROVED:
        die("Device Edge ingress IPv4 approval is not granted")
    if INGRESS_IPV4_APPROVAL != "approved-outside-dhcp-pool":
        die("Device Edge ingress IPv4 approval contract mismatch")
    assert_new_identity(manifest["id"], artifact_sha256)
    current_source_state()
    validate_predecessor_runtime()
    preserved = preserved_runtime_snapshot()
    validate_host_network_boundary()
    return preserved


def compose_command(*args, baseline=False):
    command = [
        DOCKER,
        "compose",
        "--project-name",
        COMPOSE_PROJECT,
        "--file",
        str(BASE_COMPOSE),
    ]
    if not baseline:
        command.extend(["--file", str(INGRESS_COMPOSE)])
    command.extend(args)
    return command


def ensure_state_directories():
    for path in (
        APPLIED_ROOT,
        FAILED_ROOT,
        BACKUP_ROOT,
        APPLIED_JOURNAL.parent,
    ):
        path.mkdir(parents=True, exist_ok=True, mode=0o750)
        os.chmod(path, 0o750)


def acquire_lock():
    ensure_state_directories()
    try:
        descriptor = os.open(
            DEPLOY_LOCK,
            os.O_WRONLY | os.O_CREAT | os.O_EXCL,
            0o600,
        )
    except FileExistsError:
        die("Device Edge deploy lock is present")
    os.write(descriptor, f"pid={os.getpid()}\n".encode())
    os.close(descriptor)


def release_lock():
    try:
        DEPLOY_LOCK.unlink()
    except FileNotFoundError:
        pass


def create_backup(patch_id: str):
    timestamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
    backup_id = f"{patch_id}-{timestamp}"
    backup = BACKUP_ROOT / backup_id
    backup.mkdir(parents=False, mode=0o750)
    present = []
    absent = []
    for relative in ENTRIES:
        source = LIVE_ROOT / relative
        target = backup / "payload" / relative
        if not source.exists():
            absent.append(relative)
            continue
        present.append(relative)
        target.parent.mkdir(parents=True, exist_ok=True)
        if source.is_dir():
            shutil.copytree(source, target, symlinks=False)
        else:
            shutil.copy2(source, target, follow_symlinks=False)
    (backup / "backup.json").write_text(json.dumps({
        "schemaVersion": "nodedc.device-edge.backup.v1",
        "patch": patch_id,
        "present": present,
        "absent": absent,
    }, sort_keys=True, indent=2) + "\n", encoding="utf-8")
    return backup_id, backup


def publish_payload(payload: Path):
    for relative in ENTRIES:
        source = payload / relative
        target = LIVE_ROOT / relative
        if target.exists():
            if target.is_dir():
                shutil.rmtree(target)
            else:
                target.unlink()
        target.parent.mkdir(parents=True, exist_ok=True)
        if source.is_dir():
            shutil.copytree(source, target, symlinks=False)
        else:
            shutil.copy2(source, target, follow_symlinks=False)


def restore_backup(backup: Path):
    descriptor = json.loads((backup / "backup.json").read_text(encoding="utf-8"))
    for relative in ENTRIES:
        target = LIVE_ROOT / relative
        if target.exists():
            if target.is_dir():
                shutil.rmtree(target)
            else:
                target.unlink()
    for relative in descriptor["present"]:
        source = backup / "payload" / relative
        target = LIVE_ROOT / relative
        target.parent.mkdir(parents=True, exist_ok=True)
        if source.is_dir():
            shutil.copytree(source, target, symlinks=False)
        else:
            shutil.copy2(source, target, follow_symlinks=False)


def build_relay():
    run([
        DOCKER,
        "build",
        "--no-cache",
        "--network=host",
        "--file",
        "services/device-edge-relay/Dockerfile",
        "--tag",
        RELAY_IMAGE,
        ".",
    ], cwd=LIVE_ROOT, timeout=900, capture=False)


def wait_healthy(name: str, timeout_seconds=150):
    deadline = time.monotonic() + timeout_seconds
    while time.monotonic() < deadline:
        try:
            container = inspect_container(name)
        except DeployError:
            time.sleep(2)
            continue
        if (
            container.get("State", {}).get("Status") == "running"
            and container_health(container) == "healthy"
        ):
            return container
        if container.get("State", {}).get("Status") in {"exited", "dead"}:
            die(f"container stopped before health acceptance: {name}")
        time.sleep(2)
    die(f"container health timeout: {name}")


def validate_network_runtime(relay):
    networks = relay.get("NetworkSettings", {}).get("Networks") or {}
    if set(networks) != {"nodedc-device-edge-private", INGRESS_NETWORK}:
        die("Device Edge relay network set mismatch")
    if networks[INGRESS_NETWORK].get("IPAddress") != INGRESS_IPV4:
        die("Device Edge relay IPvlan address mismatch")
    response = docker_json("network", "inspect", INGRESS_NETWORK)
    if len(response) != 1:
        die("Device Edge ingress network cardinality mismatch")
    network = response[0]
    if network.get("Driver") != "ipvlan" or network.get("Internal") is True:
        die("Device Edge ingress network driver mismatch")
    options = network.get("Options") or {}
    if options.get("parent") != INGRESS_PARENT or options.get("ipvlan_mode") != "l2":
        die("Device Edge ingress network option mismatch")
    configs = network.get("IPAM", {}).get("Config") or []
    if len(configs) != 1:
        die("Device Edge ingress IPAM cardinality mismatch")
    if configs[0].get("Subnet") != INGRESS_SUBNET or configs[0].get("Gateway") != INGRESS_GATEWAY:
        die("Device Edge ingress IPAM mismatch")


def validate_relay_runtime(preserved):
    relay = wait_healthy(RELAY_CONTAINER)
    if relay.get("Config", {}).get("User") != "1000:1000":
        die("Device Edge relay user mismatch")
    host = relay.get("HostConfig", {})
    if host.get("ReadonlyRootfs") is not True or host.get("Privileged") is not False:
        die("Device Edge relay filesystem/privilege mismatch")
    if set(host.get("CapDrop") or []) != {"ALL"}:
        die("Device Edge relay capability mismatch")
    if host.get("PortBindings") not in ({}, None):
        die("Device Edge relay host port publication detected")
    environment = set(relay.get("Config", {}).get("Env") or [])
    required = {
        "DEVICE_EDGE_RELAY_HEALTH_HOST=127.0.0.1",
        "DEVICE_EDGE_RELAY_INGRESS_ENABLED=true",
        "DEVICE_EDGE_RELAY_TCP_HOST=0.0.0.0",
        "DEVICE_EDGE_RELAY_TCP_PORT=9921",
        "DEVICE_EDGE_RELAY_UPSTREAM_HOST=device-edge-backhaul",
        "DEVICE_EDGE_RELAY_UPSTREAM_PORT=19921",
        "DEVICE_EDGE_RELAY_SOURCE_POLICY=public-ipv4-only",
        "DEVICE_EDGE_RELAY_MAX_TRACKED_SOURCE_ADDRESSES=2048",
        "DEVICE_EDGE_RELAY_MAX_BYTES_PER_DIRECTION=262144",
    }
    if not required.issubset(environment):
        die("Device Edge relay environment mismatch")
    validate_network_runtime(relay)
    health_result = run([
        DOCKER,
        "exec",
        RELAY_CONTAINER,
        "node",
        "-e",
        "fetch('http://127.0.0.1:18221/healthz').then(async r=>{if(!r.ok)process.exit(2);console.log(await r.text())}).catch(()=>process.exit(3))",
    ])
    try:
        health = json.loads(health_result.stdout)
    except json.JSONDecodeError:
        die("Device Edge relay health JSON invalid")
    expected_health = {
        "ok": True,
        "service": "nodedc-device-edge-relay",
        "ingress": "relay-only",
        "protocolInspection": "disabled",
        "commandTransport": "disabled",
        "sourceAdmission": "public-ipv4-only",
    }
    for key, expected in expected_health.items():
        if health.get(key) != expected:
            die(f"Device Edge relay health contract mismatch: {key}")
    run([
        DOCKER,
        "exec",
        RELAY_CONTAINER,
        "node",
        "-e",
        "const n=require('node:net');const s=n.connect({host:'device-edge-backhaul',port:19921});s.setTimeout(5000);s.once('connect',()=>{s.destroy();process.exit(0)});s.once('timeout',()=>process.exit(2));s.once('error',()=>process.exit(3))",
    ])
    validate_host_network_boundary()
    assert_preserved_runtime(preserved)


def write_journal(path: Path, record):
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8") as handle:
        handle.write(json.dumps(record, sort_keys=True) + "\n")


def archive_artifact(artifact: Path, destination_root: Path):
    destination = destination_root / artifact.name
    if destination.exists():
        die("Device Edge artifact archive collision")
    os.replace(artifact, destination)
    return destination


def rollback(backup: Path, preserved):
    restore_backup(backup)
    run(compose_command(
        "up",
        "--detach",
        "--no-deps",
        "--force-recreate",
        "--pull",
        "never",
        RELAY_SERVICE,
    ), cwd=LIVE_ROOT, timeout=300, capture=False)
    wait_healthy(RELAY_CONTAINER)
    current_source_state()
    validate_predecessor_runtime()
    assert_preserved_runtime(preserved)


def plan_artifact(artifact_argument: str):
    assert_root()
    artifact = Path(artifact_argument)
    with tempfile.TemporaryDirectory(prefix="nodedc-edge-plan-") as directory:
        manifest, entries, _payload, digest, resolved = load_artifact(
            artifact,
            Path(directory),
        )
        preflight(manifest, digest)
    print("== plan ==")
    print(f"artifact={resolved.name}")
    print(f"sha256={digest}")
    print(f"id={manifest['id']}")
    print(f"component={COMPONENT}")
    print(f"type={ARTIFACT_TYPE}")
    print(f"payload_root={LIVE_ROOT}")
    print(f"compose_root={LIVE_ROOT}")
    print(f"compose_project={COMPOSE_PROJECT}")
    print("compose_files=docker-compose.device-edge.yml docker-compose.device-edge.ingress.yml")
    print("build=/usr/bin/docker build --no-cache --network=host -f services/device-edge-relay/Dockerfile -t nodedc/device-edge-relay:local .")
    print("services=device-edge-relay")
    print("preserved_services=device-edge-backhaul tailnet")
    print(f"device_edge_ingress=ipvlan:l2:{INGRESS_PARENT}:{INGRESS_IPV4}:{INGRESS_PORT}/tcp")
    print(f"device_edge_lan={INGRESS_SUBNET}:gateway:{INGRESS_GATEWAY}")
    print(f"device_edge_ingress_ipv4_approval={INGRESS_IPV4_APPROVAL}")
    print("device_edge_host_port_publication=disabled")
    print("device_edge_health_publication=disabled")
    print("device_edge_private_upstream=device-edge-backhaul:19921")
    print("device_edge_source_admission=public-ipv4-only")
    print("device_edge_source_table_limit=2048")
    print("device_edge_byte_limit_per_direction=262144")
    print("device_edge_command_transport=disabled")
    print("device_edge_discovery_lifecycle=quarantine")
    print("device_edge_gelios=untouched")
    print("device_edge_amnezia=preserved:active:host-full-tunnel")
    print("device_edge_router_nat_firewall=unchanged")
    print("device_edge_rollback=restore-reviewed-ipvlan-predecessor-no-router-mutation")
    print("state=new")
    print("== files ==")
    for entry in entries:
        print(f"  {entry}")


def apply_artifact(artifact_argument: str):
    assert_root()
    artifact = Path(artifact_argument)
    acquire_lock()
    manifest = None
    digest = None
    resolved = None
    backup_id = None
    backup = None
    preserved = None
    try:
        with tempfile.TemporaryDirectory(prefix="nodedc-edge-apply-") as directory:
            manifest, _entries, payload, digest, resolved = load_artifact(
                artifact,
                Path(directory),
            )
            preserved = preflight(manifest, digest)
            backup_id, backup = create_backup(manifest["id"])
            publish_payload(payload)
            build_relay()
            run(compose_command(
                "up",
                "--detach",
                "--no-deps",
                "--force-recreate",
                "--pull",
                "never",
                RELAY_SERVICE,
            ), cwd=LIVE_ROOT, timeout=300, capture=False)
            validate_relay_runtime(preserved)
        archived = archive_artifact(resolved, APPLIED_ROOT)
        write_journal(APPLIED_JOURNAL, {
            "status": "ok",
            "patch": manifest["id"],
            "component": COMPONENT,
            "sha256": digest,
            "artifact": archived.name,
            "backup": backup_id,
            "appliedAt": datetime.now(timezone.utc).isoformat(),
        })
        print(
            f"deploy-ok patch={manifest['id']} component={COMPONENT} "
            f"backup={backup_id}"
        )
    except Exception as error:
        rollback_status = "not-started"
        if backup is not None and preserved is not None:
            try:
                rollback(backup, preserved)
                rollback_status = "ok"
            except Exception as rollback_error:
                rollback_status = f"failed:{type(rollback_error).__name__}"
        if resolved is not None and resolved.exists():
            failed_name = (
                FAILED_ROOT
                / f"{resolved.name}.{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')}"
            )
            os.replace(resolved, failed_name)
        if manifest is not None and digest is not None:
            write_journal(FAILED_JOURNAL, {
                "status": "failed",
                "patch": manifest["id"],
                "component": COMPONENT,
                "sha256": digest,
                "backup": backup_id,
                "rollback": rollback_status,
                "error": type(error).__name__,
                "failedAt": datetime.now(timezone.utc).isoformat(),
            })
        if rollback_status.startswith("failed"):
            die(f"apply failed and rollback failed: {error}")
        die(f"apply failed; automatic rollback={rollback_status}: {error}")
    finally:
        release_lock()


def verify_install():
    assert_root()
    path = RUNNER_PATH if RUNNER_PATH.exists() else Path(__file__).resolve()
    assert_regular_nonsymlink(path, "runner")
    docker_version = run([DOCKER, "version", "--format", "{{.Server.Version}}"]).stdout.strip()
    compose_version = run([DOCKER, "compose", "version", "--short"]).stdout.strip()
    print(f"path={path}")
    print(f"sha256={sha256_file(path)}")
    print(f"python={sys.version.split()[0]}")
    print(f"docker={docker_version}")
    print(f"compose={compose_version}")
    print(f"device_edge_ingress_ipv4={INGRESS_IPV4}")
    print(f"device_edge_ingress_ipv4_approval={INGRESS_IPV4_APPROVAL}")
    print("device_edge_source_admission=public-ipv4-only")
    print("verify-install-ok")


def main(arguments):
    if len(arguments) == 1 and arguments[0] == "verify-install":
        verify_install()
        return 0
    if len(arguments) == 2 and arguments[0] == "plan":
        plan_artifact(arguments[1])
        return 0
    if len(arguments) == 2 and arguments[0] == "apply":
        apply_artifact(arguments[1])
        return 0
    print(
        "usage: nodedc-edge-deploy verify-install | plan <artifact.tgz> | apply <artifact.tgz>",
        file=sys.stderr,
    )
    return 2


if __name__ == "__main__":
    try:
        raise SystemExit(main(sys.argv[1:]))
    except DeployError as error:
        print(f"ERROR: {error}", file=sys.stderr)
        raise SystemExit(1)