vacuum-wall/lib/nginx.py

"""Nginx server-block generator for Vacuum Wall SSL proxy firewall.

Manages per-domain SSL reverse proxy configurations backed by named backends,
certificate bootstrap, basic-auth htpasswd files, and nginx reload cycles.
"""

import logging
import os
import subprocess
from copy import deepcopy
from pathlib import Path
from typing import Any

from jinja2 import Environment, FileSystemLoader

from lib.acme import find_cert_dir
from lib.common import _hash_password, ensure_dirs, load_json, save_json

logger = logging.getLogger(__name__)

PROJECT_DIR = Path(__file__).resolve().parent.parent
CONFIG_DIR = PROJECT_DIR / "config" / "nginx"
DATA_DIR = PROJECT_DIR / "data" / "nginx"
SITES_DIR = DATA_DIR / "sites-enabled"
CONFIG_FILE = CONFIG_DIR / "config.json"
INCLUDE_FILE = Path("/etc/nginx/conf.d/vacuum-wall.conf")
SSL_SNIPPET = Path("/etc/nginx/snippets/vacuum-wall-ssl.conf")
HTPASSWD_FILE = DATA_DIR / ".htpasswd"

ENV = Environment(
    loader=FileSystemLoader(str(PROJECT_DIR / "system")),
    autoescape=False,
    lstrip_blocks=True,
    trim_blocks=True,
)

DEFAULT_SSL: dict[str, Any] = {
    "protocols": "TLSv1.2 TLSv1.3",
    "ciphers": (
        "ECDHE-ECDSA-AES128-GCM-SHA256:"
        "ECDHE-RSA-AES128-GCM-SHA256:"
        "ECDHE-ECDSA-AES256-GCM-SHA384:"
        "ECDHE-RSA-AES256-GCM-SHA384:"
        "ECDHE-ECDSA-CHACHA20-POLY1305:"
        "ECDHE-RSA-CHACHA20-POLY1305"
    ),
    "prefer_server_ciphers": False,
}

WEBUI_BACKEND: dict[str, Any] = {
    "label": "Vacuum Wall WebUI",
    "builtin": True,
    "paths": {
        "/": {
            "backend": {"host": "127.0.0.1", "port": 9090, "proto": "http"},
            "is_management": True,
            # The WebUI is protected by JWT at the Flask layer; nginx must
            # not gate it with auth_basic (the SPA sends Bearer tokens, which
            # suppress the browser's automatic Basic credentials). auth=None
            # renders `auth_basic off` even if legacy auth was harvested.
            "auth": None,
        },
        "/ws": {
            "backend": {"host": "127.0.0.1", "port": 9091, "proto": "http"},
            "is_websocket": True,
        },
    },
}

DEFAULT_CONFIG: dict[str, Any] = {
    "backends": {},
    "domains": {},
    "ssl": {**DEFAULT_SSL},
}


# ---------------------------------------------------------------------------
# Resolution


def _resolve_paths(
    domain_cfg: dict[str, Any], backends: dict[str, Any]
) -> dict[str, Any]:
    """Resolve effective paths from backends[domain_cfg['backend']].paths."""
    backend_name = domain_cfg.get("backend", "")
    if backend_name and backend_name in backends:
        return backends[backend_name].get("paths", {})
    return {}


def _resolve_auth(
    domain_cfg: dict[str, Any], backends: dict[str, Any]
) -> dict[str, Any] | None:
    """Resolve effective auth: domain auth -> backend auth -> None."""
    if "auth" in domain_cfg:
        return domain_cfg.get("auth")
    backend_name = domain_cfg.get("backend", "")
    if backend_name and backend_name in backends:
        backend_auth = backends[backend_name].get("auth")
        if backend_auth is not None:
            return backend_auth
    return None


# ---------------------------------------------------------------------------
# Migration


def _migrate_config(raw: dict[str, Any]) -> dict[str, Any]:
    """Migrate legacy config to backends model."""
    _ensure_webui_backend(raw)
    _migrate_mgmt_domains(raw)
    return raw


def _ensure_webui_backend(raw: dict[str, Any]) -> None:
    """Create the builtin webui backend if not yet migrated."""
    backends = raw.setdefault("backends", {})
    webui = backends.get("webui")
    if webui and webui.get("_migrated"):
        return
    backends["webui"] = deepcopy(WEBUI_BACKEND)
    backends["webui"]["_migrated"] = True


def _migrate_mgmt_domains(raw: dict[str, Any]) -> None:
    """Migrate legacy management domains to backend references.

    Legacy format: management domains had inline paths pointing to
    127.0.0.1:9090 (Flask) and 127.0.0.1:9091 (WebSocket).
    New format: domains reference the "webui" backend by name.

    Detection heuristic: if both "/" path points to 127.0.0.1:9090
    (is_management) and "/ws" path points to 127.0.0.1:9091
    (is_websocket), the domain is a management domain and gets migrated.
    """
    backends = raw.get("backends", {})
    if not backends.get("webui", {}).get("_migrated"):
        return
    domains = raw.setdefault("domains", {})
    for _name, dom in list(domains.items()):
        if dom.get("backend") == "webui":
            continue  # Already migrated
        if dom.get("application") == "webui":
            del dom["application"]
        paths = dom.get("paths", {})
        root = paths.get("/", {})
        ws = paths.get("/ws", {})
        root_backend = root.get("backend", {})
        ws_backend = ws.get("backend", {})
        # Check if root path points to Flask management backend
        is_mgmt_root = root.get("is_management") or (
            root_backend.get("host") == "127.0.0.1" and root_backend.get("port") == 9090
        )
        # Check if WS path points to WebSocket management backend
        is_mgmt_ws = ws.get("is_websocket") or (
            ws_backend.get("host") == "127.0.0.1" and ws_backend.get("port") == 9091
        )
        # If both match, migrate: set backend reference, remove inline paths/auth
        if is_mgmt_root and is_mgmt_ws:
            dom["backend"] = "webui"
            dom.pop("paths", None)
            dom.pop("auth", None)


# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------


def get_config() -> dict[str, Any]:
    """Load the current nginx config, initializing with defaults if needed.

    Ensures config and sites directories exist, applies migrations for
    legacy formats, then returns the config dict.

    Returns:
        The complete config dict with ``backends``, ``domains``, and ``ssl`` keys.
    """
    ensure_dirs(CONFIG_DIR, SITES_DIR)
    raw = load_json(CONFIG_FILE)
    if not raw:
        raw = deepcopy(DEFAULT_CONFIG)
    if "ssl" not in raw:
        raw["ssl"] = deepcopy(DEFAULT_SSL)
    if "backends" not in raw:
        raw["backends"] = {}
    raw = _migrate_config(raw)
    save_config(raw)
    return raw


def save_config(cfg: dict[str, Any]) -> None:
    """Persist *cfg* to the nginx config file atomically."""
    save_json(CONFIG_FILE, cfg)


def get_domains() -> list[dict[str, Any]]:
    """Return a list of all configured proxy domains flattened by path.

    Each path within a domain becomes a separate entry. Paths are resolved
    from the domain's referenced backend.

    Returns:
        List of dicts with ``domain``, ``path``, ``backend``, ``online``,
        ``force_ssl``, ``backend_name``, and path-level flags.
    """
    cfg = get_config()
    backends = cfg.get("backends", {})
    result: list[dict[str, Any]] = []
    for name, dom in cfg.get("domains", {}).items():
        if "backend" not in dom:
            continue
        site = SITES_DIR / f"{name}.conf"
        paths = _resolve_paths(dom, backends)
        for ppath, pcfg in paths.items():
            entry: dict[str, Any] = {
                "domain": name,
                "path": ppath,
                "backend": pcfg.get("backend", {}),
                "online": site.exists(),
                "force_ssl": dom.get("force_ssl", True),
                "backend_name": dom["backend"],
            }
            if pcfg.get("is_management"):
                entry["is_management"] = True
            if pcfg.get("is_websocket"):
                entry["is_websocket"] = True
            result.append(entry)
    return result


def get_management_domains() -> list[str]:
    """Return domain names that serve the management UI.

    Checks both backend-referenced paths (for migrated configs) and
    inline paths (for legacy configs pending migration).

    Returns:
        List of domain name strings.
    """
    cfg = get_config()
    backends = cfg.get("backends", {})
    domains: list[str] = []
    for name, dom in cfg.get("domains", {}).items():
        # Check inline paths (pre-migration format)
        inline_paths = dom.get("paths", {})
        if any(p.get("is_management") for p in inline_paths.values()):
            domains.append(name)
            continue
        # Check backend-referenced paths
        backend_name = dom.get("backend", "")
        if backend_name and backend_name in backends:
            paths = backends[backend_name].get("paths", {})
            if any(p.get("is_management") for p in paths.values()):
                domains.append(name)
    return domains


# ------------------------------------------------------------------
# Domain CRUD
# ------------------------------------------------------------------


def add_domain(
    domain: str,
    backend_name: str,
    cert: str | None = None,
    force_ssl: bool = True,
    auth: dict[str, Any] | None = None,
) -> None:
    """Add a new proxy domain that references an existing backend.

    Args:
        domain: Domain name to add.
        backend_name: Name of the backend to proxy through.
        cert: Optional certificate type identifier.
        force_ssl: Whether to enforce HTTPS redirect.
        auth: Optional domain-level auth override.

    Raises:
        ValueError: If the domain is already configured or backend not found.
    """
    cfg = get_config()
    if domain in cfg["domains"]:
        raise ValueError(f"Domain {domain!r} already configured")
    if backend_name not in cfg.get("backends", {}):
        raise ValueError(f"Backend {backend_name!r} not found")

    entry: dict[str, Any] = {
        "backend": backend_name,
        "force_ssl": force_ssl,
    }
    if cert is not None:
        entry["cert"] = cert
    if auth is not None:
        entry["auth"] = auth

    cfg["domains"][domain] = entry
    save_config(cfg)
    logger.info("Proxy domain '%s' added", domain)


def remove_domain(domain: str) -> None:
    """Remove *domain* from the config and delete its site file."""
    cfg = get_config()
    cfg["domains"].pop(domain, None)
    save_config(cfg)
    site = SITES_DIR / f"{domain}.conf"
    if site.exists():
        site.unlink()
    logger.info("Proxy domain '%s' removed", domain)


def update_domain(domain: str, **kwargs: Any) -> None:
    """Update domain-level fields of an existing domain entry.

    Only domain-level keys are accepted: ``backend``, ``cert``, ``force_ssl``,
    ``auth``. Path changes must be made on the backend.

    Args:
        domain: Domain name to update.
        **kwargs: Key-value pairs to merge into the domain config.

    Raises:
        KeyError: If the domain is not configured.
        ValueError: If a new backend is specified but doesn't exist.
    """
    cfg = get_config()
    if domain not in cfg["domains"]:
        raise KeyError(f"Domain {domain!r} not configured")
    entry = cfg["domains"][domain]

    new_backend = kwargs.get("backend")
    if new_backend:
        if new_backend not in cfg.get("backends", {}):
            raise ValueError(f"Backend {new_backend!r} not found")
        entry["backend"] = new_backend

    allowed = ("backend", "cert", "force_ssl", "auth")
    for key in allowed:
        if key in kwargs and key != "backend":
            if key == "auth" and kwargs[key] is None:
                entry.pop("auth", None)
            else:
                entry[key] = kwargs[key]

    save_config(cfg)
    logger.info("Proxy domain '%s' updated: %s", domain, list(kwargs.keys()))


# ------------------------------------------------------------------
# Nginx config generation
# ------------------------------------------------------------------


def generate_server_conf(
    domain_cfg: dict[str, Any], backends: dict[str, Any] | None = None
) -> str:
    """Render the Jinja template for a domain server block.

    Args:
        domain_cfg: Domain entry dict including the ``domain`` key.
        backends: Optional backends dict for path/auth resolution.
            When omitted, falls back to reading from domain inline paths.

    Returns:
        The complete nginx server-block configuration as a string.
    """
    tmpl = ENV.get_template("nginx/server_block.conf")
    acme_home_path = PROJECT_DIR / "data" / "acme"
    acme_cert_dir = str(find_cert_dir(domain_cfg["domain"], acme_home_path))

    if backends is not None:
        paths = _resolve_paths(domain_cfg, backends)
        domain_auth = _resolve_auth(domain_cfg, backends)
    else:
        paths = domain_cfg.get("paths", {})
        domain_auth = domain_cfg.get("auth")

    has_management = any(p.get("is_management") for p in paths.values())
    # Resolve custom cert paths for cert=="file"
    cert_cfg = domain_cfg.get("cert")
    if isinstance(cert_cfg, dict):
        cert_path = cert_cfg.get("cert_path", "")
        cert_key_path = cert_cfg.get("cert_key_path", "")
    else:
        cert_path = domain_cfg.get("cert_path", "")
        cert_key_path = domain_cfg.get("cert_key_path", "")
    return tmpl.render(
        domain=domain_cfg["domain"],
        paths=paths,
        force_ssl=domain_cfg.get("force_ssl", True),
        cert=domain_cfg.get("cert"),
        cert_path=cert_path,
        cert_key_path=cert_key_path,
        domain_auth=domain_auth,
        has_management=has_management,
        acme_cert_dir=acme_cert_dir,
        certs_dir=str(PROJECT_DIR / "data" / "certs"),
        acme_webroot=str(PROJECT_DIR / "data" / "acme" / "www"),
    )


# ------------------------------------------------------------------
# File writers
# ------------------------------------------------------------------


def write_site(domain: str, conf_text: str) -> None:
    """Atomically write *conf_text* to the site config file for *domain*.

    Args:
        domain: Domain name (becomes the ``<domain>.conf`` file).
        conf_text: Nginx server-block configuration text.
    """
    ensure_dirs(SITES_DIR)
    path = SITES_DIR / f"{domain}.conf"
    tmp = path.with_suffix(".tmp")
    with open(tmp, "w") as f:
        f.write(conf_text)
        f.write("\n")
    os.chmod(tmp, 0o644)
    os.replace(tmp, path)


def write_acme_challenge() -> None:
    """Write the catch-all nginx config for ACME HTTP-01 challenges.

    Serves ``/.well-known/acme-challenge/`` on port 80 from the ACME
    webroot for any domain not yet covered by a dedicated server block.
    """
    tmpl = ENV.get_template("nginx/acme-challenge.conf")
    content = tmpl.render(
        acme_webroot=str(PROJECT_DIR / "data" / "acme" / "www"),
    )
    site = SITES_DIR / "_acme-challenge.conf"
    tmp = site.with_suffix(".tmp")
    with open(tmp, "w") as f:
        f.write(content)
        f.write("\n")
    os.chmod(tmp, 0o644)
    os.replace(tmp, site)


def write_all_sites() -> None:
    """Regenerate all site configs from the current config state.

    Writes server blocks for every configured domain using backend-resolved
    paths, removes orphaned site files, and ensures the ACME challenge
    config is present.
    """
    ensure_dirs(SITES_DIR)
    cfg = get_config()
    backends = cfg.get("backends", {})

    existing = set(SITES_DIR.iterdir()) if SITES_DIR.exists() else set()

    written: set[str] = set()
    for name, dom in cfg.get("domains", {}).items():
        dom_copy = dict(dom, domain=name)
        conf = generate_server_conf(dom_copy, backends)
        write_site(name, conf)
        written.add(f"{name}.conf")

    # Remove old management.conf if it exists
    old_mgmt = SITES_DIR / "management.conf"
    if old_mgmt.exists() and old_mgmt.name not in written:
        old_mgmt.unlink()

    for old in existing:
        if old.suffix == ".conf" and old.name not in written:
            old.unlink()

    write_acme_challenge()
    logger.info("All nginx site configs written (%d sites)", len(written))


def write_include_file() -> None:
    """Write the nginx include file that pulls in managed site configs.

    The include file is installed at ``/etc/nginx/conf.d/vacuum-wall.conf``
    and must be owned by root.
    """
    tmpl = ENV.get_template("nginx/include.conf")
    content = tmpl.render(sites_glob=str(SITES_DIR / "*.conf"))
    tmp = INCLUDE_FILE.with_suffix(".tmp")
    with open(tmp, "w") as f:
        f.write(content)
    os.chmod(tmp, 0o644)
    subprocess.run(["sudo", "cp", str(tmp), str(INCLUDE_FILE)], check=True)
    subprocess.run(["sudo", "chown", "root:root", str(INCLUDE_FILE)], check=True)
    tmp.unlink(missing_ok=True)


def write_ssl_snippet() -> None:
    """Write the shared SSL settings snippet to ``/etc/nginx/snippets/``.

    The snippet is populated from the ``ssl`` section of the nginx config
    and installed with root ownership.
    """
    cfg = get_config()
    ssl_cfg = cfg.get("ssl", {})
    ssl_cfg.setdefault("prefer_server_ciphers", DEFAULT_SSL["prefer_server_ciphers"])
    ssl_cfg.setdefault("protocols", DEFAULT_SSL["protocols"])
    ssl_cfg.setdefault("ciphers", DEFAULT_SSL["ciphers"])

    tmpl = ENV.get_template("nginx/ssl_snippet.conf")
    content = tmpl.render(ssl=ssl_cfg)
    tmp = SSL_SNIPPET.with_suffix(".tmp")
    with open(tmp, "w") as f:
        f.write(content)
    os.chmod(tmp, 0o644)
    subprocess.run(["sudo", "cp", str(tmp), str(SSL_SNIPPET)], check=True)
    subprocess.run(["sudo", "chown", "root:root", str(SSL_SNIPPET)], check=True)
    tmp.unlink(missing_ok=True)


# ------------------------------------------------------------------
# nginx lifecycle
# ------------------------------------------------------------------


def test_config() -> tuple[bool, str]:
    """Run ``nginx -t`` and return the pass/fail result.

    Returns:
        Tuple of ``(ok, message)`` where ``ok`` is ``True`` if the
        config test passed and ``message`` contains output or a summary.
    """
    result = subprocess.run(
        ["sudo", "nginx", "-t"], capture_output=True, text=True, check=False
    )
    ok = result.returncode == 0
    output = (result.stderr or result.stdout or "").strip()
    if not output and ok:
        output = "nginx configuration test passed"
    if ok:
        logger.info("nginx config test passed")
    else:
        logger.error("nginx config test failed: %s", output)
    return ok, output


def apply() -> None:
    """Generate all configs, test them, and reload nginx.

    Raises:
        RuntimeError: If the nginx config test fails.
    """
    write_ssl_snippet()
    write_all_sites()
    write_include_file()
    ok, msg = test_config()
    if not ok:
        raise RuntimeError(f"nginx config test failed: {msg}")
    result = subprocess.run(
        ["sudo", "nginx", "-s", "reload"], capture_output=True, text=True, check=False
    )
    if result.returncode != 0:
        logger.error("nginx reload failed: %s", result.stderr.strip())
    else:
        logger.info("nginx configuration applied and reloaded")


# ------------------------------------------------------------------
# htpasswd
# ------------------------------------------------------------------


def write_htpasswd(user: str, password: str) -> None:
    """Append (or create) an htpasswd entry for *user*.

    Args:
        user: Username for the htpasswd entry.
        password: Plain-text password to hash and store.
    """
    ensure_dirs(DATA_DIR)
    hashed = _hash_password(password)
    existing: dict[str, str] = {}
    if HTPASSWD_FILE.exists():
        with open(HTPASSWD_FILE) as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith("#"):
                    continue
                parts = line.split(":", 1)
                if len(parts) == 2:
                    existing[parts[0]] = line

    existing[user] = f"{user}:{hashed}"

    tmp = HTPASSWD_FILE.with_suffix(".tmp")
    with open(tmp, "w") as f:
        for _uname, entry in existing.items():
            f.write(entry + "\n")
    os.chmod(tmp, 0o640)
    os.replace(tmp, HTPASSWD_FILE)


__all__ = [
    "WEBUI_BACKEND",
    "_ensure_webui_backend",
    "_migrate_mgmt_domains",
    "_resolve_auth",
    "_resolve_paths",
    "add_domain",
    "apply",
    "generate_server_conf",
    "get_config",
    "get_domains",
    "get_management_domains",
    "remove_domain",
    "save_config",
    "test_config",
    "update_domain",
    "write_acme_challenge",
    "write_all_sites",
    "write_htpasswd",
    "write_include_file",
    "write_site",
    "write_ssl_snippet",
]