mteehan/vacuum-wall
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: unify project structure, improve security, and enhance deployment

Browse Source 
- Fix WireGuard private key leak in API responses and config updates
- Update systemd service to serve from repo root with adjusted sandbox
- Add CLI flags, idempotency, and dev mode to install.sh
- Extract common utilities to lib/common.py and webui/api/common.py
- Migrate frontend to htmx for simpler, more maintainable UI
- Update docs to reflect current architecture and deployment model
- Vendor htmx dependencies per project requirements
This commit is contained in:
Mike Teehan
2026-05-25 00:53:32 +00:00
parent 8829ac579d
commit d1ab717c0f
36 changed files with 857 additions and 626 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/nginx.py
+102 -75
View File
@@ -1,18 +1,20 @@
"""
Nginx server-block generator for Vacuum Wall SSL proxy firewall.
"""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 copy import deepcopy
from pathlib import Path
from typing import Any
from jinja2 import Environment, FileSystemLoader
from lib.common import ensure_dirs, load_json, save_json
logger = logging.getLogger(__name__)
PROJECT_DIR = Path(__file__).resolve().parent.parent
@@ -31,7 +33,7 @@ ENV = Environment(
trim_blocks=True,
)
DEFAULT_SSL = {
DEFAULT_SSL: dict[str, Any] = {
"protocols": "TLSv1.2 TLSv1.3",
"ciphers": (
"ECDHE-ECDSA-AES128-GCM-SHA256:"
@@ -44,63 +46,35 @@ DEFAULT_SSL = {
"prefer_server_ciphers": False,
}
DEFAULT_CONFIG = {
DEFAULT_CONFIG: dict[str, Any] = {
"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 get_config() -> dict[str, Any]:
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)
return raw
def save_config(cfg: dict) -> None:
_json_dump(CONFIG_FILE, cfg)
def save_config(cfg: dict[str, Any]) -> None:
save_json(CONFIG_FILE, cfg)
def get_domains() -> list[dict]:
def get_domains() -> list[dict[str, Any]]:
cfg = get_config()
result = []
result: list[dict[str, Any]] = []
for name, dom in cfg.get("domains", {}).items():
site = SITES_DIR / f"{name}.conf"
result.append(
@@ -120,17 +94,17 @@ def get_domains() -> list[dict]:
def add_domain(
domain,
backend_host,
backend_port,
backend_proto="http",
cert=None,
extra_headers=None,
domain: str,
backend_host: str,
backend_port: int,
backend_proto: str = "http",
cert: str | None = None,
extra_headers: dict[str, str] | None = None,
) -> None:
cfg = get_config()
if domain in cfg["domains"]:
raise ValueError(f"Domain {domain!r} already configured")
entry = {
entry: dict[str, Any] = {
"backend": {
"host": backend_host,
"port": int(backend_port),
@@ -153,7 +127,7 @@ def add_domain(
)
def remove_domain(domain) -> None:
def remove_domain(domain: str) -> None:
cfg = get_config()
cfg["domains"].pop(domain, None)
save_config(cfg)
@@ -163,7 +137,7 @@ def remove_domain(domain) -> None:
logger.info("Proxy domain '%s' removed", domain)
def update_domain(domain, **kwargs) -> None:
def update_domain(domain: str, **kwargs: Any) -> None:
cfg = get_config()
if domain not in cfg["domains"]:
raise KeyError(f"Domain {domain!r} not configured")
@@ -182,7 +156,7 @@ def update_domain(domain, **kwargs) -> None:
# ------------------------------------------------------------------
def generate_server_conf(domain_cfg: dict) -> str:
def generate_server_conf(domain_cfg: dict[str, Any]) -> str:
tmpl = ENV.get_template("nginx/server_block.conf")
return tmpl.render(
domain=domain_cfg["domain"],
@@ -194,10 +168,11 @@ def generate_server_conf(domain_cfg: dict) -> str:
is_management=False,
acme_home=str(PROJECT_DIR / "data" / "acme"),
certs_dir=str(PROJECT_DIR / "data" / "certs"),
acme_webroot=str(PROJECT_DIR / "data" / "acme" / "www"),
)
def _generate_management_conf(management: dict) -> str:
def _generate_management_conf(management: dict[str, Any]) -> str:
tmpl = ENV.get_template("nginx/server_block.conf")
return tmpl.render(
domain=management.get("domain"),
@@ -211,6 +186,7 @@ def _generate_management_conf(management: dict) -> str:
is_management=True,
acme_home=str(PROJECT_DIR / "data" / "acme"),
certs_dir=str(PROJECT_DIR / "data" / "certs"),
acme_webroot=str(PROJECT_DIR / "data" / "acme" / "www"),
)
@@ -219,8 +195,8 @@ def _generate_management_conf(management: dict) -> str:
# ------------------------------------------------------------------
def write_site(domain, conf_text) -> None:
_ensure_dirs()
def write_site(domain: str, conf_text: str) -> None:
ensure_dirs(SITES_DIR)
path = SITES_DIR / f"{domain}.conf"
tmp = path.with_suffix(".tmp")
with open(tmp, "w") as f:
@@ -230,13 +206,32 @@ def write_site(domain, conf_text) -> None:
os.replace(tmp, path)
def write_acme_challenge() -> None:
"""Write the ACME HTTP-01 challenge catch-all nginx config.
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:
_ensure_dirs()
ensure_dirs(SITES_DIR)
cfg = get_config()
existing = set(SITES_DIR.iterdir()) if SITES_DIR.exists() else set()
written = 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)
@@ -252,6 +247,7 @@ def write_all_sites() -> None:
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))
@@ -262,15 +258,15 @@ def write_include_file() -> None:
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)
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:
cfg = get_config()
ssl_cfg = cfg.get("ssl", DEFAULT_SSL.copy())
ssl_cfg.setdefault("prefer_server_ciphers", False)
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"])
@@ -280,8 +276,8 @@ def write_ssl_snippet() -> None:
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)
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)
@@ -291,7 +287,9 @@ def write_ssl_snippet() -> None:
def test_config() -> tuple[bool, str]:
result = _run(["sudo", "nginx", "-t"])
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:
@@ -310,8 +308,13 @@ def apply() -> None:
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")
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")
# ------------------------------------------------------------------
@@ -320,10 +323,14 @@ def apply() -> None:
def set_management_proxy(
domain, flask_host="127.0.0.1", flask_port=9090, auth_user=None, auth_pass=None
domain: str,
flask_host: str = "127.0.0.1",
flask_port: int = 9090,
auth_user: str | None = None,
auth_pass: str | None = None,
) -> None:
cfg = get_config()
entry = {
entry: dict[str, Any] = {
"domain": domain,
"backend": {
"host": flask_host,
@@ -348,11 +355,11 @@ def set_management_proxy(
# ------------------------------------------------------------------
def write_htpasswd(user, password) -> None:
def write_htpasswd(user: str, password: str) -> None:
"""Append (or create) an htpasswd entry for *user*."""
_ensure_dirs()
ensure_dirs(DATA_DIR)
hashed = _hash_password(password)
existing = {}
existing: dict[str, str] = {}
if HTPASSWD_FILE.exists():
with open(HTPASSWD_FILE) as f:
for line in f:
@@ -373,7 +380,7 @@ def write_htpasswd(user, password) -> None:
os.replace(tmp, HTPASSWD_FILE)
def _hash_password(password):
def _hash_password(password: str) -> str:
try:
from passlib.hash import apache_passwd
@@ -383,3 +390,23 @@ def _hash_password(password):
salt = os.urandom(16).hex()[:16]
return _crypt.crypt(password, f"$5${salt}")
__all__ = [
"add_domain",
"apply",
"generate_server_conf",
"get_config",
"get_domains",
"remove_domain",
"save_config",
"set_management_proxy",
"test_config",
"update_domain",
"write_acme_challenge",
"write_all_sites",
"write_htpasswd",
"write_include_file",
"write_site",
"write_ssl_snippet",
]