#!/usr/bin/env python3 """ Secrets-issuance HTTP service. Lives on LXC 105 alongside the MCP server. Mesh-bound (nftables restricts to Netbird + Tailscale subnets). Identifies callers by source mesh IP and returns the per-client age private key. State: /var/lib/secrets-issuance/keys/.key (private, 0600 root) /var/lib/secrets-issuance/keys/.pub (public) /var/lib/secrets-issuance/denylist.txt (one hostname per line) Inventory lookup: /opt/homelab-context/inventory.yaml — the issuance server is itself a homelab-context client, so it sees inventory updates within 5 min. Endpoints: POST /issue body: {"hostname": "..."} -> raw age private key POST /revoke body: {"hostname": "..."} -> shred key + add to denylist (called by 'homelab client remove') GET /health """ from __future__ import annotations import ipaddress import json import logging import os import secrets as sysrandom import subprocess import sys from http.server import BaseHTTPRequestHandler, HTTPServer from pathlib import Path import yaml BIND_HOST = "0.0.0.0" BIND_PORT = 9820 CONTEXT_DIR = Path(os.environ.get("HOMELAB_CONTEXT_DIR", "/opt/homelab-context")) STATE_DIR = Path(os.environ.get("SECRETS_ISSUANCE_STATE", "/var/lib/secrets-issuance")) KEYS_DIR = STATE_DIR / "keys" DENYLIST = STATE_DIR / "denylist.txt" ADMIN_TOKEN_PATH = Path(os.environ.get("SECRETS_ISSUANCE_ADMIN_TOKEN", "/etc/secrets-issuance/admin-token")) MESH_SUBNETS = [ ipaddress.ip_network(s.strip()) for s in os.environ.get("MESH_SUBNETS", "100.64.0.0/10").split(",") if s.strip() ] logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") log = logging.getLogger("secrets-issuance") def load_inventory() -> dict: return yaml.safe_load((CONTEXT_DIR / "inventory.yaml").read_text()) def known_hosts() -> dict[str, dict]: return load_inventory().get("hosts", {}) def host_by_mesh_ip(ip: str) -> str | None: """Resolve a mesh source IP to a hostname per inventory.""" for name, entry in known_hosts().items(): mesh = entry.get("mesh", {}) for proto in ("netbird", "tailscale"): p = mesh.get(proto, {}) if not isinstance(p, dict): continue if p.get("ip") == ip: return name return None def in_mesh(ip: str) -> bool: addr = ipaddress.ip_address(ip) return any(addr in subnet for subnet in MESH_SUBNETS) def is_denied(hostname: str) -> bool: if not DENYLIST.exists(): return False return hostname in {line.strip() for line in DENYLIST.read_text().splitlines() if line.strip()} def deny(hostname: str) -> None: DENYLIST.parent.mkdir(parents=True, exist_ok=True) existing = set() if DENYLIST.exists(): existing = {line.strip() for line in DENYLIST.read_text().splitlines() if line.strip()} existing.add(hostname) DENYLIST.write_text("\n".join(sorted(existing)) + "\n") def generate_key(hostname: str) -> tuple[Path, str]: """Generate an age keypair for hostname; return (priv_path, pubkey).""" KEYS_DIR.mkdir(parents=True, exist_ok=True) priv = KEYS_DIR / f"{hostname}.key" pub_path = KEYS_DIR / f"{hostname}.pub" proc = subprocess.run(["age-keygen"], capture_output=True, text=True, check=True) priv.write_text(proc.stdout) priv.chmod(0o600) # Extract the public key. age-keygen writes "# public key: ageXXX" line. pubkey = "" for line in proc.stdout.splitlines(): if line.startswith("# public key:"): pubkey = line.split(":", 1)[1].strip() if not pubkey: priv.unlink(missing_ok=True) raise RuntimeError("age-keygen did not emit a public key") pub_path.write_text(pubkey + "\n") log.info("generated new age key for %s (pubkey: %s)", hostname, pubkey) return priv, pubkey def existing_pubkey(hostname: str) -> str | None: pub_path = KEYS_DIR / f"{hostname}.pub" if pub_path.exists(): return pub_path.read_text().strip() return None def read_admin_token() -> str | None: if ADMIN_TOKEN_PATH.exists(): return ADMIN_TOKEN_PATH.read_text().strip() return None class Handler(BaseHTTPRequestHandler): def log_message(self, fmt, *args): log.info("%s - %s", self.address_string(), fmt % args) def _reply(self, status: int, body: bytes | str = b"", content_type: str = "text/plain") -> None: self.send_response(status) self.send_header("Content-Type", content_type) if isinstance(body, str): body = body.encode() self.send_header("Content-Length", str(len(body))) self.end_headers() if body: self.wfile.write(body) def _read_json(self) -> dict | None: length = int(self.headers.get("Content-Length", "0")) if not length: return {} body = self.rfile.read(length) try: return json.loads(body) except json.JSONDecodeError: return None def do_GET(self) -> None: if self.path == "/health": self._reply(200, "ok") return self._reply(404, "not found") def do_POST(self) -> None: client_ip = self.client_address[0] if not in_mesh(client_ip): log.warning("rejecting non-mesh source %s", client_ip) self._reply(403, "non-mesh source") return if self.path == "/issue": self._handle_issue(client_ip) elif self.path == "/revoke": self._handle_revoke() else: self._reply(404, "not found") def _handle_issue(self, client_ip: str) -> None: payload = self._read_json() if payload is None: self._reply(400, "bad json") return hostname = (payload.get("hostname") or "").strip() if not hostname: self._reply(400, "hostname required") return # Cross-check: does inventory list this hostname, and does its mesh IP # match the caller (if known)? hosts = known_hosts() if hostname not in hosts: log.warning("unknown hostname in /issue: %s (from %s)", hostname, client_ip) self._reply(403, f"unknown hostname: {hostname} (run 'homelab client add {hostname}' first)") return if is_denied(hostname): log.warning("denied hostname %s tried to issue (from %s)", hostname, client_ip) self._reply(403, "hostname on denylist (was removed); operator must clear before re-enrolling") return inv_ip = host_by_mesh_ip(client_ip) if inv_ip is not None and inv_ip != hostname: log.warning("source %s maps to inventory host %s but body claims %s", client_ip, inv_ip, hostname) self._reply(403, "source IP / hostname mismatch") return if inv_ip is None: # Inventory has no IP for this hostname yet (first-bootstrap state). # Accept the call but log it loudly so the operator backfills. log.info("issuing for %s (source %s) — inventory has no mesh IP yet", hostname, client_ip) priv = KEYS_DIR / f"{hostname}.key" existing_pub = existing_pubkey(hostname) if priv.exists() and existing_pub: # Re-issue: client lost its key but is still authorized. Return # the existing key. log.info("returning existing key for %s", hostname) self._reply(200, priv.read_text()) return # Fresh provisioning. priv, pubkey = generate_key(hostname) # Tell the operator (via response body comment) which pubkey to commit. body = priv.read_text() body += f"\n# operator: add this pubkey to inventory.yaml under hosts.{hostname}.age_pubkey:\n" body += f"# {pubkey}\n" self._reply(200, body) def _handle_revoke(self) -> None: token = self.headers.get("X-Admin-Token", "") expected = read_admin_token() if not expected or token != expected: self._reply(403, "admin token required") return payload = self._read_json() if payload is None: self._reply(400, "bad json") return hostname = (payload.get("hostname") or "").strip() if not hostname: self._reply(400, "hostname required") return priv = KEYS_DIR / f"{hostname}.key" pub = KEYS_DIR / f"{hostname}.pub" for p in (priv, pub): if p.exists(): # shred-then-unlink subprocess.run(["shred", "-u", str(p)], check=False) if p.exists(): p.unlink(missing_ok=True) deny(hostname) log.warning("revoked %s (shredded keys, added to denylist)", hostname) self._reply(200, f"revoked {hostname}\n") def main() -> int: KEYS_DIR.mkdir(parents=True, exist_ok=True) server = HTTPServer((BIND_HOST, BIND_PORT), Handler) log.info("listening on %s:%d (context=%s, state=%s)", BIND_HOST, BIND_PORT, CONTEXT_DIR, STATE_DIR) server.serve_forever() return 0 if __name__ == "__main__": sys.exit(main())