mteehan/vacuum-wall
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
739253b2e59883fd64c1d682fb0d661dca4154af
vacuum-wall/daemon/handlers/nginx.py
T

764 lines
26 KiB
Python
Raw Blame History

"""Nginx daemon handler."""
import logging
import os
import subprocess
from copy import deepcopy
from pathlib import Path
from typing import Any
from jinja2 import Environment, FileSystemLoader
from daemon.iface import (
DELETE_NGINX_BACKENDS_REMOVE,
DELETE_NGINX_DOMAINS_REMOVE,
GET_NGINX_BACKENDS,
GET_NGINX_CONFIG,
GET_NGINX_DOMAINS,
PATCH_NGINX_BACKENDS,
PATCH_NGINX_CONFIG,
POST_NGINX_APPLY,
POST_NGINX_BACKENDS_ADD,
POST_NGINX_CONFIG,
POST_NGINX_DOMAINS_ADD,
POST_NGINX_DOMAINS_UPDATE,
POST_NGINX_RELOAD,
POST_NGINX_SSL_APPLY,
POST_NGINX_TEST,
)
from daemon.server import ConflictError, NotFoundError, refresh_state, registry
from lib.acme import find_cert_dir
from lib.common import (
_APPLY_HASH_KEY,
config_hash,
deep_merge,
ensure_dirs,
load_json,
run,
run_proc,
save_json,
)
from lib.nginx import DEFAULT_SSL, WEBUI_BACKEND
from lib.nginx import _resolve_auth as _ngx_resolve_auth
from lib.nginx import _resolve_paths as _ngx_resolve_paths
logger = logging.getLogger(__name__)
PROJECT_DIR = Path(__file__).resolve().parent.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_CONFIG: dict[str, Any] = {
"backends": {},
"domains": {},
"ssl": {**DEFAULT_SSL},
}
# ---------------------------------------------------------------------------
# Migration
def _migrate_config(raw: dict[str, Any]) -> tuple[dict[str, Any], bool]:
"""Migrate legacy config to backends model. Returns (config, changed)."""
c1 = _ensure_webui_backend(raw)
c2 = _migrate_mgmt_domains(raw)
return raw, c1 or c2
def _ensure_webui_backend(raw: dict[str, Any]) -> bool:
"""Create builtin webui backend if not yet migrated. Returns True if changed."""
backends = raw.setdefault("backends", {})
webui = backends.get("webui")
if webui and webui.get("_migrated"):
return False
backends["webui"] = deepcopy(WEBUI_BACKEND)
backends["webui"]["_migrated"] = True
# Harvest auth from legacy path-level auth if present
for dom in raw.get("domains", {}).values():
paths = dom.get("paths", {})
root = paths.get("/", {})
if root.get("auth"):
backends["webui"]["auth"] = root["auth"]
break
return True
def _migrate_mgmt_domains(raw: dict[str, Any]) -> bool:
"""Migrate legacy mgmt domains to backend refs. Returns True if anything changed."""
backends = raw.get("backends", {})
if not backends.get("webui", {}).get("_migrated"):
return False
domains = raw.setdefault("domains", {})
changed = False
for _name, dom in list(domains.items()):
if dom.get("backend") == "webui":
continue
if dom.get("application") == "webui":
del dom["application"]
changed = True
paths = dom.get("paths", {})
root = paths.get("/", {})
ws = paths.get("/ws", {})
root_backend = root.get("backend", {})
ws_backend = ws.get("backend", {})
is_mgmt_root = root.get("is_management") or (
root_backend.get("host") == "127.0.0.1" and root_backend.get("port") == 9090
)
is_mgmt_ws = ws.get("is_websocket") or (
ws_backend.get("host") == "127.0.0.1" and ws_backend.get("port") == 9091
)
if is_mgmt_root and is_mgmt_ws:
dom["backend"] = "webui"
dom.pop("paths", None)
dom.pop("auth", None)
changed = True
return changed
# ---------------------------------------------------------------------------
# Config helpers
def _get_state() -> dict[str, Any] | None:
"""Retrieve cached nginx state from the state store."""
from lib.state import state as state_store
return state_store.get("nginx")
def _get_config() -> dict[str, Any]:
"""Load the nginx config JSON, applying defaults and migrations."""
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)
cfg, changed = _migrate_config(raw)
if changed:
_save_config(cfg)
return cfg
def _save_config(cfg: dict[str, Any]) -> None:
"""Persist the nginx config dict to disk."""
save_json(CONFIG_FILE, cfg)
# ---------------------------------------------------------------------------
# Backend CRUD
def _get_backends() -> dict[str, Any]:
"""Return the backends dict from config, creating builtin webui if needed."""
cfg = _get_config()
backends = cfg.get("backends", {})
webui = backends.get("webui")
if not webui or not webui.get("_migrated"):
backends["webui"] = deepcopy(WEBUI_BACKEND)
backends["webui"]["_migrated"] = True
cfg["backends"] = backends
_save_config(cfg)
return backends
def _validate_paths(paths: dict) -> None:
"""Validate each path has backend with host, port, proto."""
for path_str, path_cfg in paths.items():
backend = path_cfg.get("backend")
if not backend:
raise ValueError(f"path {path_str!r} missing 'backend'")
if not isinstance(backend, dict):
raise ValueError(f"path {path_str!r} 'backend' must be a dict")
if not backend.get("host"):
raise ValueError(f"path {path_str!r} 'backend' missing 'host'")
if not backend.get("port"):
raise ValueError(f"path {path_str!r} 'backend' missing 'port'")
if not backend.get("proto"):
raise ValueError(f"path {path_str!r} 'backend' missing 'proto'")
def _add_backend(
name: str, label: str, paths: dict, auth: dict | None = None, builtin: bool = False
) -> None:
"""Add a backend. Validate name uniqueness and path schema."""
_validate_paths(paths)
cfg = _get_config()
backends = cfg.setdefault("backends", {})
if name in backends:
raise ValueError(f"Backend {name!r} already exists")
entry: dict[str, Any] = {"label": label, "paths": paths}
if builtin:
entry["builtin"] = True
if auth is not None:
if "htpasswd" in auth:
htpasswd_path = Path(auth["htpasswd"])
if not htpasswd_path.is_absolute():
auth = {**auth, "htpasswd": str(PROJECT_DIR / htpasswd_path)}
entry["auth"] = auth
backends[name] = entry
_save_config(cfg)
def _update_backend(
name: str,
label: str | None = None,
paths: dict | None = None,
auth: dict | None | bool = None,
) -> None:
"""Update a backend. Cannot edit builtin backends. auth=False removes auth."""
cfg = _get_config()
backends = cfg.setdefault("backends", {})
if name not in backends:
raise KeyError(name)
if backends[name].get("builtin"):
raise ValueError("Cannot modify builtin backend")
entry = backends[name]
if label is not None:
entry["label"] = label
if paths is not None:
_validate_paths(paths)
entry["paths"] = paths
if auth is False or auth is None:
entry.pop("auth", None)
elif isinstance(auth, dict):
if "htpasswd" in auth:
htpasswd_path = Path(auth["htpasswd"])
if not htpasswd_path.is_absolute():
auth = {**auth, "htpasswd": str(PROJECT_DIR / htpasswd_path)}
entry["auth"] = auth
_save_config(cfg)
def _remove_backend(name: str) -> None:
"""Remove a non-builtin backend. Raise ConflictError if domains reference it."""
cfg = _get_config()
backends = cfg.setdefault("backends", {})
if name not in backends:
raise KeyError(name)
if backends[name].get("builtin"):
raise ValueError("Cannot remove builtin backend")
for dom_name, dom in cfg.get("domains", {}).items():
if dom.get("backend") == name:
raise ConflictError(
f"Backend {name!r} is referenced by domain {dom_name!r}"
)
del backends[name]
_save_config(cfg)
# ---------------------------------------------------------------------------
# Site generation
def _generate_server_conf(domain_cfg: dict[str, Any], backends: dict[str, Any]) -> str:
"""Render an nginx server block config from a domain entry via Jinja."""
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 = _ngx_resolve_paths(domain_cfg, backends)
has_management = any(p.get("is_management") for p in paths.values())
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=_ngx_resolve_auth(domain_cfg, backends),
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"),
)
def _write_site(domain: str, conf_text: str) -> None:
"""Atomically write a single site config file into sites-enabled."""
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_include_file() -> None:
"""Write the system include file that references all per-site configs."""
tmpl = ENV.get_template("nginx/include.conf")
content = tmpl.render(sites_glob=str(SITES_DIR / "*.conf"))
tmp = Path("/run/vacuum-wall/include.tmp")
tmp.parent.mkdir(exist_ok=True)
tmp.write_text(content)
os.chmod(tmp, 0o644)
run(["cp", "--", str(tmp), str(INCLUDE_FILE)], sudo=True)
run(["chown", "root:root", str(INCLUDE_FILE)], sudo=True)
tmp.unlink(missing_ok=True)
def _write_ssl_snippet() -> None:
"""Render and install the shared SSL snippet to /etc/nginx/snippets/."""
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 = Path("/run/vacuum-wall/ssl-snippet.tmp")
tmp.parent.mkdir(exist_ok=True)
tmp.write_text(content)
os.chmod(tmp, 0o644)
run(["cp", "--", str(tmp), str(SSL_SNIPPET)], sudo=True)
run(["chown", "root:root", str(SSL_SNIPPET)], sudo=True)
tmp.unlink(missing_ok=True)
def _test_config() -> tuple[bool, str]:
"""Run `nginx -t` to validate the current config."""
result = run_proc(["nginx", "-t"], sudo=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"
return ok, output
def _reload_nginx() -> None:
"""Send SIGHUP to nginx to reload its configuration."""
result = run_proc(["nginx", "-s", "reload"], sudo=True, check=False)
if result.returncode != 0:
logger.error("nginx reload failed: %s", result.stderr.strip())
else:
logger.info("nginx configuration applied and reloaded")
def _ensure_self_signed_cert(domain: str) -> None:
"""Auto-generate a self-signed cert for *domain* if not yet present."""
certs_dir = PROJECT_DIR / "data" / "certs"
certs_dir.mkdir(parents=True, exist_ok=True)
cert_file = certs_dir / f"{domain}.crt"
key_file = certs_dir / f"{domain}.key"
if cert_file.is_file() and key_file.is_file():
return
logger.info("Auto-generating self-signed cert for %s", domain)
subprocess.run(
[
"openssl",
"req",
"-x509",
"-newkey",
"rsa:2048",
"-keyout",
str(key_file),
"-out",
str(cert_file),
"-days",
"365",
"-nodes",
"-subj",
f"/CN={domain}",
],
capture_output=True,
text=True,
check=True,
timeout=30,
)
cert_file.chmod(0o644)
key_file.chmod(0o600)
def _write_all_sites() -> None:
"""Regenerate all site configs and ACME challenge site."""
ensure_dirs(SITES_DIR)
cfg = _get_config()
backends = cfg.get("backends", {})
# Auto-generate self-signed certs for management domains that need them
for name, dom in cfg.get("domains", {}).items():
cert = dom.get("cert")
paths = _ngx_resolve_paths(dom, backends)
has_management = any(p.get("is_management") for p in paths.values())
if has_management and (cert == "selfsigned" or cert is None):
_ensure_self_signed_cert(name)
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")
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()
tmpl = ENV.get_template("nginx/acme-challenge.conf")
acme_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(acme_content)
f.write("\n")
os.chmod(tmp, 0o644)
os.replace(tmp, site)
# ---------------------------------------------------------------------------
# Auth helpers
def _hash_password(password: str) -> str:
"""Hash *password* using SHA-256 crypt via passlib."""
from passlib.hash import sha256_crypt
return sha256_crypt.hash(password)
def _write_htpasswd(
user: str, password: str, htpasswd_path: Path | None = None
) -> None:
"""Add or update a user entry in the .htpasswd file using SHA-256 hashing.
If *htpasswd_path* is not given, defaults to :data:`HTPASSWD_FILE`.
"""
target = htpasswd_path or HTPASSWD_FILE
ensure_dirs(target.parent)
hashed = _hash_password(password)
existing: dict[str, str] = {}
if target.exists():
with open(target) 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 = target.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, target)
def _get_nginx_state() -> dict[str, Any]:
"""Return a shallow copy of the cached nginx state, or empty dict if unset."""
ng = _get_state()
if ng is None:
return {}
return ng
# ---------------------------------------------------------------------------
# Routes
@registry.register(GET_NGINX_CONFIG)
def get_config(_request: Any, _body: Any) -> dict[str, Any]:
"""GET /nginx/config — return current nginx config."""
ng = _get_nginx_state()
if ng:
return ng.get("config", {})
cfg = _get_config()
return {k: v for k, v in cfg.items() if k != _APPLY_HASH_KEY}
@registry.register(POST_NGINX_CONFIG)
def save_config_handler(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
"""POST /nginx/config — replace the entire nginx config and refresh state."""
if not body:
raise ValueError("Request body required")
_save_config(body)
refresh_state(["nginx"])
return {"config_saved": True}
@registry.register(PATCH_NGINX_CONFIG)
def patch_config(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
"""PATCH /nginx/config — deep-merge partial updates into current config."""
if not body:
raise ValueError("Request body required")
current = _get_config()
merged = deep_merge(current, body)
_save_config(merged)
refresh_state(["nginx"])
return {"config_saved": True}
@registry.register(GET_NGINX_DOMAINS)
def get_domains(_request: Any, _body: Any) -> list[dict[str, Any]]:
"""GET /nginx/domains — return the list of configured proxy domains."""
ng = _get_nginx_state()
if ng:
return ng.get("domains", [])
return []
@registry.register(POST_NGINX_DOMAINS_ADD)
def add_domain(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
"""POST /nginx/domains/add — add a new reverse-proxy domain entry."""
if not body:
raise ValueError("Request body required")
domain = body.get("domain", "").strip()
if not domain:
raise ValueError("'domain' is required")
cfg = _get_config()
if domain in cfg["domains"]:
raise ValueError(f"Domain {domain!r} already configured")
backend_name = body.get("backend", "").strip()
if not backend_name:
raise ValueError("'backend' is required")
if backend_name not in cfg.get("backends", {}):
raise ValueError(f"Backend {backend_name!r} not found")
cert = body.get("cert")
force_ssl = body.get("force_ssl", True)
entry: dict[str, Any] = {
"backend": backend_name,
"force_ssl": force_ssl,
}
if cert is not None:
entry["cert"] = cert
auth = body.get("auth")
if auth is not None:
# Write htpasswd file when a password is provided, then store
# only {user, htpasswd path} — never persist the raw password.
if auth.get("user") and auth.get("pass"):
htpasswd_path = auth.get(
"htpasswd", str(PROJECT_DIR / "data" / "nginx" / ".htpasswd")
)
if isinstance(htpasswd_path, str) and not Path(htpasswd_path).is_absolute():
htpasswd_path = PROJECT_DIR / htpasswd_path
_write_htpasswd(auth["user"], auth["pass"], Path(htpasswd_path))
entry["auth"] = {"user": auth["user"], "htpasswd": str(htpasswd_path)}
else:
entry["auth"] = auth
cfg["domains"][domain] = entry
_save_config(cfg)
refresh_state(["nginx"])
return {"domain": domain}
@registry.register(DELETE_NGINX_DOMAINS_REMOVE)
def remove_domain(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
"""DELETE /nginx/domains/remove — remove a domain from the proxy config."""
if not body:
raise ValueError("Request body required")
domain = body.get("domain", "").strip()
if not domain:
raise ValueError("'domain' is required")
cfg = _get_config()
if domain not in cfg["domains"]:
raise NotFoundError(f"Domain {domain!r} not found")
del cfg["domains"][domain]
_save_config(cfg)
site = SITES_DIR / f"{domain}.conf"
if site.exists():
site.unlink()
refresh_state(["nginx"])
return {"domain": domain}
@registry.register(POST_NGINX_DOMAINS_UPDATE)
def update_domain(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
"""POST /nginx/domains/update — patch fields of an existing domain entry."""
if not body:
raise ValueError("Request body required")
domain = body.get("domain", "").strip()
if not domain:
raise ValueError("'domain' is required")
cfg = _get_config()
if domain not in cfg["domains"]:
raise NotFoundError(f"Domain {domain!r} not configured")
entry = cfg["domains"][domain]
new_backend = body.get("backend")
if new_backend:
new_backend = new_backend.strip()
if new_backend not in cfg.get("backends", {}):
raise ValueError(f"Backend {new_backend!r} not found")
entry["backend"] = new_backend
if "cert" in body:
if body["cert"] is None:
entry.pop("cert", None)
else:
entry["cert"] = body["cert"]
if "force_ssl" in body:
entry["force_ssl"] = body["force_ssl"]
if "auth" in body:
if body["auth"] is None:
entry.pop("auth", None)
else:
# Normalize: if auth has `pass`, write htpasswd and store only
# `{user, htpasswd path}` — never persist the raw password.
if body["auth"].get("user") and body["auth"].get("pass"):
htpasswd_path = body["auth"].get(
"htpasswd", str(PROJECT_DIR / "data" / "nginx" / ".htpasswd")
)
if (
isinstance(htpasswd_path, str)
and not Path(htpasswd_path).is_absolute()
):
htpasswd_path = PROJECT_DIR / htpasswd_path
_write_htpasswd(
body["auth"]["user"], body["auth"]["pass"], Path(htpasswd_path)
)
entry["auth"] = {
"user": body["auth"]["user"],
"htpasswd": str(htpasswd_path),
}
else:
entry["auth"] = body["auth"]
_save_config(cfg)
refresh_state(["nginx"])
return {"domain": domain}
@registry.register(GET_NGINX_BACKENDS)
def get_backends(_request: Any, _body: Any) -> dict[str, Any]:
"""GET /nginx/backends — return backends with secrets stripped."""
backends = _get_backends()
result: dict[str, Any] = {}
for name, be in backends.items():
entry = deepcopy(be)
entry.pop("_migrated", None)
if "auth" in entry:
entry["has_auth"] = entry["auth"] is not None
del entry["auth"]
else:
entry["has_auth"] = False
result[name] = entry
return result
@registry.register(PATCH_NGINX_BACKENDS)
def patch_backends(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
"""PATCH /nginx/backends — deep-merge partial updates into a backend entry."""
if not body:
raise ValueError("Request body required")
name = body.get("name", "").strip()
if not name:
raise ValueError("'name' is required")
cfg = _get_config()
backends = cfg.setdefault("backends", {})
if name not in backends:
raise KeyError(name)
if backends[name].get("builtin"):
raise ValueError("Cannot modify builtin backend")
update_data = {k: v for k, v in body.items() if k != "name"}
if "auth" in update_data and (
update_data["auth"] is False or update_data["auth"] is None
):
backends[name].pop("auth", None)
update_data.pop("auth")
backends[name] = deep_merge(backends[name], update_data)
_save_config(cfg)
refresh_state(["nginx"])
return {"backend": name}
@registry.register(POST_NGINX_BACKENDS_ADD)
def add_backend(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
"""POST /nginx/backends/add — add a new backend."""
if not body:
raise ValueError("Request body required")
name = body.get("name", "").strip()
label = body.get("label", "").strip()
paths = body.get("paths")
auth = body.get("auth")
if not name:
raise ValueError("'name' is required")
if not label:
raise ValueError("'label' is required")
if not paths:
raise ValueError("'paths' is required")
_add_backend(name, label, paths, auth=auth if auth else None)
refresh_state(["nginx"])
return {"backend": name}
@registry.register(DELETE_NGINX_BACKENDS_REMOVE)
def remove_backend(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
"""DELETE /nginx/backends/remove — remove a non-builtin backend."""
if not body:
raise ValueError("Request body required")
name = body.get("name", "").strip()
if not name:
raise ValueError("'name' is required")
_remove_backend(name)
refresh_state(["nginx"])
return {"backend": name}
@registry.register(POST_NGINX_APPLY)
def apply(_request: Any, _body: Any) -> dict[str, Any]:
"""POST /nginx/apply — render all configs, test, and reload nginx."""
_write_ssl_snippet()
_write_all_sites()
_write_include_file()
ok, msg = _test_config()
if not ok:
raise RuntimeError(f"nginx config test failed: {msg}")
_reload_nginx()
cfg_after = _get_config()
cfg_after[_APPLY_HASH_KEY] = config_hash(cfg_after)
_save_config(cfg_after)
refresh_state(["nginx"])
return {"applied": True}
@registry.register(POST_NGINX_TEST)
def test(_request: Any, _body: Any) -> dict[str, Any]:
"""POST /nginx/test — dry-run validate the live nginx config without applying."""
valid, output = _test_config()
return {"valid": valid, "output": output}
@registry.register(POST_NGINX_SSL_APPLY)
def ssl_apply(_request: Any, _body: Any) -> dict[str, Any]:
"""POST /nginx/ssl-apply — re-render and install only the SSL snippet."""
_write_ssl_snippet()
refresh_state(["nginx"])
return {"applied": True}
@registry.register(POST_NGINX_RELOAD)
def reload_nginx(_request: Any, _body: Any) -> dict[str, Any]:
"""POST /nginx/reload — trigger an nginx reload (SIGHUP)."""
_reload_nginx()
return {"reloaded": True}
Reference in New Issue View Git Blame Copy Permalink