"""Nginx server-block generator for Vacuum Wall SSL proxy firewall. Manages per-domain SSL reverse proxy configurations, 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, } DEFAULT_CONFIG: dict[str, Any] = { "domains": {}, "ssl": {**DEFAULT_SSL}, } # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def _migrate_config(raw: dict[str, Any]) -> dict[str, Any]: """Migrate legacy config formats to the new paths-based model. Handles two migrations: 1. Legacy ``management`` top-level key -> path entry under its domain. 2. Domain entries without ``paths`` -> wrap ``backend`` inside ``paths["/"]``. Args: raw: Config dict as loaded from disk. Returns: The migrated config dict. """ # Migrate management key if "management" in raw and raw["management"] is not None: mgmt = raw["management"] mgmt_domain = mgmt.get("domain", "") if mgmt_domain: domains = raw.setdefault("domains", {}) if mgmt_domain not in domains: domains[mgmt_domain] = { "force_ssl": True, "paths": {}, } dom = domains[mgmt_domain] paths = dom.setdefault("paths", {}) if "/" not in paths: paths["/"] = { "backend": { "host": mgmt.get("backend", {}).get("host", "127.0.0.1"), "port": mgmt.get("backend", {}).get("port", 9090), "proto": "http", }, "is_management": True, } if mgmt.get("auth"): paths["/"]["auth"] = mgmt["auth"] # Add WebSocket path if not present if "/ws" not in paths: paths["/ws"] = { "backend": {"host": "127.0.0.1", "port": 9091, "proto": "http"}, "is_websocket": True, } del raw["management"] # Migrate domain entries without paths for dom in raw.get("domains", {}).values(): if "paths" not in dom and "backend" in dom: dom["paths"] = { "/": { "backend": dom.pop("backend"), "headers": dom.pop("headers", {}), } } return raw 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 ``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) 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 with domain-level settings repeated. Returns: List of dicts with ``domain``, ``path``, ``backend``, ``online``, ``force_ssl``, and path-level flags. """ cfg = get_config() result: list[dict[str, Any]] = [] for name, dom in cfg.get("domains", {}).items(): site = SITES_DIR / f"{name}.conf" paths = dom.get("paths", {}) if not paths: continue 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), } 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_host: str | None = None, backend_port: int | None = None, backend_proto: str = "http", cert: str | None = None, extra_headers: dict[str, str] | None = None, paths: dict[str, dict[str, Any]] | None = None, ) -> None: """Add a new proxy domain with the given backend and optional settings. Args: domain: Domain name to add. backend_host: Upstream host to proxy to (legacy mode). backend_port: Upstream port (legacy mode). backend_proto: Protocol (``http`` or ``https``; legacy mode). cert: Optional certificate type identifier. extra_headers: Optional dict of extra headers to forward (legacy mode). paths: Optional path-to-config map (new mode). Each path entry must have a ``backend`` key with ``host``, ``port``, and ``proto``. Raises: ValueError: If the domain is already configured. """ cfg = get_config() if domain in cfg["domains"]: raise ValueError(f"Domain {domain!r} already configured") if paths is not None: entry: dict[str, Any] = { "paths": paths, "force_ssl": True, } if cert is not None: entry["cert"] = cert else: if not backend_host or backend_port is None: raise ValueError("'backend_host' and 'backend_port' are required") entry = { "paths": { "/": { "backend": { "host": backend_host, "port": int(backend_port), "proto": backend_proto, }, "headers": extra_headers or {}, } }, "force_ssl": True, } if cert is not None: entry["cert"] = cert 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 fields of an existing domain entry in-place. Supports both domain-level keys (``force_ssl``, ``cert``, ``auth``, ``paths``) and paths-level shorthand (``backend``, ``headers`` for the root path). When ``paths`` is provided, it is fully replaced. When ``backend`` is provided, it updates ``paths["/"]["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. """ cfg = get_config() if domain not in cfg["domains"]: raise KeyError(f"Domain {domain!r} not configured") entry = cfg["domains"][domain] # If paths is given, replace entirely if "paths" in kwargs: entry["paths"] = kwargs["paths"] else: # Legacy: top-level backend/headers -> paths["/"] paths = entry.setdefault("paths", {}) if "backend" in kwargs: root = paths.setdefault("/", {"backend": {}, "headers": {}}) root["backend"] = kwargs["backend"] if "headers" in kwargs: root = paths.setdefault("/", {"backend": {}, "headers": {}}) root["headers"] = kwargs["headers"] # Remove legacy top-level keys from domain entry entry.pop("backend", None) entry.pop("headers", None) for key, val in kwargs.items(): if key in ("backend", "headers", "paths"): continue if isinstance(val, dict) and key in entry: entry[key].update(val) else: entry[key] = val 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]) -> str: """Render the Jinja template for a domain server block. Args: domain_cfg: Domain entry dict including the ``domain`` key. 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)) paths = domain_cfg.get("paths", {}) 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_cfg.get("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 (now unified, including any management paths), removes orphaned site files, and ensures the ACME challenge config is present. """ ensure_dirs(SITES_DIR) cfg = get_config() 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) 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__ = [ "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", ]