oikos/archive/secrets-issuance/server.py

#!/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/<hostname>.key       (private, 0600 root)
  /var/lib/secrets-issuance/keys/<hostname>.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())