Temporary additive watchdog workaround for Augani/dory#45 until the gvproxy reconciliation fix ships

dory-port-watchdog.py

#!/usr/bin/python3
"""Temporary additive port-forward watchdog for pre-fix Dory builds.

This is intentionally not a replacement for Dory's reconciler. It restores container
forwards that Docker still wants but gvproxy has dropped. It removes only forwards that
this watchdog created, and only when the same dory-hv instance no longer wants them.
"""

from __future__ import annotations

import argparse
import dataclasses
import fcntl
import glob
import ipaddress
import json
import os
import pathlib
import plistlib
import re
import shutil
import signal
import subprocess
import sys
import time
from typing import Any, Iterable


LABEL = "dev.dory.port-watchdog"
DEFAULT_STATE_DIR = pathlib.Path.home() / ".dory" / "hv"
INSTALL_DIR = pathlib.Path.home() / "Library" / "Application Support" / "Dory Workarounds" / "port-watchdog"
INSTALLED_SCRIPT = pathlib.Path.home() / "bin.shared" / "dory-port-watchdog.py"
LEGACY_INSTALLED_SCRIPT = INSTALL_DIR / "dory-port-watchdog.py"
STATE_FILE = INSTALL_DIR / "state.json"
LOCK_FILE = INSTALL_DIR / "lock"
LOG_FILE = INSTALL_DIR / "watchdog.log"
PLIST_FILE = pathlib.Path.home() / "Library" / "LaunchAgents" / f"{LABEL}.plist"
LOOPBACK_INTENT_LABEL = "dev.dory.internal.loopback-port-intent"
SUPPORTED_PROTOCOLS = {"tcp", "udp"}


@dataclasses.dataclass(frozen=True, order=True)
class Forward:
    protocol: str
    local: str
    remote: str

    def as_json(self) -> dict[str, str]:
        return dataclasses.asdict(self)

    @classmethod
    def from_json(cls, value: dict[str, Any]) -> "Forward":
        return cls(
            protocol=str(value["protocol"]),
            local=str(value["local"]),
            remote=str(value["remote"]),
        )


def log(message: str) -> None:
    print(f"dory-port-watchdog: {message}", flush=True)


def curl_json(socket_path: pathlib.Path, url: str, timeout: float = 3) -> Any:
    result = subprocess.run(
        [
            "/usr/bin/curl",
            "-sS",
            "-f",
            "--max-time",
            str(timeout),
            "--unix-socket",
            str(socket_path),
            url,
        ],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        check=False,
    )
    if result.returncode != 0:
        detail = result.stderr.decode("utf-8", "replace").strip()
        raise RuntimeError(detail or f"curl exited {result.returncode}")
    return json.loads(result.stdout)


def curl_post(socket_path: pathlib.Path, endpoint: str, body: dict[str, str]) -> None:
    result = subprocess.run(
        [
            "/usr/bin/curl",
            "-sS",
            "-f",
            "--max-time",
            "3",
            "--unix-socket",
            str(socket_path),
            "-X",
            "POST",
            "-H",
            "Content-Type: application/json",
            "-d",
            json.dumps(body, separators=(",", ":")),
            f"http://gvproxy/services/forwarder/{endpoint}",
        ],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.PIPE,
        check=False,
    )
    if result.returncode != 0:
        detail = result.stderr.decode("utf-8", "replace").strip()
        raise RuntimeError(detail or f"gvproxy {endpoint} exited {result.returncode}")


def split_endpoint(value: str) -> tuple[str, int] | None:
    match = re.fullmatch(r"(?:tcp://)?(\[[^]]+\]|[^:]+):(\d+)", value)
    if not match:
        return None
    host = match.group(1).strip("[]")
    port = int(match.group(2))
    if not 1 <= port <= 65535:
        return None
    return host, port


def parse_registry(value: Any) -> set[Forward]:
    result: set[Forward] = set()
    if not isinstance(value, list):
        return result
    for item in value:
        if not isinstance(item, dict):
            continue
        protocol = str(item.get("protocol", "")).lower()
        local = str(item.get("local", ""))
        remote = str(item.get("remote", ""))
        if protocol not in SUPPORTED_PROTOCOLS:
            continue
        if split_endpoint(local) is None or split_endpoint(remote) is None:
            continue
        result.add(Forward(protocol, local, remote.removeprefix("tcp://")))
    return result


def infer_guest_host(registry: Any) -> str:
    if isinstance(registry, list):
        for item in registry:
            if not isinstance(item, dict):
                continue
            parsed = split_endpoint(str(item.get("remote", "")))
            if parsed and parsed[1] in {2377, 2380}:
                return parsed[0]
        for item in registry:
            if not isinstance(item, dict):
                continue
            parsed = split_endpoint(str(item.get("remote", "")))
            if parsed:
                return parsed[0]
    return "192.168.127.2"


def parse_loopback_intents(raw: Any) -> dict[str, dict[str, str]]:
    if not isinstance(raw, str):
        return {}
    try:
        value = json.loads(raw)
    except json.JSONDecodeError:
        return {}
    if not isinstance(value, dict):
        return {}
    result: dict[str, dict[str, str]] = {}
    for key, entry in value.items():
        if not re.fullmatch(r"[1-9]\d{0,4}/(?:tcp|tcp6|udp|udp6)", str(key), re.IGNORECASE):
            continue
        if isinstance(entry, str):
            entry = {"": entry}
        if not isinstance(entry, dict):
            continue
        valid = {
            str(port): str(intent)
            for port, intent in entry.items()
            if (str(port) == "" or str(port).isdigit())
            and str(intent) in {"ipv4", "ipv6", "localhost"}
        }
        if valid:
            result[str(key).lower()] = valid
    return result


def requested_host(
    docker_host: Any,
    private_port: Any,
    public_port: Any,
    docker_type: str,
    intents: dict[str, dict[str, str]],
) -> str | None:
    host = str(docker_host).strip() if docker_host is not None else ""
    mapping = intents.get(f"{private_port}/{docker_type.lower()}")
    if not mapping:
        return host or None
    intent = mapping.get(str(public_port), mapping.get(""))
    return {
        "ipv4": "127.0.0.1",
        "ipv6": "::1",
        "localhost": "localhost",
    }.get(intent, host or None)


def valid_ip(value: str) -> bool:
    candidate = value.strip("[]")
    if "%" in candidate:
        address, zone = candidate.split("%", 1)
        if not zone or not re.fullmatch(r"[A-Za-z0-9_-]+", zone):
            return False
        candidate = address
    try:
        ipaddress.ip_address(candidate)
        return True
    except ValueError:
        return False


def local_hosts(publish_host: str, requested: str | None) -> list[str]:
    value = requested.strip() if requested else ""
    lan_enabled = publish_host == "0.0.0.0"
    if value == "127.0.0.1":
        return ["127.0.0.1"]
    if value in {"::1", "[::1]"}:
        return ["[::1]"]
    if value in {"", "0.0.0.0", "::", "[::]"}:
        return ["0.0.0.0", "[::1]"] if lan_enabled else ["127.0.0.1", "[::1]"]
    if not lan_enabled or not valid_ip(value):
        return ["127.0.0.1", "[::1]"]
    return [f"[{value.strip('[]')}]" if ":" in value else value]


def desired_forwards(
    containers: Any,
    publish_host: str,
    guest_host: str,
) -> set[Forward]:
    result: set[Forward] = set()
    if not isinstance(containers, list):
        return result
    for container in containers:
        if not isinstance(container, dict):
            continue
        labels = container.get("Labels") if isinstance(container.get("Labels"), dict) else {}
        intents = parse_loopback_intents(labels.get(LOOPBACK_INTENT_LABEL))
        ports = container.get("Ports")
        if not isinstance(ports, list):
            continue
        for item in ports:
            if not isinstance(item, dict):
                continue
            protocol = str(item.get("Type", "tcp")).strip().lower()
            if protocol in {"tcp6", "udp6"}:
                protocol = protocol[:-1]
            public_port = item.get("PublicPort")
            if protocol not in SUPPORTED_PROTOCOLS or not isinstance(public_port, int):
                continue
            if not 1 <= public_port <= 65535:
                continue
            requested = requested_host(
                item.get("IP"),
                item.get("PrivatePort"),
                public_port,
                str(item.get("Type", "tcp")),
                intents,
            )
            local_port = public_port if public_port >= 1024 else 60000 + public_port
            for host in local_hosts(publish_host, requested):
                result.add(Forward(protocol, f"{host}:{local_port}", f"{guest_host}:{public_port}"))
    return result


def engine_pid(api_socket: pathlib.Path) -> int | None:
    match = re.fullmatch(r"n(\d+)", api_socket.parent.name)
    return int(match.group(1)) if match else None


def publish_host_for(api_socket: pathlib.Path) -> str:
    pid = engine_pid(api_socket)
    if pid is None:
        return "127.0.0.1"
    result = subprocess.run(
        ["/bin/ps", "-p", str(pid), "-ww", "-o", "command="],
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL,
        text=True,
        check=False,
    )
    match = re.search(r"(?:^|\s)--publish-host\s+(\S+)", result.stdout)
    return match.group(1) if match else "127.0.0.1"


def active_api(state_dir: pathlib.Path) -> tuple[pathlib.Path, Any] | None:
    candidates = sorted(
        (pathlib.Path(path) for path in glob.glob(str(state_dir / "n*" / "a.sock"))),
        key=lambda path: path.stat().st_mtime if path.exists() else 0,
        reverse=True,
    )
    for candidate in candidates:
        try:
            return candidate, curl_json(candidate, "http://gvproxy/services/forwarder/all")
        except (OSError, RuntimeError, json.JSONDecodeError):
            continue
    return None


def load_owned(engine_id: str) -> set[Forward]:
    try:
        value = json.loads(STATE_FILE.read_text())
    except (OSError, json.JSONDecodeError):
        return set()
    if value.get("engine_id") != engine_id or not isinstance(value.get("owned"), list):
        return set()
    try:
        return {Forward.from_json(item) for item in value["owned"]}
    except (KeyError, TypeError, ValueError):
        return set()


def save_owned(engine_id: str, owned: Iterable[Forward]) -> None:
    INSTALL_DIR.mkdir(mode=0o700, parents=True, exist_ok=True)
    temporary = STATE_FILE.with_suffix(".tmp")
    temporary.write_text(json.dumps({
        "engine_id": engine_id,
        "owned": [item.as_json() for item in sorted(owned)],
    }, indent=2) + "\n")
    os.chmod(temporary, 0o600)
    os.replace(temporary, STATE_FILE)


def reconcile_once(state_dir: pathlib.Path, dry_run: bool = False) -> tuple[int, int]:
    active = active_api(state_dir)
    if active is None:
        return 0, 0
    api_socket, registry_json = active
    engine_socket = state_dir / "engine.sock"
    try:
        containers = curl_json(engine_socket, "http://d/v1.41/containers/json")
    except (OSError, RuntimeError, json.JSONDecodeError):
        return 0, 0

    actual = parse_registry(registry_json)
    wanted = desired_forwards(
        containers,
        publish_host_for(api_socket),
        infer_guest_host(registry_json),
    )
    engine_id = str(api_socket.parent)
    owned = load_owned(engine_id)
    removed = 0
    restored = 0

    for forward in sorted((owned & actual) - wanted):
        if dry_run:
            log(f"would remove sidecar-owned {forward.local}/{forward.protocol}")
            continue
        try:
            curl_post(api_socket, "unexpose", {
                "local": forward.local,
                "protocol": forward.protocol,
            })
            owned.discard(forward)
            removed += 1
            log(f"removed sidecar-owned {forward.local}/{forward.protocol}")
        except RuntimeError as error:
            log(f"could not remove {forward.local}/{forward.protocol}: {error}")

    owned.intersection_update(actual)
    for forward in sorted(wanted - actual):
        if dry_run:
            log(f"would restore {forward.local}/{forward.protocol} -> {forward.remote}")
            continue
        try:
            curl_post(api_socket, "expose", {
                "local": forward.local,
                "remote": forward.remote,
                "protocol": forward.protocol,
            })
            owned.add(forward)
            restored += 1
            log(f"restored {forward.local}/{forward.protocol} -> {forward.remote}")
        except RuntimeError as error:
            log(f"could not restore {forward.local}/{forward.protocol}: {error}")

    if not dry_run:
        save_owned(engine_id, owned)
    return restored, removed


def with_lock() -> Any:
    INSTALL_DIR.mkdir(mode=0o700, parents=True, exist_ok=True)
    handle = LOCK_FILE.open("a+")
    fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
    return handle


def run_loop(state_dir: pathlib.Path, interval: float) -> None:
    stopping = False

    def stop(_signum: int, _frame: Any) -> None:
        nonlocal stopping
        stopping = True

    signal.signal(signal.SIGTERM, stop)
    signal.signal(signal.SIGINT, stop)
    lock = with_lock()
    try:
        while not stopping:
            try:
                reconcile_once(state_dir)
            except Exception as error:
                log(f"cycle failed: {error}")
            deadline = time.monotonic() + interval
            while not stopping and time.monotonic() < deadline:
                time.sleep(min(0.25, deadline - time.monotonic()))
    finally:
        lock.close()


def launchctl(*arguments: str, check: bool = True) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        ["/bin/launchctl", *arguments],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        check=check,
    )


def install(interval: float) -> None:
    if not pathlib.Path("/usr/bin/python3").exists():
        raise RuntimeError("/usr/bin/python3 is unavailable")
    INSTALL_DIR.mkdir(mode=0o700, parents=True, exist_ok=True)
    INSTALLED_SCRIPT.parent.mkdir(mode=0o755, parents=True, exist_ok=True)
    PLIST_FILE.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
    source = pathlib.Path(__file__).resolve()
    if source != INSTALLED_SCRIPT.resolve():
        shutil.copy2(source, INSTALLED_SCRIPT)
    os.chmod(INSTALLED_SCRIPT, 0o700)
    payload = {
        "Label": LABEL,
        "ProgramArguments": [
            "/usr/bin/python3",
            str(INSTALLED_SCRIPT),
            "run",
            "--interval",
            str(interval),
        ],
        "RunAtLoad": True,
        "KeepAlive": True,
        "ThrottleInterval": 10,
        "ProcessType": "Background",
        "StandardOutPath": str(LOG_FILE),
        "StandardErrorPath": str(LOG_FILE),
    }
    with PLIST_FILE.open("wb") as handle:
        plistlib.dump(payload, handle, sort_keys=True)
    os.chmod(PLIST_FILE, 0o600)
    domain = f"gui/{os.getuid()}"
    launchctl("bootout", domain, str(PLIST_FILE), check=False)
    launchctl("bootstrap", domain, str(PLIST_FILE))
    launchctl("kickstart", "-k", f"{domain}/{LABEL}")
    if LEGACY_INSTALLED_SCRIPT != INSTALLED_SCRIPT:
        try:
            LEGACY_INSTALLED_SCRIPT.unlink()
        except FileNotFoundError:
            pass
    log(f"installed LaunchAgent {LABEL}; log: {LOG_FILE}")


def uninstall() -> None:
    domain = f"gui/{os.getuid()}"
    launchctl("bootout", domain, str(PLIST_FILE), check=False)
    try:
        PLIST_FILE.unlink()
    except FileNotFoundError:
        pass
    for path in [STATE_FILE, LOCK_FILE, LOG_FILE, INSTALLED_SCRIPT, LEGACY_INSTALLED_SCRIPT]:
        try:
            path.unlink()
        except FileNotFoundError:
            pass
    try:
        INSTALL_DIR.rmdir()
    except OSError:
        pass
    log(f"uninstalled LaunchAgent {LABEL}")


def status() -> int:
    domain = f"gui/{os.getuid()}"
    result = launchctl("print", f"{domain}/{LABEL}", check=False)
    if result.returncode == 0:
        print(result.stdout, end="")
        return 0
    print(f"{LABEL} is not installed")
    return 1


def positive_interval(value: str) -> float:
    parsed = float(value)
    if parsed < 1:
        raise argparse.ArgumentTypeError("interval must be at least one second")
    return parsed


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--state-dir",
        type=pathlib.Path,
        default=DEFAULT_STATE_DIR,
        help=f"Dory runtime state directory (default: {DEFAULT_STATE_DIR})",
    )
    subparsers = parser.add_subparsers(dest="command", required=True)
    once = subparsers.add_parser("once", help="Run one reconciliation")
    once.add_argument("--dry-run", action="store_true")
    run = subparsers.add_parser("run", help="Run in the foreground")
    run.add_argument("--interval", type=positive_interval, default=2.0)
    install_parser = subparsers.add_parser("install", help="Install and start the LaunchAgent")
    install_parser.add_argument("--interval", type=positive_interval, default=2.0)
    subparsers.add_parser("uninstall", help="Stop and remove the LaunchAgent")
    subparsers.add_parser("status", help="Show LaunchAgent status")
    arguments = parser.parse_args()

    if arguments.command == "once":
        restored, removed = reconcile_once(arguments.state_dir, arguments.dry_run)
        log(f"complete: restored={restored} removed={removed}")
        return 0
    if arguments.command == "run":
        run_loop(arguments.state_dir, arguments.interval)
        return 0
    if arguments.command == "install":
        install(arguments.interval)
        return 0
    if arguments.command == "uninstall":
        uninstall()
        return 0
    if arguments.command == "status":
        return status()
    return 2


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except BlockingIOError:
        log("another watchdog instance is already running")
        raise SystemExit(0)
    except (OSError, RuntimeError, subprocess.CalledProcessError) as error:
        log(f"fatal: {error}")
        raise SystemExit(1)