#!/usr/bin/env python3 """Read-only candidate inspection helper for Overdeck K3s Phase 2. The workstation uploads this file to the candidate, verifies its digest, and invokes only ``inspect`` through ``sudo -n``. It has no mutation subcommands. """ from __future__ import annotations import argparse import json import os import platform import re import shutil import socket import stat import subprocess import sys from pathlib import Path from typing import Any, Mapping, Sequence MACHINE_ID_RE = re.compile(r"^[0-9a-f]{32}$") class InspectError(RuntimeError): pass def root_path(root: Path, absolute: str) -> Path: return root / absolute.lstrip("/") if root != Path("/") else Path(absolute) def read_text(root: Path, absolute: str, *, required: bool = True) -> str: path = root_path(root, absolute) if path.is_symlink(): raise InspectError(f"refusing symlink: {absolute}") try: info = path.stat() except FileNotFoundError: if required: raise InspectError(f"required file is missing: {absolute}") return "" if not stat.S_ISREG(info.st_mode): raise InspectError(f"required regular file: {absolute}") return path.read_text(encoding="utf-8", errors="strict") def parse_os_release(text: str) -> dict[str, str]: out: dict[str, str] = {} for line in text.splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) value = value.strip() if len(value) >= 2 and value[0] == value[-1] and value[0] in {'"', "'"}: value = value[1:-1] out[key] = value return out def parse_meminfo(text: str) -> int: for line in text.splitlines(): if line.startswith("MemTotal:"): fields = line.split() if len(fields) >= 2 and fields[1].isdigit(): return int(fields[1]) * 1024 raise InspectError("cannot read MemTotal from /proc/meminfo") def run(argv: Sequence[str], *, timeout: int = 20) -> subprocess.CompletedProcess[str]: return subprocess.run( list(argv), check=False, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout, env={"PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin", "LANG": "C"}, ) def command_path(name: str) -> str | None: return shutil.which(name, path="/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin") def systemd_state(unit: str) -> str: systemctl = command_path("systemctl") if systemctl is None: return "systemctl-unavailable" completed = run([systemctl, "is-active", unit]) value = completed.stdout.strip() return value or ("inactive" if completed.returncode else "active") def listening_ports() -> set[int]: ports: set[int] = set() for absolute in ("/proc/net/tcp", "/proc/net/tcp6"): try: text = Path(absolute).read_text(encoding="ascii", errors="replace") except OSError: continue for line in text.splitlines()[1:]: fields = line.split() if len(fields) < 4 or fields[3] != "0A": continue try: ports.add(int(fields[1].rsplit(":", 1)[1], 16)) except (ValueError, IndexError): continue return ports def tailscale_state(root: Path) -> tuple[bool, bool]: interface = root_path(root, "/sys/class/net/tailscale0").exists() if root != Path("/"): marker = root_path(root, "/run/overdeck-fixture/tailscale-online") return marker.exists(), interface tailscale = command_path("tailscale") if tailscale is None: return False, interface completed = run([tailscale, "status", "--json"]) if completed.returncode != 0: return False, interface try: document = json.loads(completed.stdout) except json.JSONDecodeError: return False, interface backend = str(document.get("BackendState") or "").lower() return backend == "running", interface def clock_synchronized(root: Path) -> bool: if root != Path("/"): return root_path(root, "/run/systemd/timesync/synchronized").exists() timedatectl = command_path("timedatectl") if timedatectl: completed = run([timedatectl, "show", "-p", "NTPSynchronized", "--value"]) if completed.returncode == 0: return completed.stdout.strip().lower() == "yes" return Path("/run/systemd/timesync/synchronized").exists() def rustdesk_peer(root: Path) -> str | None: candidates = ( "/var/lib/rustdesk/id", "/root/.config/rustdesk/id", "/home/user/.config/rustdesk/id", ) for absolute in candidates: try: value = read_text(root, absolute, required=False).strip() except InspectError: continue if re.fullmatch(r"[0-9]{6,16}", value): return value return None def inspect(root: Path) -> dict[str, Any]: if root == Path("/") and os.geteuid() != 0: raise InspectError("candidate inspection must run as root") machine_id = read_text(root, "/etc/machine-id").strip().lower() if not MACHINE_ID_RE.fullmatch(machine_id): raise InspectError("/etc/machine-id is not 32 lowercase hexadecimal characters") os_release = parse_os_release(read_text(root, "/etc/os-release")) os_id = os_release.get("ID", "").lower() os_version = os_release.get("VERSION_ID", "") architecture = platform.machine() if root == Path("/") else read_text(root, "/run/overdeck-fixture/architecture").strip() memory = parse_meminfo(read_text(root, "/proc/meminfo")) disk_path = root_path(root, "/var/lib") disk_free = shutil.disk_usage(disk_path).free ts_online, ts_interface = tailscale_state(root) systemd = root_path(root, "/run/systemd/system").is_dir() cgroup_v2 = root_path(root, "/sys/fs/cgroup/cgroup.controllers").is_file() if root == Path("/"): k3s_present = any(Path(path).exists() for path in ("/usr/local/bin/k3s", "/usr/bin/k3s")) state = systemd_state("k3s-agent.service") else: k3s_present = root_path(root, "/usr/local/bin/k3s").exists() or root_path(root, "/usr/bin/k3s").exists() state_file = root_path(root, "/run/overdeck-fixture/k3s-agent-state") state = state_file.read_text().strip() if state_file.is_file() else "inactive" if not k3s_present and state in {"inactive", "unknown", "not-found", "systemctl-unavailable"}: k3s_state = "absent" elif state == "active": k3s_state = "active" else: k3s_state = "inactive" ports = listening_ports() if root == Path("/") else set() recovery = { "primary_2222": "listening" if 2222 in ports else "absent-before-converge", "rescue_2223": "listening" if 2223 in ports else "absent-before-converge", "tailscale_ssh_22": "tailscale-online" if ts_online else "unavailable", } return { "schema_version": 1, "status": "ok", "identity": { "machine_id": machine_id, "os_id": os_id, "os_version_id": os_version, "architecture": architecture, "rustdesk": rustdesk_peer(root), }, "preflight": { "systemd": systemd, "cgroup_v2": cgroup_v2, "tailscale_online": ts_online, "tailscale_interface": ts_interface, "sudo_noninteractive": os.geteuid() == 0, "clock_synchronized": clock_synchronized(root), "memory_bytes": memory, "disk_free_bytes": disk_free, "k3s_agent_state": k3s_state, "recovery_doors": recovery, }, "mutation_surface": "none", } def parser() -> argparse.ArgumentParser: result = argparse.ArgumentParser(description=__doc__) result.add_argument("command", choices=["inspect"]) result.add_argument("--fixture-root", default="/") return result def main(argv: Sequence[str] | None = None) -> int: args = parser().parse_args(argv) try: payload = inspect(Path(args.fixture_root).resolve()) print(json.dumps(payload, sort_keys=True)) return 0 except (Exception, KeyboardInterrupt) as exc: print(json.dumps({"schema_version": 1, "status": "error", "error": str(exc) or type(exc).__name__}, sort_keys=True)) return 2 if __name__ == "__main__": raise SystemExit(main())