"""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 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, "auth": { "user": "admin", "htpasswd": str(HTPASSWD_FILE), }, "paths": { "/": { "backend": {"host": "127.0.0.1", "port": 9090, "proto": "http"}, "is_management": True, }, "/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 # Harvest auth from legacy path-level auth if present for dom in raw.get("domains", {}).values(): paths = dom.get("paths", {}) root = paths.get("/", {}) if root.get("auth"): backends["webui"]["auth"] = root["auth"] break def _migrate_mgmt_domains(raw: dict[str, Any]) -> None: """Migrate legacy management domains to backend references.""" 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 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", {}) is_mgmt_root = root.get("is_management") or ( root_backend.get("host") == "127.0.0.1" and root_backend.get("port") == 9090 ) is_mgmt_ws = ws.get("is_websocket") or ( ws_backend.get("host") == "127.0.0.1" and ws_backend.get("port") == 9091 ) 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 # ------------------------------------------------------------------ # 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 ``.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) def _hash_password(password: str) -> str: """Hash *password* using SHA-256 crypt (``$5$`` format) via passlib. Args: password: Plain-text password to hash. Returns: The hashed password string suitable for ``.htpasswd`` (e.g. ``$5$rounds=…$…``). """ from passlib.hash import sha256_crypt return sha256_crypt.hash(password) __all__ = [ "WEBUI_BACKEND", "_ensure_webui_backend", "_migrate_mgmt_domains", "_resolve_auth", "_resolve_paths", "add_domain", "apply", "generate_server_conf", "get_config", "get_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", ]