df6aca888c
/etc/age/key.txt is 0600 root and /etc/age is 0700 root, so the CLI's Path.exists() check was returning False under regular users — making the subcommand look broken when bootstrap had actually written the key fine. Re-exec via sudo preserves the existing UX (one password prompt, then plaintext) without loosening the key's perms. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
504 lines
18 KiB
Python
Executable File
504 lines
18 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:
|
|
h = host(name)
|
|
mesh = h.get("mesh", {})
|
|
candidates = []
|
|
if prefer_lan and h.get("lan_ip"):
|
|
candidates.append(h["lan_ip"])
|
|
candidates.extend([
|
|
mesh.get("netbird", {}).get("fqdn") if isinstance(mesh.get("netbird"), dict) else None,
|
|
mesh.get("netbird", {}).get("ip") if isinstance(mesh.get("netbird"), dict) else None,
|
|
h.get("lan_ip"),
|
|
mesh.get("tailscale", {}).get("fqdn") if isinstance(mesh.get("tailscale"), dict) else None,
|
|
mesh.get("tailscale", {}).get("ip") if isinstance(mesh.get("tailscale"), dict) else None,
|
|
])
|
|
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) -> None:
|
|
"""Stage + commit + push inventory + regenerated hosts/."""
|
|
subprocess.run(["python3", str(CONTEXT / "mcp" / "build_host_files.py")],
|
|
check=True, cwd=CONTEXT)
|
|
subprocess.run(["git", "add", "inventory.yaml", "hosts/"], 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)
|
|
|
|
|
|
# ---------- 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_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:
|
|
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"
|
|
entry = {
|
|
"kind": kind,
|
|
"os": os_name,
|
|
"role": role,
|
|
"mesh": {"netbird": {"fqdn": netbird_fqdn}},
|
|
"age_pubkey": "",
|
|
}
|
|
inv["hosts"][name] = entry
|
|
INVENTORY.write_text(yaml.safe_dump(inv, sort_keys=False))
|
|
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)")
|
|
inv["hosts"][name]["age_pubkey"] = args.finalize_pubkey
|
|
INVENTORY.write_text(yaml.safe_dump(inv, sort_keys=False))
|
|
push_inventory(f"client-add: {name} (finalize age_pubkey)")
|
|
print(f"finalized age_pubkey for {name}.")
|
|
return 0
|
|
|
|
|
|
def cmd_client_remove(args: argparse.Namespace) -> int:
|
|
name = args.name
|
|
inv = inventory()
|
|
if name not in inv["hosts"]:
|
|
die(f"{name} not in inventory")
|
|
if not args.yes:
|
|
print(f"This will:")
|
|
print(f" 1. Remove {name} from inventory.yaml and hosts/")
|
|
print(f" 2. Re-encrypt every secret without {name} as recipient")
|
|
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
|
|
del inv["hosts"][name]
|
|
INVENTORY.write_text(yaml.safe_dump(inv, sort_keys=False))
|
|
|
|
# 2. SOPS — remove recipient. Requires `sops updatekeys` after we edit .sops.yaml.
|
|
# We don't try to programmatically edit .sops.yaml because the recipient list
|
|
# there is keyed by path-glob rules; the operator must remove the pubkey line
|
|
# if it's listed by-name. We'll trigger updatekeys after the operator confirms.
|
|
secrets_dir = CONTEXT / "secrets"
|
|
if secrets_dir.exists():
|
|
print()
|
|
print("[remove] re-encrypting secrets without removed recipient (sops updatekeys)")
|
|
sops_yaml = CONTEXT / ".sops.yaml"
|
|
if sops_yaml.exists():
|
|
print(f" Note: review {sops_yaml} for hard-coded recipients of '{name}' "
|
|
"and remove them before sops updatekeys.")
|
|
for f in sorted(secrets_dir.glob("*.yaml")):
|
|
subprocess.run(["sops", "updatekeys", "-y", str(f)], check=False)
|
|
|
|
# 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
|
|
push_inventory(f"client-remove: {name}")
|
|
|
|
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("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())
|