mteehan/vacuum-wall
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
master
vacuum-wall/lib/nginx.py
T
mteehan fc478a016e nginx: serve mgmt /static/ assets from disk (no Flask round-trip) 
Generated management-domain server blocks now include a
location /static/ aliasing webui/static/ with Cache-Control: no-cache
(ETag revalidation -> 304), nosniff, and a restrictive CSP, so SPA
asset requests no longer reach Flask. Flask's static route remains
the dev-mode fallback.

- lib/nginx.py + daemon/handlers/nginx.py: static_root render context
  (the handler renders the template directly, so both paths need it)
- template: alias uses a trailing slash (nginx concatenates the
  location remainder onto the alias value)
- install.sh: a+rX on webui/static plus execute-up-the-parent-chain
  so the nginx worker (www-data) can traverse repo-in-$HOME installs
- tests: mgmt static location assertions (positive + non-mgmt negative)
- docs: AGENTS.md, architecture.md, security.md static-serving notes
2026-09-03 14:58:57 +00:00

646 lines
21 KiB
Python
Raw Permalink Blame History

"""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 (pure read, in-memory migration).
Never writes or creates directories. Returns the in-memory default when
the file is missing and applies legacy-format migration in memory, so
read paths (state collectors, apply-time checks) stay side-effect free.
The one-shot on-disk migration runs at daemon startup via
``migrate_config_file``.
Returns:
The complete config dict with ``backends``, ``domains``, and ``ssl`` keys.
"""
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"] = {}
return _migrate_config(raw)
def migrate_config_file() -> bool:
"""Persist the one-shot legacy-format migration, if the file needs it.
Runs at daemon startup so ``get_config`` reads stay pure. Rewrites the
on-disk file only when migration actually changes it.
Returns:
True when the on-disk file was rewritten, False otherwise.
"""
raw = load_json(CONFIG_FILE)
if not raw:
return False
pre = deepcopy(raw)
raw = _migrate_config(raw)
if raw != pre:
save_config(raw)
return True
return False
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,
static_root=str(PROJECT_DIR / "webui" / "static"),
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",
"migrate_config_file",
"remove_domain",
"save_config",
"test_config",
"update_domain",
"write_acme_challenge",
"write_all_sites",
"write_htpasswd",
"write_include_file",
"write_site",
"write_ssl_snippet",
]
Reference in New Issue View Git Blame Copy Permalink