mteehan
835326311b
Replace the legacy top-level management key with a unified paths-based model. Each domain now contains a paths map where each entry defines its own backend, auth, headers, and flags (is_management, is_websocket). - Add _migrate_config() to auto-migrate legacy formats on first load - Remove set_management_proxy() and POST_NGINX_MANAGEMENT endpoint - Update server_block.conf template to iterate paths with per-location auth - Update daemon handler, API blueprint, state collector, and install script - Add server config generation tests for paths, WebSocket, auth inheritance - Update frontend proxy page to display per-path rows with flags
567 lines
18 KiB
Python
567 lines
18 KiB
Python
"""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 ``<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 (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",
|
|
]
|