#!/usr/bin/env python3
"""Canonical data-only deploy runner for the Robot2B public Device Edge VPS."""

from __future__ import annotations

import hashlib
import json
import os
import pwd
import re
import shutil
import socket
import subprocess
import sys
import tarfile
import tempfile
import time
import urllib.request
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath


RUNNER_PATH = Path("/usr/local/sbin/nodedc-b2-vps-deploy")
LIVE_ROOT = Path("/opt/nodedc-b2-vps")
INBOX_ROOT = Path("/var/lib/nodedc-b2-vps-deploy/inbox")
STATE_ROOT = Path("/var/lib/nodedc-b2-vps-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"

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

RUNTIME_HOST = "koffyvngij"
PUBLIC_IPV4 = "155.212.211.15"
SERVICE_USER = "nodedc-edge"
SERVICE_GROUP = "nodedc-edge"
BACKHAUL_USER = "nodedc-backhaul"
BACKHAUL_GROUP = "nodedc-backhaul"
RELAY_USER = "nodedc-relay"
RELAY_GROUP = "nodedc-relay"
CHANNEL_USER = "nodedc-channel"
CHANNEL_GROUP = "nodedc-channel"
TAILSCALE_REQUIRED_TAG = "tag:device-edge-vps"
MANAGEMENT_KEY_FINGERPRINT = (
    "SHA256:DYYy1E3DaxIQGC0jnsW6SP7gXdBHUy3A1zn4pvgVUEw"
)
SERVER_HOST_KEY_FINGERPRINT = (
    "SHA256:mhqNn2S6zstkYL7VFdvt3SYHv1nLjB4J7/s57RrKG6w"
)

NODE_VERSION = "22.23.2"
NODE_ARCHIVE = "node-v22.23.2-linux-x64.tar.xz"
NODE_ARCHIVE_SHA256 = (
    "d60acfe00a2932254bb0ad20e01b0d74397a0875595de719654b214f4b03f307"
)
TAILSCALE_VERSION = "1.102.2"
TAILSCALE_ARCHIVE = "tailscale_1.102.2_amd64.tgz"
TAILSCALE_ARCHIVE_SHA256 = (
    "ad2cde12f8de95f7b93a1e0401e652291c603d42b9d60a33fb1741eb38ab04d8"
)
NODE_BIN_SHA256 = "3517c2df0b2f8cd7f422b4b8450ef81c6889f08eb03e281d6de9079b15e6a327"
TAILSCALE_BIN_SHA256 = "58b0fa0907677ea6afe0d3022cc3e99b1a03f39a7ed60144843ed38252e00c80"
TAILSCALED_BIN_SHA256 = "5f17b092bac92326325f6c4ffd9991fad3c073975abe412d02ee68721a500394"

NODE_BIN = LIVE_ROOT / "runtime/node/bin/node"
TAILSCALE_BIN = LIVE_ROOT / "runtime/tailscale/tailscale"
TAILSCALED_BIN = LIVE_ROOT / "runtime/tailscale/tailscaled"
TAILSCALE_SOCKET = Path("/run/nodedc-b2-vps/tailscaled.sock")
TAILSCALE_STATE = Path("/var/lib/nodedc-b2-vps/tailscale/tailscaled.state")
TRUST_ROOT = Path("/var/lib/nodedc-b2-vps/trust")
BACKHAUL_PRIVATE_KEY = TRUST_ROOT / "backhaul_ed25519"
BACKHAUL_PUBLIC_KEY = TRUST_ROOT / "backhaul_ed25519.pub"
BACKHAUL_KNOWN_HOSTS = TRUST_ROOT / "backhaul_known_hosts"
BACKHAUL_KEY_COMMENT = "nodedc-device-edge-vps-backhaul"

BACKHAUL_TARGET_IP = "100.109.216.21"
BACKHAUL_TARGET_PORT = 2222
BACKHAUL_TARGET_HOST_KEY = (
    "ssh-ed25519 "
    "AAAAC3NzaC1lZDI1NTE5AAAAIJsmoyS+0Tbhz9VXxrSxXwNMFpfbdTckCilObOnKdlEc"
)
BACKHAUL_TARGET_FINGERPRINT = (
    "SHA256:QERJ5CIUXRj0nLChGT6HMtoX+WTaeaEY5ZgaWqT8d30"
)

SSHD_DROPIN = Path("/etc/ssh/sshd_config.d/00-nodedc-b2-vps.conf")
NFTABLES_CONFIG = Path("/etc/nftables.conf")
TAILSCALE_UNIT = Path("/etc/systemd/system/nodedc-b2-tailscaled.service")
BACKHAUL_UNIT = Path("/etc/systemd/system/nodedc-b2-backhaul.service")
RELAY_UNIT = Path("/etc/systemd/system/nodedc-b2-relay.service")
CHANNEL_UNIT = Path("/etc/systemd/system/nodedc-device-edge-channel.service")
CHANNEL_TRUST_ROOT = Path("/var/lib/nodedc-b2-vps/channel-trust")
CHANNEL_PRIVATE_KEY = CHANNEL_TRUST_ROOT / "edge-private-key.pem"
CHANNEL_CERTIFICATE = CHANNEL_TRUST_ROOT / "edge-certificate.pem"
CHANNEL_CORE_CERTIFICATE = CHANNEL_TRUST_ROOT / "core-certificate.pem"
CHANNEL_RUNTIME_CONFIG = CHANNEL_TRUST_ROOT / "runtime.json"
CHANNEL_HEALTH_PORT = 18222
CHANNEL_PUBLIC_PORT = 8443

FOUNDATION_ENTRIES = (
    "vps/config/00-nodedc-b2-vps.conf",
    "vps/config/nftables-foundation.conf",
    "vps/systemd/nodedc-b2-tailscaled.service",
    "deployment/device-edge-vps-foundation-v1.json",
    f"vendor/{NODE_ARCHIVE}",
    f"vendor/{TAILSCALE_ARCHIVE}",
)
RUNTIME_RECONCILIATION_ENTRIES = (
    "deployment/device-edge-vps-runtime-reconciliation-v1.json",
)
BACKHAUL_ENTRIES = (
    "vps/config/backhaul_ssh_config",
    "vps/systemd/nodedc-b2-backhaul.service",
    "deployment/device-edge-vps-backhaul-v1.json",
)
RELAY_ENTRIES = (
    "vps/config/nftables-relay.conf",
    "vps/systemd/nodedc-b2-relay.service",
    "services/device-edge-relay/src",
    "deployment/device-edge-vps-relay-v1.json",
)
CORE_CHANNEL_ENTRIES = (
    "packages/device-protocol-contract/package.json",
    "packages/device-protocol-contract/src",
    "packages/device-edge-channel-contract/package.json",
    "packages/device-edge-channel-contract/src",
    "services/device-edge-channel/package.json",
    "services/device-edge-channel/src",
    "vps/config/nftables-core-channel.conf",
    "vps/systemd/nodedc-device-edge-channel.service",
    "deployment/device-edge-vps-core-channel-v1.json",
)

PHASE_ENTRIES = {
    "foundation": FOUNDATION_ENTRIES,
    "runtime-reconciliation": RUNTIME_RECONCILIATION_ENTRIES,
    "backhaul": BACKHAUL_ENTRIES,
    "relay": RELAY_ENTRIES,
    "core-channel": CORE_CHANNEL_ENTRIES,
}

SUPERSEDED_TRANSPORT_PHASES = frozenset({"backhaul", "relay"})

PHASE_FILE_SHA256 = {
    "foundation": {
        "vps/config/00-nodedc-b2-vps.conf":
            "2079748f48b2297ecb067e16ae46248e3a981b26331aa0998dc14fdf7454cd4a",
        "vps/config/nftables-foundation.conf":
            "bce5b8c6e2226d47d8d322f8ee9e7158f2d7a553c5e2c4bc2a722d6067f46e28",
        "vps/systemd/nodedc-b2-tailscaled.service":
            "de147d29bc1759f56533d31058993df55e1200973f2899903bdbf3d13ff579da",
        "deployment/device-edge-vps-foundation-v1.json":
            "5ae007195b17d6cfaeb564abbed6dc9234eb36bf62db23bf2b9ace539c398898",
        f"vendor/{NODE_ARCHIVE}": NODE_ARCHIVE_SHA256,
        f"vendor/{TAILSCALE_ARCHIVE}": TAILSCALE_ARCHIVE_SHA256,
    },
    "runtime-reconciliation": {
        "deployment/device-edge-vps-runtime-reconciliation-v1.json":
            "edf7e05918efa9c9ab6d759ecdcc493ec6a2c92ed5b486772b3629cd5fa981ab",
    },
    "backhaul": {
        "vps/config/backhaul_ssh_config":
            "d0df8b70dda025b7c1c3fecd2dafffe60a5bb753650d3bc37db65db626cfc1af",
        "vps/systemd/nodedc-b2-backhaul.service":
            "64c26cad21cc17675c67ae4a57fc43b129065a8d22fda894648340b310d3aa8c",
        "deployment/device-edge-vps-backhaul-v1.json":
            "d7bac0132f7940cf8d4c59f1ae0d627c4921b63fa4728c31d2f95cfa35f21cf5",
    },
    "relay": {
        "vps/config/nftables-relay.conf":
            "497485a2fb1b79faa95eb67ca30a89a3b65c3d4d8ddcb14cd1dc9acfeb2bb6f2",
        "vps/systemd/nodedc-b2-relay.service":
            "12a927a4cb42016229ac438f6e75969e1bf015037f2b5b0a60ab0591bdf424d2",
        "services/device-edge-relay/src/runtime.mjs":
            "21e83678980aa61127bf9f3d77982dd485c4aaae208c43818db7bb1cc150b83a",
        "services/device-edge-relay/src/server.mjs":
            "1b99ec944f1d3fbadded045b159f08624e2829620cec39a97f6b4b8cdcd2be22",
        "deployment/device-edge-vps-relay-v1.json":
            "cb3c2fff4878021783efef4f4a4d1ec31c8e3d8e6325fe33657fec22bec20165",
    },
    "core-channel": {
        "packages/device-protocol-contract/package.json":
            "19d0d07da0341e8c2e8d3485400767566b18f6270245f0e9a9681c95013ba3b7",
        "packages/device-protocol-contract/src/index.mjs":
            "21a8b2b85a807899f946387c7976eaffcdc438b7db5e3d41fc3930f4437f0ec7",
        "packages/device-edge-channel-contract/package.json":
            "57d5349b5dcef2cacd4f3e4fad010359a65d59f5f903eff07d89f67c497f97c0",
        "packages/device-edge-channel-contract/src/index.mjs":
            "09d45e6104779212605fed650544794aafe1c2b1ab66f33f07bef4d92f430c80",
        "services/device-edge-channel/package.json":
            "bdf502be43b62bdd6db05b022a532d93ba954277ac5143d6058d2f27f6a2e9d2",
        "services/device-edge-channel/src/runtime.mjs":
            "cbb07f7e644a68e9c8c36c1c2ecf0224c06c46c08c49339d62637b10a1495501",
        "services/device-edge-channel/src/server.mjs":
            "ea891634a18efb9eb44f17b56c95ba97527215c4d0a6147cc2b7bad1d7356e36",
        "vps/config/nftables-core-channel.conf":
            "1a04a5450042e80b8a20da3c6634dd6bc68693f191a463ba9a62984279d81d0c",
        "vps/systemd/nodedc-device-edge-channel.service":
            "afb2842db18d1863b2a089b4e89693f1ec17c78921457774ac9652f26b804856",
        "deployment/device-edge-vps-core-channel-v1.json":
            "938f6f7959b78e6be3a6e91a54dca2922fbd813f1a33dca5cebe0dc256a83a14",
    },
}

# Exact immutable baselines from terminally accepted predecessor artifacts.
# This is intentionally keyed by both patch id and artifact digest so a later
# artifact cannot inherit compatibility from a merely similar deployment.
ACCEPTED_PREDECESSOR_FILE_SHA256 = {
    "foundation": {
        (
            "device-edge-vps-foundation-20260806-003",
            "1be852f144e9f0fea32af70bebd07a2607b6a1818825094bd4c1b4062064716a",
        ): {
            "vps/config/00-nodedc-b2-vps.conf":
                "cc94d0579f85d0af9746b9ce760bc72980f4a22fb59027e1f5f9c7bf3aaebd64",
            "vps/config/nftables-foundation.conf":
                "4d44f902d8d98d1aa8506fca9d9582e700f6424def2b1d667ab2cd5a5ee84934",
            "deployment/device-edge-vps-foundation-v1.json":
                "317c98b42520fff3238275908482de7aa611b4ee41c6b1f8062abd2730ab072a",
        },
    },
}


class DeployError(RuntimeError):
    pass


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


def run(command, *, check=True, capture=True, timeout=180, cwd=None):
    result = subprocess.run(
        [str(value) for value in command],
        check=False,
        capture_output=capture,
        text=True,
        timeout=timeout,
        cwd=str(cwd) if cwd else None,
    )
    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 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 assert_root():
    if os.geteuid() != 0:
        die("nodedc-b2-vps-deploy must run as root")


def assert_regular_nonsymlink(path: Path, label: str):
    try:
        path_stat = path.lstat()
    except FileNotFoundError:
        die(f"{label} is missing")
    if path.is_symlink() or not path.is_file():
        die(f"{label} must be a regular non-symlink file")
    return path_stat


def assert_directory_nonsymlink(path: Path, label: str):
    try:
        path_stat = path.lstat()
    except FileNotFoundError:
        die(f"{label} is missing")
    if path.is_symlink() or not path.is_dir():
        die(f"{label} must be a non-symlink directory")
    return path_stat


def assert_executable_command_path(path: Path, label: str):
    try:
        resolved = path.resolve(strict=True)
        resolved_stat = resolved.stat()
    except (FileNotFoundError, OSError, RuntimeError):
        die(f"{label} is missing or has an invalid symlink chain")
    if not resolved.is_file():
        die(f"{label} must resolve to a regular file")
    if not (resolved_stat.st_mode & 0o111):
        die(f"{label} is not executable")
    return resolved


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 phase_from_entries(entries):
    for phase, expected in PHASE_ENTRIES.items():
        if tuple(entries) == expected:
            return phase
    die("VPS artifact file selection mismatch")


def validate_payload(payload: Path, phase: str):
    actual = {
        path.relative_to(payload).as_posix(): sha256_file(path)
        for path in payload.rglob("*")
        if path.is_file()
    }
    if actual != PHASE_FILE_SHA256[phase]:
        die(f"VPS {phase} payload digest set mismatch")
    descriptor = json.loads(
        (
            payload
            / f"deployment/device-edge-vps-{phase}-v1.json"
        ).read_text(encoding="utf-8")
    )
    if (
        descriptor.get("component") != COMPONENT
        or descriptor.get("runtimeHost") != RUNTIME_HOST
        or descriptor.get("commandTransport") != "disabled"
        or descriptor.get("gelios") != "untouched"
        or not descriptor.get("rollback")
    ):
        die(f"VPS {phase} descriptor mismatch")
    return descriptor


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 VPS 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 len(entries) != len(set(entries)):
        die("artifact files list contains duplicates")
    phase = phase_from_entries(entries)
    payload = extraction_root / "payload"
    descriptor = validate_payload(payload, phase)
    return {
        "manifest": manifest,
        "entries": entries,
        "phase": phase,
        "payload": payload,
        "descriptor": descriptor,
        "sha256": sha256_file(artifact),
        "artifact": artifact,
    }


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("VPS patch id is terminally recorded")
    if any(record.get("sha256") == artifact_sha256 for record in records):
        die("VPS artifact digest is terminally recorded")


def applied_phase_record(phase: str):
    matches = [
        record
        for record in journal_records(APPLIED_JOURNAL)
        if record.get("phase") == phase and record.get("status") == "ok"
    ]
    if len(matches) != 1:
        die(f"exactly one accepted {phase} predecessor is required")
    return matches[0]


def assert_host_identity():
    if socket.gethostname() != RUNTIME_HOST:
        die("VPS runtime hostname mismatch")
    os_release = Path("/etc/os-release").read_text(encoding="utf-8")
    if 'VERSION_ID="24.04"' not in os_release or 'ID=ubuntu' not in os_release:
        die("VPS operating system predecessor mismatch")
    address = run(["/usr/sbin/ip", "-4", "-brief", "address", "show", "dev", "eth0"]).stdout
    if f"{PUBLIC_IPV4}/32" not in address or "UP" not in address:
        die("VPS public interface predecessor mismatch")


def port_is_open(host: str, port: int, timeout=1.5):
    try:
        connection = socket.create_connection((host, port), timeout=timeout)
    except OSError:
        return False
    connection.close()
    return True


def assert_port_closed(port: int):
    if port_is_open("127.0.0.1", port):
        die(f"unexpected loopback listener is open: {port}")


def assert_management_key():
    result = run(["/usr/bin/ssh-keygen", "-lf", "/root/.ssh/authorized_keys"])
    if MANAGEMENT_KEY_FINGERPRINT not in result.stdout:
        die("verified Mac management key is missing")


def phase_file_sha256(phase: str):
    expected = dict(PHASE_FILE_SHA256[phase])
    accepted = [
        record
        for record in journal_records(APPLIED_JOURNAL)
        if record.get("phase") == phase and record.get("status") == "ok"
    ]
    if len(accepted) == 1:
        record = accepted[0]
        identity = (record.get("patch"), record.get("sha256"))
        expected.update(
            ACCEPTED_PREDECESSOR_FILE_SHA256.get(phase, {}).get(identity, {})
        )
    return expected


def source_file_state(phase: str):
    expected = phase_file_sha256(phase)
    actual = {}
    for relative, digest in expected.items():
        path = LIVE_ROOT / relative
        assert_regular_nonsymlink(path, f"installed {phase} file {relative}")
        actual[relative] = sha256_file(path)
        if actual[relative] != digest:
            die(f"installed {phase} source drift: {relative}")
    return actual


def assert_runtime_reconciliation_predecessor(foundation_record):
    if (
        foundation_record.get("patch") != "device-edge-vps-foundation-20260806-003"
        or foundation_record.get("sha256")
        != "1be852f144e9f0fea32af70bebd07a2607b6a1818825094bd4c1b4062064716a"
    ):
        die("runtime reconciliation foundation predecessor mismatch")
    failed = [
        record
        for record in journal_records(FAILED_JOURNAL)
        if (
            record.get("patch") == "device-edge-vps-core-channel-20260812-001"
            and record.get("phase") == "core-channel"
            and record.get("sha256")
            == "c199980e5754cf3e874a09f42e408fc88885cdbf7c872eac36c0ab768a7bab00"
            and record.get("status") == "failed"
            and record.get("rollback") == "ok"
            and record.get("error") == "PermissionError"
        )
    ]
    if len(failed) != 1:
        die("runtime reconciliation failed Core channel predecessor mismatch")
    for path, digest in (
        (NODE_BIN, NODE_BIN_SHA256),
        (TAILSCALE_BIN, TAILSCALE_BIN_SHA256),
        (TAILSCALED_BIN, TAILSCALED_BIN_SHA256),
    ):
        state = assert_regular_nonsymlink(path, f"runtime reconciliation {path.name}")
        if (
            state.st_uid != 0
            or state.st_gid != 0
            or (state.st_mode & 0o777) != 0o644
            or sha256_file(path) != digest
        ):
            die(f"runtime reconciliation binary predecessor mismatch: {path.name}")
    if not service_active("nodedc-b2-tailscaled.service"):
        die("runtime reconciliation Tailscale predecessor is inactive")
    if service_active("nodedc-device-edge-channel.service") or user_exists(CHANNEL_USER):
        die("runtime reconciliation Core channel predecessor is not absent")
    assert_channel_trust(require_runtime_owner=False)
    for port in (CHANNEL_HEALTH_PORT, CHANNEL_PUBLIC_PORT, 9921):
        assert_port_closed(port)
    nft = run(["/usr/sbin/nft", "list", "table", "inet", "nodedc_b2_vps"]).stdout
    if (
        "policy drop" not in nft
        or "tcp dport 22" not in nft
        or "tcp dport 8443" in nft
        or "tcp dport 9921" in nft
    ):
        die("runtime reconciliation firewall predecessor mismatch")


def systemctl(*args, check=True):
    return run(["/usr/bin/systemctl", *args], check=check)


def service_active(name: str):
    return systemctl("is-active", name, check=False).returncode == 0


def tailscale_status():
    result = run(
        [str(TAILSCALE_BIN), f"--socket={TAILSCALE_SOCKET}", "status", "--json"],
        check=False,
    )
    if result.returncode != 0:
        return {"BackendState": "Unavailable", "Error": (result.stderr or result.stdout).strip()}
    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        die("Tailscale status JSON is invalid")


def current_phase_preflight(phase: str):
    assert_host_identity()
    assert_management_key()
    if phase == "foundation":
        if LIVE_ROOT.exists() or TRUST_ROOT.exists():
            die("VPS foundation live/runtime root already exists")
        if any(path.exists() for path in (SSHD_DROPIN, TAILSCALE_UNIT, BACKHAUL_UNIT, RELAY_UNIT)):
            die("VPS foundation system path already exists")
        for port in (1055, 18221, CHANNEL_HEALTH_PORT, 19921, 9921, CHANNEL_PUBLIC_PORT):
            assert_port_closed(port)
        return {"predecessor": "clean-ubuntu-24.04.4"}

    foundation_record = applied_phase_record("foundation")
    source_file_state("foundation")
    if phase == "runtime-reconciliation":
        assert_runtime_reconciliation_predecessor(foundation_record)
        return {
            "predecessor": "failed-core-channel-001-rollback-runtime-mode-drift",
        }
    validate_foundation_runtime(
        require_running_tailnet=phase in {"backhaul", "relay"},
        expected_key_user=BACKHAUL_USER if phase == "relay" else SERVICE_USER,
    )
    if phase == "core-channel":
        if service_active("nodedc-b2-backhaul.service"):
            die("frozen VPS backhaul service must remain inactive")
        if service_active("nodedc-b2-relay.service"):
            die("frozen VPS relay service must remain inactive")
        if CHANNEL_UNIT.exists() or (LIVE_ROOT / CORE_CHANNEL_ENTRIES[-1]).exists():
            die("VPS Core channel target path already exists")
        if user_exists(CHANNEL_USER):
            die("VPS Core channel runtime user already exists")
        assert_channel_trust(require_runtime_owner=False)
        for port in (CHANNEL_HEALTH_PORT, CHANNEL_PUBLIC_PORT, 9921):
            assert_port_closed(port)
        return {"predecessor": "accepted-foundation-closed-channel"}
    if phase == "backhaul":
        for tool in (Path("/usr/bin/ssh"), Path("/usr/bin/nc")):
            assert_executable_command_path(
                tool,
                f"VPS backhaul prerequisite {tool}",
            )
        if BACKHAUL_UNIT.exists() or (LIVE_ROOT / BACKHAUL_ENTRIES[-1]).exists():
            die("VPS backhaul target path already exists")
        if user_exists(BACKHAUL_USER):
            die("VPS backhaul runtime user already exists")
        assert_port_closed(19921)
        return {"predecessor": "accepted-foundation"}

    applied_phase_record("backhaul")
    source_file_state("backhaul")
    validate_backhaul_runtime()
    if RELAY_UNIT.exists() or (LIVE_ROOT / RELAY_ENTRIES[-1]).exists():
        die("VPS relay target path already exists")
    if user_exists(RELAY_USER):
        die("VPS relay runtime user already exists")
    assert_port_closed(9921)
    assert_port_closed(18221)
    return {"predecessor": "accepted-backhaul"}


def preflight(loaded):
    if loaded["phase"] in SUPERSEDED_TRANSPORT_PHASES:
        die("vps_initiated_transport_frozen:ADR-0001")
    assert_new_identity(loaded["manifest"]["id"], loaded["sha256"])
    return current_phase_preflight(loaded["phase"])


def ensure_state_directories():
    for path in (
        INBOX_ROOT,
        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("VPS 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 backup_targets_for_phase(phase: str):
    common = [LIVE_ROOT / entry for entry in PHASE_ENTRIES[phase]]
    if phase == "foundation":
        return common + [SSHD_DROPIN, NFTABLES_CONFIG, TAILSCALE_UNIT]
    if phase == "runtime-reconciliation":
        return common + [NODE_BIN, TAILSCALE_BIN, TAILSCALED_BIN]
    if phase == "backhaul":
        return common + [BACKHAUL_UNIT, BACKHAUL_KNOWN_HOSTS]
    if phase == "core-channel":
        return common + [CHANNEL_UNIT, NFTABLES_CONFIG, CHANNEL_TRUST_ROOT]
    return common + [RELAY_UNIT, NFTABLES_CONFIG]


def path_backup_relative(path: Path):
    return path.as_posix().lstrip("/")


def create_backup(patch_id: str, phase: 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 source in backup_targets_for_phase(phase):
        relative = path_backup_relative(source)
        target = backup / "filesystem" / relative
        if not source.exists() and not source.is_symlink():
            absent.append(relative)
            continue
        if source.is_symlink():
            die(f"backup target symlink rejected: {source}")
        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)
    nft_rules = run(["/usr/sbin/nft", "list", "ruleset"], check=False).stdout
    (backup / "nft-ruleset-before.nft").write_text(nft_rules, encoding="utf-8")
    metadata = {
        "schemaVersion": "nodedc.device-edge-vps.backup.v1",
        "patch": patch_id,
        "phase": phase,
        "present": present,
        "absent": absent,
        "serviceUserExisted": user_exists(),
        "serviceUsersExisted": {
            name: user_exists(name)
            for name in (SERVICE_USER, BACKHAUL_USER, RELAY_USER, CHANNEL_USER)
        },
        "services": {
            name: {
                "active": service_active(name),
                "enabled": systemctl("is-enabled", name, check=False).returncode == 0,
            }
            for name in (
                "nftables.service",
                "ufw.service",
                "fail2ban.service",
                "nodedc-b2-tailscaled.service",
                "nodedc-b2-backhaul.service",
                "nodedc-b2-relay.service",
                "nodedc-device-edge-channel.service",
            )
        },
    }
    (backup / "backup.json").write_text(
        json.dumps(metadata, sort_keys=True, indent=2) + "\n",
        encoding="utf-8",
    )
    return backup_id, backup


def publish_payload(payload: Path, entries):
    for relative in entries:
        source = payload / relative
        target = LIVE_ROOT / relative
        if target.exists() or target.is_symlink():
            if target.is_symlink():
                die(f"live payload target symlink rejected: {target}")
            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)
        published = [target]
        if target.is_dir():
            published.extend(target.rglob("*"))
        for path in published:
            if path.is_symlink():
                die(f"published live source contains symlink: {path}")
            if path.is_dir():
                os.chmod(path, 0o755)
            else:
                os.chmod(path, 0o644)
    os.chown(LIVE_ROOT, 0, 0)


def restore_backup(backup: Path, phase: str):
    metadata = json.loads((backup / "backup.json").read_text(encoding="utf-8"))
    if metadata.get("phase") != phase:
        die("rollback backup phase mismatch")
    targets = backup_targets_for_phase(phase)
    for target in targets:
        if target.exists() or target.is_symlink():
            if target.is_symlink() or target.is_file():
                target.unlink()
            else:
                shutil.rmtree(target)
    for relative in metadata["present"]:
        source = backup / "filesystem" / relative
        target = Path("/") / 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 install_file(source: Path, target: Path, mode=0o644):
    assert_regular_nonsymlink(source, f"install source {source}")
    target.parent.mkdir(parents=True, exist_ok=True)
    temporary = target.with_name(f".{target.name}.installing")
    if temporary.exists() or temporary.is_symlink():
        die(f"install staging path exists: {temporary}")
    shutil.copy2(source, temporary, follow_symlinks=False)
    os.chown(temporary, 0, 0)
    os.chmod(temporary, mode)
    os.replace(temporary, target)


def user_exists(name=SERVICE_USER):
    try:
        pwd.getpwnam(name)
        return True
    except KeyError:
        return False


def ensure_service_user(name=SERVICE_USER, home_dir="/var/lib/nodedc-b2-vps"):
    if not user_exists(name):
        run([
            "/usr/sbin/useradd",
            "--system",
            "--user-group",
            "--home-dir",
            home_dir,
            "--shell",
            "/usr/sbin/nologin",
            name,
        ])
    account = pwd.getpwnam(name)
    if account.pw_shell != "/usr/sbin/nologin":
        die("VPS service user shell mismatch")
    return account


def extract_vendor_binary(archive: Path, member_name: str, target: Path, mode=0o755):
    with tarfile.open(archive, "r:*") as package:
        try:
            member = package.getmember(member_name)
        except KeyError:
            die(f"vendor binary member missing: {member_name}")
        if not member.isfile() or member.issym() or member.islnk():
            die(f"vendor binary member unsafe: {member_name}")
        source = package.extractfile(member)
        if source is None:
            die(f"vendor binary cannot be read: {member_name}")
        target.parent.mkdir(parents=True, exist_ok=True)
        temporary = target.with_name(f".{target.name}.installing")
        with temporary.open("wb") as handle:
            shutil.copyfileobj(source, handle)
        os.chown(temporary, 0, 0)
        os.chmod(temporary, mode)
        os.replace(temporary, target)


def ensure_backhaul_key(account):
    TRUST_ROOT.mkdir(parents=True, exist_ok=True, mode=0o700)
    os.chown(TRUST_ROOT, account.pw_uid, account.pw_gid)
    os.chmod(TRUST_ROOT, 0o700)
    if not BACKHAUL_PRIVATE_KEY.exists():
        run([
            "/usr/bin/ssh-keygen",
            "-q",
            "-t",
            "ed25519",
            "-N",
            "",
            "-C",
            BACKHAUL_KEY_COMMENT,
            "-f",
            str(BACKHAUL_PRIVATE_KEY),
        ])
    for path, mode in ((BACKHAUL_PRIVATE_KEY, 0o400), (BACKHAUL_PUBLIC_KEY, 0o444)):
        stat_result = assert_regular_nonsymlink(path, f"backhaul key {path.name}")
        if stat_result.st_size > 2048:
            die("backhaul key file is unexpectedly large")
        os.chown(path, account.pw_uid, account.pw_gid)
        os.chmod(path, mode)
    fingerprint = run(["/usr/bin/ssh-keygen", "-lf", str(BACKHAUL_PUBLIC_KEY)]).stdout.strip()
    if "ED25519" not in fingerprint or BACKHAUL_KEY_COMMENT not in fingerprint:
        die("backhaul public key fingerprint mismatch")
    return fingerprint


def assign_backhaul_trust(account):
    assert_directory_nonsymlink(TRUST_ROOT, "backhaul trust root")
    os.chown(TRUST_ROOT, account.pw_uid, account.pw_gid)
    os.chmod(TRUST_ROOT, 0o700)
    for path, mode in ((BACKHAUL_PRIVATE_KEY, 0o400), (BACKHAUL_PUBLIC_KEY, 0o444)):
        assert_regular_nonsymlink(path, f"backhaul trust {path.name}")
        os.chown(path, account.pw_uid, account.pw_gid)
        os.chmod(path, mode)
    if BACKHAUL_KNOWN_HOSTS.exists():
        assert_regular_nonsymlink(BACKHAUL_KNOWN_HOSTS, "backhaul known_hosts")
        os.chown(BACKHAUL_KNOWN_HOSTS, account.pw_uid, account.pw_gid)
        os.chmod(BACKHAUL_KNOWN_HOSTS, 0o444)


def certificate_fingerprint(path: Path):
    output = run([
        "/usr/bin/openssl",
        "x509",
        "-in",
        str(path),
        "-noout",
        "-fingerprint",
        "-sha256",
    ]).stdout.strip()
    prefix = "sha256 Fingerprint="
    if not output.lower().startswith(prefix.lower()):
        die("VPS certificate fingerprint output is invalid")
    fingerprint = output.split("=", 1)[1].upper()
    if not re.fullmatch(r"(?:[A-F0-9]{2}:){31}[A-F0-9]{2}", fingerprint):
        die("VPS certificate fingerprint is invalid")
    return fingerprint


def assert_channel_trust(*, require_runtime_owner: bool):
    directory = assert_directory_nonsymlink(
        CHANNEL_TRUST_ROOT,
        "Core channel trust root",
    )
    expected_uid = 0
    expected_gid = 0
    if require_runtime_owner:
        account = pwd.getpwnam(CHANNEL_USER)
        expected_uid = account.pw_uid
        expected_gid = account.pw_gid
    if (
        directory.st_uid != expected_uid
        or directory.st_gid != expected_gid
        or (directory.st_mode & 0o777) != 0o700
    ):
        die("Core channel trust root ownership/mode mismatch")

    for path, maximum, mode in (
        (CHANNEL_PRIVATE_KEY, 32 * 1024, 0o400),
        (CHANNEL_CERTIFICATE, 32 * 1024, 0o444),
        (CHANNEL_CORE_CERTIFICATE, 32 * 1024, 0o444),
        (CHANNEL_RUNTIME_CONFIG, 32 * 1024, 0o444),
    ):
        state = assert_regular_nonsymlink(path, f"Core channel trust {path.name}")
        if state.st_size < 1 or state.st_size > maximum:
            die(f"Core channel trust file size mismatch: {path.name}")
        if (
            state.st_uid != expected_uid
            or state.st_gid != expected_gid
            or (state.st_mode & 0o777) != mode
        ):
            die(f"Core channel trust file ownership/mode mismatch: {path.name}")

    private_text = CHANNEL_PRIVATE_KEY.read_text(encoding="ascii")
    public_text = (
        CHANNEL_CERTIFICATE.read_text(encoding="ascii")
        + CHANNEL_CORE_CERTIFICATE.read_text(encoding="ascii")
    )
    if "PRIVATE KEY" not in private_text or "PRIVATE KEY" in public_text:
        die("Core channel private/public trust boundary mismatch")
    for path in (
        CHANNEL_CERTIFICATE,
        CHANNEL_CORE_CERTIFICATE,
    ):
        if path.read_text(encoding="ascii").count("-----BEGIN CERTIFICATE-----") != 1:
            die(f"Core channel certificate cardinality mismatch: {path.name}")

    run(["/usr/bin/openssl", "pkey", "-in", str(CHANNEL_PRIVATE_KEY), "-check", "-noout"])
    run(["/usr/bin/openssl", "x509", "-in", str(CHANNEL_CERTIFICATE), "-noout"])
    run(["/usr/bin/openssl", "x509", "-in", str(CHANNEL_CORE_CERTIFICATE), "-noout"])
    run([
        "/usr/bin/openssl", "verify", "-purpose", "sslserver",
        "-CAfile", str(CHANNEL_CERTIFICATE), str(CHANNEL_CERTIFICATE),
    ])
    run([
        "/usr/bin/openssl", "verify", "-purpose", "sslclient",
        "-CAfile", str(CHANNEL_CORE_CERTIFICATE), str(CHANNEL_CORE_CERTIFICATE),
    ])
    run([
        "/usr/bin/openssl", "x509", "-in", str(CHANNEL_CERTIFICATE),
        "-noout", "-checkip", PUBLIC_IPV4,
    ])
    certificate_key = run([
        "/usr/bin/openssl", "x509", "-in", str(CHANNEL_CERTIFICATE), "-pubkey", "-noout",
    ]).stdout.strip()
    private_key = run([
        "/usr/bin/openssl", "pkey", "-in", str(CHANNEL_PRIVATE_KEY), "-pubout",
    ]).stdout.strip()
    if certificate_key != private_key:
        die("Core channel Edge certificate/private key mismatch")

    try:
        document = json.loads(CHANNEL_RUNTIME_CONFIG.read_text(encoding="utf-8"))
    except json.JSONDecodeError:
        die("Core channel runtime configuration is invalid JSON")
    if set(document) != {
        "schemaVersion",
        "edgeRegistrationId",
        "channelGeneration",
        "trustGeneration",
        "allowedCoreFingerprints",
    }:
        die("Core channel runtime configuration key set mismatch")
    if document.get("schemaVersion") != "nodedc.device-edge.channel-runtime.v1":
        die("Core channel runtime configuration schema mismatch")
    for key in ("edgeRegistrationId", "channelGeneration", "trustGeneration"):
        if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}", str(document.get(key, ""))):
            die(f"Core channel runtime configuration ref mismatch: {key}")
    fingerprints = document.get("allowedCoreFingerprints")
    if (
        not isinstance(fingerprints, list)
        or not 1 <= len(fingerprints) <= 2
        or len(fingerprints) != len(set(fingerprints))
        or any(
            not isinstance(value, str)
            or not re.fullmatch(r"(?:[A-F0-9]{2}:){31}[A-F0-9]{2}", value)
            for value in fingerprints
        )
    ):
        die("Core channel Core identity allowlist mismatch")
    if certificate_fingerprint(CHANNEL_CORE_CERTIFICATE) not in fingerprints:
        die("Core channel Core trust fingerprint mismatch")
    return document


def assign_channel_trust(account):
    assert_channel_trust(require_runtime_owner=False)
    os.chown(CHANNEL_TRUST_ROOT, account.pw_uid, account.pw_gid)
    os.chmod(CHANNEL_TRUST_ROOT, 0o700)
    for path, mode in (
        (CHANNEL_PRIVATE_KEY, 0o400),
        (CHANNEL_CERTIFICATE, 0o444),
        (CHANNEL_CORE_CERTIFICATE, 0o444),
        (CHANNEL_RUNTIME_CONFIG, 0o444),
    ):
        os.chown(path, account.pw_uid, account.pw_gid)
        os.chmod(path, mode)
    return assert_channel_trust(require_runtime_owner=True)


def apply_nftables(source: Path):
    install_file(source, NFTABLES_CONFIG, 0o644)
    run(["/usr/sbin/nft", "-c", "-f", str(NFTABLES_CONFIG)])
    run(["/usr/sbin/nft", "-f", str(NFTABLES_CONFIG)])
    systemctl("enable", "nftables.service")
    systemctl("disable", "ufw.service", check=False)
    if systemctl("is-enabled", "fail2ban.service", check=False).returncode == 0:
        systemctl("restart", "fail2ban.service")


def apply_foundation(payload: Path):
    account = ensure_service_user()
    extract_vendor_binary(
        LIVE_ROOT / f"vendor/{NODE_ARCHIVE}",
        f"node-v{NODE_VERSION}-linux-x64/bin/node",
        NODE_BIN,
    )
    extract_vendor_binary(
        LIVE_ROOT / f"vendor/{TAILSCALE_ARCHIVE}",
        f"tailscale_{TAILSCALE_VERSION}_amd64/tailscale",
        TAILSCALE_BIN,
    )
    extract_vendor_binary(
        LIVE_ROOT / f"vendor/{TAILSCALE_ARCHIVE}",
        f"tailscale_{TAILSCALE_VERSION}_amd64/tailscaled",
        TAILSCALED_BIN,
    )
    ensure_backhaul_key(account)
    install_file(LIVE_ROOT / FOUNDATION_ENTRIES[0], SSHD_DROPIN, 0o644)
    install_file(LIVE_ROOT / FOUNDATION_ENTRIES[2], TAILSCALE_UNIT, 0o644)
    run(["/usr/sbin/sshd", "-t"])
    apply_nftables(LIVE_ROOT / FOUNDATION_ENTRIES[1])
    systemctl("daemon-reload")
    systemctl("enable", "--now", "nodedc-b2-tailscaled.service")
    systemctl("reload", "ssh.service")
    validate_foundation_runtime(require_running_tailnet=False)


def apply_runtime_reconciliation(_payload: Path):
    for path in (NODE_BIN, TAILSCALE_BIN, TAILSCALED_BIN):
        os.chown(path, 0, 0)
        os.chmod(path, 0o755)
    validate_runtime_reconciliation()


def materialize_known_hosts(account):
    line = (
        f"[{BACKHAUL_TARGET_IP}]:{BACKHAUL_TARGET_PORT} "
        f"{BACKHAUL_TARGET_HOST_KEY}\n"
    )
    TRUST_ROOT.mkdir(parents=True, exist_ok=True, mode=0o700)
    temporary = BACKHAUL_KNOWN_HOSTS.with_name(".backhaul_known_hosts.installing")
    temporary.write_text(line, encoding="ascii")
    os.chown(temporary, account.pw_uid, account.pw_gid)
    os.chmod(temporary, 0o444)
    os.replace(temporary, BACKHAUL_KNOWN_HOSTS)
    fingerprint = run(["/usr/bin/ssh-keygen", "-lf", str(BACKHAUL_KNOWN_HOSTS)]).stdout
    if BACKHAUL_TARGET_FINGERPRINT not in fingerprint:
        die("backhaul target host key fingerprint mismatch")


def apply_backhaul(payload: Path):
    account = ensure_service_user(BACKHAUL_USER, "/var/lib/nodedc-b2-vps/backhaul")
    assign_backhaul_trust(account)
    materialize_known_hosts(account)
    install_file(LIVE_ROOT / BACKHAUL_ENTRIES[1], BACKHAUL_UNIT, 0o644)
    systemctl("daemon-reload")
    systemctl("enable", "--now", "nodedc-b2-backhaul.service")
    validate_backhaul_runtime()


def apply_relay(payload: Path):
    ensure_service_user(RELAY_USER, "/var/lib/nodedc-b2-vps/relay")
    install_file(LIVE_ROOT / RELAY_ENTRIES[1], RELAY_UNIT, 0o644)
    apply_nftables(LIVE_ROOT / RELAY_ENTRIES[0])
    systemctl("daemon-reload")
    systemctl("enable", "--now", "nodedc-b2-relay.service")
    validate_relay_runtime()


def apply_core_channel(payload: Path):
    account = ensure_service_user(
        CHANNEL_USER,
        "/var/lib/nodedc-b2-vps/channel-runtime",
    )
    assign_channel_trust(account)
    install_file(LIVE_ROOT / CORE_CHANNEL_ENTRIES[7], CHANNEL_UNIT, 0o644)
    apply_nftables(LIVE_ROOT / CORE_CHANNEL_ENTRIES[6])
    systemctl("daemon-reload")
    systemctl("enable", "--now", "nodedc-device-edge-channel.service")
    validate_core_channel_runtime()


def sshd_effective():
    return run(["/usr/sbin/sshd", "-T"]).stdout.lower()


def validate_foundation_runtime(*, require_running_tailnet: bool, expected_key_user=SERVICE_USER):
    source_file_state("foundation")
    if run([str(NODE_BIN), "--version"]).stdout.strip() != f"v{NODE_VERSION}":
        die("Node runtime version mismatch")
    tailscale_version = run([str(TAILSCALE_BIN), "version"]).stdout.splitlines()[0].strip()
    if tailscale_version != TAILSCALE_VERSION:
        die("Tailscale runtime version mismatch")
    if not service_active("nodedc-b2-tailscaled.service"):
        die("VPS tailscaled service is not active")
    effective = sshd_effective()
    for required in (
        "permitrootlogin without-password",
        "passwordauthentication no",
        "kbdinteractiveauthentication no",
        "pubkeyauthentication yes",
        "x11forwarding no",
        "allowagentforwarding no",
        "allowtcpforwarding no",
        "gatewayports no",
        "permittunnel no",
        "permituserenvironment no",
        "maxauthtries 3",
        "logingracetime 20",
    ):
        if required not in effective:
            die(f"VPS effective SSH boundary mismatch: {required}")
    nft = run(["/usr/sbin/nft", "list", "table", "inet", "nodedc_b2_vps"]).stdout
    if "policy drop" not in nft or "tcp dport 22" not in nft:
        die("VPS foundation firewall contract mismatch")
    if "tcp dport 9921" in nft:
        die("VPS foundation unexpectedly opens B2 ingress")
    for port in (18221, 19921, 9921):
        assert_port_closed(port)
    tailscale_account = pwd.getpwnam(SERVICE_USER)
    if tailscale_account.pw_shell != "/usr/sbin/nologin":
        die("VPS service user runtime mismatch")
    account = pwd.getpwnam(expected_key_user)
    if account.pw_shell != "/usr/sbin/nologin":
        die("VPS backhaul credential owner shell mismatch")
    trust_stat = assert_directory_nonsymlink(
        TRUST_ROOT,
        "runtime trust root",
    )
    if trust_stat.st_uid != account.pw_uid or (trust_stat.st_mode & 0o777) != 0o700:
        die("VPS backhaul trust root ownership/mode mismatch")
    for path, mode in ((BACKHAUL_PRIVATE_KEY, 0o400), (BACKHAUL_PUBLIC_KEY, 0o444)):
        path_stat = assert_regular_nonsymlink(path, f"runtime trust {path.name}")
        if path_stat.st_uid != account.pw_uid or (path_stat.st_mode & 0o777) != mode:
            die("VPS backhaul key ownership/mode mismatch")
    status = tailscale_status()
    state = status.get("BackendState")
    if require_running_tailnet:
        if state != "Running":
            die(f"VPS Tailscale node is not enrolled/running: {state}")
        self_state = status.get("Self") or {}
        if self_state.get("HostName") != "nodedc-b2-vps":
            die("VPS Tailscale node name mismatch")
        if self_state.get("Online") is not True:
            die("VPS Tailscale node is not online")
        if sorted(self_state.get("Tags") or []) != [TAILSCALE_REQUIRED_TAG]:
            die("VPS Tailscale service tag mismatch")
    elif state not in {"NeedsLogin", "Stopped", "Running", "NoState", "Starting"}:
        die(f"VPS Tailscale foundation state is unexpected: {state}")
    return status


def validate_runtime_reconciliation():
    source_file_state("runtime-reconciliation")
    for path, digest in (
        (NODE_BIN, NODE_BIN_SHA256),
        (TAILSCALE_BIN, TAILSCALE_BIN_SHA256),
        (TAILSCALED_BIN, TAILSCALED_BIN_SHA256),
    ):
        state = assert_regular_nonsymlink(path, f"reconciled runtime {path.name}")
        if (
            state.st_uid != 0
            or state.st_gid != 0
            or (state.st_mode & 0o777) != 0o755
            or sha256_file(path) != digest
        ):
            die(f"reconciled runtime binary mismatch: {path.name}")
    validate_foundation_runtime(require_running_tailnet=False)
    for port in (CHANNEL_HEALTH_PORT, CHANNEL_PUBLIC_PORT, 9921):
        assert_port_closed(port)


def validate_backhaul_runtime():
    validate_foundation_runtime(
        require_running_tailnet=True,
        expected_key_user=BACKHAUL_USER,
    )
    source_file_state("backhaul")
    if not service_active("nodedc-b2-backhaul.service"):
        die("VPS backhaul service is not active")
    if not port_is_open("127.0.0.1", 19921, timeout=5):
        die("VPS backhaul local forward is unavailable")
    assert_port_closed(9921)
    assert_port_closed(18221)
    account = pwd.getpwnam(BACKHAUL_USER)
    tailscale_account = pwd.getpwnam(SERVICE_USER)
    if account.pw_uid == tailscale_account.pw_uid:
        die("backhaul and Tailscale runtime identities are not isolated")
    known = assert_regular_nonsymlink(BACKHAUL_KNOWN_HOSTS, "backhaul known_hosts")
    if known.st_uid != account.pw_uid or (known.st_mode & 0o777) != 0o444:
        die("backhaul known_hosts ownership/mode mismatch")
    return True


def relay_health():
    last_error = None
    for _attempt in range(60):
        try:
            with urllib.request.urlopen("http://127.0.0.1:18221/healthz", timeout=3) as response:
                payload = json.loads(response.read(65537).decode("utf-8"))
                if response.status == 200:
                    return payload
        except Exception as error:
            last_error = str(error)
        time.sleep(2)
    die(f"VPS relay health timeout: {last_error}")


def core_channel_health(*, require_accepted: bool):
    last_error = None
    for _attempt in range(60):
        try:
            with urllib.request.urlopen(
                f"http://127.0.0.1:{CHANNEL_HEALTH_PORT}/healthz",
                timeout=3,
            ) as response:
                payload = json.loads(response.read(65537).decode("utf-8"))
                if (
                    response.status == 200
                    and payload.get("ok") is True
                    and (
                        not require_accepted
                        or payload.get("channel") == "accepted"
                    )
                ):
                    return payload
                last_error = f"channel={payload.get('channel')}"
        except Exception as error:
            last_error = str(error)
        time.sleep(2)
    die(f"VPS Core channel health timeout: {last_error}")


def validate_core_channel_runtime():
    validate_foundation_runtime(
        require_running_tailnet=False,
        expected_key_user=SERVICE_USER,
    )
    source_file_state("core-channel")
    assert_channel_trust(require_runtime_owner=True)
    if service_active("nodedc-b2-backhaul.service"):
        die("frozen VPS backhaul service became active")
    if service_active("nodedc-b2-relay.service"):
        die("frozen VPS relay service became active")
    if not service_active("nodedc-device-edge-channel.service"):
        die("VPS Core channel service is not active")
    health = core_channel_health(require_accepted=True)
    expected = {
        "ok": True,
        "service": "nodedc-device-edge-channel",
        "channel": "accepted",
        "trackerIngress": "disabled",
        "commandTransport": "disabled",
    }
    for key, value in expected.items():
        if health.get(key) != value:
            die(f"VPS Core channel health contract mismatch: {key}")
    if not port_is_open(PUBLIC_IPV4, CHANNEL_PUBLIC_PORT, timeout=5):
        die("VPS public Core channel listener is unavailable")
    assert_port_closed(9921)
    nft = run(["/usr/sbin/nft", "list", "table", "inet", "nodedc_b2_vps"]).stdout
    if (
        "policy drop" not in nft
        or "tcp dport 8443" not in nft
        or "tcp dport 9921" in nft
    ):
        die("VPS Core channel firewall contract mismatch")
    unit = run([
        "/usr/bin/systemctl",
        "show",
        "nodedc-device-edge-channel.service",
        "--property=User,Group,NoNewPrivileges,MemoryMax,MemorySwapMax,CPUQuotaPerSecUSec,TasksMax,LimitNOFILE",
    ]).stdout
    for required in (
        "User=nodedc-channel",
        "Group=nodedc-channel",
        "NoNewPrivileges=yes",
        "MemoryMax=134217728",
        "MemorySwapMax=0",
        "TasksMax=64",
        "LimitNOFILE=1024",
    ):
        if required not in unit:
            die(f"VPS Core channel resource boundary mismatch: {required}")
    return health


def validate_relay_runtime():
    validate_backhaul_runtime()
    source_file_state("relay")
    if not service_active("nodedc-b2-relay.service"):
        die("VPS relay service is not active")
    relay_account = pwd.getpwnam(RELAY_USER)
    backhaul_account = pwd.getpwnam(BACKHAUL_USER)
    tailscale_account = pwd.getpwnam(SERVICE_USER)
    if len({relay_account.pw_uid, backhaul_account.pw_uid, tailscale_account.pw_uid}) != 3:
        die("relay, backhaul, and Tailscale runtime identities are not isolated")
    health = relay_health()
    expected = {
        "ok": True,
        "service": "nodedc-device-edge-relay",
        "ingress": "relay-only",
        "protocolInspection": "disabled",
        "commandTransport": "disabled",
        "sourceAdmission": "public-ipv4-only",
    }
    for key, value in expected.items():
        if health.get(key) != value:
            die(f"VPS relay health contract mismatch: {key}")
    if not port_is_open(PUBLIC_IPV4, 9921, timeout=5):
        die("VPS public B2 listener is unavailable")
    nft = run(["/usr/sbin/nft", "list", "table", "inet", "nodedc_b2_vps"]).stdout
    if "policy drop" not in nft or "tcp dport 9921" not in nft:
        die("VPS relay firewall contract mismatch")
    return health


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("VPS artifact archive collision")
    os.replace(artifact, destination)
    return destination


def restore_service_enablement(metadata):
    for name, state in metadata.get("services", {}).items():
        if state.get("enabled"):
            systemctl("enable", name, check=False)
        else:
            systemctl("disable", name, check=False)
        if state.get("active"):
            systemctl("start", name, check=False)
        else:
            systemctl("stop", name, check=False)


def rollback(backup: Path, phase: str):
    for service in (
        "nodedc-b2-relay.service",
        "nodedc-b2-backhaul.service",
        "nodedc-device-edge-channel.service",
        "nodedc-b2-tailscaled.service",
    ):
        if phase == "foundation" or service != "nodedc-b2-tailscaled.service":
            systemctl("disable", "--now", service, check=False)
    restore_backup(backup, phase)
    metadata = json.loads((backup / "backup.json").read_text(encoding="utf-8"))
    systemctl("daemon-reload", check=False)
    if NFTABLES_CONFIG.exists():
        run(["/usr/sbin/nft", "-f", str(NFTABLES_CONFIG)], check=False)
    else:
        rules = backup / "nft-ruleset-before.nft"
        if rules.exists() and rules.stat().st_size:
            run(["/usr/sbin/nft", "-f", str(rules)], check=False)
    restore_service_enablement(metadata)
    systemctl("restart", "fail2ban.service", check=False)
    run(["/usr/sbin/sshd", "-t"], check=False)
    systemctl("reload", "ssh.service", check=False)
    users_before = metadata.get("serviceUsersExisted", {})
    if phase == "backhaul":
        assign_backhaul_trust(pwd.getpwnam(SERVICE_USER))
        if not users_before.get(BACKHAUL_USER, False) and user_exists(BACKHAUL_USER):
            run(["/usr/sbin/userdel", BACKHAUL_USER], check=False)
    if phase == "relay":
        if not users_before.get(RELAY_USER, False) and user_exists(RELAY_USER):
            run(["/usr/sbin/userdel", RELAY_USER], check=False)
    if phase == "core-channel":
        if not users_before.get(CHANNEL_USER, False) and user_exists(CHANNEL_USER):
            run(["/usr/sbin/userdel", CHANNEL_USER], check=False)
    if phase == "foundation" and not metadata.get("serviceUserExisted"):
        runtime_state_root = Path("/var/lib/nodedc-b2-vps")
        if LIVE_ROOT.exists() and not LIVE_ROOT.is_symlink():
            shutil.rmtree(LIVE_ROOT)
        if runtime_state_root.exists() and not runtime_state_root.is_symlink():
            shutil.rmtree(runtime_state_root)
        if user_exists():
            run(["/usr/sbin/userdel", SERVICE_USER], check=False)


def plan_artifact(artifact_argument: str):
    assert_root()
    artifact = Path(artifact_argument)
    with tempfile.TemporaryDirectory(prefix="nodedc-b2-vps-plan-") as directory:
        loaded = load_artifact(artifact, Path(directory))
        evidence = preflight(loaded)
    phase = loaded["phase"]
    print("== plan ==")
    print(f"artifact={loaded['artifact'].name}")
    print(f"sha256={loaded['sha256']}")
    print(f"id={loaded['manifest']['id']}")
    print(f"component={COMPONENT}")
    print(f"type={ARTIFACT_TYPE}")
    print(f"phase={phase}")
    print(f"predecessor={evidence['predecessor']}")
    print(f"payload_root={LIVE_ROOT}")
    print(f"runtime_host={RUNTIME_HOST}")
    print(f"public_ipv4={PUBLIC_IPV4}")
    print("management_ssh=root-key-only:tcp/22")
    print(f"management_key_fingerprint={MANAGEMENT_KEY_FINGERPRINT}")
    print(f"server_host_key_fingerprint={SERVER_HOST_KEY_FINGERPRINT}")
    if phase == "foundation":
        print(f"node_runtime={NODE_VERSION}:sha256:{NODE_ARCHIVE_SHA256}")
        print(f"tailscale_runtime={TAILSCALE_VERSION}:sha256:{TAILSCALE_ARCHIVE_SHA256}")
        print("firewall=default-deny:public-tcp/22-only")
        print("tailscale=enrollment-required-after-deploy-ok")
        print("backhaul_key=runner-managed-new-ed25519")
        print("public_b2_ingress=disabled")
        print("services=nodedc-b2-tailscaled")
        print(f"tailscale_runtime_identity={SERVICE_USER}")
    elif phase == "runtime-reconciliation":
        print("runtime_reconciliation=exact-known-binaries:0644=>0755")
        print("failed_patch=device-edge-vps-core-channel-20260812-001")
        print("public_core_channel=disabled")
        print("public_b2_ingress=disabled")
        print("tracker_tcp_9921=closed")
        print("services=preserved:nodedc-b2-tailscaled")
    elif phase == "backhaul":
        print(f"target={BACKHAUL_TARGET_IP}:{BACKHAUL_TARGET_PORT}")
        print(f"target_host_key_fingerprint={BACKHAUL_TARGET_FINGERPRINT}")
        print("local_forward=127.0.0.1:19921=>127.0.0.1:9921")
        print("proxy=tailscale-userspace-socks5:127.0.0.1:1055")
        print("public_b2_ingress=disabled")
        print("services=nodedc-b2-backhaul")
        print(f"backhaul_runtime_identity={BACKHAUL_USER}:private-key-owner")
    elif phase == "relay":
        print("public_b2_ingress=155.212.211.15:9921/tcp")
        print("health=127.0.0.1:18221")
        print("private_upstream=127.0.0.1:19921")
        print("source_admission=public-ipv4-only")
        print("services=nodedc-b2-relay")
        print(f"relay_runtime_identity={RELAY_USER}:no-credentials")
    else:
        print("public_core_channel=155.212.211.15:8443/tcp:tls13-mtls-h2")
        print(f"health=127.0.0.1:{CHANNEL_HEALTH_PORT}")
        print("public_b2_ingress=disabled")
        print("tracker_tcp_9921=closed")
        print("services=nodedc-device-edge-channel")
        print(f"channel_runtime_identity={CHANNEL_USER}:host-local-private-key")
        print("peer_trust=preprovisioned-pinned-self-signed-core-certificate+fingerprint")
    print("command_transport=disabled")
    print("gelios=untouched")
    print("dns=unchanged")
    print("b2_routes=unchanged")
    print("state=new")
    print("== files ==")
    for entry in loaded["entries"]:
        print(f"  {entry}")


def apply_artifact(artifact_argument: str):
    assert_root()
    acquire_lock()
    loaded = None
    backup_id = None
    backup = None
    try:
        with tempfile.TemporaryDirectory(prefix="nodedc-b2-vps-apply-") as directory:
            loaded = load_artifact(Path(artifact_argument), Path(directory))
            preflight(loaded)
            backup_id, backup = create_backup(
                loaded["manifest"]["id"],
                loaded["phase"],
            )
            publish_payload(loaded["payload"], loaded["entries"])
            if loaded["phase"] == "foundation":
                apply_foundation(loaded["payload"])
            elif loaded["phase"] == "runtime-reconciliation":
                apply_runtime_reconciliation(loaded["payload"])
            elif loaded["phase"] == "backhaul":
                apply_backhaul(loaded["payload"])
            elif loaded["phase"] == "relay":
                apply_relay(loaded["payload"])
            else:
                apply_core_channel(loaded["payload"])

        archived = archive_artifact(loaded["artifact"], APPLIED_ROOT)
        record = {
            "status": "ok",
            "patch": loaded["manifest"]["id"],
            "component": COMPONENT,
            "phase": loaded["phase"],
            "sha256": loaded["sha256"],
            "artifact": archived.name,
            "backup": backup_id,
            "appliedAt": datetime.now(timezone.utc).isoformat(),
        }
        write_journal(APPLIED_JOURNAL, record)
        print(
            f"deploy-ok patch={record['patch']} component={COMPONENT} "
            f"backup={backup_id}"
        )
    except Exception as error:
        rollback_status = "not-required"
        if backup is not None and loaded is not None:
            try:
                rollback(backup, loaded["phase"])
                rollback_status = "ok"
            except Exception as rollback_error:
                rollback_status = f"failed:{type(rollback_error).__name__}"
        if loaded is not None and loaded["artifact"].exists():
            destination = FAILED_ROOT / (
                f"{loaded['artifact'].name}."
                f"{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')}"
            )
            os.replace(loaded["artifact"], destination)
        if loaded is not None:
            write_journal(FAILED_JOURNAL, {
                "status": "failed",
                "patch": loaded["manifest"]["id"],
                "component": COMPONENT,
                "phase": loaded["phase"],
                "sha256": loaded["sha256"],
                "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")
    print(f"path={path}")
    print(f"sha256={sha256_file(path)}")
    print(f"python={sys.version.split()[0]}")
    print(f"runtime_host={socket.gethostname()}")
    print(f"component={COMPONENT}")
    print(f"live_root={LIVE_ROOT}")
    print(f"inbox_root={INBOX_ROOT}")
    print(f"node_runtime={'present' if NODE_BIN.exists() else 'absent'}")
    print(f"tailscale_runtime={'present' if TAILSCALE_BIN.exists() else 'absent'}")
    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-b2-vps-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)
