36a686d953
Audit against actual netbird+tailscale peer lists:
- hubris is the only LXC-host on Netbird; only workstations + hubris
have netbird entries
- 10 LXCs+VMs have real Tailscale FQDNs: apps, jellyfin, paperless,
gitea, nextcloud, elementsynapse, sophia, mule-images→muleimage,
arriman→arr, haos→homeassistant
- 7 hosts are LAN-only (no mesh block): nfs-export, caddy, claudio-bot,
authentik, plato, mule-photos-new, zimaos
- mac-mini's netbird FQDN corrected to the actual peer name
(mac-mini-234-17.netbird.selfhosted)
Also: bin/homelab host_address() now prefers lan_ip first — universally
reachable from any LAN client and from any Netbird peer via the
192.168.8.0/24 network resource routed through hubris. Mesh FQDNs are
fallbacks for roaming workstations without a fixed lan_ip.
This makes 'homelab status' from republic show all backends 'ok' instead
of falsely reporting them 'down' against unresolvable netbird FQDNs.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
842 lines
32 KiB
Python
Executable File
842 lines
32 KiB
Python
Executable File
#!/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 <subcommand> --help` for usage.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import urllib.request
|
|
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}")
|
|
|
|
|
|
def hubris_ssh() -> list[str]:
|
|
"""SSH base command for hubris, honoring its non-default Netbird SSH port."""
|
|
h = host("hubris")
|
|
port = h.get("ssh", {}).get("netbird_port", 22)
|
|
return ["ssh", "-p", str(port), f"root@{host_address('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$"),
|
|
]
|
|
|
|
|
|
# -------- 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 ` <name>:` 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.<name>.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.<name>: 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.<name>: 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) -> None:
|
|
"""Add `pubkey` to the recipient list of every shared secret + re-key."""
|
|
sops_path = CONTEXT / ".sops.yaml"
|
|
env = {**os.environ, "SOPS_AGE_KEY_FILE": str(AGE_KEY)}
|
|
for rel_path, pattern in SHARED_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) -> None:
|
|
"""Remove `pubkey` from every shared-secret rule + re-key the files."""
|
|
sops_path = CONTEXT / ".sops.yaml"
|
|
env = {**os.environ, "SOPS_AGE_KEY_FILE": str(AGE_KEY)}
|
|
for rel_path, pattern in SHARED_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
|
|
h = host(name)
|
|
addr = host_address(name)
|
|
cmd = ["ssh"]
|
|
# hubris uses a non-default Netbird SSH port for the mesh path
|
|
if name == "hubris":
|
|
port = h.get("ssh", {}).get("netbird_port", 22)
|
|
cmd.extend(["-p", str(port)])
|
|
user = args.user or "root"
|
|
cmd.append(f"{user}@{addr}")
|
|
if args.command:
|
|
cmd.append(" ".join(args.command))
|
|
os.execvp(cmd[0], 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)
|
|
if host_name == "hubris":
|
|
base = hubris_ssh()
|
|
else:
|
|
base = ["ssh", f"root@{host_address(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
|
|
if host_name == "hubris":
|
|
base = hubris_ssh()
|
|
else:
|
|
base = ["ssh", f"root@{host_address(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 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)
|
|
|
|
|
|
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:
|
|
if sys.platform == "darwin":
|
|
return subprocess.call(
|
|
["sudo", "launchctl", "kickstart", "-k",
|
|
"system/network.hubris.homelab-context-sync"]
|
|
)
|
|
return subprocess.call(
|
|
["sudo", "systemctl", "start", "homelab-context-sync.service"]
|
|
)
|
|
|
|
|
|
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 []))
|
|
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 <gitea>/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 <age1...>")
|
|
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)
|
|
push_inventory(
|
|
f"client-add: {name} (finalize age_pubkey + grant shared secrets)",
|
|
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)
|
|
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
|
|
addr = host_address(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(["ssh", f"root@{addr}", remote])
|
|
|
|
|
|
# ---------- argparse ----------
|
|
|
|
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/<hostname>.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("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("secret", help="decrypt a secret (sops -d wrapper)")
|
|
sp.add_argument("name")
|
|
sp.set_defaults(func=cmd_secret)
|
|
|
|
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("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.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())
|