mteehan/vacuum-wall
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
8829ac579d2fb21a1543afd9a0b1e5e17dc5e78d
vacuum-wall/lib/nginx.py
T
mteehan 37039351be fix htmx refactor route mismatches and remaining TODO items 
- wireguard: POST /peers with JSON encoding (was /add-peer)
- rules: delete by rule_id in URL path (was JSON body); pass rule objects with id from server; add hx-disable to initial render
- nat: port forward delete uses URL path params to match blueprint
- nat: masquerade toggle uses native hx-post/hx-vals (was inline fetch)
- app.js renderers updated to use URL path deletes for rules and forwards
- remove TODO.md
2026-05-17 01:17:12 +00:00

386 lines
11 KiB
Python
Raw Blame History

"""
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 json
import logging
import os
import subprocess
from pathlib import Path
from jinja2 import Environment, FileSystemLoader
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 = {
"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 = {
"domains": {},
"management": None,
"ssl": {**DEFAULT_SSL},
}
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
def _ensure_dirs():
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
SITES_DIR.mkdir(parents=True, exist_ok=True)
def _run(cmd, **kw):
return subprocess.run(cmd, capture_output=True, text=True, check=False, **kw)
def _json_load(path):
_ensure_dirs()
if not path.exists():
return DEFAULT_CONFIG.copy()
with open(path) as f:
data = json.load(f)
if "ssl" not in data:
data["ssl"] = DEFAULT_SSL.copy()
return data
def _json_dump(path, data):
_ensure_dirs()
tmp = path.with_suffix(".tmp")
with open(tmp, "w") as f:
json.dump(data, f, indent=4)
f.write("\n")
os.replace(tmp, path)
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def get_config() -> dict:
return _json_load(CONFIG_FILE)
def save_config(cfg: dict) -> None:
_json_dump(CONFIG_FILE, cfg)
def get_domains() -> list[dict]:
cfg = get_config()
result = []
for name, dom in cfg.get("domains", {}).items():
site = SITES_DIR / f"{name}.conf"
result.append(
{
"domain": name,
"backend": dom.get("backend", {}),
"online": site.exists(),
"force_ssl": dom.get("force_ssl", True),
}
)
return result
# ------------------------------------------------------------------
# Domain CRUD
# ------------------------------------------------------------------
def add_domain(
domain,
backend_host,
backend_port,
backend_proto="http",
cert=None,
extra_headers=None,
) -> None:
cfg = get_config()
if domain in cfg["domains"]:
raise ValueError(f"Domain {domain!r} already configured")
entry = {
"backend": {
"host": backend_host,
"port": int(backend_port),
"proto": backend_proto,
},
"force_ssl": True,
}
if cert is not None:
entry["cert"] = cert
if extra_headers is not None:
entry["headers"] = extra_headers
cfg["domains"][domain] = entry
save_config(cfg)
logger.info(
"Proxy domain '%s' added -> %s:%d (%s)",
domain,
backend_host,
backend_port,
backend_proto,
)
def remove_domain(domain) -> None:
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, **kwargs) -> None:
cfg = get_config()
if domain not in cfg["domains"]:
raise KeyError(f"Domain {domain!r} not configured")
entry = cfg["domains"][domain]
for key, val in kwargs.items():
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:
tmpl = ENV.get_template("nginx/server_block.conf")
return tmpl.render(
domain=domain_cfg["domain"],
backend=domain_cfg.get("backend", {}),
headers=domain_cfg.get("headers", {}),
force_ssl=domain_cfg.get("force_ssl", True),
cert=domain_cfg.get("cert"),
auth=domain_cfg.get("auth"),
is_management=False,
acme_home=str(PROJECT_DIR / "data" / "acme"),
certs_dir=str(PROJECT_DIR / "data" / "certs"),
)
def _generate_management_conf(management: dict) -> str:
tmpl = ENV.get_template("nginx/server_block.conf")
return tmpl.render(
domain=management.get("domain"),
backend=dict(
management.get("backend", {}), host="127.0.0.1", port=9090, proto="http"
),
headers={},
force_ssl=True,
cert=None,
auth=management.get("auth"),
is_management=True,
acme_home=str(PROJECT_DIR / "data" / "acme"),
certs_dir=str(PROJECT_DIR / "data" / "certs"),
)
# ------------------------------------------------------------------
# File writers
# ------------------------------------------------------------------
def write_site(domain, conf_text) -> None:
_ensure_dirs()
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_all_sites() -> None:
_ensure_dirs()
cfg = get_config()
existing = set(SITES_DIR.iterdir()) if SITES_DIR.exists() else set()
written = 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")
if cfg.get("management"):
mgmt_conf = _generate_management_conf(cfg["management"])
write_site("management", mgmt_conf)
written.add("management.conf")
for old in existing:
if old.suffix == ".conf" and old.name not in written:
old.unlink()
logger.info("All nginx site configs written (%d sites)", len(written))
def write_include_file() -> None:
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), INCLUDE_FILE], check=True)
subprocess.run(["sudo", "chown", "root:root", INCLUDE_FILE], check=True)
tmp.unlink(missing_ok=True)
def write_ssl_snippet() -> None:
cfg = get_config()
ssl_cfg = cfg.get("ssl", DEFAULT_SSL.copy())
ssl_cfg.setdefault("prefer_server_ciphers", False)
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), SSL_SNIPPET], check=True)
subprocess.run(["sudo", "chown", "root:root", SSL_SNIPPET], check=True)
tmp.unlink(missing_ok=True)
# ------------------------------------------------------------------
# nginx lifecycle
# ------------------------------------------------------------------
def test_config() -> tuple[bool, str]:
result = _run(["sudo", "nginx", "-t"])
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:
write_ssl_snippet()
write_all_sites()
write_include_file()
ok, msg = test_config()
if not ok:
raise RuntimeError(f"nginx config test failed: {msg}")
_run(["sudo", "nginx", "-s", "reload"])
logger.info("nginx configuration applied and reloaded")
# ------------------------------------------------------------------
# Management WebUI
# ------------------------------------------------------------------
def set_management_proxy(
domain, flask_host="127.0.0.1", flask_port=9090, auth_user=None, auth_pass=None
) -> None:
cfg = get_config()
entry = {
"domain": domain,
"backend": {
"host": flask_host,
"port": int(flask_port),
"proto": "http",
},
}
if auth_user:
entry["auth"] = {
"user": auth_user,
"htpasswd": str(HTPASSWD_FILE),
}
cfg["management"] = entry
save_config(cfg)
if auth_user and auth_pass:
write_htpasswd(auth_user, auth_pass)
logger.info("Management proxy set to '%s'", domain)
# ------------------------------------------------------------------
# htpasswd
# ------------------------------------------------------------------
def write_htpasswd(user, password) -> None:
"""Append (or create) an htpasswd entry for *user*."""
_ensure_dirs()
hashed = _hash_password(password)
existing = {}
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):
try:
from passlib.hash import apache_passwd
return apache_passwd.using(rounds=12).hash(password)
except Exception:
import crypt as _crypt
salt = os.urandom(16).hex()[:16]
return _crypt.crypt(password, f"$5${salt}")
Reference in New Issue View Git Blame Copy Permalink