#!/usr/bin/env python3 """ homelab — single-binary CLI for managing the hubris homelab from any client. Reads /opt/homelab-context/inventory.yaml as the source of truth, resolves hostnames to mesh addresses, and wraps the common ops (ssh, pct, logs, restart, status, open, secret, client add/remove, sync, mcp). Mutating subcommands prompt unless -y / --yes is passed. Run `homelab --help` or `homelab --help` for usage. """ from __future__ import annotations import argparse import json import os import shlex import shutil import socket import subprocess import sys import urllib.request from datetime import datetime from functools import lru_cache from pathlib import Path try: import yaml except ImportError: print("PyYAML required (apt install python3-yaml or pip install pyyaml)", file=sys.stderr) sys.exit(2) CONTEXT = Path(os.environ.get("HOMELAB_CONTEXT_DIR", "/opt/homelab-context")) INVENTORY = CONTEXT / "inventory.yaml" HOSTS_DIR = CONTEXT / "hosts" AGE_KEY = Path(os.environ.get("SOPS_AGE_KEY_FILE", "/etc/age/key.txt")) # ---------- helpers ---------- def die(msg: str, code: int = 1) -> None: print(f"homelab: {msg}", file=sys.stderr) sys.exit(code) def inventory() -> dict: if not INVENTORY.exists(): die(f"no inventory at {INVENTORY} — has bootstrap run?") return yaml.safe_load(INVENTORY.read_text()) def host(name: str) -> dict: inv = inventory() if name not in inv.get("hosts", {}): die(f"unknown host: {name}") return inv["hosts"][name] def host_address(name: str, prefer_lan: bool = False) -> str: """Best-effort single address for `name`. Preference order is lan_ip first because: - From any LAN client it's a direct route. - From any Netbird peer it's routed through hubris's 192.168.8.0/24 network resource. - From any Tailscale peer with subnet routing it works too. Mesh FQDNs are fallbacks for hosts without a lan_ip (e.g. roaming workstations). """ h = host(name) mesh = h.get("mesh", {}) nb = mesh.get("netbird") if isinstance(mesh.get("netbird"), dict) else {} ts = mesh.get("tailscale") if isinstance(mesh.get("tailscale"), dict) else {} candidates = [ h.get("lan_ip"), nb.get("fqdn"), nb.get("ip"), ts.get("fqdn"), ts.get("ip"), ] for c in candidates: if c: return c die(f"no reachable address for host {name}") @lru_cache(maxsize=None) def _can_tcp_connect(addr: str, port: int) -> bool: """Cached 1.5s TCP probe. Used to detect LAN unreachability before falling back to the mesh FQDN — cheaper and more predictable than waiting for ssh to timeout.""" try: with socket.create_connection((addr, port), timeout=1.5): return True except OSError: return False def ssh_target(name: str, *, force_mesh: bool = False) -> tuple[str, int, str]: """Resolve (addr, port, user) for ssh-ing into `name`. Address rules: - If `ssh.netbird_port` is set (e.g. hubris with 22022), use it + force the netbird FQDN/IP since netbird-ssh-server binds only the netbird interface, not the LAN IP. - Otherwise: try the LAN IP first. If a 1.5s TCP probe fails (off-LAN, VPN/symmetric-NAT path, etc.), fall back to the mesh FQDN. - `force_mesh=True` skips the LAN probe entirely. User: `ssh.user` from inventory if set, else `root`. Mismatched user is the netbird-ssh "user not found" gotcha — defaults baked into the inventory prevent it. """ h = host(name) ssh = h.get("ssh", {}) or {} user = ssh.get("user", "root") nb = h.get("mesh", {}).get("netbird") or {} mesh_addr = nb.get("fqdn") or nb.get("ip") nb_port = ssh.get("netbird_port") if nb_port is not None: addr = mesh_addr or host_address(name) return addr, nb_port, user port = ssh.get("port", 22) lan = h.get("lan_ip") if force_mesh or not lan: return mesh_addr or host_address(name), port, user if _can_tcp_connect(lan, port): return lan, port, user if mesh_addr: return mesh_addr, port, user return lan, port, user # last resort; ssh will surface the real error def ssh_base(name: str, *, force_mesh: bool = False) -> list[str]: """Construct an `ssh ... user@addr` invocation for `name`, honoring port and per-host user from inventory.""" addr, port, user = ssh_target(name, force_mesh=force_mesh) cmd = ["ssh"] if port != 22: cmd.extend(["-p", str(port)]) cmd.append(f"{user}@{addr}") return cmd def hubris_ssh() -> list[str]: """Back-compat shim — prefer `ssh_base("hubris")` in new code.""" return ssh_base("hubris") def confirm(question: str, default_no: bool = True) -> bool: suffix = "[y/N]" if default_no else "[Y/n]" try: answer = input(f"{question} {suffix} ").strip().lower() except (EOFError, KeyboardInterrupt): return False if not answer: return not default_no return answer in ("y", "yes") def service(name: str) -> dict: inv = inventory() svc = inv.get("services", {}).get(name) if not svc: die(f"unknown service: {name}") return svc def service_backend_host(name: str) -> str: return service(name)["backend"] def push_inventory(message: str, extra_paths: list[str] | None = None) -> None: """Stage + commit + push inventory + regenerated hosts/ (+ any extras).""" subprocess.run(["python3", str(CONTEXT / "mcp" / "build_host_files.py")], check=True, cwd=CONTEXT) paths = ["inventory.yaml", "hosts/"] if extra_paths: paths.extend(extra_paths) subprocess.run(["git", "add"] + paths, check=True, cwd=CONTEXT) if subprocess.run(["git", "diff", "--cached", "--quiet"], cwd=CONTEXT).returncode == 0: print("(no changes to commit)") return subprocess.run(["git", "commit", "-m", message], check=True, cwd=CONTEXT) subprocess.run(["git", "push"], check=True, cwd=CONTEXT) # Secrets every enrolled client should be a recipient on. Each entry is # (secret-file-path-relative-to-CONTEXT, path_regex used in .sops.yaml). SHARED_SECRETS = [ ("secrets/hello.yaml", "^secrets/hello\\.yaml$"), ("secrets/gitea-pat.yaml", "^secrets/gitea-pat\\.yaml$"), # VPS-render secrets — only strictly required on hubris (the rendering # proxy), but kept on the SHARED list so any operator can `homelab secret # turn-shared-secret` / `... netbird-authentik-oidc` for debugging. ("secrets/turn-shared-secret.yaml", "^secrets/turn-shared-secret\\.yaml$"), ("secrets/netbird-authentik-oidc.yaml", "^secrets/netbird-authentik-oidc\\.yaml$"), ] # Secrets granted only to hosts that opt into running the Hermes agent # (via `homelab client add --finalize-pubkey ... --with-hermes`). HERMES_SECRETS = [ ("secrets/openrouter-api-key.yaml", "^secrets/openrouter-api-key\\.yaml$"), ] # -------- comment-preserving inventory.yaml edits -------- # yaml.safe_load + safe_dump round-trips strip every comment, which is fine # for hosts/*.yaml (generated anyway) but unfriendly for inventory.yaml where # we want the doc comments at the top + per-section to survive. These helpers # do line-based surgical edits instead. import re as _re def _find_host_block(lines: list[str], name: str) -> tuple[int, int] | None: """Return (start, end_exclusive) line range for ` :` in inventory.yaml. Range covers the host's body up to the next 2-space-indented key (next host) or the next 0-indented top-level key, whichever comes first. """ target_re = _re.compile(rf"^ {_re.escape(name)}:\s*$") host_at_2sp = _re.compile(r"^ [A-Za-z][A-Za-z0-9_-]*:\s*$") toplevel = _re.compile(r"^[A-Za-z]") start = None for i, line in enumerate(lines): if start is None: if target_re.match(line): start = i continue # We're inside the target block; look for the next sibling host # or a top-level key to mark the end. if host_at_2sp.match(line): return (start, i) if toplevel.match(line): return (start, i) if start is not None: return (start, len(lines)) return None def _inventory_set_age_pubkey(name: str, pubkey: str) -> bool: """Surgically set hosts..age_pubkey in inventory.yaml. Preserves every comment outside the modified line. Returns True if updated, False if the host block or age_pubkey line wasn't found. """ lines = INVENTORY.read_text().splitlines(keepends=True) block = _find_host_block(lines, name) if block is None: return False start, end = block pubkey_re = _re.compile(r"^(\s+age_pubkey:\s*).*$") for i in range(start, end): m = pubkey_re.match(lines[i]) if m: lines[i] = f"{m.group(1)}{pubkey}\n" INVENTORY.write_text("".join(lines)) return True # No existing age_pubkey line; insert one at the end of the block. # Use 4-space indent (matching the other host fields). insert_at = end while insert_at > start and lines[insert_at - 1].strip() == "": insert_at -= 1 lines.insert(insert_at, f" age_pubkey: {pubkey}\n") INVENTORY.write_text("".join(lines)) return True def _inventory_remove_host(name: str) -> bool: """Delete the hosts.: block. Preserves comments outside the block.""" lines = INVENTORY.read_text().splitlines(keepends=True) block = _find_host_block(lines, name) if block is None: return False start, end = block del lines[start:end] INVENTORY.write_text("".join(lines)) return True def _inventory_append_host(name: str, kind: str, os_name: str, role: str, netbird_fqdn: str) -> bool: """Append a new hosts.: block at the end of the hosts: section. Returns True on success, False if no hosts: section was found. """ lines = INVENTORY.read_text().splitlines(keepends=True) # Find the line "hosts:" at column 0. hosts_idx = None for i, line in enumerate(lines): if line.startswith("hosts:"): hosts_idx = i break if hosts_idx is None: return False # Find the end of the hosts: section (next top-level key or EOF). insert_at = len(lines) for i in range(hosts_idx + 1, len(lines)): if _re.match(r"^[A-Za-z]", lines[i]): insert_at = i break # Trim trailing blanks before insertion point so we don't double-space. while insert_at > hosts_idx + 1 and lines[insert_at - 1].strip() == "": insert_at -= 1 block = ( f" {name}:\n" f" kind: {kind}\n" f" os: {os_name}\n" f" role: {role}\n" f" mesh:\n" f" netbird:\n" f" fqdn: {netbird_fqdn}\n" f" age_pubkey: \"\"\n" ) lines.insert(insert_at, block) INVENTORY.write_text("".join(lines)) return True def _add_recipient_to_sops_policy(sops_path: Path, path_regex_pattern: str, pubkey: str) -> bool: """Append `pubkey` to the `age:` list of the .sops.yaml rule whose `path_regex:` line contains `path_regex_pattern`. Preserves comments. Returns True if added (or already present), False if no matching rule. """ if not sops_path.exists(): return False lines = sops_path.read_text().splitlines(keepends=True) in_target_rule = False age_block_start = None age_block_last_idx = None for i, line in enumerate(lines): stripped = line.strip() if stripped.startswith("- path_regex:"): # If we've already collected what we need from the target rule, # stop — don't let the next rule wipe age_block_last_idx. if in_target_rule and age_block_last_idx is not None: break in_target_rule = path_regex_pattern in line age_block_start = None age_block_last_idx = None continue if not in_target_rule: continue if "age: >-" in line: age_block_start = i continue if age_block_start is None: continue if "age1" in stripped: if pubkey in line: return True # already a recipient age_block_last_idx = i elif stripped == "" or stripped.startswith("#"): continue else: break # next key, age block ended if age_block_last_idx is None: return False last_line = lines[age_block_last_idx] indent = last_line[: len(last_line) - len(last_line.lstrip())] if not last_line.rstrip().endswith(","): lines[age_block_last_idx] = last_line.rstrip() + ",\n" lines.insert(age_block_last_idx + 1, f"{indent}{pubkey}\n") sops_path.write_text("".join(lines)) return True def _grant_shared_secrets(pubkey: str, secrets: list[tuple[str, str]] = SHARED_SECRETS) -> None: """Add `pubkey` to the recipient list of every listed secret + re-key. `secrets` defaults to SHARED_SECRETS; pass HERMES_SECRETS to grant the Hermes-only set. """ sops_path = CONTEXT / ".sops.yaml" env = {**os.environ, "SOPS_AGE_KEY_FILE": str(AGE_KEY)} for rel_path, pattern in secrets: target = CONTEXT / rel_path if not target.exists(): print(f" skipping {rel_path}: file does not exist yet") continue added = _add_recipient_to_sops_policy(sops_path, pattern, pubkey) if not added: print(f" warning: no matching rule in .sops.yaml for {rel_path} — skipping") continue proc = subprocess.run( ["sops", "updatekeys", "-y", rel_path], capture_output=True, text=True, env=env, cwd=str(CONTEXT), ) if proc.returncode == 0: print(f" re-keyed {rel_path} (added {pubkey[:20]}…)") else: print(f" warning: sops updatekeys failed for {rel_path}: {proc.stderr.strip()}") def _remove_recipient_from_sops_policy(sops_path: Path, path_regex_pattern: str, pubkey: str) -> bool: """Remove `pubkey` from the `age:` list of the .sops.yaml rule whose `path_regex:` line contains `path_regex_pattern`. Preserves comments. Returns True if removed (or already absent), False if rule not found. """ if not sops_path.exists(): return False lines = sops_path.read_text().splitlines(keepends=True) def find_target_age_lines() -> tuple[list[int], bool]: """Scan and return (indices_of_age1_lines_in_target_rule, found_rule).""" in_target = False age_block_start = None age_idxs: list[int] = [] for i, line in enumerate(lines): stripped = line.strip() if stripped.startswith("- path_regex:"): if in_target and age_idxs: return age_idxs, True in_target = path_regex_pattern in line age_block_start = None age_idxs = [] continue if not in_target: continue if "age: >-" in line: age_block_start = i continue if age_block_start is None: continue if "age1" in stripped: age_idxs.append(i) elif stripped == "" or stripped.startswith("#"): continue else: break return age_idxs, in_target age_idxs, found = find_target_age_lines() if not found: return False target_idx = next((i for i in age_idxs if pubkey in lines[i]), None) if target_idx is None: return True # already absent del lines[target_idx] # After deletion, fix trailing comma on the new last age line. new_age_idxs, _ = find_target_age_lines() if new_age_idxs: last_age_idx = new_age_idxs[-1] last_line = lines[last_age_idx] if last_line.rstrip().endswith(","): lines[last_age_idx] = last_line.rstrip().rstrip(",") + "\n" sops_path.write_text("".join(lines)) return True def _revoke_shared_secrets(pubkey: str, secrets: list[tuple[str, str]] = SHARED_SECRETS) -> None: """Remove `pubkey` from every listed secret rule + re-key the files. Defaults to SHARED_SECRETS; pass HERMES_SECRETS to revoke the Hermes-only set. """ sops_path = CONTEXT / ".sops.yaml" env = {**os.environ, "SOPS_AGE_KEY_FILE": str(AGE_KEY)} for rel_path, pattern in secrets: target = CONTEXT / rel_path if not target.exists(): continue removed = _remove_recipient_from_sops_policy(sops_path, pattern, pubkey) if not removed: print(f" warning: no matching rule in .sops.yaml for {rel_path} — skipping") continue proc = subprocess.run( ["sops", "updatekeys", "-y", rel_path], capture_output=True, text=True, env=env, cwd=str(CONTEXT), ) if proc.returncode == 0: print(f" re-keyed {rel_path} (removed {pubkey[:20]}…)") else: print(f" warning: sops updatekeys failed for {rel_path}: {proc.stderr.strip()}") # ---------- subcommands ---------- def cmd_whoami(args: argparse.Namespace) -> int: name = args.hostname or os.uname().nodename.split(".")[0] path = HOSTS_DIR / f"{name}.yaml" if not path.exists(): die(f"no hosts/{name}.yaml — has this client been enrolled?") print(path.read_text()) return 0 def cmd_list(args: argparse.Namespace) -> int: inv = inventory() print(f"{'NAME':<22} {'KIND':<14} {'OS':<6} {'ROLE':<22} ADDRESS") for name, entry in inv.get("hosts", {}).items(): addr = "" try: addr = host_address(name) except SystemExit: addr = "?" print(f"{name:<22} {entry.get('kind','?'):<14} {entry.get('os','?'):<6} " f"{entry.get('role','?'):<22} {addr}") print() print("Services:") for svc, entry in inv.get("services", {}).items(): url = entry.get("url") or entry.get("endpoint") or "" backend = entry.get("backend", "?") print(f" {svc:<22} backend={backend:<15} {url}") return 0 def cmd_ssh(args: argparse.Namespace) -> int: name = args.host cmd = ssh_base(name) # Optional --user override replaces the resolved user in the last arg if args.user: cmd[-1] = f"{args.user}@{cmd[-1].split('@', 1)[1]}" if args.command: cmd.append(" ".join(args.command)) os.execvp(cmd[0], cmd) def cmd_ssh_config(args: argparse.Namespace) -> int: """Generate SSH config from inventory.yaml.""" script = CONTEXT / "ssh" / "gen-config.py" if not script.exists(): die(f"ssh-config generator not found: {script}") cmd = [sys.executable or "python3", str(script)] if args.install: cmd.append("--install") return subprocess.call(cmd) def cmd_pct(args: argparse.Namespace) -> int: lxc = args.lxc inv = inventory() entry = inv.get("hosts", {}).get(lxc) if not entry or entry.get("kind") != "lxc": die(f"{lxc} is not an LXC in inventory") pve_id = str(entry["pve_id"]) action = args.action if action == "exec": if not args.rest: die("pct exec needs a command") remote_cmd = ["pct", "exec", pve_id, "--"] + args.rest elif action in ("status", "config", "start", "stop", "reboot", "shutdown"): remote_cmd = ["pct", action, pve_id] elif action == "enter": remote_cmd = ["pct", "enter", pve_id] else: die(f"unknown pct action: {action}") if action in ("stop", "reboot", "shutdown") and not args.yes: if not confirm(f"run 'pct {action} {pve_id}' (={lxc}) on hubris?"): return 1 cmd = hubris_ssh() + ["--"] + remote_cmd return subprocess.call(cmd) def cmd_logs(args: argparse.Namespace) -> int: svc = args.service host_name = service_backend_host(svc) unit = service(svc).get("systemd_unit", svc) base = ssh_base(host_name) remote = ["journalctl", "-u", unit, "-n", str(args.lines), "--no-pager"] if args.follow: remote.append("-f") return subprocess.call(base + ["--"] + remote) def cmd_restart(args: argparse.Namespace) -> int: svc = args.service host_name = service_backend_host(svc) unit = service(svc).get("systemd_unit", svc) if not args.yes: if not confirm(f"restart systemd unit '{unit}' on {host_name}?"): return 1 base = ssh_base(host_name) return subprocess.call(base + ["--", "systemctl", "restart", unit]) def cmd_open(args: argparse.Namespace) -> int: url = service(args.service).get("url") if not url: die(f"service {args.service} has no url in inventory") opener = "open" if sys.platform == "darwin" else "xdg-open" if shutil.which(opener) is None: print(url) return 0 return subprocess.call([opener, url]) def cmd_status(args: argparse.Namespace) -> int: inv = inventory() print(f"{'NAME':<22} REACH") for name, entry in inv.get("hosts", {}).items(): try: addr = host_address(name) except SystemExit: print(f"{name:<22} no-address") continue # Quick ping-ish check via mesh. proc = subprocess.run( ["ping", "-c", "1", "-W", "2", addr], capture_output=True, text=True, ) reach = "ok" if proc.returncode == 0 else "down" print(f"{name:<22} {reach:<5} {addr}") print() print(f"{'SERVICE':<22} URL HEALTH") for svc, entry in inv.get("services", {}).items(): url = entry.get("url") or entry.get("endpoint") if not url: continue proc = subprocess.run( ["curl", "-sS", "-o", "/dev/null", "-w", "%{http_code}", "--max-time", "3", url], capture_output=True, text=True, ) code = proc.stdout.strip() or "---" print(f"{svc:<22} {url:<50} {code}") return 0 def _decrypt_secret(name: str) -> dict: """Decrypt secrets/.yaml and parse as YAML. Re-execs via sudo when invoked as a non-root user (the age key at /etc/age/key.txt is 0600 root).""" path = CONTEXT / "secrets" / f"{name}.yaml" if not path.exists(): die(f"no secret '{name}' (looked for {path})") if os.geteuid() != 0: proc = subprocess.run( ["sudo", "-E", "env", f"SOPS_AGE_KEY_FILE={AGE_KEY}", "sops", "-d", str(path)], capture_output=True, text=True, ) else: if not AGE_KEY.exists(): die(f"no age key at {AGE_KEY} — has bootstrap run?") env = {**os.environ, "SOPS_AGE_KEY_FILE": str(AGE_KEY)} proc = subprocess.run( ["sops", "-d", str(path)], capture_output=True, text=True, env=env, ) if proc.returncode != 0: die(f"sops decrypt of '{name}' failed: {proc.stderr.strip() or '(no stderr)'}") return yaml.safe_load(proc.stdout) def cmd_secret(args: argparse.Namespace) -> int: name = args.name path = CONTEXT / "secrets" / f"{name}.yaml" if not path.exists(): die(f"no secret '{name}' (looked for {path})") # The age key lives at /etc/age/key.txt (root:root 0600) so non-root # users can't read it — or even stat it, since /etc/age is 0700 root. # Re-exec via sudo when invoked as a regular user. if os.geteuid() != 0: return subprocess.call([ "sudo", "-E", "env", f"SOPS_AGE_KEY_FILE={AGE_KEY}", "sops", "-d", str(path), ]) if not AGE_KEY.exists(): die(f"no age key at {AGE_KEY} — has bootstrap run?") env = {**os.environ, "SOPS_AGE_KEY_FILE": str(AGE_KEY)} return subprocess.call(["sops", "-d", str(path)], env=env) # ---------- render-vps-configs ---------- # # The IONOS netbird VPS hosts plaintext credentials that we now keep encrypted # in secrets/. This command re-renders the affected config files on the VPS by # decrypting the secrets locally and pushing the rendered output. Idempotent — # safe to re-run; it shows a diff and prompts before applying. # Map of (template path in repo, target path on VPS, mode, restart cmd after). _VPS_RENDER_TARGETS = [ { "tmpl": "vps/turnserver.conf.tmpl", "remote": "/etc/turnserver.conf", "mode": "0644", "restart": "systemctl restart coturn", }, { "tmpl": "vps/management.json.tmpl", "remote": "/opt/management.json", "mode": "0600", "restart": "docker restart netbird-mgmt", }, ] def _vps_remote_cmd(remote_cmd: str) -> list[str]: """Build the argv that, when run, executes `remote_cmd` on the IONOS VPS. The VPS only accepts hubris's pubkey (per infrastructure/vps-hardening.md), so when we're not on hubris we hop through it: laptop → ssh hubris → ssh netbird-vps. On hubris we go direct. """ direct = ssh_base("netbird-vps") # ["ssh", "root@netbird-ionos..."] try: on_hubris = socket.gethostname() == "hubris" except Exception: on_hubris = False if on_hubris: return direct + [remote_cmd] # laptop → hubris → vps. shlex-quote the remote_cmd so it survives # hubris's shell parsing as a single argument to the inner ssh. inner = " ".join(direct) + " " + shlex.quote(remote_cmd) return ssh_base("hubris") + [inner] def _vps_send(remote_path: str, content: str, mode: str) -> None: """Atomically write `content` to `remote_path` on the VPS with given mode. Writes to .new, chmods, then mv-replaces.""" new_path = f"{remote_path}.new" write_cmd = f"umask 077 && cat > {new_path} && chmod {mode} {new_path}" proc = subprocess.run( _vps_remote_cmd(write_cmd), input=content, text=True, capture_output=True, ) if proc.returncode != 0: die(f"failed writing {new_path}: {proc.stderr.strip()}") mv = subprocess.run(_vps_remote_cmd(f"mv {new_path} {remote_path}"), capture_output=True, text=True) if mv.returncode != 0: die(f"failed renaming {new_path} → {remote_path}: {mv.stderr.strip()}") def cmd_render_vps_configs(args: argparse.Namespace) -> int: """Re-render /etc/turnserver.conf + /opt/management.json on the IONOS netbird VPS from templates in vps/, substituting secrets decrypted from sops.""" import difflib # Decrypt the two sops files we need. turn = _decrypt_secret("turn-shared-secret") auth = _decrypt_secret("netbird-authentik-oidc") subs = { "{{TURN_PASSWORD}}": turn["password"], "{{AUTHENTIK_CLIENT_SECRET}}": auth["client_secret"], } # Render each template and compare against current VPS content. plans = [] for t in _VPS_RENDER_TARGETS: tmpl_path = CONTEXT / t["tmpl"] if not tmpl_path.exists(): die(f"missing template {tmpl_path}") rendered = tmpl_path.read_text() for needle, value in subs.items(): rendered = rendered.replace(needle, value) cat = subprocess.run(_vps_remote_cmd(f"cat {t['remote']} 2>/dev/null"), capture_output=True, text=True) current = cat.stdout plans.append({ **t, "rendered": rendered, "current": current, "changed": current != rendered, "current_present": cat.returncode == 0 and bool(current), }) # Summarize. print("Render plan for IONOS netbird VPS:") any_changes = False for p in plans: status = "(changed)" if p["changed"] else "(unchanged)" print(f" {p['remote']:<28} {status}") if p["changed"]: any_changes = True if not any_changes: print("Nothing to do — every target matches the rendered template.") return 0 # Show diffs (always — both dry-run and live). print() for p in plans: if not p["changed"]: continue print(f"--- diff for {p['remote']} ---") # Mask any sops-decrypted secret values so they don't print to stdout. def _mask(s): for v in subs.values(): s = s.replace(v, "") return s diff = difflib.unified_diff( _mask(p["current"]).splitlines(keepends=True), _mask(p["rendered"]).splitlines(keepends=True), fromfile=f"vps:{p['remote']}", tofile=f"rendered:{p['tmpl']}", n=3, ) sys.stdout.writelines(diff) print() if args.dry_run: print("--- dry-run: not applying ---") return 0 if not args.yes: if not confirm("apply changes + restart services on netbird-vps?"): return 1 # Apply changed targets, then restart their services. for p in plans: if not p["changed"]: continue print(f"writing {p['remote']} ({p['mode']}) ...") _vps_send(p["remote"], p["rendered"], p["mode"]) print(f" → {p['restart']}") rc = subprocess.run(_vps_remote_cmd(p["restart"]), capture_output=True, text=True) if rc.returncode != 0: print(f" ! restart failed: {rc.stderr.strip()}", file=sys.stderr) return 1 print("done.") return 0 def cmd_doctor(args: argparse.Namespace) -> int: """Run health checks for the homelab-context client setup.""" results: list[tuple[str, str, str]] = [] # (status, label, detail) def ok(label, detail=""): results.append(("ok", label, detail)) def warn(label, detail=""): results.append(("warn", label, detail)) def fail(label, detail=""): results.append(("fail", label, detail)) # 1. /opt/homelab-context is a git clone. if (CONTEXT / ".git").is_dir(): try: head = subprocess.run( ["git", "-C", str(CONTEXT), "log", "--oneline", "-1"], capture_output=True, text=True, check=True).stdout.strip() ok("clone present", f"head: {head[:60]}") except Exception as e: warn("clone present but git unhappy", str(e)) else: fail("clone missing", f"{CONTEXT}/.git does not exist") # 2. Sync mechanism — systemd timer (Linux) or launchd plist (macOS). if sys.platform == "darwin": try: out = subprocess.run( ["launchctl", "list"], capture_output=True, text=True, ).stdout if "network.hubris.homelab-context-sync" in out: ok("launchd job loaded") else: fail("launchd job missing", "expected network.hubris.homelab-context-sync") except FileNotFoundError: fail("launchctl missing", "is this really macOS?") else: proc = subprocess.run( ["systemctl", "list-timers", "homelab-context-sync.timer", "--no-pager", "--no-legend"], capture_output=True, text=True, ) if "homelab-context-sync.timer" in proc.stdout: ok("sync timer active", proc.stdout.strip()[:80]) else: fail("sync timer not active", "systemctl enable --now homelab-context-sync.timer") # 3. Age key. if AGE_KEY.exists(): try: st = AGE_KEY.stat() mode = st.st_mode & 0o777 if mode != 0o600: warn("age key permissions loose", f"{AGE_KEY} mode={oct(mode)} (expected 0o600)") else: ok("age key present", f"{AGE_KEY} 0600 root") except PermissionError: ok("age key present", f"{AGE_KEY} (can't stat — expected as non-root)") else: warn("age key missing", "rerun bootstrap or call /issue manually") # 4. /usr/local/bin/homelab is a symlink into the repo. cli = Path("/usr/local/bin/homelab") if cli.is_symlink() and cli.resolve() == (CONTEXT / "bin" / "homelab").resolve(): ok("CLI is symlinked into sync target") elif cli.exists(): warn("CLI is a copy, not a symlink", f"sudo ln -sfn {CONTEXT}/bin/homelab {cli}") else: fail("CLI not installed", f"{cli} missing") # 5. AGENTS.md symlink. expected_link = Path("/root/AGENTS.md") if sys.platform != "darwin" else Path("/etc/AGENTS.md") if expected_link.is_symlink(): ok("AGENTS.md symlink", str(expected_link)) else: warn("AGENTS.md not symlinked", f"expected {expected_link}") # 6. hosts/.yaml present (own identity). hname = os.uname().nodename.split(".")[0] if (HOSTS_DIR / f"{hname}.yaml").exists(): ok("inventory entry", f"hosts/{hname}.yaml") else: fail("no inventory entry for this hostname", f"need hosts/{hname}.yaml — run 'homelab client add {hname}' on an enrolled client") # 7. MCP reachability. inv = inventory() mcp_url = inv.get("services", {}).get("homelab_mcp", {}).get("endpoint") if mcp_url: proc = subprocess.run( ["curl", "-sS", "-o", "/dev/null", "-w", "%{http_code}", "-H", "Accept: text/event-stream", "--max-time", "3", mcp_url], capture_output=True, text=True, ) code = proc.stdout.strip() if code.startswith("2"): ok("MCP reachable", f"{mcp_url} -> {code}") else: fail("MCP unreachable", f"{mcp_url} -> {code or 'no response'}") # 8. Secrets-issuance health. issue_url = inv.get("services", {}).get("secrets_issuance", {}).get("endpoint", "") if issue_url: health_url = issue_url.rstrip("/").rsplit("/", 1)[0] + "/health" proc = subprocess.run( ["curl", "-sS", "-o", "/dev/null", "-w", "%{http_code}", "--max-time", "3", health_url], capture_output=True, text=True, ) code = proc.stdout.strip() if code == "200": ok("secrets-issuance healthy", health_url) else: warn("secrets-issuance not responding 200", f"{health_url} -> {code or 'no response'}") # 9. sops can decrypt the canary secret (needs age key access — root only). if (CONTEXT / "secrets" / "hello.yaml").exists(): if os.geteuid() != 0: ok("sops canary skipped (run as root to test)", "homelab secret hello (will sudo)") elif AGE_KEY.exists(): env = {**os.environ, "SOPS_AGE_KEY_FILE": str(AGE_KEY)} proc = subprocess.run( ["sops", "-d", str(CONTEXT / "secrets" / "hello.yaml")], capture_output=True, text=True, env=env, timeout=5, ) if proc.returncode == 0: ok("sops canary decrypts") else: warn("sops canary decrypt failed", "this client may not be a recipient of secrets/hello.yaml") else: warn("sops canary not tested", "age key missing") # 10. Git credentials — read-only or write-scoped? creds = Path("/etc/homelab-context/git-credentials") if creds.exists() and os.geteuid() == 0: try: # The write-scoped PAT comes from secrets/gitea-pat.yaml. We can't # check scope from the file alone, but we can at least confirm # presence + format. content = creds.read_text().strip() if "@" in content and "://" in content: ok("git credentials file present", "/etc/homelab-context/git-credentials") else: warn("git credentials malformed", content[:80]) except PermissionError: ok("git credentials present (can't read as non-root)") elif creds.exists(): ok("git credentials present (can't read as non-root)") # Print results table. for status, label, detail in results: icon = {"ok": "✓", "warn": "!", "fail": "✗"}[status] line = f" {icon} {label}" if detail: line += f" — {detail}" print(line) counts = {"ok": 0, "warn": 0, "fail": 0} for s, *_ in results: counts[s] += 1 print() print(f"{counts['ok']} ok · {counts['warn']} warn · {counts['fail']} fail") return 1 if counts["fail"] else 0 def cmd_refresh_creds(args: argparse.Namespace) -> int: """Replace /etc/homelab-context/git-credentials with the write-scoped PAT from secrets/gitea-pat.yaml so push (not just pull) works from this client. """ if os.geteuid() != 0: # Need root for the decrypt + creds-file write. return subprocess.call(["sudo", "-E", sys.argv[0], "refresh-creds"]) pat_file = CONTEXT / "secrets" / "gitea-pat.yaml" if not pat_file.exists(): die(f"no {pat_file} — has the sync pulled it yet? Try `homelab sync`.") if not AGE_KEY.exists(): die(f"no age key at {AGE_KEY} — bootstrap first.") env = {**os.environ, "SOPS_AGE_KEY_FILE": str(AGE_KEY)} proc = subprocess.run(["sops", "-d", str(pat_file)], capture_output=True, text=True, env=env) if proc.returncode != 0: die(f"could not decrypt {pat_file} — is this client a recipient? " f"sops error: {proc.stderr.strip()}") pat_data = yaml.safe_load(proc.stdout) or {} user = pat_data.get("user") token = pat_data.get("token") if not user or not token: die("decrypted gitea-pat.yaml missing user or token") # Find the host from the existing remote. remote_url = subprocess.run( ["git", "-C", str(CONTEXT), "remote", "get-url", "origin"], capture_output=True, text=True, ).stdout.strip() # Match http(s)://host or scp-style gitea@host:path proto = "https" host = "git.hubris.network" if remote_url.startswith(("http://", "https://")): proto = remote_url.split("://", 1)[0] host = remote_url.split("://", 1)[1].split("/", 1)[0] creds_dir = Path("/etc/homelab-context") creds_dir.mkdir(parents=True, exist_ok=True) creds_file = creds_dir / "git-credentials" creds_file.write_text(f"{proto}://{user}:{token}@{host}\n") creds_file.chmod(0o600) subprocess.run(["git", "config", "--system", "credential.helper", f"store --file={creds_file}"], check=True) print(f"refreshed {creds_file} (user={user}, scope=write)") print("`git push` from /opt/homelab-context now works.") return 0 def cmd_sync(args: argparse.Namespace) -> int: # Unlike every other mutating command here, this one had no os.geteuid() # guard — always shelled out to sudo. Fails outright with "No such file # or directory: 'sudo'" on minimal root-only Linux images (no sudo # binary installed at all) reached via `ssh root@host`, e.g. strong. needs_sudo = os.geteuid() != 0 if sys.platform == "darwin": cmd = ["launchctl", "kickstart", "-k", "system/network.hubris.homelab-context-sync"] else: cmd = ["systemctl", "start", "homelab-context-sync.service"] if needs_sudo: cmd = ["sudo"] + cmd return subprocess.call(cmd) def cmd_mcp(args: argparse.Namespace) -> int: svc = service("homelab_mcp") endpoint = svc.get("endpoint") if not endpoint: die("no MCP endpoint in inventory") # Minimal SSE invocation via mcp-cli if available; otherwise instruct. if shutil.which("mcp"): return subprocess.call(["mcp", "call", endpoint, args.tool, *(args.args or [])]) die("'mcp' CLI not installed. Install with: pip install 'mcp[cli]'") def cmd_client_add(args: argparse.Namespace) -> int: # client add edits root-owned files and may need to read the age key # for sops updatekeys. Re-exec under sudo if not root. if os.geteuid() != 0: return subprocess.call(["sudo", "-E", sys.argv[0], "client", "add"] + ([args.name] if args.name else []) + (["--finalize-pubkey", args.finalize_pubkey] if args.finalize_pubkey else []) + (["--with-hermes"] if args.with_hermes else [])) name = args.name inv = inventory() if not args.finalize_pubkey: if name in inv["hosts"]: die(f"{name} already in inventory (use --finalize-pubkey to update age_pubkey)") print(f"Adding new client '{name}' to inventory.yaml.") kind = input(" kind [workstation/lxc/vm] (default: workstation): ").strip() or "workstation" os_name = input(f" os [linux/macos] (default: linux): ").strip() or "linux" netbird_fqdn = input(f" netbird FQDN (default: {name}.netbird.selfhosted): ").strip() \ or f"{name}.netbird.selfhosted" role = input(" role (e.g. primary-dev, dev): ").strip() or "dev" if not _inventory_append_host(name, kind, os_name, role, netbird_fqdn): die("could not locate 'hosts:' section in inventory.yaml") push_inventory(f"client-add: {name}") print() print("Next steps:") print(f" 1. Join {name} to Netbird (out-of-band, Netbird console / setup key).") print(f" 2. On {name}: curl -fsSL /dtoro/Homelab-Docs/raw/main/bootstrap.sh | sudo bash") print(f" 3. bootstrap prints an age pubkey — bring it back here and run:") print(f" homelab client add {name} --finalize-pubkey ") print(f" (append --with-hermes to also grant the Hermes agent's OpenRouter key.)") return 0 # finalize_pubkey path if name not in inv["hosts"]: die(f"{name} not in inventory — run 'homelab client add {name}' first (no --finalize-pubkey)") pubkey = args.finalize_pubkey if not _inventory_set_age_pubkey(name, pubkey): die(f"could not find hosts.{name} block to update age_pubkey") print(f"set age_pubkey for {name}") print("granting shared secrets...") _grant_shared_secrets(pubkey) commit_subject = f"client-add: {name} (finalize age_pubkey + grant shared secrets)" if args.with_hermes: print("granting hermes-only secrets...") _grant_shared_secrets(pubkey, HERMES_SECRETS) commit_subject = f"client-add: {name} (finalize age_pubkey + grant shared + hermes secrets)" push_inventory( commit_subject, extra_paths=[".sops.yaml", "secrets/"], ) print(f"finalized {name}.") return 0 def cmd_client_remove(args: argparse.Namespace) -> int: # Needs root for the age key + root-owned file writes (same as client add). if os.geteuid() != 0: extra = ["--yes"] if args.yes else [] return subprocess.call(["sudo", "-E", sys.argv[0], "client", "remove", args.name, *extra]) name = args.name inv = inventory() if name not in inv["hosts"]: die(f"{name} not in inventory") pubkey = inv["hosts"][name].get("age_pubkey") or "" if not args.yes: print(f"This will:") print(f" 1. Remove {name} from inventory.yaml and hosts/") print(f" 2. Remove {name}'s pubkey from .sops.yaml shared-secret rules + re-key") print(f" 3. Revoke {name}'s age key on the issuance server (shred + denylist)") print(f" 4. Commit + push the change") print(f"After: rotate any credentials inside secrets {name} previously had access to,") print(f" and revoke {name}'s Netbird peer in the console.") if not confirm(f"proceed removing {name}?"): return 1 # 1. Inventory — surgical block delete (preserves comments) if not _inventory_remove_host(name): die(f"could not locate hosts.{name} block to delete") # 2. SOPS — remove the pubkey from shared-secret rules and re-key. if pubkey: print("revoking shared secrets...") _revoke_shared_secrets(pubkey) # Also revoke from hermes-only secrets; idempotent if the pubkey # was never on those rules (logs a "not present" warning, no harm). _revoke_shared_secrets(pubkey, HERMES_SECRETS) else: print(f" note: no age_pubkey recorded for {name} — skipping sops re-key") # 3. Revoke on issuance server admin_token_url = inv["services"].get("secrets_issuance", {}).get("endpoint", "").replace("/issue", "") if admin_token_url: revoke_url = admin_token_url.rstrip("/") + "/revoke" token = os.environ.get("HOMELAB_ISSUANCE_ADMIN_TOKEN", "") if not token: print(f" WARNING: set HOMELAB_ISSUANCE_ADMIN_TOKEN to call {revoke_url} — skipping revocation") else: req = urllib.request.Request( revoke_url, data=json.dumps({"hostname": name}).encode(), headers={"X-Admin-Token": token, "Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=10) as resp: print(f" issuance revoke: {resp.status} {resp.read().decode().strip()}") except Exception as e: print(f" issuance revoke failed: {e}") # 4. Commit + push (extras: .sops.yaml + secrets/ may also have changed). push_inventory( f"client-remove: {name}", extra_paths=[".sops.yaml", "secrets/"], ) print() print("Follow-up checklist (the CLI cannot do these automatically):") print(f" [ ] Revoke {name}'s Netbird peer in the Netbird console.") print(f" [ ] Rotate any credentials whose ciphertext {name} already has on disk") print(f" — the only real revocation for past-disclosed secrets is rotation.") print(f" [ ] If the machine is reachable and decommissioned, run:") print(f" homelab nuke {name} (shreds /etc/age/key.txt, removes /opt/homelab-context)") return 0 def cmd_nuke(args: argparse.Namespace) -> int: name = args.name if not args.yes: if not confirm(f"destroy /etc/age/key.txt + /opt/homelab-context on {name}?"): return 1 base = ssh_base(name) remote = ("set -euo pipefail; " "shred -u /etc/age/key.txt 2>/dev/null || true; " "rm -rf /opt/homelab-context; " "systemctl disable --now homelab-context-sync.timer 2>/dev/null || true; " "launchctl bootout system/network.hubris.homelab-context-sync 2>/dev/null || true; " "echo nuked") return subprocess.call(base + [remote]) # ---------- argparse ---------- # ====== apt fleet operations (audit + upgrade wrapper) ====== # # Both subcommands fan out to "apt targets": hubris (the PVE host) + every LXC. # VMs (haos, zimaos) and workstations are intentionally excluded — they have # their own update flows. # # Designed to survive ssh teardown: `apt-upgrade` wraps the remote apt run in # `screen -dmS` so a flaky ssh control socket can't kill it mid-transaction # (the failure mode from the 2026-05-21 fleet sweep). All output is tee'd to # /var/log/homelab-apt-upgrade.log on the target; check via `apt-upgrade # --status`. _APT_AUDIT_PROBE = r""" audit=$(dpkg --audit 2>&1 | grep -c '^ ') holds=$(apt-mark showhold 2>/dev/null | wc -l) upgr=$(apt list --upgradable 2>/dev/null | grep -cv '^Listing') nonapt=$(for f in /usr/bin/caddy /usr/bin/docker /usr/bin/jellyfin /usr/local/bin/*; do [ -e "$f" ] && (dpkg -S "$f" 2>/dev/null >/dev/null || echo "$f") done 2>/dev/null | wc -l) if getent hosts deb.debian.org >/dev/null 2>&1; then dns=ok; else dns=fail; fi echo "$audit $holds $upgr $nonapt $dns" """ _APT_UPGRADE_WRAPPER = r""" set -e command -v systemd-run >/dev/null || { echo "systemd-run not available on target" >&2; exit 3; } mkdir -p /var/log LOG=/var/log/homelab-apt-upgrade.log UNIT="apt-upgrade-$(hostname -s)" if systemctl is-active --quiet "$UNIT.service" 2>/dev/null; then echo "unit $UNIT already running on $(hostname -s); not relaunching" >&2 exit 4 fi # systemd-run creates a transient unit that survives our ssh teardown. # --collect cleans up the unit after the command finishes (no stale units). systemd-run --unit="$UNIT" --collect --quiet bash -c ' exec > >(tee -a '"$LOG"') 2>&1 echo "=== START $(date -Is) host=$(hostname -s) ===" export DEBIAN_FRONTEND=noninteractive apt -o Acquire::Retries=3 -o Acquire::ForceIPv4=true -y update apt -o Acquire::Retries=3 -o Dpkg::Options::=--force-confold -y upgrade rc=$? echo "=== END rc=$rc $(date -Is) ===" exit $rc ' echo "launched unit=$UNIT log=$LOG host=$(hostname -s)" """ def _apt_targets(include_hubris: bool = True, include_lxcs: bool = True) -> list[tuple[str, str | None]]: """Standard apt targets: hubris (None pve_id) + every LXC (pve_id as str).""" inv = inventory() targets: list[tuple[str, str | None]] = [] if include_hubris: targets.append(("hubris", None)) if include_lxcs: for name, entry in inv.get("hosts", {}).items(): if entry.get("kind") == "lxc": pid = entry.get("pve_id") if pid: targets.append((name, str(pid))) return targets def _run_on_target(name: str, pve_id: str | None, remote_argv: list[str], stdin_text: str | None = None) -> subprocess.CompletedProcess: """Run `remote_argv` on the target. pve_id=None → on hubris directly. pve_id= → via `pct exec` inside that LXC. Stdin (typically a heredoc'd bash script) can be piped via stdin_text. """ if pve_id is None: cmd = hubris_ssh() + ["--"] + remote_argv else: cmd = hubris_ssh() + ["--", "pct", "exec", pve_id, "--"] + remote_argv return subprocess.run(cmd, input=stdin_text, capture_output=True, text=True) def _audit_one(name: str, pve_id: str | None) -> dict | None: """Run the audit probe on one target. Returns metrics dict or None if unreachable.""" res = _run_on_target(name, pve_id, ["bash", "-s"], stdin_text=_APT_AUDIT_PROBE) if res.returncode != 0: return None parts = res.stdout.strip().split() if len(parts) != 5: return None try: return { "dpkg_dirty": int(parts[0]), "holds": int(parts[1]), "upgradable": int(parts[2]), "nonapt_bins": int(parts[3]), "dns": parts[4], } except ValueError: return None def _snapshot_target(name: str, pve_id: str | None, snap_name: str) -> tuple[str, str]: """Take a pre-upgrade snapshot of a target. Tries `pct snapshot` first (CoW, near-instant). Falls back to `vzdump --mode snapshot` for LXCs where pct snapshot refuses due to host bind- mounts. Hubris is skipped (no PVE-host-level snapshot supported here). Returns (method, detail) where method is one of: "pct" -> pct snapshot succeeded; detail = snapshot name "vzdump" -> vzdump succeeded; detail = backup file path "skip" -> hubris (skipped intentionally) "fail" -> both failed; detail = error message """ if pve_id is None: return ("skip", "hubris (no host-level snapshot)") # pct snapshot first res = subprocess.run(hubris_ssh() + ["--", "pct", "snapshot", pve_id, snap_name, "--description", f"homelab apt-upgrade --safe ({name})"], capture_output=True, text=True) if res.returncode == 0: return ("pct", snap_name) err = (res.stderr or res.stdout or "").strip() # Common refusal for LXCs with host bind-mounts. bind_mount_refused = ( "snapshot feature is not available" in err.lower() or "is not snapshottable" in err.lower() or "snapshots are not supported" in err.lower() ) if not bind_mount_refused: return ("fail", err) # Fall back to vzdump res = subprocess.run(hubris_ssh() + ["--", "vzdump", pve_id, "--mode", "snapshot", "--storage", "local", "--compress", "zstd", "--notes-template", f"homelab apt-upgrade --safe ({name})"], capture_output=True, text=True) if res.returncode == 0: # Extract the file path from vzdump output (line like "creating archive '/var/lib/vz/dump/...vma.zst'"). m = _re.search(r"creating[^']*'([^']+)'", res.stdout) path = m.group(1) if m else "(vzdump complete)" return ("vzdump", path) return ("fail", (res.stderr or res.stdout or "").strip()) def cmd_apt_audit(args: argparse.Namespace) -> int: """Per-host pre-flight: dpkg state, holds, upgradable count, non-apt binaries, DNS health. Exit nonzero if any host shows dpkg-interrupted state (would block apt-upgrade).""" targets = _apt_targets() if args.target: targets = [(n, i) for n, i in targets if n == args.target] if not targets: die(f"target '{args.target}' is not a known apt target") print(f"{'HOST':<20} {'DPKG':<7} {'HOLDS':<6} {'UPGR':<6} {'NONAPT':<7} {'DNS':<5}") any_dirty = False for name, pve_id in targets: m = _audit_one(name, pve_id) if m is None: print(f"{name:<20} unreachable") continue if m["dpkg_dirty"]: dpkg_s = f"DIRTY({m['dpkg_dirty']})" any_dirty = True else: dpkg_s = "ok" dns_s = m["dns"] if m["dns"] == "ok" else "FAIL" print(f"{name:<20} {dpkg_s:<7} {m['holds']:<6} {m['upgradable']:<6} {m['nonapt_bins']:<7} {dns_s:<5}") if any_dirty: print() print("DIRTY hosts have unconfigured packages — recover with: ssh dpkg --configure -a", file=sys.stderr) return 1 if any_dirty else 0 def cmd_apt_upgrade(args: argparse.Namespace) -> int: """Launch apt update+upgrade inside a detached `screen` on each target. Survives ssh teardown. Output tee'd to /var/log/homelab-apt-upgrade.log on the target. Refuses if pre-flight audit finds dpkg-interrupted state (override with --force).""" targets = _apt_targets() if args.target: targets = [(n, i) for n, i in targets if n == args.target] if not targets: die(f"target '{args.target}' is not a known apt target") elif not args.all: die("specify --target or --all") if args.status: status_probe = r""" u="apt-upgrade-$(hostname -s)" if systemctl is-active --quiet "$u.service" 2>/dev/null; then echo "$u: ACTIVE" else echo "$u: not running" fi echo --- if [ -f /var/log/homelab-apt-upgrade.log ]; then tail -20 /var/log/homelab-apt-upgrade.log else echo "(no log)" fi """ for name, pve_id in targets: print(f"=== {name} ===") res = _run_on_target(name, pve_id, ["bash", "-s"], stdin_text=status_probe) print(res.stdout.rstrip()) return 0 # Pre-flight audit gate if not args.force: dirty = [] for name, pve_id in targets: m = _audit_one(name, pve_id) if m and m["dpkg_dirty"]: dirty.append(name) if dirty: print(f"refusing: pre-existing dpkg-interrupted state on: {', '.join(dirty)}", file=sys.stderr) print(" recover with: homelab ssh -- dpkg --configure -a", file=sys.stderr) print(" or rerun with --force to skip the audit gate", file=sys.stderr) return 2 # Optional snapshot pass — LXC-only; hubris is skipped. snap_results: dict[str, tuple[str, str]] = {} if args.safe: # pct snapshot names must match [a-zA-Z][a-zA-Z0-9_]* — underscores only. snap_name = "preupgrade_" + datetime.now().strftime("%Y%m%d_%H%M") print(f"snapshot pass ({snap_name}):") any_fail = False for name, pve_id in targets: method, detail = _snapshot_target(name, pve_id, snap_name) snap_results[name] = (method, detail) print(f" {name:<20} {method:<7} {detail}") if method == "fail": any_fail = True if any_fail and not args.force: print("refusing: at least one snapshot failed; rerun with --force to upgrade anyway", file=sys.stderr) return 3 # Launch on each target rc = 0 for name, pve_id in targets: res = _run_on_target(name, pve_id, ["bash", "-s"], stdin_text=_APT_UPGRADE_WRAPPER) if res.returncode != 0: print(f"{name}: launch failed (rc={res.returncode}): {res.stderr.strip()}", file=sys.stderr) rc = 1 continue print(f"{name}: {res.stdout.strip()}") # Reminder of snapshot rollback paths after launch. if snap_results: print() print("snapshots created (rollback path on failure):") for name, (method, detail) in snap_results.items(): if method == "pct": print(f" {name}: pct rollback {detail} (or pct delsnapshot {detail})") elif method == "vzdump": print(f" {name}: pct restore {detail} (or rm {detail} when no longer needed)") return rc def main() -> int: p = argparse.ArgumentParser(prog="homelab", description=__doc__) sub = p.add_subparsers(dest="cmd", required=True) sp = sub.add_parser("whoami", help="print this host's hosts/.yaml") sp.add_argument("hostname", nargs="?") sp.set_defaults(func=cmd_whoami) sp = sub.add_parser("list", help="list hosts and services from inventory") sp.set_defaults(func=cmd_list) sp = sub.add_parser("ssh", help="ssh to a host via mesh") sp.add_argument("host") sp.add_argument("--user", "-u", default=None) sp.add_argument("command", nargs=argparse.REMAINDER) sp.set_defaults(func=cmd_ssh) sp = sub.add_parser("ssh-config", help="generate ~/.ssh/config.d/homelab from inventory.yaml") sp.add_argument("--install", "-i", action="store_true", help=f"write to ~/.ssh/config.d/homelab and wire Include into main config") sp.set_defaults(func=cmd_ssh_config) sp = sub.add_parser("pct", help="proxy pct commands via ssh to hubris") sp.add_argument("lxc") sp.add_argument("action") sp.add_argument("rest", nargs=argparse.REMAINDER) sp.add_argument("--yes", "-y", action="store_true") sp.set_defaults(func=cmd_pct) sp = sub.add_parser("logs", help="journalctl for a service") sp.add_argument("service") sp.add_argument("--lines", "-n", type=int, default=200) sp.add_argument("--follow", "-f", action="store_true") sp.set_defaults(func=cmd_logs) sp = sub.add_parser("restart", help="restart a service") sp.add_argument("service") sp.add_argument("--yes", "-y", action="store_true") sp.set_defaults(func=cmd_restart) sp = sub.add_parser("open", help="open a service's URL in browser") sp.add_argument("service") sp.set_defaults(func=cmd_open) sp = sub.add_parser("status", help="ping every host + HTTP-check every service") sp.set_defaults(func=cmd_status) sp = sub.add_parser("doctor", help="run health checks on this client's enrollment") sp.set_defaults(func=cmd_doctor) sp = sub.add_parser("secret", help="decrypt a secret (sops -d wrapper)") sp.add_argument("name") sp.set_defaults(func=cmd_secret) sp = sub.add_parser("render-vps-configs", help="re-render /etc/turnserver.conf + /opt/management.json on the IONOS netbird VPS from templates + sops secrets") sp.add_argument("--dry-run", action="store_true", help="show what would change; don't apply or restart") sp.add_argument("-y", "--yes", action="store_true", help="skip the confirmation prompt") sp.set_defaults(func=cmd_render_vps_configs) sp = sub.add_parser("sync", help="manually trigger homelab-context-sync") sp.set_defaults(func=cmd_sync) sp = sub.add_parser("refresh-creds", help="swap the read-only bootstrap PAT for the write-scoped one from secrets/gitea-pat.yaml") sp.set_defaults(func=cmd_refresh_creds) sp = sub.add_parser("mcp", help="call an MCP tool (requires 'mcp' CLI installed)") sp.add_argument("tool") sp.add_argument("args", nargs=argparse.REMAINDER) sp.set_defaults(func=cmd_mcp) sp = sub.add_parser("apt-audit", help="per-host pre-flight: dpkg state, holds, upgradable count, non-apt binaries, DNS") sp.add_argument("--target", default=None, help="only audit one host (default: hubris + every LXC)") sp.set_defaults(func=cmd_apt_audit) sp = sub.add_parser("apt-upgrade", help="launch `apt update && apt upgrade` inside detached screen on each target") grp = sp.add_mutually_exclusive_group() grp.add_argument("--target", default=None, help="upgrade one host") grp.add_argument("--all", action="store_true", help="upgrade hubris + every LXC") sp.add_argument("--status", action="store_true", help="show running screen sessions + tail the upgrade log on each target") sp.add_argument("--safe", action="store_true", help="take a pre-upgrade snapshot per LXC (pct snapshot, vzdump fallback)") sp.add_argument("--force", action="store_true", help="skip the pre-flight dpkg-audit gate AND proceed past snapshot failures") sp.set_defaults(func=cmd_apt_upgrade) sp = sub.add_parser("nuke", help="shred /etc/age/key.txt + /opt/homelab-context on a host") sp.add_argument("name") sp.add_argument("--yes", "-y", action="store_true") sp.set_defaults(func=cmd_nuke) client = sub.add_parser("client", help="client lifecycle (add/remove)") csub = client.add_subparsers(dest="action", required=True) csub_add = csub.add_parser("add") csub_add.add_argument("name") csub_add.add_argument("--finalize-pubkey", default=None, help="set/update age_pubkey for an already-added client") csub_add.add_argument("--with-hermes", action="store_true", help="also grant secrets/openrouter-api-key.yaml so this " "host can run the Hermes agent (see " "operations/hermes-agent.md). Combine with --finalize-pubkey.") csub_add.set_defaults(func=cmd_client_add) csub_rm = csub.add_parser("remove") csub_rm.add_argument("name") csub_rm.add_argument("--yes", "-y", action="store_true") csub_rm.set_defaults(func=cmd_client_remove) args = p.parse_args() return args.func(args) if __name__ == "__main__": sys.exit(main())