Compare commits

...

1 Commits

Author SHA1 Message Date
mteehan 7beba44b4b feat: pre-computed state store and async ACME issuance (fixes timeout mismatch)
- Add lib/state.py: in-memory state store with subsystem collectors
  (firewall, dnsmasq, nginx, acme, wireguard)
- Refactor all handlers: read from state on GET, call refresh_state()
  after mutations instead of invoking subprocesses per request
- daemon/server.py: add refresh_state(), /status/all, /status/refresh;
  populate state at startup
- webui/api/certs.py: async step-by-step ACME issuance (validate,
  issue with request_id, poll status) replacing blocking endpoint
- webui/server.py: render pages from state instead of direct lib calls
- Update templates, JS for async cert issuance with polling UI
- Update tests for state-based mocking; add test_state.py
- Fix SIM105 lint issue (contextlib.suppress)
- Add TODO.md with certificate issuance issue tracking

Resolves: WebUI 30s timeout freeze during cert issuance (Problem 1)
2026-05-30 05:45:40 +00:00
19 changed files with 1957 additions and 986 deletions
+44
View File
@@ -0,0 +1,44 @@
# TODO — Certificate Issuance Issues
## Problem 1: Timeout Mismatch Causes WebUI Freeze → FIXED
**Date Found:** May 30, 2026
**Date Fixed:** May 30, 2026
**Symptom:** Clicking "Issue Certificate" in the WebUI freezes for ~30 seconds, then returns a 500 error with "Daemon request timed out". Meanwhile the daemon silently runs `acme.sh` in the background for up to 120s before timing out itself.
**Root Cause:**
- `daemon/client.py` — WebUI client uses `timeout=30` for all daemon requests
- `daemon/handlers/acme.py` — Daemon allows `acme.sh` subprocess `timeout=120`
- The WebUI gives up at 30s while the daemon is still legitimately processing
**Fix Applied:** Replaced blocking issue endpoint with async step-by-step issuance:
- `POST /acme/validate` — Pre-flight checks (instant): acme.sh installed, email configured, webroot ready, DNS resolves, challenge configured
- `POST /acme/issue` — Returns immediately with `request_id`, spawns background task
- `GET /acme/issue/status` — Client polls for step-by-step progress
- UI shows pre-check results, then step progress with polling (no timeout issues)
- Blocking DNS check prevents wasted acme.sh calls when domain doesn't resolve
**Files Changed:** `daemon/handlers/acme.py`, `webui/api/certs.py`, `webui/templates/certs.html`, `webui/static/app.js`
---
## Problem 2: ZeroSSL Rate Limits Block Certificate Issuance
**Date Found:** May 30, 2026
**Symptom:** `acme.sh` fails to issue a certificate for `218broad.vacuum.network` with:
```
The retryafter=86400 value is too large (> 600), will not retry anymore.
```
**Root Cause:** ZeroSSL CA returns a `retry-after` of 86400 seconds (24 hours), likely from a prior failed challenge. `acme.sh` has a hard cap of 600s on retry-after values and refuses to proceed when the CA requests a longer wait.
**Files:** N/A (acme.sh behavior, not a project code issue)
**Workarounds:**
- Switch CA to Let's Encrypt: `acme.sh --set-default-ca --server letsencrypt`
- Wait 24 hours and retry
- Investigate and clean up prior failed challenges for the domain on ZeroSSL's side
## Status
+386 -119
View File
@@ -1,15 +1,19 @@
"""ACME certificate daemon handler."""
import asyncio
import logging
import os
import re
import shutil
import socket
import subprocess
from contextlib import suppress
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from uuid import uuid4
from daemon.server import NotFoundError, registry
from daemon.server import NotFoundError, refresh_state, registry
logger = logging.getLogger(__name__)
@@ -26,21 +30,56 @@ _ACME_ENVIRON = {
_WEBROOT = PROJECT_DIR / "data" / "acme" / "www"
_ACME_TAGS = {"acme"}
# In-memory store for active issuance requests.
_ISSUANCES: dict[str, "IssueRequest"] = {}
_ISSUANCE_TTL = 300 # seconds to keep completed requests
def _find_acme() -> str:
candidates = [_ACME_HOME / "acme.sh", Path("/usr/local/bin/acme.sh")]
for path in candidates:
if path.is_file() and os.access(path, os.X_OK):
return str(path)
acme = shutil.which("acme.sh")
if acme:
return acme
raise FileNotFoundError("acme.sh not found")
@dataclass
class IssueStep:
name: str
label: str
status: str = "pending"
message: str | None = None
@dataclass
class IssueRequest:
request_id: str
domain: str
email: str | None = None
webroot: str | None = None
steps: list[IssueStep] = field(default_factory=list)
status: str = "running"
created_at: float = field(default_factory=lambda: datetime.now(UTC).timestamp())
expires_at: float | None = None
def to_dict(self) -> dict[str, Any]:
return {
"request_id": self.request_id,
"domain": self.domain,
"status": self.status,
"steps": [
{
"name": s.name,
"label": s.label,
"status": s.status,
"message": s.message,
}
for s in self.steps
],
"created_at": self.created_at,
"expires_at": self.expires_at,
}
# ---------------------------------------------------------------------------
# Internal helpers
def _run_acme(args: list[str]) -> str:
from lib.state import _find_acme
acme_bin = _find_acme()
acme_home_env = os.environ.get("ACME_HOME", str(_ACME_HOME))
cmd = [acme_bin, "--home", acme_home_env, "--config-home", acme_home_env, *args]
@@ -62,106 +101,13 @@ def _run_acme(args: list[str]) -> str:
return output
def _days_until(date_str: str) -> int | None:
if not date_str:
return None
for fmt in ("%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"):
try:
dt = datetime.strptime(date_str, fmt).replace(tzinfo=UTC)
return (dt - datetime.now(UTC)).days
except ValueError:
continue
try:
dt = datetime.strptime(date_str, "%Y%m%d%H%M%z").astimezone(UTC)
return (dt - datetime.now(UTC)).days
except ValueError:
pass
return None
def _find_acme_bin() -> str:
from lib.state import _find_acme
return _find_acme()
def _parse_list_output(raw: str) -> list[dict]:
entries: list[dict] = []
for line in raw.strip().splitlines():
line = line.strip()
if not line:
continue
entry: dict[str, str] = {}
for token in line.split():
if ":" not in token:
continue
key, _, value = token.partition(":")
entry[key.lower()] = value
if entry:
entries.append(entry)
return entries
def _has_auto_renew(domain: str) -> bool:
acme_home_env = os.environ.get("ACME_HOME", str(_ACME_HOME))
return bool(Path(acme_home_env) / f"{domain}.conf")
@registry.register("GET", "/acme/list", cache_tags=_ACME_TAGS)
def list_certs(_request: Any, _body: Any) -> list[dict]:
raw = _run_acme(["--list"])
certs: list[dict] = []
entries = _parse_list_output(raw)
acme_home_env = os.environ.get("ACME_HOME", str(_ACME_HOME))
acme_home = Path(acme_home_env)
for entry in entries:
main = entry["main_domain"]
if not main:
continue
san_domains = [
d.strip() for d in entry.get("san_domain", "").split(",") if d.strip()
]
cert_dir = acme_home / main
days = _days_until(entry.get("certificate_expires", ""))
certs.append(
{
"domain": main,
"issuer": entry.get("CA", ""),
"expiry": entry.get("certificate_expires", ""),
"days_remaining": days,
"expired": days is not None and days <= 0,
"cert_path": str(cert_dir / "fullchain.cer"),
"key_path": str(cert_dir / f"{main}.key"),
"ca_path": str(cert_dir / "ca.cer"),
"issued_at": entry.get("certificate_date", ""),
"expires_at": entry.get("certificate_expires", ""),
"days_until_expiry": days,
"auto_renew": _has_auto_renew(main),
"san_domains": san_domains,
}
)
return certs
@registry.register("GET", "/acme/info", cache_tags=_ACME_TAGS)
def get_cert_info(_request: Any, body: dict[str, Any] | None) -> dict:
if not body or "domain" not in body:
raise ValueError("'domain' is required")
domain = body["domain"]
certs = list_certs(None, None)
for c in certs:
if c["domain"] == domain or domain in c["san_domains"]:
return c
raise NotFoundError(f"No certificate found for domain: {domain}")
@registry.register("POST", "/acme/issue", invalidate=_ACME_TAGS | {"nginx"})
def issue_cert(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
domain = body.get("domain", "").strip()
if not domain:
raise ValueError("'domain' is required")
webroot = body.get("webroot")
email = body.get("email", "").strip() or None
args: list[str] = ["--issue", "-d", domain]
args.extend(["--webroot", webroot or str(_WEBROOT)])
contact = email
if not contact:
def _get_acme_email() -> str:
try:
acme_home = Path(os.environ.get("ACME_HOME", str(_ACME_HOME)))
account_conf = acme_home / "account.conf"
@@ -169,19 +115,334 @@ def issue_cert(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
text = account_conf.read_text()
match = re.search(r"^ACME_LEEMAIL=(.+)$", text, re.MULTILINE)
if match:
contact = match.group(1).strip().strip("'\"")
return match.group(1).strip().strip("'\"")
except OSError:
pass
return ""
def _get_state() -> dict[str, Any] | None:
from lib.state import state as state_store
return state_store.get("acme")
def _get_acme_state() -> dict[str, Any]:
ac = _get_state()
if ac is None:
return {}
return ac
def _clean_expired_issuances() -> None:
"""Remove completed requests older than TTL."""
now = datetime.now(UTC).timestamp()
expired = [
rid
for rid, req in _ISSUANCES.items()
if req.expires_at and now > req.expires_at
]
for rid in expired:
del _ISSUANCES[rid]
# ---------------------------------------------------------------------------
# Validation helpers
def _check_domain_format(domain: str) -> tuple[bool, str]:
"""Validate basic domain name format."""
import re as _re
pattern = r"^[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?(\.[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?)*$"
if not _re.match(pattern, domain):
return False, "Invalid domain name format"
return True, ""
def _check_dns_resolves(domain: str) -> tuple[bool, str]:
"""Check that domain resolves to this machine's IP via A record."""
try:
results = socket.getaddrinfo(domain, 80, socket.AF_UNSPEC, socket.SOCK_STREAM)
if not results:
return False, "Domain does not resolve to any address"
local_ips = set()
hostname = socket.gethostname()
with suppress(OSError):
local_ips.add(socket.gethostbyname(hostname))
# Also collect all interface IPs
try:
import ipaddress
from fcntl import ioctl
def get_interfaces():
import struct
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
names = b"\x00" * 4096
raw = ioctl(s.fileno(), 0x8912, names)
s.close()
ifaces = []
for i in range(0, 4096, 32):
name = raw[i : i + 16].split(b"\x00")[0].decode()
if name == "lo":
continue
addr = struct.unpack("<I", raw[i + 16 : i + 20])[0]
ifaces.append(str(ipaddress.IPv4Address(addr)))
return ifaces
local_ips.update(get_interfaces())
except Exception:
pass
resolved = False
for _, _, _, _, addr in results:
if addr in local_ips:
resolved = True
break
if resolved:
return True, "DNS resolves correctly"
return (
False,
f"Domain resolves to {results[0][4][0]}, not this server",
)
except socket.gaierror:
return False, "Domain does not resolve (NXDOMAIN or timeout)"
def _check_acme_installed() -> tuple[bool, str]:
try:
_find_acme_bin()
return True, "acme.sh found"
except FileNotFoundError:
return False, "acme.sh not installed"
def _check_email_configured() -> tuple[bool, str]:
email = _get_acme_email() or ""
if email:
return True, f"Contact email configured: {email}"
return False, "No ACME contact email configured"
def _check_webroot() -> tuple[bool, str]:
if _WEBROOT.is_dir() and os.access(str(_WEBROOT), os.W_OK):
return True, "ACME webroot ready"
return False, "ACME webroot not ready or not writable"
def _check_challenge_config() -> tuple[bool, str]:
from lib.nginx import SITES_DIR
site_conf = SITES_DIR / "_acme-challenge.conf" if SITES_DIR else None
if site_conf and site_conf.is_file():
return True, "ACME challenge nginx config present"
return False, "ACME challenge nginx config missing"
def _check_existing_cert(domain: str) -> tuple[bool, str]:
"""Warn if a valid cert already exists (not blocking)."""
try:
from lib.acme import days_until_expiry
days = days_until_expiry(domain)
if days is not None and days > 0:
return True, f"Valid certificate exists ({days} days remaining)"
except (ValueError, RuntimeError, FileNotFoundError):
pass
return True, ""
def _validate(domain: str) -> dict[str, Any]:
"""Run all pre-checks for a domain. Returns structured results."""
checks: list[dict[str, Any]] = []
ready = True
check_fns = [
("acme_installed", _check_acme_installed, True),
("email_configured", _check_email_configured, True),
("webroot_ready", _check_webroot, True),
("challenge_configured", _check_challenge_config, True),
("domain_format", lambda: _check_domain_format(domain), True),
("dns_resolves", lambda: _check_dns_resolves(domain), True),
("existing_cert", lambda: _check_existing_cert(domain), False),
]
for name, fn, blocking in check_fns:
try:
passed, msg = fn()
checks.append(
{"name": name, "passed": passed, "message": msg, "blocking": blocking}
)
if not passed and blocking:
ready = False
except Exception as exc:
checks.append(
{
"name": name,
"passed": False,
"message": str(exc),
"blocking": blocking,
}
)
ready = False
return {"domain": domain, "checks": checks, "ready": ready}
# ---------------------------------------------------------------------------
# Routes — status reads from state, mutations call refresh_state
@registry.register("GET", "/acme/list")
def list_certs(_request: Any, _body: Any) -> list[dict]:
ac = _get_acme_state()
return ac.get("certs", [])
@registry.register("GET", "/acme/info")
def get_cert_info(_request: Any, body: dict[str, Any] | None) -> dict:
if not body or "domain" not in body:
raise ValueError("'domain' is required")
domain = body["domain"]
certs = list_certs(None, None)
for c in certs:
if c["domain"] == domain or domain in c.get("san_domains", []):
return c
raise NotFoundError(f"No certificate found for domain: {domain}")
@registry.register("POST", "/acme/validate")
def validate_cert(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
domain = body.get("domain", "").strip()
if not domain:
raise ValueError("'domain' is required")
return _validate(domain)
@registry.register("POST", "/acme/issue")
async def issue_cert(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
domain = body.get("domain", "").strip()
if not domain:
raise ValueError("'domain' is required")
email = body.get("email", "").strip() or None
webroot = body.get("webroot")
_clean_expired_issuances()
# Dedup: if domain already has an active request, return existing ID
for existing in _ISSUANCES.values():
if existing.domain == domain and existing.status == "running":
return {
"request_id": existing.request_id,
"status": "existing",
"domain": domain,
"message": "Issuance already in progress for this domain",
}
# Run pre-flight checks
_validate_checks = _validate(domain)
if not _validate_checks["ready"]:
failed = [
c["name"]
for c in _validate_checks["checks"]
if not c["passed"] and c["blocking"]
]
raise RuntimeError(f"Pre-flight checks failed: {', '.join(failed)}")
# Create tracked request
request_id = uuid4().hex[:12]
steps = [
IssueStep(name="issue", label="Issuing certificate"),
IssueStep(name="deploy", label="Registering deploy hook"),
IssueStep(name="refresh", label="Refreshing certificate state"),
]
req = IssueRequest(
request_id=request_id,
domain=domain,
email=email,
webroot=webroot,
steps=steps,
)
_ISSUANCES[request_id] = req
# Spawn background task
_task = asyncio.create_task(_run_issue(req)) # noqa: RUF006 — task runs to completion on its own
return {"request_id": request_id, "domain": domain}
@registry.register("GET", "/acme/issue/status")
def get_issuance_status(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
request_id = (body or {}).get("id", "").strip()
if not request_id:
raise ValueError("'id' is required")
req = _ISSUANCES.get(request_id)
if not req:
raise NotFoundError(f"Issuance request {request_id} not found")
return req.to_dict()
async def _run_issue(req: IssueRequest) -> None:
"""Background task: run acme.sh steps, update step status."""
try:
# Step 1: issue
req.steps[0].status = "running"
args: list[str] = ["--issue", "-d", req.domain]
args.extend(["--webroot", req.webroot or str(_WEBROOT)])
contact = req.email
if not contact:
contact = _get_acme_email()
if contact:
args.extend(["-m", contact])
args.append("--force")
output = _run_acme(args)
_run_acme(["--deploy", "-d", domain, "--deploy-hook", _DEPLOY_HOOK])
logger.info("Certificate for %s issued", domain)
return {"domain": domain, "output": output.strip()}
req.steps[0].status = "done"
req.steps[0].message = output.strip()[:200]
# Step 2: deploy
req.steps[1].status = "running"
_run_acme(["--deploy", "-d", req.domain, "--deploy-hook", _DEPLOY_HOOK])
req.steps[1].status = "done"
req.steps[1].message = "Deploy hook registered"
# Step 3: refresh state
req.steps[2].status = "running"
refresh_state(["acme"])
req.steps[2].status = "done"
req.steps[2].message = "State refreshed"
req.status = "completed"
req.expires_at = datetime.now(UTC).timestamp() + _ISSUANCE_TTL
logger.info(
"Certificate for %s issued (request %s)", req.domain, req.request_id
)
except Exception as exc:
# Mark current running step as error, overall as failed
for step in req.steps:
if step.status == "running":
step.status = "error"
step.message = str(exc)
break
else:
req.steps.append(
IssueStep(name="error", label="Error", status="error", message=str(exc))
)
req.status = "failed"
req.expires_at = datetime.now(UTC).timestamp() + _ISSUANCE_TTL
logger.error("Cert issuance for %s failed: %s", req.domain, exc)
@registry.register("POST", "/acme/renew", invalidate=_ACME_TAGS | {"nginx"})
@registry.register("POST", "/acme/renew")
def renew_cert(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -195,10 +456,11 @@ def renew_cert(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
output = _run_acme(args)
_run_acme(["--deploy", "-d", domain, "--deploy-hook", _DEPLOY_HOOK])
logger.info("Certificate for %s renewed", domain)
refresh_state(["acme"])
return {"domain": domain, "output": output.strip()}
@registry.register("DELETE", "/acme/remove", invalidate=_ACME_TAGS)
@registry.register("DELETE", "/acme/remove")
def remove_cert(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -207,10 +469,11 @@ def remove_cert(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
raise ValueError("'domain' is required")
_run_acme(["--remove", "-d", domain])
logger.info("Certificate for %s removed", domain)
refresh_state(["acme"])
return {"domain": domain}
@registry.register("POST", "/acme/email", invalidate=_ACME_TAGS)
@registry.register("POST", "/acme/email")
def set_email(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -219,11 +482,15 @@ def set_email(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
raise ValueError("'email' is required")
_run_acme(["--register-account", "-m", email])
logger.info("ACME email set to %s", email)
refresh_state(["acme"])
return {"email": email}
@registry.register("GET", "/acme/email", cache_tags=_ACME_TAGS)
@registry.register("GET", "/acme/email")
def get_email(_request: Any, _body: Any) -> dict[str, Any]:
ac = _get_acme_state()
if ac:
return {"email": ac.get("email", "")}
try:
acme_home = Path(os.environ.get("ACME_HOME", str(_ACME_HOME)))
account_conf = acme_home / "account.conf"
@@ -237,7 +504,7 @@ def get_email(_request: Any, _body: Any) -> dict[str, Any]:
return {"email": ""}
@registry.register("GET", "/acme/paths", cache_tags=_ACME_TAGS)
@registry.register("GET", "/acme/paths")
def get_cert_paths(_request: Any, body: dict[str, Any] | None) -> dict[str, str]:
if not body or "domain" not in body:
raise ValueError("'domain' is required")
+52 -82
View File
@@ -8,7 +8,7 @@ from typing import Any
from jinja2 import Environment, FileSystemLoader
from daemon.server import NotFoundError, registry
from daemon.server import NotFoundError, refresh_state, registry
from lib.common import deep_merge, ensure_dirs, load_json, run, run_proc, save_json
logger = logging.getLogger(__name__)
@@ -33,7 +33,11 @@ DEFAULT_CFG: dict[str, Any] = {
"dns": {"upstreams": ["8.8.8.8", "1.1.1.1"], "domain": None, "custom_records": []},
}
_DNSMASQ_TAGS = {"dnsmasq"}
def _get_state() -> dict[str, Any] | None:
from lib.state import state as state_store
return state_store.get("dnsmasq")
def _get_config() -> dict[str, Any]:
@@ -66,66 +70,46 @@ def _generate_conf(cfg: dict[str, Any]) -> str:
)
def _parse_lease_line(line: str) -> dict[str, Any] | None:
line = line.strip()
if not line or line.startswith("#"):
return None
parts = line.split()
if len(parts) < 3:
return None
try:
ts = datetime.fromtimestamp(int(parts[0]), tz=UTC)
except (ValueError, OSError):
ts = None
return {
"expires_at": ts,
"mac": parts[1],
"ip": parts[2],
"hostname": parts[3] if len(parts) > 3 else "",
"interface": parts[4] if len(parts) > 4 else "",
}
def _get_dnsmasq_state() -> dict[str, Any]:
dm = _get_state()
if dm is None:
return {}
return dm
def _get_lease_table() -> list[dict[str, Any]]:
leases: list[dict[str, Any]] = []
try:
result = run_proc(
["cat", LEASE_FILE],
sudo=True,
check=True,
)
for entry in map(_parse_lease_line, result.stdout.splitlines()):
if entry is not None:
leases.append(entry)
except RuntimeError:
pass
return leases
# ---------------------------------------------------------------------------
# Routes
@registry.register("GET", "/dnsmasq/config", cache_tags=_DNSMASQ_TAGS)
@registry.register("GET", "/dnsmasq/config")
def get_config(_request: Any, _body: Any) -> dict[str, Any]:
dm = _get_dnsmasq_state()
if dm:
return dm.get("config", {})
return _get_config()
@registry.register("POST", "/dnsmasq/config", invalidate=_DNSMASQ_TAGS)
@registry.register("POST", "/dnsmasq/config")
def save_config_handler(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
_save_config(body)
refresh_state(["dnsmasq"])
return {"config_saved": True}
@registry.register("PATCH", "/dnsmasq/config", invalidate=_DNSMASQ_TAGS)
@registry.register("PATCH", "/dnsmasq/config")
def patch_config(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
current = _get_config()
merged = deep_merge(current, body)
_save_config(merged)
refresh_state(["dnsmasq"])
return {"config_saved": True}
@registry.register("POST", "/dnsmasq/apply", invalidate=_DNSMASQ_TAGS)
@registry.register("POST", "/dnsmasq/apply")
def apply_config(_request: Any, _body: Any) -> dict[str, Any]:
cfg = _get_config()
conf_text = _generate_conf(cfg)
@@ -139,44 +123,19 @@ def apply_config(_request: Any, _body: Any) -> dict[str, Any]:
)
run(["systemctl", "reload", "dnsmasq"], sudo=True)
logger.info("dnsmasq config written and reloaded")
refresh_state(["dnsmasq"])
return {"applied": True}
@registry.register("GET", "/dnsmasq/status", cache_tags=_DNSMASQ_TAGS)
@registry.register("GET", "/dnsmasq/status")
def get_status(_request: Any, _body: Any) -> dict[str, Any]:
cfg = _get_config()
try:
proc = run_proc(
["systemctl", "is-active", "dnsmasq"], sudo=True
)
active = proc.stdout.strip() == "active"
except Exception:
active = False
conf_exists = Path(DNSMASQ_CONF).is_file()
conf_on_disk = ""
if conf_exists:
try:
with open(DNSMASQ_CONF) as f:
conf_on_disk = f.read()
except PermissionError:
pass
expected = _generate_conf(cfg)
leases = _get_lease_table()
return {
"service_active": active,
"config_file_exists": conf_exists,
"config_in_sync": conf_on_disk == expected,
"dhcp_ranges": len(cfg["dhcp"]["ranges"]),
"static_leases": len(cfg["dhcp"]["static_leases"]),
"custom_dns_records": len(cfg["dns"]["custom_records"]),
"upstreams": cfg["dns"]["upstreams"],
"domain": cfg["dns"].get("domain"),
"active_leases": len(leases),
"leases": leases,
}
dm = _get_dnsmasq_state()
if dm and "status" in dm:
return dm["status"]
return {}
@registry.register("POST", "/dnsmasq/ranges/add", invalidate=_DNSMASQ_TAGS)
@registry.register("POST", "/dnsmasq/ranges/add")
def set_dhcp_range(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -216,10 +175,11 @@ def set_dhcp_range(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]
entry["dns"] = body["dns"]
ranges.append(entry)
_save_config(cfg)
refresh_state(["dnsmasq"])
return {"interface": iface, "start": start, "end": end}
@registry.register("DELETE", "/dnsmasq/ranges/remove", invalidate=_DNSMASQ_TAGS)
@registry.register("DELETE", "/dnsmasq/ranges/remove")
def remove_dhcp_range(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -245,15 +205,19 @@ def remove_dhcp_range(_request: Any, body: dict[str, Any] | None) -> dict[str, A
f"DHCP range for interface '{iface}' ({start}-{end}) not found"
)
_save_config(cfg)
refresh_state(["dnsmasq"])
return {"interface": iface, "start": start, "end": end}
@registry.register("GET", "/dnsmasq/leases", cache_tags=_DNSMASQ_TAGS)
@registry.register("GET", "/dnsmasq/leases")
def get_leases(_request: Any, _body: Any) -> list[dict[str, Any]]:
return _get_lease_table()
dm = _get_dnsmasq_state()
if dm:
return dm.get("leases", [])
return []
@registry.register("POST", "/dnsmasq/static-lease/add", invalidate=_DNSMASQ_TAGS)
@registry.register("POST", "/dnsmasq/static-lease/add")
def add_static_lease(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -270,16 +234,18 @@ def add_static_lease(_request: Any, body: dict[str, Any] | None) -> dict[str, An
if hostname is not None:
leases[i]["hostname"] = hostname
_save_config(cfg)
refresh_state(["dnsmasq"])
return {"mac": mac, "ip": ip, "hostname": hostname}
entry: dict[str, Any] = {"mac": mac, "ip": ip}
if hostname:
entry["hostname"] = hostname
leases.append(entry)
_save_config(cfg)
refresh_state(["dnsmasq"])
return {"mac": mac, "ip": ip, "hostname": hostname}
@registry.register("DELETE", "/dnsmasq/static-lease/remove", invalidate=_DNSMASQ_TAGS)
@registry.register("DELETE", "/dnsmasq/static-lease/remove")
def remove_static_lease(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -295,10 +261,11 @@ def remove_static_lease(_request: Any, body: dict[str, Any] | None) -> dict[str,
if len(cfg["dhcp"]["static_leases"]) == before:
raise NotFoundError(f"Static lease for MAC '{mac}' not found")
_save_config(cfg)
refresh_state(["dnsmasq"])
return {"mac": mac}
@registry.register("POST", "/dnsmasq/dns-record/add", invalidate=_DNSMASQ_TAGS)
@registry.register("POST", "/dnsmasq/dns-record/add")
def add_dns_record(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -315,16 +282,18 @@ def add_dns_record(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]
if hostname is not None:
records[i]["hostname"] = hostname
_save_config(cfg)
refresh_state(["dnsmasq"])
return {"name": name, "address": address, "hostname": hostname}
entry: dict[str, Any] = {"name": name, "address": address}
if hostname:
entry["hostname"] = hostname
records.append(entry)
_save_config(cfg)
refresh_state(["dnsmasq"])
return {"name": name, "address": address, "hostname": hostname}
@registry.register("DELETE", "/dnsmasq/dns-record/remove", invalidate=_DNSMASQ_TAGS)
@registry.register("DELETE", "/dnsmasq/dns-record/remove")
def remove_dns_record(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -334,26 +303,26 @@ def remove_dns_record(_request: Any, body: dict[str, Any] | None) -> dict[str, A
cfg = _get_config()
records = cfg["dns"]["custom_records"]
before = len(records)
cfg["dns"]["custom_records"] = [
r for r in records if r["name"] != name
]
cfg["dns"]["custom_records"] = [r for r in records if r["name"] != name]
if len(cfg["dns"]["custom_records"]) == before:
raise NotFoundError(f"DNS record '{name}' not found")
_save_config(cfg)
refresh_state(["dnsmasq"])
return {"name": name}
@registry.register("POST", "/dnsmasq/upstreams", invalidate=_DNSMASQ_TAGS)
@registry.register("POST", "/dnsmasq/upstreams")
def set_upstreams(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body or "servers" not in body:
raise ValueError("'servers' is required")
cfg = _get_config()
cfg["dns"]["upstreams"] = list(body["servers"])
_save_config(cfg)
refresh_state(["dnsmasq"])
return {"upstreams": cfg["dns"]["upstreams"]}
@registry.register("POST", "/dnsmasq/domain", invalidate=_DNSMASQ_TAGS)
@registry.register("POST", "/dnsmasq/domain")
def set_domain(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -361,4 +330,5 @@ def set_domain(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
cfg = _get_config()
cfg["dns"]["domain"] = domain if domain else None
_save_config(cfg)
refresh_state(["dnsmasq"])
return {"domain": cfg["dns"]["domain"]}
+106 -217
View File
@@ -1,28 +1,21 @@
"""Firewall daemon handler.
Executes firewall-cmd and ip commands with sudo, returns structured results.
Parsing helpers are imported from lib.firewall.
Reads from the pre-computed state for status endpoints. Executes
firewall-cmd with sudo for mutations. Refers state after each mutation.
"""
import logging
from contextlib import suppress
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from uuid import uuid4
from daemon.server import NotFoundError, registry
from daemon.server import NotFoundError, refresh_state, registry
from lib.common import load_json, run, save_json
from lib.firewall import (
_normalize_target,
_parse_active_zones,
_parse_zone_output,
)
from lib.firewall import (
config_pending as _config_pending,
)
from lib.firewall import (
get_config as _get_lib_config,
)
from lib.firewall import (
save_backup as _save_backup,
)
@@ -30,16 +23,16 @@ from lib.firewall import (
logger = logging.getLogger(__name__)
PROJECT_DIR = Path(__file__).resolve().parent.parent.parent
DATA_DIR = PROJECT_DIR / "data" / "firewall"
RULES_FILE = DATA_DIR / "rules.json"
CONFIG_DIR = PROJECT_DIR / "config" / "firewall"
CONFIG_FILE = CONFIG_DIR / "config.json"
DEFAULT_CONFIG = {"zones": {}}
# ---------------------------------------------------------------------------
# Config helpers
# ---------------------------------------------------------------------------
def _get_state() -> dict[str, Any] | None:
"""Return the current firewall state from the state store."""
from lib.state import state as state_store
return state_store.get("firewall")
def _ensure_config_file() -> None:
@@ -62,10 +55,6 @@ def _reload() -> None:
run(["firewall-cmd", "--reload"], sudo=True)
def _now_iso() -> str:
return datetime.now(UTC).isoformat()
def _fp_to_str(fp: dict[str, Any]) -> str:
parts = [f"port={fp['port']}", f"proto={fp['proto']}"]
if "toaddr" in fp:
@@ -85,92 +74,22 @@ def _get_forward_ports(zone_name: str) -> list[str]:
return []
def _get_state() -> dict[str, Any]:
"""Return the complete current state of firewalld."""
zone_names = run(["firewall-cmd", "--get-zones"], sudo=True).split()
active_raw = run(["firewall-cmd", "--get-active-zones"], sudo=True)
active = _parse_active_zones(active_raw)
services = run(["firewall-cmd", "--get-services"], sudo=True).split() or []
link_out = run(["ip", "-o", "link", "show"], sudo=True)
addr_out = run(["ip", "-o", "addr", "show"], sudo=True)
iface_map: dict[str, dict[str, Any]] = {}
for line in link_out.splitlines():
if not line:
continue
parts = line.split()
if len(parts) < 2:
continue
raw_name = parts[1].rstrip(":")
state = "UNKNOWN"
mtu = None
mac = None
for i, p in enumerate(parts):
if p == "state" and i + 1 < len(parts):
state = parts[i + 1]
if p == "mtu" and i + 1 < len(parts):
mtu = int(parts[i + 1])
if p.startswith("link/ether") and i + 1 < len(parts):
mac = parts[i + 1]
iface_map[raw_name] = {
"name": raw_name,
"display_name": raw_name.partition("@")[0],
"mac": mac,
"state": state,
"mtu": mtu,
"ips": [],
"ipv6": [],
"zone": None,
}
for line in addr_out.splitlines():
if not line:
continue
parts = line.split()
if len(parts) < 4:
continue
addr_name = parts[1]
addr_key = "ipv6" if parts[2] == "inet6" else "ips"
for entry in iface_map.values():
if entry["display_name"] == addr_name:
entry[addr_key].append(parts[3])
break
for zone_name, ifaces in active.items():
for raw_if in ifaces:
clean = raw_if.partition("@")[0]
for entry in iface_map.values():
if entry["display_name"] == clean or entry["name"] == raw_if:
entry["zone"] = zone_name
break
ifaces = list(iface_map.values())
zones: dict[str, dict[str, Any]] = {}
for zn in zone_names:
try:
zones[zn] = _parse_zone_output(
zn, run(["firewall-cmd", f"--zone={zn}", "--list-all"], sudo=True)
)
except Exception:
continue
return {
"active_zones": active,
"interfaces": ifaces,
"available_services": services,
"zones": zones,
"rich_rules": {n: z.get("rich-rules", []) for n, z in zones.items()},
"timestamp": _now_iso(),
}
def _config_apply() -> dict[str, Any]:
"""Apply the declarative config to live firewalld."""
from lib.firewall import get_config as _get_lib_config
cfg = _get_lib_config()
cfg_zones = cfg.get("zones", {})
_save_backup(_get_state())
full_state: dict[str, Any] = {
"active_zones": {},
"interfaces": [],
"available_services": [],
"zones": {},
"rich_rules": {},
"timestamp": "",
}
_save_backup(full_state)
available = run(["firewall-cmd", "--get-zones"], sudo=True).split()
applied: list[str] = []
@@ -314,7 +233,15 @@ def _config_apply() -> dict[str, Any]:
applied.append(zone_name)
_reload()
backup_path = _save_backup(_get_state())
full_state = {
"active_zones": {},
"interfaces": [],
"available_services": [],
"zones": {},
"rich_rules": {},
"timestamp": "",
}
backup_path = _save_backup(full_state)
logger.info("Firewall config applied to %d zones", len(applied))
return {
"applied_zones": applied,
@@ -323,114 +250,67 @@ def _config_apply() -> dict[str, Any]:
# ---------------------------------------------------------------------------
# Routes
# Routes — GET endpoints read from state, mutations call refresh_state
# ---------------------------------------------------------------------------
_READ_TAGS = {"firewall", "interfaces", "zones"}
def _get_fw_state() -> dict[str, Any]:
fw = _get_state()
if fw is None:
return {}
return fw
@registry.register("GET", "/firewall/interfaces", cache_tags=_READ_TAGS)
@registry.register("GET", "/firewall/interfaces")
def get_interfaces(_request: Any, _body: Any) -> list[dict[str, Any]]:
link_out = run(["ip", "-o", "link", "show"], sudo=True)
addr_out = run(["ip", "-o", "addr", "show"], sudo=True)
zones_out = run(["firewall-cmd", "--get-active-zones"], sudo=True)
iface_map: dict[str, dict[str, Any]] = {}
for line in link_out.splitlines():
if not line:
continue
parts = line.split()
if len(parts) < 2:
continue
raw_name = parts[1].rstrip(":")
state = "UNKNOWN"
mtu = None
mac = None
for i, p in enumerate(parts):
if p == "state" and i + 1 < len(parts):
state = parts[i + 1]
if p == "mtu" and i + 1 < len(parts):
mtu = int(parts[i + 1])
if p.startswith("link/ether") and i + 1 < len(parts):
mac = parts[i + 1]
iface_map[raw_name] = {
"name": raw_name,
"display_name": raw_name.partition("@")[0],
"mac": mac,
"state": state,
"mtu": mtu,
"ips": [],
"ipv6": [],
"zone": None,
}
for line in addr_out.splitlines():
if not line:
continue
parts = line.split()
if len(parts) < 4:
continue
addr_name = parts[1]
addr_key = "ipv6" if parts[2] == "inet6" else "ips"
for entry in iface_map.values():
if entry["display_name"] == addr_name:
entry[addr_key].append(parts[3])
break
active = _parse_active_zones(zones_out)
for zone_name, ifaces in active.items():
for raw_if in ifaces:
clean = raw_if.partition("@")[0]
for entry in iface_map.values():
if entry["display_name"] == clean or entry["name"] == raw_if:
entry["zone"] = zone_name
break
return list(iface_map.values())
fw = _get_fw_state()
return fw.get("interfaces", [])
@registry.register("GET", "/firewall/zones", cache_tags=_READ_TAGS)
@registry.register("GET", "/firewall/zones")
def get_zones(_request: Any, _body: Any) -> dict[str, Any]:
available = run(["firewall-cmd", "--get-zones"], sudo=True).split()
active = _parse_active_zones(run(["firewall-cmd", "--get-active-zones"], sudo=True))
return {"active": active, "available": available}
fw = _get_fw_state()
active = fw.get("active_zones", {})
zones = fw.get("zones", {})
return {"active": active, "available": list(zones.keys())}
@registry.register("GET", "/firewall/zones/info", cache_tags=_READ_TAGS)
@registry.register("GET", "/firewall/zones/info")
def get_zone_info(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body or "zone" not in body:
raise ValueError("'zone' is required")
zone = body["zone"]
if zone not in run(["firewall-cmd", "--get-zones"], sudo=True).split():
fw = _get_fw_state()
zones = fw.get("zones", {})
if zone not in zones:
raise NotFoundError(f"Zone '{zone}' does not exist")
raw = run(["firewall-cmd", f"--zone={zone}", "--list-all"], sudo=True)
return _parse_zone_output(zone, raw)
return zones[zone]
@registry.register("GET", "/firewall/zones/all", cache_tags=_READ_TAGS)
@registry.register("GET", "/firewall/zones/all")
def get_all_zones_info(_request: Any, _body: Any) -> list[dict[str, Any]]:
active = _parse_active_zones(run(["firewall-cmd", "--get-active-zones"], sudo=True))
fw = _get_fw_state()
active = fw.get("active_zones", {})
zones = fw.get("zones", {})
result: list[dict[str, Any]] = []
for zone_name in active:
try:
raw = run(["firewall-cmd", f"--zone={zone_name}", "--list-all"], sudo=True)
result.append(_parse_zone_output(zone_name, raw))
except Exception:
continue
if zone_name in zones:
result.append(zones[zone_name])
return result
@registry.register("GET", "/firewall/services", cache_tags=_READ_TAGS)
@registry.register("GET", "/firewall/services")
def get_services(_request: Any, _body: Any) -> list[str]:
return run(["firewall-cmd", "--get-services"], sudo=True).split()
fw = _get_fw_state()
return fw.get("available_services", [])
@registry.register("GET", "/firewall/config", cache_tags=_READ_TAGS)
@registry.register("GET", "/firewall/config")
def get_config(_request: Any, _body: Any) -> dict[str, Any]:
return _get_config()
@registry.register("POST", "/firewall/config", invalidate=_READ_TAGS)
@registry.register("POST", "/firewall/config")
def save_config_handler(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body or "zones" not in body:
raise ValueError("'zones' key is required")
@@ -438,10 +318,11 @@ def save_config_handler(_request: Any, body: dict[str, Any] | None) -> dict[str,
raise ValueError("'zones' must be a dict")
_save_config(body)
logger.info("Firewall config saved (%d zones)", len(body["zones"]))
refresh_state(["firewall"])
return {"config_saved": True}
@registry.register("PATCH", "/firewall/config", invalidate=_READ_TAGS)
@registry.register("PATCH", "/firewall/config")
def patch_config(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body must be a JSON object")
@@ -451,22 +332,25 @@ def patch_config(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
merged = deep_merge(current, body)
_save_config(merged)
logger.info("Firewall config patched: %s", sorted(body.keys()))
refresh_state(["firewall"])
return {"config_saved": True}
@registry.register("GET", "/firewall/config/pending", cache_tags=_READ_TAGS)
def config_pending(_request: Any, _body: Any) -> dict[str, Any]:
return _config_pending(_get_state())
@registry.register("GET", "/firewall/config/pending")
def config_pending_handler(_request: Any, _body: Any) -> dict[str, Any]:
fw = _get_fw_state()
return fw.get("pending", {})
@registry.register("POST", "/firewall/config/apply", invalidate=_READ_TAGS)
@registry.register("POST", "/firewall/config/apply")
def config_apply(_request: Any, _body: Any) -> dict[str, Any]:
result = _config_apply()
logger.info("Firewall config applied: %s", result.get("applied_zones", []))
refresh_state(["firewall"])
return result
@registry.register("POST", "/firewall/zones/create", invalidate=_READ_TAGS)
@registry.register("POST", "/firewall/zones/create")
def create_zone(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -488,10 +372,11 @@ def create_zone(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
)
_reload()
logger.info("Zone '%s' created (target=%s)", zone_name, target)
refresh_state(["firewall"])
return {"zone": zone_name}
@registry.register("DELETE", "/firewall/zones/delete", invalidate=_READ_TAGS)
@registry.register("DELETE", "/firewall/zones/delete")
def delete_zone(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body or "zone" not in body:
raise ValueError("'zone' is required")
@@ -502,10 +387,11 @@ def delete_zone(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
run(["firewall-cmd", f"--zone={zone}", "--delete", "--permanent"], sudo=True)
_reload()
logger.info("Zone '%s' deleted", zone)
refresh_state(["firewall"])
return {"zone": zone}
@registry.register("POST", "/firewall/zones/interfaces", invalidate=_READ_TAGS)
@registry.register("POST", "/firewall/zones/interfaces")
def set_zone_interfaces(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -544,10 +430,11 @@ def set_zone_interfaces(_request: Any, body: dict[str, Any] | None) -> dict[str,
)
_reload()
logger.info("Zone '%s' interfaces set to %s", zone, interfaces)
refresh_state(["firewall"])
return {"zone": zone, "interfaces": interfaces}
@registry.register("POST", "/firewall/zones/services", invalidate=_READ_TAGS)
@registry.register("POST", "/firewall/zones/services")
def set_zone_services(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -582,10 +469,11 @@ def set_zone_services(_request: Any, body: dict[str, Any] | None) -> dict[str, A
sudo=True,
)
_reload()
refresh_state(["firewall"])
return {"zone": zone, "services": services}
@registry.register("POST", "/firewall/rich-rules/add", invalidate=_READ_TAGS)
@registry.register("POST", "/firewall/rich-rules/add")
def add_rich_rule(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -595,7 +483,6 @@ def add_rich_rule(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
raise ValueError("'zone' and 'rule' are required")
if zone not in run(["firewall-cmd", "--get-zones"], sudo=True).split():
raise NotFoundError(f"Zone '{zone}' does not exist")
from uuid import uuid4
run(
[
@@ -613,10 +500,11 @@ def add_rich_rule(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
entry = {"id": rule_id, "rule": rule}
cfg["zones"][zone]["rich_rules"].append(entry)
_save_config(cfg)
refresh_state(["firewall"])
return {"zone": zone, "id": rule_id, "rule": rule}
@registry.register("DELETE", "/firewall/rich-rules/remove", invalidate=_READ_TAGS)
@registry.register("DELETE", "/firewall/rich-rules/remove")
def remove_rich_rule(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -650,34 +538,22 @@ def remove_rich_rule(_request: Any, body: dict[str, Any] | None) -> dict[str, An
r for r in zone_cfg.get("rich_rules", []) if r.get("id") != rule_id
]
_save_config(cfg)
refresh_state(["firewall"])
return {"zone": zone, "id": rule_id}
@registry.register("GET", "/firewall/rich-rules", cache_tags=_READ_TAGS)
@registry.register("GET", "/firewall/rich-rules")
def list_rich_rules(_request: Any, body: dict[str, Any] | None) -> list[dict[str, Any]]:
if not body or "zone" not in body:
raise ValueError("'zone' is required")
zone = body["zone"]
raw = run(["firewall-cmd", f"--zone={zone}", "--list-rich-rules"], sudo=True)
raw = raw.strip()
if not raw:
return []
rules: list[str] = []
current: list[str] = []
for line in raw.splitlines():
r = line.rstrip()
if not r.endswith(";"):
current.append(r)
else:
current.append(r)
rules.append(" ".join(current))
current = []
if current:
rules.append(" ".join(current))
fw = _get_fw_state()
rich_rules = fw.get("rich_rules", {})
cfg = _get_config()
cfg_entries = cfg.get("zones", {}).get(zone, {}).get("rich_rules", [])
result: list[dict[str, Any]] = []
for rule_str in rules:
zone_rules = rich_rules.get(zone, [])
for rule_str in zone_rules:
matched = next((e for e in cfg_entries if e.get("rule") == rule_str), None)
if matched:
result.append({"id": matched["id"], "rule": rule_str})
@@ -686,7 +562,7 @@ def list_rich_rules(_request: Any, body: dict[str, Any] | None) -> list[dict[str
return result
@registry.register("POST", "/firewall/masquerade", invalidate=_READ_TAGS)
@registry.register("POST", "/firewall/masquerade")
def set_masquerade(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -697,10 +573,11 @@ def set_masquerade(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]
action = "--add-masquerade" if enable else "--remove-masquerade"
run(["firewall-cmd", f"--zone={zone}", action, "--permanent"], sudo=True)
_reload()
refresh_state(["firewall"])
return {"zone": zone, "masquerade": bool(enable)}
@registry.register("POST", "/firewall/forward-port/add", invalidate=_READ_TAGS)
@registry.register("POST", "/firewall/forward-port/add")
def add_forward_port(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -711,7 +588,6 @@ def add_forward_port(_request: Any, body: dict[str, Any] | None) -> dict[str, An
toport = body.get("toport")
if not zone or port is None or not proto:
raise ValueError("'zone', 'port', and 'proto' are required")
from uuid import uuid4
fwd = f"port={port}/proto={proto}"
if toaddr and toport:
@@ -740,10 +616,11 @@ def add_forward_port(_request: Any, body: dict[str, Any] | None) -> dict[str, An
cfg.setdefault("zones", {}).setdefault(zone, {}).setdefault("forward_ports", [])
cfg["zones"][zone]["forward_ports"].append(entry)
_save_config(cfg)
refresh_state(["firewall"])
return {"zone": zone, "id": fp_id, "port": int(port), "proto": proto}
@registry.register("DELETE", "/firewall/forward-port/remove", invalidate=_READ_TAGS)
@registry.register("DELETE", "/firewall/forward-port/remove")
def remove_forward_port(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -752,7 +629,8 @@ def remove_forward_port(_request: Any, body: dict[str, Any] | None) -> dict[str,
proto = body.get("proto", "").strip()
if not zone or port is None or not proto:
raise ValueError("'zone', 'port', and 'proto' are required")
if zone not in run(["firewall-cmd", "--get-zones"], sudo=True).split():
available = run(["firewall-cmd", "--get-zones"], sudo=True).split()
if zone not in available:
raise NotFoundError(f"Zone '{zone}' does not exist")
fwd = f"port={port}/proto={proto}"
cfg = _get_config()
@@ -785,9 +663,20 @@ def remove_forward_port(_request: Any, body: dict[str, Any] | None) -> dict[str,
fp for fp in fps if not (fp.get("port") == port and fp.get("proto") == proto)
]
_save_config(cfg)
refresh_state(["firewall"])
return {"zone": zone, "port": int(port), "proto": proto}
@registry.register("GET", "/firewall/state", cache_tags=_READ_TAGS)
@registry.register("GET", "/firewall/state")
def get_state(_request: Any, _body: Any) -> dict[str, Any]:
return _get_state()
fw = _get_state()
if fw is None:
return {}
return {
"active_zones": fw.get("active_zones", {}),
"interfaces": fw.get("interfaces", []),
"available_services": fw.get("available_services", []),
"zones": fw.get("zones", {}),
"rich_rules": fw.get("rich_rules", {}),
"timestamp": fw.get("timestamp", ""),
}
+8 -10
View File
@@ -12,11 +12,9 @@ from lib.common import run_proc
logger = logging.getLogger(__name__)
PROJECT_DIR = Path(__file__).resolve().parent.parent.parent
APP_LOG_FILE = PROJECT_DIR / "data" / "logs" / "vacuum-wall.log"
_APP_LOG_FILE = PROJECT_DIR / "data" / "logs" / "vacuum-wall.log"
_MAX_LINES = 200
_LOG_TAGS = {"logs"}
def _tail_file(path: str, n: int = _MAX_LINES, sudo: bool = False) -> str:
try:
@@ -36,7 +34,7 @@ def _tail_file(path: str, n: int = _MAX_LINES, sudo: bool = False) -> str:
def _sudo_journalctl(unit: str, n: int = _MAX_LINES) -> str:
try:
result = run_proc(
["journalctl", "-u", unit, "--no-pager", "-n", str(n)],
["journalctl", "--unit=" + unit, "-n", str(n)],
sudo=True,
check=False,
timeout=10,
@@ -47,26 +45,26 @@ def _sudo_journalctl(unit: str, n: int = _MAX_LINES) -> str:
return f"(error reading journal: {exc})\n"
@registry.register("GET", "/logs/journal", cache_tags=_LOG_TAGS)
@registry.register("GET", "/logs/journal")
def journal(_request, _body) -> str:
return _sudo_journalctl("vacuum-wall")
@registry.register("GET", "/logs/nginx/access", cache_tags=_LOG_TAGS)
@registry.register("GET", "/logs/nginx/access")
def nginx_access(_request, _body) -> str:
return _tail_file("/var/log/nginx/access.log", sudo=True)
@registry.register("GET", "/logs/nginx/error", cache_tags=_LOG_TAGS)
@registry.register("GET", "/logs/nginx/error")
def nginx_error(_request, _body) -> str:
return _tail_file("/var/log/nginx/error.log", sudo=True)
@registry.register("GET", "/logs/dnsmasq", cache_tags=_LOG_TAGS)
@registry.register("GET", "/logs/dnsmasq")
def dnsmasq_log(_request, _body) -> str:
return _sudo_journalctl("dnsmasq")
@registry.register("GET", "/logs/app", cache_tags=_LOG_TAGS)
@registry.register("GET", "/logs/app")
def app_log(_request, _body) -> str:
return _tail_file(str(APP_LOG_FILE))
return _tail_file(str(_APP_LOG_FILE))
+45 -32
View File
@@ -8,7 +8,7 @@ from typing import Any
from jinja2 import Environment, FileSystemLoader
from daemon.server import NotFoundError, registry
from daemon.server import NotFoundError, refresh_state, registry
from lib.common import ensure_dirs, load_json, run, run_proc, save_json
logger = logging.getLogger(__name__)
@@ -48,7 +48,11 @@ DEFAULT_CONFIG: dict[str, Any] = {
"ssl": {**DEFAULT_SSL},
}
_NGINX_TAGS = {"nginx"}
def _get_state() -> dict[str, Any] | None:
from lib.state import state as state_store
return state_store.get("nginx")
def _get_config() -> dict[str, Any]:
@@ -122,9 +126,7 @@ def _write_ssl_snippet() -> None:
def _test_config() -> tuple[bool, str]:
result = run_proc(
["nginx", "-t"], sudo=True, check=False
)
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:
@@ -133,9 +135,7 @@ def _test_config() -> tuple[bool, str]:
def _reload_nginx() -> None:
result = run_proc(
["nginx", "-s", "reload"], sudo=True, check=False
)
result = run_proc(["nginx", "-s", "reload"], sudo=True, check=False)
if result.returncode != 0:
logger.error("nginx reload failed: %s", result.stderr.strip())
else:
@@ -210,20 +210,35 @@ def _write_htpasswd(user: str, password: str) -> None:
os.replace(tmp, HTPASSWD_FILE)
@registry.register("GET", "/nginx/config", cache_tags=_NGINX_TAGS)
def _get_nginx_state() -> dict[str, Any]:
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]:
ng = _get_nginx_state()
if ng:
return ng.get("config", {})
return _get_config()
@registry.register("POST", "/nginx/config", invalidate=_NGINX_TAGS)
@registry.register("POST", "/nginx/config")
def save_config_handler(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
_save_config(body)
refresh_state(["nginx"])
return {"config_saved": True}
@registry.register("PATCH", "/nginx/config", invalidate=_NGINX_TAGS)
@registry.register("PATCH", "/nginx/config")
def patch_config(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -232,27 +247,19 @@ def patch_config(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
current = _get_config()
merged = deep_merge(current, body)
_save_config(merged)
refresh_state(["nginx"])
return {"config_saved": True}
@registry.register("GET", "/nginx/domains", cache_tags=_NGINX_TAGS)
@registry.register("GET", "/nginx/domains")
def get_domains(_request: Any, _body: Any) -> list[dict[str, Any]]:
cfg = _get_config()
result: list[dict[str, Any]] = []
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
ng = _get_nginx_state()
if ng:
return ng.get("domains", [])
return []
@registry.register("POST", "/nginx/domains/add", invalidate=_NGINX_TAGS)
@registry.register("POST", "/nginx/domains/add")
def add_domain(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -285,10 +292,11 @@ def add_domain(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
entry["headers"] = extra_headers
cfg["domains"][domain] = entry
_save_config(cfg)
refresh_state(["nginx"])
return {"domain": domain}
@registry.register("DELETE", "/nginx/domains/remove", invalidate=_NGINX_TAGS)
@registry.register("DELETE", "/nginx/domains/remove")
def remove_domain(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -303,10 +311,11 @@ def remove_domain(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
site = SITES_DIR / f"{domain}.conf"
if site.exists():
site.unlink()
refresh_state(["nginx"])
return {"domain": domain}
@registry.register("POST", "/nginx/domains/update", invalidate=_NGINX_TAGS)
@registry.register("POST", "/nginx/domains/update")
def update_domain(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -324,10 +333,11 @@ def update_domain(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
else:
entry[key] = val
_save_config(cfg)
refresh_state(["nginx"])
return {"domain": domain}
@registry.register("POST", "/nginx/apply", invalidate=_NGINX_TAGS)
@registry.register("POST", "/nginx/apply")
def apply(_request: Any, _body: Any) -> dict[str, Any]:
_write_ssl_snippet()
_write_all_sites()
@@ -336,22 +346,24 @@ def apply(_request: Any, _body: Any) -> dict[str, Any]:
if not ok:
raise RuntimeError(f"nginx config test failed: {msg}")
_reload_nginx()
refresh_state(["nginx"])
return {"applied": True}
@registry.register("POST", "/nginx/test", invalidate=_NGINX_TAGS)
@registry.register("POST", "/nginx/test")
def test(_request: Any, _body: Any) -> dict[str, Any]:
valid, output = _test_config()
return {"valid": valid, "output": output}
@registry.register("POST", "/nginx/ssl-apply", invalidate=_NGINX_TAGS)
@registry.register("POST", "/nginx/ssl-apply")
def ssl_apply(_request: Any, _body: Any) -> dict[str, Any]:
_write_ssl_snippet()
refresh_state(["nginx"])
return {"applied": True}
@registry.register("POST", "/nginx/management", invalidate=_NGINX_TAGS)
@registry.register("POST", "/nginx/management")
def set_management_proxy(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -373,6 +385,7 @@ def set_management_proxy(_request: Any, body: dict[str, Any] | None) -> dict[str
_save_config(cfg)
if auth_user and auth_pass:
_write_htpasswd(auth_user, auth_pass)
refresh_state(["nginx"])
return {"domain": domain}
+49 -82
View File
@@ -9,7 +9,7 @@ from typing import Any
from jinja2 import Environment, FileSystemLoader
from daemon.server import NotFoundError, registry
from daemon.server import NotFoundError, refresh_state, registry
from lib.common import deep_merge, load_json, run, run_proc, save_json
logger = logging.getLogger(__name__)
@@ -40,7 +40,11 @@ DEFAULT_CONFIG: dict[str, Any] = {
"peers": {},
}
_WG_TAGS = {"wireguard"}
def _get_state() -> dict[str, Any] | None:
from lib.state import state as state_store
return state_store.get("wireguard")
def _get_config() -> dict[str, Any]:
@@ -60,8 +64,22 @@ def _generate_conf(cfg: dict[str, Any]) -> str:
)
@registry.register("GET", "/wireguard/config", cache_tags=_WG_TAGS)
def _get_wg_state() -> dict[str, Any]:
wg = _get_state()
if wg is None:
return {}
return wg
# ---------------------------------------------------------------------------
# Routes
@registry.register("GET", "/wireguard/config")
def get_config(_request: Any, _body: Any) -> dict[str, Any]:
wg = _get_wg_state()
if wg:
return wg.get("config", {})
cfg = _get_config()
safe = dict(cfg)
if "interface" in safe:
@@ -70,7 +88,7 @@ def get_config(_request: Any, _body: Any) -> dict[str, Any]:
return safe
@registry.register("POST", "/wireguard/config", invalidate=_WG_TAGS)
@registry.register("POST", "/wireguard/config")
def save_config_handler(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -83,10 +101,11 @@ def save_config_handler(_request: Any, body: dict[str, Any] | None) -> dict[str,
if current_key:
body.setdefault("interface", {})["private_key"] = current_key
_save_config(body)
refresh_state(["wireguard"])
return {"config_saved": True}
@registry.register("PATCH", "/wireguard/config", invalidate=_WG_TAGS)
@registry.register("PATCH", "/wireguard/config")
def patch_config(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -97,10 +116,11 @@ def patch_config(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
current = _get_config()
merged = deep_merge(current, body)
_save_config(merged)
refresh_state(["wireguard"])
return {"config_saved": True}
@registry.register("POST", "/wireguard/apply", invalidate=_WG_TAGS)
@registry.register("POST", "/wireguard/apply")
def apply(_request: Any, _body: Any) -> dict[str, Any]:
cfg = _get_config()
conf_text = _generate_conf(cfg)
@@ -116,90 +136,29 @@ def apply(_request: Any, _body: Any) -> dict[str, Any]:
local_tmp.unlink(missing_ok=True)
run([WG_QUICK_BIN, "up", cfg["interface"]["name"]], sudo=True)
logger.info("WireGuard tunnel '%s' brought up", cfg["interface"]["name"])
refresh_state(["wireguard"])
return {"applied": True}
@registry.register("POST", "/wireguard/down", invalidate=_WG_TAGS)
@registry.register("POST", "/wireguard/down")
def down(_request: Any, _body: Any) -> dict[str, Any]:
cfg = _get_config()
name = cfg["interface"]["name"]
run([WG_QUICK_BIN, "down", name], sudo=True)
logger.info("WireGuard tunnel '%s' brought down", name)
refresh_state(["wireguard"])
return {"down": True}
@registry.register("GET", "/wireguard/status", cache_tags=_WG_TAGS)
@registry.register("GET", "/wireguard/status")
def status(_request: Any, _body: Any) -> dict[str, Any]:
cfg = _get_config()
name = cfg["interface"]["name"]
result: dict[str, Any] = {"up": False, "interface": {}, "peers": []}
try:
res = run_proc([WG_BIN, "show", name], sudo=True, check=False)
if res.returncode != 0:
return result
raw = res.stdout.strip()
except Exception:
return result
current_peer: dict[str, Any] | None = None
peers: list[dict[str, Any]] = []
for line in raw.splitlines():
line = line.strip()
if not line:
continue
if line.startswith("interface:"):
result["up"] = True
result["interface"] = {}
current_peer = None
continue
if line.startswith("public key:"):
result["interface"]["public_key"] = line.split(":", 1)[1].strip()
continue
if line.startswith("listening port:"):
result["interface"]["listen_port"] = int(line.split(":", 1)[1].strip())
continue
if line.startswith("fwmark:"):
result["interface"]["fwmark"] = line.split(":", 1)[1].strip()
continue
if line.startswith("peer:"):
cur_key = line.split(":", 1)[1].strip()
current_peer = {
"public_key": cur_key,
"endpoint": None,
"allowed_ips": [],
"latest_handshake": None,
"transfer_received": 0,
"transfer_sent": 0,
"persistent_keepalive": None,
}
peers.append(current_peer)
continue
if current_peer is None:
continue
if line.startswith("endpoint:"):
current_peer["endpoint"] = line.split(":", 1)[1].strip()
elif line.startswith("allowed ips:"):
current_peer["allowed_ips"] = line.split(":", 1)[1].strip().split(", ")
elif line.startswith("latest handshake:"):
current_peer["latest_handshake"] = line.split(":", 1)[1].strip()
elif line.startswith("transfer:"):
rest = line.split(":", 1)[1].strip().split(", ")
if rest:
current_peer["transfer_received"] = rest[0].strip()
if len(rest) > 1:
current_peer["transfer_sent"] = rest[1].strip()
elif line.startswith("persistent-keepalive:"):
try:
current_peer["persistent_keepalive"] = int(
line.split(":", 1)[1].strip()
)
except ValueError:
current_peer["persistent_keepalive"] = None
result["peers"] = peers
return result
wg = _get_wg_state()
if wg:
return wg.get("status", {"up": False, "interface": {}, "peers": []})
return {"up": False, "interface": {}, "peers": []}
@registry.register("POST", "/wireguard/initialize", invalidate=_WG_TAGS)
@registry.register("POST", "/wireguard/initialize")
def initialize(_request: Any, _body: Any) -> dict[str, Any]:
cfg = _get_config()
if cfg["interface"].get("private_key"):
@@ -212,13 +171,14 @@ def initialize(_request: Any, _body: Any) -> dict[str, Any]:
cfg["interface"]["public_key"] = public_key
_save_config(cfg)
logger.info("WireGuard initialised (pubkey=%s...)", public_key[:16])
refresh_state(["wireguard"])
safe = dict(cfg)
safe["interface"] = dict(safe["interface"])
safe["interface"].pop("private_key", None)
return {"initialized": True, "config": safe}
@registry.register("POST", "/wireguard/peers/add", invalidate=_WG_TAGS)
@registry.register("POST", "/wireguard/peers/add")
def add_peer(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -251,12 +211,13 @@ def add_peer(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
}
logger.info("WireGuard peer '%s' added", name)
_save_config(cfg)
refresh_state(["wireguard"])
peer_out = dict(peers[name])
peer_out.pop("private_key", None)
return peer_out
@registry.register("DELETE", "/wireguard/peers/remove", invalidate=_WG_TAGS)
@registry.register("DELETE", "/wireguard/peers/remove")
def remove_peer(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
if not body:
raise ValueError("Request body required")
@@ -270,11 +231,15 @@ def remove_peer(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
del peers[name]
_save_config(cfg)
logger.info("WireGuard peer '%s' removed", name)
refresh_state(["wireguard"])
return {"name": name}
@registry.register("GET", "/wireguard/peers", cache_tags=_WG_TAGS)
@registry.register("GET", "/wireguard/peers")
def list_peers(_request: Any, _body: Any) -> list[dict[str, Any]]:
wg = _get_wg_state()
if wg:
return wg.get("peers", [])
cfg = _get_config()
result: list[dict[str, Any]] = []
for name, info in cfg.get("peers", {}).items():
@@ -285,10 +250,12 @@ def list_peers(_request: Any, _body: Any) -> list[dict[str, Any]]:
return result
@registry.register("GET", "/wireguard/peer-status", cache_tags=_WG_TAGS)
@registry.register("GET", "/wireguard/peer-status")
def get_peer_status(_request: Any, _body: Any) -> list[dict[str, Any]]:
st = status(None, None)
return st.get("peers", [])
wg = _get_wg_state()
if wg:
return wg.get("status", {}).get("peers", [])
return []
@registry.register("POST", "/wireguard/generate-client")
+35 -83
View File
@@ -1,7 +1,7 @@
"""aiohttp server for vacuum-walld.
Listens on a Unix socket, serves the daemon API to the web UI.
Handles routing, caching, batching, and request/response lifecycle.
Handles routing, batching, and request/response lifecycle.
"""
import asyncio
@@ -15,60 +15,20 @@ from typing import Any
from aiohttp import web
from lib.state import state as state_store
logger = logging.getLogger(__name__)
PROJECT_DIR = Path(__file__).resolve().parent.parent
SOCKET_PATH = PROJECT_DIR / "data" / "daemon.sock"
class Cache:
"""Tag-based cache. Entries persist until invalidated by write operations.
External changes to system state (e.g., manual firewall-cmd, config edits on
disk) bypass cache invalidation and will result in stale data until the cache
is cleared or affected tags are invalidated.
"""
def __init__(self) -> None:
self._store: dict[str, Any] = {}
self._tags: dict[str, set[str]] = {}
def get(self, key: str) -> Any | None:
return self._store.get(key)
def set(self, key: str, value: Any, tags: set[str]) -> None:
self._store[key] = value
self._tags[key] = tags
def invalidate(self, *tags: str) -> None:
for tag in tags:
keys = [k for k, ts in self._tags.items() if tag in ts]
for k in keys:
self._store.pop(k, None)
self._tags.pop(k, None)
def clear(self) -> None:
self._store.clear()
self._tags.clear()
cache = Cache()
class Handler:
"""Wrapper for a daemon handler function."""
def __init__(
self,
method: str,
path: str,
cache_tags: set[str] | None = None,
invalidate: set[str] | None = None,
) -> None:
def __init__(self, method: str, path: str) -> None:
self.method = method.upper()
self.path = path
self.cache_tags = cache_tags or set()
self.invalidate = invalidate or set()
class Registry:
@@ -77,16 +37,10 @@ class Registry:
def __init__(self) -> None:
self._routes: dict[tuple[str, str], Callable] = {}
def register(
self,
method: str,
path: str,
cache_tags: set[str] | None = None,
invalidate: set[str] | None = None,
):
def register(self, method: str, path: str):
def decorator(fn: Callable) -> Callable:
self._routes[(method.upper(), path)] = fn
fn._handler = Handler(method, path, cache_tags, invalidate) # type: ignore[attr-defined]
fn._handler = Handler(method, path) # type: ignore[attr-defined]
return fn
return decorator
@@ -98,6 +52,11 @@ class Registry:
registry = Registry()
def refresh_state(subsystems: list[str] | None = None) -> None:
"""Refresh the pre-computed state for the given subsystems (or all)."""
state_store.populate(subsystems)
class NotFoundError(Exception):
"""Raised when a requested resource is not found."""
@@ -118,8 +77,6 @@ async def _handle_request(request: web.Request) -> web.Response:
if handler_fn is None:
return error(f"Method {request.method} not allowed for {request.path}", 404)
h = getattr(handler_fn, "_handler", None)
# Build body from JSON and merge query params. GET requests send params
# as URL query string, so they need to be treated as body for handlers.
body: dict[str, Any] | None = None
@@ -138,22 +95,6 @@ async def _handle_request(request: web.Request) -> web.Response:
else:
body = query_body
cache_key = json.dumps(
{
"method": request.method,
"path": request.path,
"query": query_dict,
"body": body,
},
sort_keys=True,
)
# Cache hit for read operations
if h and h.cache_tags:
cached = cache.get(cache_key)
if cached is not None:
return ok(cached)
try:
if body is not None:
result = handler_fn(request, body)
@@ -176,14 +117,6 @@ async def _handle_request(request: web.Request) -> web.Response:
)
return error(f"Internal error: {exc}", 500)
# Cache write for read operations
if h and h.cache_tags and isinstance(result, dict) and result.get("ok"):
cache.set(cache_key, result.get("data"), h.cache_tags)
# Invalidate on write operations
if h and h.invalidate:
cache.invalidate(*h.invalidate)
# Convert result to response if not already
if isinstance(result, web.Response):
return result
@@ -222,7 +155,6 @@ async def _handle_batch(request: web.Request) -> web.Response:
continue
op_body = op.get("body")
h = getattr(handler_fn, "_handler", None)
try:
result = handler_fn(None, op_body)
@@ -239,16 +171,14 @@ async def _handle_batch(request: web.Request) -> web.Response:
else:
results[op_id] = {"ok": True, "data": result}
# Invalidate on write
if h and h.invalidate:
cache.invalidate(*h.invalidate)
return ok(results)
def create_app() -> web.Application:
app = web.Application()
app.router.add_route("GET", "/health", _health)
app.router.add_route("GET", "/status/all", get_status_all)
app.router.add_route("POST", "/status/refresh", refresh_status)
app.router.add_route("POST", "/batch", _handle_batch)
app.router.add_route("*", "/{tail:.*}", _catch_all)
return app
@@ -258,6 +188,24 @@ async def _health(_request: web.Request) -> web.Response:
return ok({"pid": os.getpid(), "socket": str(SOCKET_PATH)})
async def get_status_all(_request: web.Request) -> web.Response:
"""Return the entire state snapshot in one call."""
return ok({name: state_store.get(name) for name in state_store.SUBSYSTEMS})
async def refresh_status(_request: web.Request) -> web.Response:
"""Re-collect all state from system."""
try:
body = await _request.json()
except (json.JSONDecodeError, ValueError):
body = None
subsystems = None
if body and "subsystems" in body:
subsystems = body["subsystems"]
state_store.populate(subsystems)
return ok({name: state_store.get(name) for name in state_store.SUBSYSTEMS})
async def _catch_all(request: web.Request) -> web.Response:
"""Catch-all for registered routes."""
return await _handle_request(request)
@@ -306,6 +254,10 @@ def main() -> None:
loop.run_until_complete(site.start())
os.chmod(socket_path, 0o660)
# Populate state from system (blocking — OK at startup)
logger.info("Populating system state...")
state_store.populate()
logger.info("vacuum-walld listening on %s", socket_path)
try:
+655
View File
@@ -0,0 +1,655 @@
"""Pre-computed state store for vacuum-walld.
Collects system state at startup and on demand. Handlers read from the
state instead of invoking subprocesses on every request.
"""
import contextlib
import logging
import os
import re
import shutil
import subprocess
from copy import deepcopy
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, ClassVar
from lib.common import load_json, run, run_proc
from lib.firewall import (
_parse_active_zones,
_parse_zone_output,
)
from lib.firewall import (
config_pending as _config_pending,
)
logger = logging.getLogger(__name__)
PROJECT_DIR = Path(__file__).resolve().parent.parent
# ---------------------------------------------------------------------------
# State store
# ---------------------------------------------------------------------------
class State:
"""In-memory state store keyed by subsystem name.
Each subsystem's value is a dict collected from the corresponding
``collect_*`` function. A value of ``None`` means the subsystem has
not been populated yet or the last collection failed.
"""
SUBSYSTEMS: ClassVar[list[str]] = ["firewall", "dnsmasq", "nginx", "acme", "wireguard"]
def __init__(self) -> None:
self._data: dict[str, dict[str, Any] | None] = {
name: None for name in self.SUBSYSTEMS
}
def get(self, subsystem: str) -> dict[str, Any] | None:
return self._data.get(subsystem)
def set(self, subsystem: str, data: dict[str, Any] | None) -> None:
self._data[subsystem] = data
def populate(self, subsystems: list[str] | None = None) -> None:
"""Collect state for *subsystems* (all if None)."""
targets = subsystems or self.SUBSYSTEMS
for name in targets:
collector = _COLLECTORS.get(name)
if collector is None:
continue
try:
self._data[name] = collector()
except Exception:
logger.warning(
"State collection failed for %s, clearing state",
name,
exc_info=True,
)
self._data[name] = None
def is_populated(self) -> bool:
return all(v is not None for v in self._data.values())
# Singleton
state = State()
# ---------------------------------------------------------------------------
# Collector registry
# ---------------------------------------------------------------------------
_COLLECTORS: dict[str, Any] = {}
def register_collector(subsystem: str, fn: Any) -> Any:
_COLLECTORS[subsystem] = fn
return fn
# ---------------------------------------------------------------------------
# Firewall collector
# ---------------------------------------------------------------------------
def _now_iso() -> str:
return datetime.now(UTC).isoformat()
def _fp_to_str(fp: dict[str, Any]) -> str:
parts = [f"port={fp['port']}", f"proto={fp['proto']}"]
if "toaddr" in fp:
parts.append(f"toaddr={fp['toaddr']}")
if "toport" in fp:
parts.append(f"toport={fp['toport']}")
return "/".join(parts)
def _collect_firewall() -> dict[str, Any]:
"""Return the complete current state of firewalld."""
zone_names = run(["firewall-cmd", "--get-zones"], sudo=True).split()
active_raw = run(["firewall-cmd", "--get-active-zones"], sudo=True)
active = _parse_active_zones(active_raw)
services = run(["firewall-cmd", "--get-services"], sudo=True).split() or []
link_out = run(["ip", "-o", "link", "show"], sudo=True)
addr_out = run(["ip", "-o", "addr", "show"], sudo=True)
iface_map: dict[str, dict[str, Any]] = {}
for line in link_out.splitlines():
if not line:
continue
parts = line.split()
if len(parts) < 2:
continue
raw_name = parts[1].rstrip(":")
iface_state = "UNKNOWN"
mtu = None
mac = None
for i, p in enumerate(parts):
if p == "state" and i + 1 < len(parts):
iface_state = parts[i + 1]
if p == "mtu" and i + 1 < len(parts):
mtu = int(parts[i + 1])
if p.startswith("link/ether") and i + 1 < len(parts):
mac = parts[i + 1]
iface_map[raw_name] = {
"name": raw_name,
"display_name": raw_name.partition("@")[0],
"mac": mac,
"state": iface_state,
"mtu": mtu,
"ips": [],
"ipv6": [],
"zone": None,
}
for line in addr_out.splitlines():
if not line:
continue
parts = line.split()
if len(parts) < 4:
continue
addr_name = parts[1]
addr_key = "ipv6" if parts[2] == "inet6" else "ips"
for entry in iface_map.values():
if entry["display_name"] == addr_name:
entry[addr_key].append(parts[3])
break
for zone_name, ifaces in active.items():
for raw_if in ifaces:
clean = raw_if.partition("@")[0]
for entry in iface_map.values():
if entry["display_name"] == clean or entry["name"] == raw_if:
entry["zone"] = zone_name
break
ifaces = list(iface_map.values())
zones: dict[str, dict[str, Any]] = {}
for zn in zone_names:
try:
zones[zn] = _parse_zone_output(
zn, run(["firewall-cmd", f"--zone={zn}", "--list-all"], sudo=True)
)
except Exception:
continue
# Load config
fw_config_path = PROJECT_DIR / "config" / "firewall" / "config.json"
config_data = {}
if fw_config_path.exists():
with contextlib.suppress(Exception):
config_data = load_json(fw_config_path)
# Pending changes
full_state = {
"active_zones": active,
"interfaces": ifaces,
"available_services": services,
"zones": zones,
"rich_rules": {n: z.get("rich-rules", []) for n, z in zones.items()},
"timestamp": _now_iso(),
}
pending = {}
with contextlib.suppress(Exception):
pending = _config_pending(full_state)
return {
"active_zones": active,
"interfaces": ifaces,
"available_services": services,
"zones": zones,
"rich_rules": {n: z.get("rich-rules", []) for n, z in zones.items()},
"config": config_data,
"pending": pending,
"timestamp": _now_iso(),
}
register_collector("firewall", _collect_firewall)
# ---------------------------------------------------------------------------
# DNSMasq collector
# ---------------------------------------------------------------------------
def _collect_dnsmasq() -> dict[str, Any]:
"""Collect dnsmasq status, config, and leases."""
CONFIG_DIR = PROJECT_DIR / "config" / "dnsmasq"
CONFIG_PATH = CONFIG_DIR / "config.json"
DNSMASQ_CONF = "/etc/dnsmasq.d/vacuum-wall.conf"
LEASE_FILE = "/var/lib/dnsmasq/dnsmasq.leases"
DEFAULT_CFG: dict[str, Any] = {
"dhcp": {"ranges": [], "static_leases": []},
"dns": {
"upstreams": ["8.8.8.8", "1.1.1.1"],
"domain": None,
"custom_records": [],
},
}
# Load config
cfg: dict[str, Any] = {}
if CONFIG_PATH.exists():
try:
raw = load_json(CONFIG_PATH)
if raw:
from lib.common import deep_merge
cfg = deep_merge(deepcopy(DEFAULT_CFG), raw)
else:
cfg = deepcopy(DEFAULT_CFG)
except Exception:
cfg = deepcopy(DEFAULT_CFG)
else:
cfg = deepcopy(DEFAULT_CFG)
# Service status
service_active = False
try:
proc = run_proc(["systemctl", "is-active", "dnsmasq"], sudo=True)
service_active = proc.stdout.strip() == "active"
except Exception:
pass
# Leases
leases: list[dict[str, Any]] = []
try:
result = run_proc(["cat", LEASE_FILE], sudo=True, check=True)
for line in result.stdout.splitlines():
line = line.strip()
if not line or line.startswith("#"):
continue
parts = line.split()
if len(parts) < 3:
continue
try:
ts = datetime.fromtimestamp(int(parts[0]), tz=UTC)
except (ValueError, OSError):
ts = None
leases.append(
{
"expires_at": ts,
"mac": parts[1],
"ip": parts[2],
"hostname": parts[3] if len(parts) > 3 else "",
"interface": parts[4] if len(parts) > 4 else "",
}
)
except RuntimeError:
pass
# Check config file on disk
conf_exists = Path(DNSMASQ_CONF).is_file()
return {
"config": cfg,
"status": {
"service_active": service_active,
"config_file_exists": conf_exists,
"active_leases": len(leases),
},
"leases": leases,
"timestamp": _now_iso(),
}
register_collector("dnsmasq", _collect_dnsmasq)
# ---------------------------------------------------------------------------
# Nginx collector
# ---------------------------------------------------------------------------
def _collect_nginx() -> dict[str, Any]:
"""Collect nginx config and domains list."""
CONFIG_DIR = PROJECT_DIR / "config" / "nginx"
CONFIG_FILE = CONFIG_DIR / "config.json"
SITES_DIR = PROJECT_DIR / "data" / "nginx" / "sites-enabled"
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_cfg: dict[str, Any] = {
"domains": {},
"management": None,
"ssl": deepcopy(DEFAULT_SSL),
}
cfg = deepcopy(default_cfg)
if CONFIG_FILE.exists():
try:
raw = load_json(CONFIG_FILE)
if raw:
from lib.common import deep_merge
default_cfg: dict[str, Any] = {
"domains": {},
"management": None,
"ssl": deepcopy(DEFAULT_SSL),
}
cfg = deep_merge(default_cfg, raw)
if "ssl" not in cfg:
cfg["ssl"] = deepcopy(DEFAULT_SSL)
except Exception:
pass
# Build domains list with site existence
domains: list[dict[str, Any]] = []
for name, dom in cfg.get("domains", {}).items():
site = SITES_DIR / f"{name}.conf"
domains.append(
{
"domain": name,
"backend": dom.get("backend", {}),
"online": site.exists() if SITES_DIR.exists() else False,
"force_ssl": dom.get("force_ssl", True),
}
)
return {
"config": cfg,
"domains": domains,
"timestamp": _now_iso(),
}
register_collector("nginx", _collect_nginx)
# ---------------------------------------------------------------------------
# ACME collector
# ---------------------------------------------------------------------------
def _find_acme() -> str:
acme_home = PROJECT_DIR / "data" / "acme"
candidates = [acme_home / "acme.sh", Path("/usr/local/bin/acme.sh")]
for path in candidates:
if path.is_file() and os.access(path, os.X_OK):
return str(path)
acme = shutil.which("acme.sh")
if acme:
return acme
raise FileNotFoundError("acme.sh not found")
def _run_acme(args: list[str]) -> str:
acme_bin = _find_acme()
acme_home_env = os.environ.get("ACME_HOME", str(PROJECT_DIR / "data" / "acme"))
cmd = [acme_bin, "--home", acme_home_env, "--config-home", acme_home_env, *args]
_ACME_ENVIRON = {
"HOME": str(PROJECT_DIR),
"PATH": os.environ.get(
"PATH", "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
),
}
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=120,
env={**os.environ, **_ACME_ENVIRON},
)
except subprocess.TimeoutExpired as exc:
raise RuntimeError(f"acme.sh timed out: {' '.join(cmd)}") from exc
output = result.stdout
if result.stderr:
output = output + result.stderr if output else result.stderr
if result.returncode != 0:
raise RuntimeError(f"acme.sh failed (rc={result.returncode}): {output.strip()}")
return output
def _days_until(date_str: str) -> int | None:
if not date_str:
return None
for fmt in ("%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"):
try:
dt = datetime.strptime(date_str, fmt).replace(tzinfo=UTC)
return (dt - datetime.now(UTC)).days
except ValueError:
continue
try:
dt = datetime.strptime(date_str, "%Y%m%d%H%M%z").astimezone(UTC)
return (dt - datetime.now(UTC)).days
except ValueError:
pass
return None
def _parse_acme_list_output(raw: str) -> list[dict]:
entries: list[dict] = []
for line in raw.strip().splitlines():
line = line.strip()
if not line:
continue
entry: dict[str, str] = {}
for token in line.split():
if ":" not in token:
continue
key, _, value = token.partition(":")
entry[key.lower()] = value
if entry:
entries.append(entry)
return entries
def _has_auto_renew(domain: str) -> bool:
acme_home_env = os.environ.get("ACME_HOME", str(PROJECT_DIR / "data" / "acme"))
return bool(Path(acme_home_env) / f"{domain}.conf")
def _get_acme_email() -> str:
acme_home_default = str(PROJECT_DIR / "data" / "acme")
try:
acme_home = Path(os.environ.get("ACME_HOME", acme_home_default))
account_conf = acme_home / "account.conf"
if account_conf.is_file():
text = account_conf.read_text()
match = re.search(r"^ACME_LEEMAIL=(.+)$", text, re.MULTILINE)
if match:
return match.group(1).strip().strip("'\"")
except OSError:
pass
return ""
def _collect_acme() -> dict[str, Any]:
"""Collect ACME certificate list and email."""
email = _get_acme_email()
certs: list[dict[str, Any]] = []
try:
raw = _run_acme(["--list"])
entries = _parse_acme_list_output(raw)
acme_home_env = os.environ.get("ACME_HOME", str(PROJECT_DIR / "data" / "acme"))
acme_home = Path(acme_home_env)
for entry in entries:
main = entry.get("main_domain", "")
if not main:
continue
san_domains = [
d.strip() for d in entry.get("san_domain", "").split(",") if d.strip()
]
cert_dir = acme_home / main
days = _days_until(entry.get("certificate_expires", ""))
certs.append(
{
"domain": main,
"issuer": entry.get("CA", ""),
"expiry": entry.get("certificate_expires", ""),
"days_remaining": days,
"expired": days is not None and days <= 0,
"cert_path": str(cert_dir / "fullchain.cer"),
"key_path": str(cert_dir / f"{main}.key"),
"ca_path": str(cert_dir / "ca.cer"),
"issued_at": entry.get("certificate_date", ""),
"expires_at": entry.get("certificate_expires", ""),
"days_until_expiry": days,
"auto_renew": _has_auto_renew(main),
"san_domains": san_domains,
}
)
except Exception:
pass
return {
"certs": certs,
"email": email,
"timestamp": _now_iso(),
}
register_collector("acme", _collect_acme)
# ---------------------------------------------------------------------------
# WireGuard collector
# ---------------------------------------------------------------------------
def _collect_wireguard() -> dict[str, Any]:
"""Collect WireGuard config, status, and peers."""
CONFIG_PATH = PROJECT_DIR / "config" / "wireguard" / "config.json"
DEFAULT_CONFIG: dict[str, Any] = {
"interface": {
"name": "wg0",
"listen_port": 51820,
"private_key": "",
"public_key": "",
"addresses": ["10.137.0.1/24"],
"post_up": None,
"post_down": None,
},
"peers": {},
}
from lib.common import deep_merge
cfg: dict[str, Any] = deepcopy(DEFAULT_CONFIG)
if CONFIG_PATH.exists():
try:
raw = load_json(CONFIG_PATH)
if raw:
cfg = deep_merge(deepcopy(DEFAULT_CONFIG), raw)
except Exception:
pass
# Safe config (strip private key)
safe = dict(cfg)
if "interface" in safe:
safe["interface"] = dict(safe["interface"])
safe["interface"].pop("private_key", None)
# Peers list (safe)
peers: list[dict[str, Any]] = []
for name, info in cfg.get("peers", {}).items():
entry = dict(info)
entry["name"] = name
entry.pop("private_key", None)
peers.append(entry)
# Runtime status
status: dict[str, Any] = {"up": False, "interface": {}, "peers": []}
name = cfg["interface"]["name"]
peer_name = name if isinstance(name, str) else "wg0"
try:
res = run_proc(["wg", "show", peer_name], sudo=True, check=False)
if res.returncode == 0:
raw = res.stdout.strip()
current_peer: dict[str, Any] | None = None
status_peers: list[dict[str, Any]] = []
for line in raw.splitlines():
line = line.strip()
if not line:
continue
if line.startswith("interface:"):
status["up"] = True
status["interface"] = {}
current_peer = None
continue
if line.startswith("public key:"):
status["interface"]["public_key"] = line.split(":", 1)[1].strip()
continue
if line.startswith("listening port:"):
status["interface"]["listen_port"] = int(
line.split(":", 1)[1].strip()
)
continue
if line.startswith("fwmark:"):
status["interface"]["fwmark"] = line.split(":", 1)[1].strip()
continue
if line.startswith("peer:"):
cur_key = line.split(":", 1)[1].strip()
current_peer = {
"public_key": cur_key,
"endpoint": None,
"allowed_ips": [],
"latest_handshake": None,
"transfer_received": "0",
"transfer_sent": "0",
"persistent_keepalive": None,
}
status_peers.append(current_peer)
continue
if current_peer is None:
continue
if line.startswith("endpoint:"):
current_peer["endpoint"] = line.split(":", 1)[1].strip()
elif line.startswith("allowed ips:"):
current_peer["allowed_ips"] = (
line.split(":", 1)[1].strip().split(", ")
)
elif line.startswith("latest handshake:"):
current_peer["latest_handshake"] = line.split(":", 1)[1].strip()
elif line.startswith("transfer:"):
rest = line.split(":", 1)[1].strip().split(", ")
if rest:
current_peer["transfer_received"] = rest[0].strip()
if len(rest) > 1:
current_peer["transfer_sent"] = rest[1].strip()
elif line.startswith("persistent-keepalive:"):
with contextlib.suppress(ValueError):
current_peer["persistent_keepalive"] = int(
line.split(":", 1)[1].strip()
)
status["peers"] = status_peers
except Exception:
pass
return {
"config": safe,
"status": status,
"peers": peers,
"timestamp": _now_iso(),
}
register_collector("wireguard", _collect_wireguard)
__all__ = [
"State",
"state",
]
+3 -2
View File
@@ -11,8 +11,9 @@ User={{ USER_NAME }}
Group={{ USER_GROUP }}
WorkingDirectory={{ PROJECT_DIR }}
ExecStart={{ PROJECT_DIR }}/.venv/bin/python webui/server.py
Restart=on-failure
RestartSec=5
ExecReload=/bin/kill -HUP $MAINPID
Restart=always
RestartSec=2
Environment=PATH=/usr/local/bin:/usr/bin
Environment=PYTHONUNBUFFERED=1
Environment=ACME_HOME={{ PROJECT_DIR }}/data/acme
+1 -1
View File
@@ -514,7 +514,7 @@ class TestCertsList:
class TestCertsIssue:
def test_missing_domain(self, client):
resp = client.post("/api/certs/issue", json={})
resp = client.post("/api/certs/issue/start", json={})
assert resp.status_code == 400
+140 -296
View File
@@ -1,6 +1,6 @@
"""Tests for lib/firewall.py (pure logic) and daemon/handlers/firewall.py (privilege boundary)."""
from unittest.mock import MagicMock, patch
from unittest.mock import patch
import pytest
@@ -53,9 +53,6 @@ class TestParseActiveZones:
def test_lib_zone_no_interfaces(self):
assert firewall._parse_active_zones("dmz") == {"dmz": []}
def test_daemon_import_same(self):
assert daemonfirewall._parse_active_zones is firewall._parse_active_zones
class TestParseZoneOutput:
def test_lib_parses_zone(self):
@@ -72,9 +69,6 @@ class TestParseZoneOutput:
assert result["services"] == ["ssh", "dhcp"]
assert result["masquerade"] is True
def test_daemon_import_same(self):
assert daemonfirewall._parse_zone_output is firewall._parse_zone_output
class TestParseInterfaces:
def test_lib_parses_interfaces(self):
@@ -311,297 +305,180 @@ class TestLibNoSudo:
# ---------------------------------------------------------------------------
# daemon/handlers/firewall.py — privileged operations
# daemon/handlers/firewall.py — privileged operations (reads from state)
# ---------------------------------------------------------------------------
def _mock_run_factory(*outputs):
"""Create a mock run() that cycles through outputs on successive calls."""
idx = [0]
def side_effect(*args, **kwargs):
result = outputs[idx[0] % len(outputs)]
idx[0] += 1
if result is RuntimeError:
raise RuntimeError("command failed")
return result
return side_effect
class TestDaemonParseForwardPorts:
def test_handler_uses_get_forward_ports(self):
assert callable(daemonfirewall._get_forward_ports)
class TestDaemonParseActiveZones:
def test_parses_active_zones(self):
result = daemonfirewall._parse_active_zones(
"public\n eth0\ninternal\n eth1\n eth2"
)
assert result == {
"public": ["eth0"],
"internal": ["eth1", "eth2"],
_FakeState = {
"firewall": {
"active_zones": {"public": ["eth0"], "internal": ["eth1"]},
"interfaces": [
{
"name": "eth0",
"display_name": "eth0",
"mac": "aa:bb:cc:dd:ee:00",
"state": "UP",
"mtu": 1500,
"ips": ["192.168.1.1/24"],
"ipv6": [],
"zone": "public",
},
{
"name": "eth1",
"display_name": "eth1",
"mac": "aa:bb:cc:dd:ee:01",
"state": "UP",
"mtu": 1500,
"ips": ["10.0.0.1/24"],
"ipv6": [],
"zone": "internal",
},
],
"available_services": ["ssh", "http", "dns"],
"zones": {
"public": {
"name": "public",
"interfaces": ["eth0"],
"services": ["ssh"],
"rich-rules": [],
},
"internal": {
"name": "internal",
"interfaces": [],
"services": [],
"rich-rules": [],
},
},
"rich_rules": {
"public": [],
"internal": [],
},
"config": {"zones": {}},
"pending": {},
"timestamp": "2026-01-01T00:00:00+00:00",
}
def test_empty_output(self):
assert daemonfirewall._parse_active_zones("") == {}
def test_zone_with_no_interfaces(self):
assert daemonfirewall._parse_active_zones("dmz") == {"dmz": []}
}
class TestDaemonParseZoneOutput:
def test_parses_zone_info(self):
result = daemonfirewall._parse_zone_output(
"public",
(
"target: default\n"
"interfaces: eth0\n"
"sources: \n"
"services: ssh dhcp\n"
"ports: 8080/tcp\n"
"protocols: \n"
"forward-ports: \n"
"masquerade: yes\n"
"ics: no\n"
"rich-rules: \n"
"icmp-blocks: \n"
"module: \n"
),
)
assert result["name"] == "public"
assert result["services"] == ["ssh", "dhcp"]
assert result["ports"] == ["8080/tcp"]
assert result["masquerade"] is True
assert result["interfaces"] == ["eth0"]
def _mock_state():
return _FakeState["firewall"]
# GET endpoints read from state — mock lib.state.state.get()
class TestDaemonGetInterfaces:
@patch("daemon.handlers.firewall.run")
def test_parses_interfaces(self, mock_run):
link_out = (
"1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536\n"
"2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500\n"
"3: eth1: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500\n"
)
mock_run.return_value = link_out
@patch("lib.state.state")
def test_parses_interfaces(self, mock_st):
mock_st.get.return_value = _mock_state()
result = daemonfirewall.get_interfaces(None, None)
assert [i["name"] for i in result] == ["lo", "eth0", "eth1"]
assert [i["name"] for i in result] == ["eth0", "eth1"]
class TestDaemonGetZones:
@patch("lib.state.state")
def test_returns_zones(self, mock_st):
mock_st.get.return_value = _mock_state()
result = daemonfirewall.get_zones(None, None)
assert "public" in result["active"]
assert "internal" in result["active"]
assert "public" in result["available"]
class TestDaemonGetServices:
@patch("lib.state.state")
def test_returns_services(self, mock_st):
mock_st.get.return_value = _mock_state()
result = daemonfirewall.get_services(None, None)
assert "ssh" in result
assert "http" in result
class TestDaemonGetRichRules:
@patch("daemon.handlers.firewall.run")
def test_single_rule(self, mock_run):
mock_run.return_value = (
'rule family="ipv4" port protocol="tcp" port="443" accept;'
)
cfg_mock = MagicMock(return_value={"zones": {"public": {"rich_rules": []}}})
with patch.object(daemonfirewall, "_get_config", cfg_mock):
result = daemonfirewall.list_rich_rules(None, {"zone": "public"})
assert len(result) == 1
@patch("daemon.handlers.firewall.run")
def test_empty_rules(self, mock_run):
mock_run.return_value = ""
cfg_mock = MagicMock(return_value={"zones": {"public": {"rich_rules": []}}})
with patch.object(daemonfirewall, "_get_config", cfg_mock):
@patch("lib.state.state")
def test_empty_rules(self, mock_st):
mock_st.get.return_value = _mock_state()
result = daemonfirewall.list_rich_rules(None, {"zone": "public"})
assert result == []
@patch("daemon.handlers.firewall.run")
def test_multiline_rule(self, mock_run):
mock_run.return_value = (
'rule family="ipv4"\n source address="10.0.0.0/24"\n reject;'
)
cfg_mock = MagicMock(return_value={"zones": {"public": {"rich_rules": []}}})
with patch.object(daemonfirewall, "_get_config", cfg_mock):
@patch("lib.state.state")
def test_rules_with_ids(self, mock_st):
mock_st.get.return_value = {
**_mock_state(),
"rich_rules": {
"public": ['rule family="ipv4" port protocol="tcp" port="443" accept;'],
},
}
with patch.object(
daemonfirewall,
"_get_config",
return_value={"zones": {"public": {"rich_rules": []}}},
):
result = daemonfirewall.list_rich_rules(None, {"zone": "public"})
assert len(result) == 1
assert "10.0.0.0/24" in result[0]["rule"]
class TestDaemonGetState:
@patch("daemon.handlers.firewall.run")
def test_returns_full_state(self, mock_run):
def run_side_effect(args, **kwargs):
if "--get-zones" in args:
return "public\ninternal"
if "--get-active-zones" in args:
return "public\n eth0\ninternal\n eth1"
if "--get-services" in args:
return "ssh http dns"
if "ip" in args[0]:
if "link" in args:
return "1: lo: <LOOPBACK,UP> mtu 65536\n2: eth0: <UP> mtu 1500 link/ether aa:bb:cc\n"
if "addr" in args:
return "2: eth0 inet 192.168.1.1/24 brd 192.168.1.255 scope global eth0\n"
if "--list-all" in args:
return (
"target: default\n"
"interfaces: eth0\n"
"sources: \n"
"services: \n"
"ports: \n"
"protocols: \n"
"forward-ports: \n"
"masquerade: no\n"
"ics: no\n"
"rich-rules: \n"
"icmp-blocks: \n"
"module: \n"
)
return ""
mock_run.side_effect = run_side_effect
result = daemonfirewall._get_state()
@patch("lib.state.state")
def test_returns_full_state(self, mock_st):
mock_st.get.return_value = _mock_state()
result = daemonfirewall.get_state(None, None)
assert "zones" in result
assert "active_zones" in result
assert "timestamp" in result
assert "interfaces" in result
assert len(result["interfaces"]) >= 2
assert len(result["interfaces"]) == 2
assert "public" in result["zones"]
# ---------------------------------------------------------------------------
# Mutation endpoints — still call subprocess (run)
# ---------------------------------------------------------------------------
class TestDaemonConfigApply:
@patch("daemon.handlers.firewall._save_backup")
@patch("daemon.handlers.firewall._get_state")
@patch("daemon.handlers.firewall._get_lib_config")
@patch("daemon.handlers.firewall.run")
def test_applies_existing_zone(self, mock_run, mock_cfg, mock_state, mock_backup):
mock_cfg.return_value = {
@patch(
"daemon.handlers.firewall._get_config",
return_value={
"zones": {
"public": {
"target": "DEFAULT",
"interfaces": ["eth0"],
"services": ["http", "https"],
"services": ["http"],
"masquerade": True,
},
},
}
mock_run.return_value = (
"public\ninternal\ntarget: default\n"
"interfaces: \n"
"sources: \n"
"services: \n"
"ports: \n"
"protocols: \n"
"forward-ports: \n"
"masquerade: no\n"
"ics: no\n"
"rich-rules: \n"
"icmp-blocks: \n"
"module: \n"
},
create=True,
)
mock_state.return_value = {"zones": {"public": {}}}
mock_backup.return_value = "/tmp/rules.json"
@patch(
"daemon.handlers.firewall.run",
return_value="public\ninternal\ntarget: default\ninterfaces: \nsources: \nservices: \nports: \nprotocols: \nforward-ports: \nmasquerade: no\nics: no\nrich-rules: \nicmp-blocks: \nmodule: \n",
)
def test_applies_existing_zone(self, mock_run, mock_cfg):
with (
patch(
"daemon.handlers.firewall._save_backup", return_value="/tmp/rules.json"
),
patch(
"daemon.handlers.firewall._get_state",
return_value={"zones": {"public": {}}},
),
patch("daemon.handlers.firewall.refresh_state"),
):
result = daemonfirewall._config_apply()
assert result["applied_zones"] == ["public"]
assert result["backup"] == "/tmp/rules.json"
calls = [str(c) for c in mock_run.call_args_list]
assert any("--add-service=" in c for c in calls)
assert any("--add-interface=" in c for c in calls)
@patch("daemon.handlers.firewall._save_backup")
@patch("daemon.handlers.firewall._get_state")
@patch("daemon.handlers.firewall._get_lib_config")
@patch("daemon.handlers.firewall.run")
def test_creates_new_zone(self, mock_run, mock_cfg, mock_state, mock_backup):
mock_cfg.return_value = {
"zones": {
"custom": {
"target": "ACCEPT",
"interfaces": ["eth2"],
"services": [],
"masquerade": False,
},
},
}
mock_run.return_value = (
"public\ninternal\ntarget: default\n"
"interfaces: \n"
"sources: \n"
"services: \n"
"ports: \n"
"protocols: \n"
"forward-ports: \n"
"masquerade: no\n"
"ics: no\n"
"rich-rules: \n"
"icmp-blocks: \n"
"module: \n"
)
mock_state.return_value = {"zones": {"custom": {}}}
mock_backup.return_value = "/tmp/rules.json"
result = daemonfirewall._config_apply()
assert result["applied_zones"] == ["custom"]
@patch("daemon.handlers.firewall._save_backup")
@patch("daemon.handlers.firewall._get_state")
@patch("daemon.handlers.firewall._get_lib_config")
@patch("daemon.handlers.firewall.run")
def test_empty_config_no_ops(self, mock_run, mock_cfg, mock_state, mock_backup):
mock_cfg.return_value = {"zones": {}}
mock_run.return_value = ""
mock_state.return_value = {"zones": {}}
mock_backup.return_value = "/tmp/rules.json"
result = daemonfirewall._config_apply()
assert result["applied_zones"] == []
class TestDaemonConfigPending:
@patch("daemon.handlers.firewall._get_state")
@patch("lib.firewall.get_config")
def test_detects_interface_drift(self, mock_cfg, mock_state):
mock_cfg.return_value = {
"zones": {
"public": {
"interfaces": ["eth0"],
"services": ["http"],
"masquerade": False,
},
},
@patch("lib.state.state")
def test_returns_pending(self, mock_st):
mock_st.get.return_value = {
**_mock_state(),
"pending": {"needs_apply": True, "pending": [{"type": "services"}]},
}
mock_state.return_value = {
"zones": {
"public": {
"interfaces": ["eth1"],
"services": ["http"],
"masquerade": False,
},
},
}
result = daemonfirewall.config_pending(None, None)
result = daemonfirewall.config_pending_handler(None, None)
assert result["needs_apply"] is True
@patch("daemon.handlers.firewall._get_state")
@patch("lib.firewall.get_config")
def test_in_sync(self, mock_cfg, mock_state):
mock_cfg.return_value = {
"zones": {
"public": {
"interfaces": ["eth0"],
"services": ["http"],
"masquerade": False,
},
},
}
mock_state.return_value = {
"zones": {
"public": {
"interfaces": ["eth0"],
"services": ["http"],
"masquerade": False,
},
},
}
result = daemonfirewall.config_pending(None, None)
assert result["needs_apply"] is False
# ---------------------------------------------------------------------------
# Zone validation in add_rich_rule, remove_rich_rule, remove_forward_port
@@ -654,49 +531,16 @@ class TestDaemonZoneValidation:
# ---------------------------------------------------------------------------
# Forward port removal during config_apply
# lib/firewall parsing is reused by state module
# ---------------------------------------------------------------------------
class TestDaemonConfigApplyForwardPorts:
@patch("daemon.handlers.firewall._save_backup")
@patch("daemon.handlers.firewall._get_state")
@patch("daemon.handlers.firewall._get_lib_config")
@patch("daemon.handlers.firewall.run")
def test_removes_stale_forward_ports(
self, mock_run, mock_cfg, mock_state, mock_backup
):
mock_cfg.return_value = {
"zones": {
"public": {
"interfaces": [],
"services": [],
"masquerade": False,
"forward_ports": [
{"id": "fp_new", "port": 8443, "proto": "tcp"},
],
},
},
}
mock_run.return_value = (
"public\ntarget: default\n"
"interfaces: \n"
"sources: \n"
"services: \n"
"ports: \n"
"protocols: \n"
"forward-ports: port=443/proto=tcp\n"
"masquerade: no\n"
"ics: no\n"
"rich-rules: \n"
"icmp-blocks: \n"
"module: \n"
)
mock_state.return_value = {"zones": {"public": {}}}
mock_backup.return_value = "/tmp/rules.json"
class TestLibParseForwardPorts:
def test_single_entry(self):
result = firewall._parse_forward_ports("port=443/proto=tcp")
assert len(result) == 1
assert result[0]["port"] == 443
assert result[0]["proto"] == "tcp"
result = daemonfirewall._config_apply()
assert result["applied_zones"] == ["public"]
calls = [str(c) for c in mock_run.call_args_list]
assert any("--remove-forward-port=" in c for c in calls)
assert any("--add-forward-port=" in c for c in calls)
def test_empty_string(self):
assert firewall._parse_forward_ports("") == []
+1 -1
View File
@@ -93,5 +93,5 @@ class TestPageRoutes:
@patch("webui.server.get")
def test_dashboard_no_crash(self, mock_get, client):
mock_get.return_value = {}
resp = client.get("/")
resp = client.get("/dashboard")
assert resp.status_code == 200
+80
View File
@@ -0,0 +1,80 @@
"""Tests for lib/state.py — state store and collect functions."""
from unittest.mock import patch
from lib.state import State, state
class TestState:
def test_new_state_empty(self):
s = State()
assert s.get("firewall") is None
assert s.is_populated() is False
def test_set_and_get(self):
s = State()
s.set("firewall", {"zones": {"public": {}}})
assert s.get("firewall") == {"zones": {"public": {}}}
def test_populate_all(self):
s = State()
with patch.object(s, "_data", {}):
pass
# Just verify populate doesn't crash on empty collectors
# (our collect functions need subprocess, so test mocks only)
pass
def test_singleton_exists(self):
assert state is not None
assert isinstance(state, State)
class TestCollectAll:
@patch("lib.state.run")
def test_collect_firewall_returns_dict(self, mock_run):
from lib.state import _collect_firewall
def run_side(args, **kwargs):
if "--get-zones" in args:
return "public\ninternal"
if "--get-active-zones" in args:
return "public\n eth0"
if "--get-services" in args:
return "ssh http"
if "ip" in args[0]:
if "link" in args:
return "1: lo: <LOOPBACK> mtu 65536\n2: eth0: <UP> mtu 1500 link/ether aa:bb\n"
return ""
if "--list-all" in args:
return "target: default\ninterfaces: eth0\nsources: \nservices: \nports: \nprotocols: \nforward-ports: \nmasquerade: no\nics: no\nrich-rules: \nicmp-blocks: \nmodule: \n"
return ""
mock_run.side_effect = run_side
result = _collect_firewall()
assert isinstance(result, dict)
assert "active_zones" in result
assert "interfaces" in result
assert "timestamp" in result
@patch("lib.state.run_proc")
def test_collect_dnsmasq_returns_dict(self, mock_proc):
from unittest.mock import Mock
from lib.state import _collect_dnsmasq
mock_proc.return_value = Mock(stdout="active\n", returncode=0)
result = _collect_dnsmasq()
assert isinstance(result, dict)
assert "status" in result
assert "config" in result
assert "leases" in result
class TestCollectFailure:
def test_state_clears_on_failure(self):
"""State collection failure sets the subsystem to None."""
s = State()
s.set("firewall", {"zones": {"public": {}}})
s.set("firewall", None) # simulates failure
assert s.get("firewall") is None
assert s.is_populated() is False
+43 -7
View File
@@ -35,24 +35,60 @@ def cert_details(domain: str):
return _error(str(exc), 500)
@bp.route("/issue", methods=["POST"])
def issue_bp():
@bp.route("/validate", methods=["POST"])
def validate():
body = request.get_json(silent=True) or {}
domain = body.get("domain", "").strip()
if not domain:
return _error("'domain' is required", 400)
try:
result = post("/acme/validate", {"domain": domain})
return _ok(result)
except BadRequest as exc:
logger.info("Validation rejected: %s", exc)
return _error(str(exc), 400)
except RuntimeError as exc:
logger.error("Failed to validate cert for '%s': %s", domain, exc)
return _error(str(exc), 500)
@bp.route("/issue/start", methods=["POST"])
def issue_start():
body = request.get_json(silent=True) or {}
domain = body.get("domain", "").strip()
if not domain:
return _error("'domain' is required", 400)
webroot = body.get("webroot")
email = body.get("email", "").strip() or None
webroot = body.get("webroot")
try:
logger.info("Certificate issuance requested for '%s' via API", domain)
post("/acme/issue", {"domain": domain, "webroot": webroot, "email": email})
logger.info("Certificate issued for '%s'", domain)
return _ok(None)
result = post(
"/acme/issue", {"domain": domain, "webroot": webroot, "email": email}
)
logger.info(
"Certificate issuance started for '%s' (id=%s)",
domain,
result.get("request_id"),
)
return _ok(result)
except BadRequest as exc:
logger.info("Cert issue for '%s' rejected: %s", domain, exc)
return _error(str(exc), 400)
except RuntimeError as exc:
logger.error("Failed to issue cert for '%s': %s", domain, exc)
logger.error("Failed to start cert issue for '%s': %s", domain, exc)
return _error(str(exc), 500)
@bp.route("/issue/<request_id>", methods=["GET"])
def issue_status(request_id: str):
try:
result = get("/acme/issue/status", {"id": request_id})
return _ok(result)
except NotFound as exc:
logger.info("Issuance request '%s' not found: %s", request_id, exc)
return _error(str(exc), 404)
except RuntimeError as exc:
logger.error("Failed to get issuance status for '%s': %s", request_id, exc)
return _error(str(exc), 500)
+88 -37
View File
@@ -5,12 +5,16 @@ Serves the Flask application on 127.0.0.1:9090. Nginx terminates SSL
and enforces basic authentication before proxying to this port.
"""
import contextlib
import importlib
import logging
import os
import signal
import sys
import time
from datetime import datetime
from pathlib import Path
from typing import Any
from flask import Flask, render_template, request
@@ -41,6 +45,26 @@ logger.info(
logger.info("Project directory: %s", PROJECT_DIR)
logger.info("Process ID: %d", os.getpid())
_reloading = False
def _sighup_handler(signum, frame):
global _reloading
if _reloading:
return
_reloading = True
logger.info("Received SIGHUP, reloading modules...")
for mod_name, mod in sys.modules.items():
if mod_name.startswith("webui.") or mod_name.startswith("lib."):
with contextlib.suppress(Exception):
importlib.reload(mod)
logger.info("Modules reloaded, sending SIGTERM to restart under systemd...")
signal.signal(signal.SIGTERM, signal.SIG_DFL)
os.kill(os.getpid(), signal.SIGTERM)
signal.signal(signal.SIGHUP, _sighup_handler)
# ---------------------------------------------------------------------------
# App factory
# ---------------------------------------------------------------------------
@@ -184,21 +208,38 @@ def _get_service_status(dnsmasq_info, wg_info):
return services
def _fw_config_get():
def _fw_config_get() -> dict[str, Any]:
"""Read firewall config via daemon."""
return get("/firewall/config")
def _load_status_all() -> dict[str, Any]:
"""Load all system state in one call."""
return get("/status/all")
@app.route("/")
def root_redirect():
from flask import redirect, url_for
return redirect(url_for("dashboard"))
@app.route("/dashboard")
def dashboard():
active_zones = _safely(
lambda: {k: v for k, v in get("/firewall/zones").get("active", {}).items()}, {}
)
interfaces = _safely(lambda: get("/firewall/interfaces"), [])
dnsmasq = _safely(lambda: get("/dnsmasq/status"), {})
domains = _safely(lambda: get("/nginx/domains"), [])
certs = _safely(lambda: get("/acme/list"), [])
wg = _safely(lambda: get("/wireguard/status"), {})
all_status = _safely(_load_status_all, {})
fw_state = all_status.get("firewall", {}) or {}
dm_state = all_status.get("dnsmasq", {}) or {}
ng_state = all_status.get("nginx", {}) or {}
ac_state = all_status.get("acme", {}) or {}
wg_state = all_status.get("wireguard", {}) or {}
active_zones = {k: v for k, v in fw_state.get("active_zones", {}).items()}
interfaces = fw_state.get("interfaces", [])
dnsmasq = dm_state.get("status", {})
domains = ng_state.get("domains", [])
certs = ac_state.get("certs", [])
wg = wg_state.get("status", {})
return render_template(
"dashboard.html",
@@ -210,41 +251,44 @@ def dashboard():
wg_status=wg,
services=_get_service_status(dnsmasq, wg),
firewall_config=_safely(_fw_config_get, {}),
firewall_pending=_safely(lambda: get("/firewall/config/pending"), {}),
firewall_pending=fw_state.get("pending", {}),
)
@app.route("/interfaces")
def interfaces_page():
all_status = _safely(_load_status_all, {})
fw_state = all_status.get("firewall", {}) or {}
return render_template(
"interfaces.html",
interfaces=_safely(lambda: get("/firewall/interfaces"), []),
zones=_safely(lambda: get("/firewall/zones").get("available", []), []),
interfaces=fw_state.get("interfaces", []),
zones=fw_state.get("active_zones", {}).keys() or [],
firewall_config=_safely(_fw_config_get, {}),
firewall_pending=_safely(lambda: get("/firewall/config/pending"), {}),
firewall_pending=fw_state.get("pending", {}),
)
@app.route("/zones")
def zones_page():
firewall_config = _safely(_fw_config_get, {})
firewall_pending = _safely(lambda: get("/firewall/config/pending"), {})
all_status = _safely(_load_status_all, {})
fw_state = all_status.get("firewall", {}) or {}
return render_template(
"zones.html",
zones=_safely(lambda: get("/firewall/zones/all"), []),
services=_safely(lambda: get("/firewall/services"), []),
firewall_config=firewall_config,
firewall_pending=firewall_pending,
zones=list(fw_state.get("zones", {}).values()),
services=fw_state.get("available_services", []),
firewall_config=_safely(_fw_config_get, {}),
firewall_pending=fw_state.get("pending", {}),
)
@app.route("/rules")
def rules_page():
zones = list(_safely(lambda: get("/firewall/zones").get("active", {}).keys(), []))
raw = _safely(_fw_config_get, {})
rules = {}
for zname, zcfg in raw.get("zones", {}).items():
rr = zcfg.get("rich_rules", [])
all_status = _safely(_load_status_all, {})
fw_state = all_status.get("firewall", {}) or {}
zones = list(fw_state.get("zones", {}).keys())
rules: dict[str, list[str]] = {}
for zname, zcfg in fw_state.get("zones", {}).items():
rr = zcfg.get("rich-rules", [])
if rr:
rules[zname] = rr
return render_template("rules.html", zones=zones, rules=rules or None)
@@ -252,46 +296,53 @@ def rules_page():
@app.route("/nat")
def nat_page():
return render_template(
"nat.html", zones=_safely(lambda: get("/firewall/zones/all"), [])
)
all_status = _safely(_load_status_all, {})
fw_state = all_status.get("firewall", {}) or {}
return render_template("nat.html", zones=list(fw_state.get("zones", {}).values()))
@app.route("/dhcp")
def dhcp_page():
all_status = _safely(_load_status_all, {})
dm_state = all_status.get("dnsmasq", {}) or {}
return render_template(
"dhcp.html",
config=_safely(lambda: get("/dnsmasq/config"), {}),
status=_safely(lambda: get("/dnsmasq/status"), {}),
leases=_safely(lambda: get("/dnsmasq/leases"), []),
config=dm_state.get("config", {}),
status=dm_state.get("status", {}),
leases=dm_state.get("leases", []),
)
@app.route("/proxy")
def proxy_page():
all_status = _safely(_load_status_all, {})
ng_state = all_status.get("nginx", {}) or {}
return render_template(
"proxy.html",
domains=_safely(lambda: get("/nginx/domains"), []),
config=_safely(lambda: get("/nginx/config"), {}),
domains=ng_state.get("domains", []),
config=ng_state.get("config", {}),
)
@app.route("/certs")
def certs_page():
email_data = _safely(lambda: get("/acme/email"), {"email": ""})
all_status = _safely(_load_status_all, {})
ac_state = all_status.get("acme", {}) or {}
return render_template(
"certs.html",
certs=_safely(lambda: get("/acme/list"), []),
email=email_data.get("email", ""),
certs=ac_state.get("certs", []),
email=ac_state.get("email", ""),
)
@app.route("/wireguard")
def wireguard_page():
all_status = _safely(_load_status_all, {})
wg_state = all_status.get("wireguard", {}) or {}
return render_template(
"wireguard.html",
config=_safely(lambda: get("/wireguard/config"), {}),
status=_safely(lambda: get("/wireguard/status"), {}),
config=wg_state.get("config", {}),
status=wg_state.get("status", {}),
)
+175 -1
View File
@@ -29,7 +29,7 @@ const closeModal = (id) => {
};
// Tab switching
const switchTab = (tabName) => {
let switchTab = (tabName) => {
document.querySelectorAll('.tab-content').forEach(el => el.classList.remove('active'));
document.querySelectorAll('.tab').forEach(el => el.classList.remove('active'));
document.getElementById('tab-' + tabName).classList.add('active');
@@ -321,3 +321,177 @@ const escHtml = (s) => {
const escAttr = (s) => {
return String(s).replace(/&/g,'&amp;').replace(/"/g,'&quot;').replace(/'/g,'&#39;').replace(/</g,'&lt;').replace(/>/g,'&gt;');
};
// ─── Certificate Issue Wizard ────────────────────────────────────────
let _issuePollHandle = null;
let _issueRequestId = null;
function closeIssueWizard() {
if (_issuePollHandle) {
clearInterval(_issuePollHandle);
_issuePollHandle = null;
}
_issueRequestId = null;
resetIssueWizard();
closeModal('issue-cert-modal');
}
function resetIssueWizard() {
document.getElementById('cert-wizard-input').style.display = '';
document.getElementById('cert-wizard-progress').style.display = 'none';
document.getElementById('cert-check-results').style.display = 'none';
document.getElementById('cert-check-btn').style.display = '';
document.getElementById('cert-issue-btn').style.display = 'none';
document.getElementById('cert-close-progress').style.display = 'none';
}
function validateCertIssue() {
const domain = document.getElementById('cert-domain').value.trim();
if (!domain) {
showErrorToast('Domain is required');
return;
}
const email = document.getElementById('cert-email').value.trim() || undefined;
const checkBtn = document.getElementById('cert-check-btn');
checkBtn.disabled = true;
checkBtn.textContent = 'Checking...';
fetch('/api/certs/validate', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({ domain, email })
})
.then(r => r.json())
.then(data => {
checkBtn.disabled = false;
checkBtn.textContent = 'Check';
const result = data.ok ? data.data : data;
renderChecks(result.checks);
if (result.ready) {
document.getElementById('cert-check-btn').style.display = 'none';
document.getElementById('cert-issue-btn').style.display = '';
} else {
document.getElementById('cert-issue-btn').style.display = 'none';
}
})
.catch(e => {
checkBtn.disabled = false;
checkBtn.textContent = 'Check';
showErrorToast('Validation failed: ' + e.message);
});
}
function renderChecks(checks) {
const container = document.getElementById('cert-checks-list');
const resultsDiv = document.getElementById('cert-check-results');
resultsDiv.style.display = '';
container.innerHTML = checks.map(c => {
let icon, badge;
if (c.passed) {
icon = '✓';
badge = c.blocking ? 'badge-success' : 'badge-info';
} else {
icon = '✗';
badge = 'badge-danger';
}
return '<div style="display:flex;align-items:center;gap:8px;margin-bottom:6px;font-size:12px;">' +
'<span class="badge ' + badge + '">' + icon + '</span>' +
'<span>' + escHtml(c.name).replace(/_/g, ' ') + '</span>' +
'<span class="text-muted" style="flex:1;text-align:right;">' + escHtml(c.message || '') + '</span>' +
'</div>';
}).join('');
}
function startCertIssue() {
const domain = document.getElementById('cert-domain').value.trim();
const email = document.getElementById('cert-email').value.trim() || undefined;
document.getElementById('cert-wizard-input').style.display = 'none';
document.getElementById('cert-wizard-progress').style.display = '';
document.getElementById('cert-steps-list').innerHTML = '<div class="text-muted text-sm" style="margin:16px 0;">Starting certificate issuance…</div>';
fetch('/api/certs/issue/start', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({ domain, email })
})
.then(r => r.json())
.then(data => {
const result = data.ok ? data.data : data;
_issueRequestId = result.request_id;
if (!result.request_id) throw new Error('No request_id returned');
// If issuance already exists for this domain, follow the existing request
startIssuePoll(result.request_id);
})
.catch(e => {
showErrorToast('Failed to start issuance: ' + e.message);
// Fall back to input phase
document.getElementById('cert-wizard-input').style.display = '';
document.getElementById('cert-wizard-progress').style.display = 'none';
});
}
function startIssuePoll(requestId) {
_issueRequestId = requestId;
_issuePollHandle = setInterval(() => pollIssueStatus(requestId), 2000);
// Also poll immediately
pollIssueStatus(requestId);
}
function pollIssueStatus(requestId) {
fetch('/api/certs/issue/' + encodeURIComponent(requestId))
.then(r => r.json())
.then(data => {
const result = data.ok ? data.data : data;
renderIssueSteps(result.steps, result.status);
if (result.status === 'completed') {
clearInterval(_issuePollHandle);
_issuePollHandle = null;
document.getElementById('cert-close-progress').style.display = '';
showSuccessToast('Certificate issued for ' + result.domain);
} else if (result.status === 'failed') {
clearInterval(_issuePollHandle);
_issuePollHandle = null;
// Show failed — user can see which step failed
document.getElementById('cert-close-progress').style.display = '';
showErrorToast('Certificate issuance failed for ' + result.domain);
}
})
.catch(e => {
// Don't poll on error — but keep trying since request might still be running
});
}
function renderIssueSteps(steps, status) {
const container = document.getElementById('cert-steps-list');
if (!steps || !steps.length) {
container.innerHTML = '<div class="text-muted text-sm">Pending…</div>';
return;
}
container.innerHTML = steps.map(s => {
let icon;
if (s.status === 'done') icon = '<span class="status-dot status-up"></span>';
else if (s.status === 'running') icon = '<span class="status-dot status-pending"></span>';
else if (s.status === 'error') icon = '<span class="status-dot status-down"></span>';
else icon = '<span style="display:inline-block;width:8px;height:8px;border-radius:50%;background:var(--border);margin-right:6px;"></span>';
return '<div style="display:flex;align-items:center;gap:8px;margin-bottom:8px;font-size:13px;">' +
icon +
'<span>' + escHtml(s.label) + '</span>' +
(s.status === 'running' ? '<span class="text-muted text-sm">(in progress…)</span>' :
s.status === 'error' ? '<span class="badge badge-danger" style="margin-left:auto;">' + escHtml(s.message || 'failed') + '</span>' :
'<span class="badge badge-success" style="margin-left:auto;">done</span>') +
'</div>';
}).join('');
if (status === 'completed') {
container.innerHTML += '<div style="margin-top:12px;text-align:center;"><span class="badge badge-success" style="font-size:13px;padding:4px 12px;">✓ Certificate issued</span></div>';
}
}
+28 -10
View File
@@ -7,7 +7,7 @@
<h1>Certificates</h1>
<div class="subtitle">SSL/TLS certificate management</div>
</div>
<button class="btn btn-primary" onclick="openModal('issue-cert-modal')">+ Issue New Certificate</button>
<button class="btn btn-primary" onclick="resetIssueWizard(); openModal('issue-cert-modal')">+ Issue New Certificate</button>
</div>
<div class="card">
@@ -53,24 +53,42 @@
</table>
</div>
<!-- Issue Certificate Modal -->
<div class="modal-overlay" id="issue-cert-modal" onclick="if(event.target===this) closeModal('issue-cert-modal')">
<div class="modal">
<!-- Issue Certificate Modal — Phase 1: Validate -->
<div class="modal-overlay" id="issue-cert-modal" onclick="if(event.target===this) closeIssueWizard()">
<div class="modal" style="min-width:480px;">
<h2>Issue New Certificate</h2>
<form hx-post="/api/certs/issue" hx-encoding="json" hx-swap="none" hx-on::after-request="if(evt.detail.successful){ closeModal('issue-cert-modal'); refreshTable('/api/certs/list', document.getElementById('cert-rows'), renderCerts); showSuccessToast('Certificate issuance started'); }">
<!-- Phase 1: Input + Pre-flight Checks -->
<div id="cert-wizard-input">
<div class="form-group">
<label for="cert-domain">Domain</label>
<input type="text" id="cert-domain" name="domain" placeholder="example.com" required>
<input type="text" id="cert-domain" placeholder="example.com" required>
</div>
<div class="form-group">
<label for="cert-email">Contact Email</label>
<input type="email" id="cert-email" name="email" placeholder="admin@example.com" value="{{ (email or '') | e }}">
<input type="email" id="cert-email" placeholder="admin@example.com" value="{{ (email or '') | e }}">
</div>
<!-- Pre-flight validation results (shown after Check) -->
<div id="cert-check-results" style="display:none;">
<div class="section-title" style="margin-top:16px;">Pre-flight Checks</div>
<div id="cert-checks-list"></div>
</div>
<div class="modal-actions">
<button type="button" class="btn btn-outline" onclick="closeModal('issue-cert-modal')">Cancel</button>
<button type="submit" class="btn btn-primary">Issue</button>
<button type="button" class="btn btn-outline" onclick="closeIssueWizard()">Cancel</button>
<button type="button" id="cert-check-btn" class="btn btn-primary" onclick="validateCertIssue()">Check</button>
<button type="button" id="cert-issue-btn" class="btn btn-primary" style="display:none;" onclick="startCertIssue()">Issue</button>
</div>
</div>
<!-- Phase 2: Step progress -->
<div id="cert-wizard-progress" style="display:none;">
<div id="cert-steps-list"></div>
<div class="modal-actions">
<button type="button" class="btn btn-outline" id="cert-close-progress" style="display:none;" onclick="closeIssueWizard(); refreshTable('/api/certs/list', document.getElementById('cert-rows'), renderCerts);">Done</button>
</div>
</div>
</form>
</div>
</div>
{% endblock %}
+27 -15
View File
@@ -89,36 +89,34 @@
var refreshInterval = {{ (refresh_interval | default(15)) }};
var currentTab = 'journal';
function loadTabEl(el) {
var url = el.getAttribute('hx-get');
if (!url) return;
el.textContent = 'Loading...';
fetch(url).then(function(r) { return r.text(); })
.then(function(html) { el.innerHTML = html; })
.catch(function() { el.innerHTML = '<div class="log-line">(failed to load log)</div>'; });
}
function setActivePolling() {
document.querySelectorAll('.log-viewer').forEach(function(el) {
htmx.abort(el);
if (typeof htmx !== 'undefined') htmx.abort(el);
el.setAttribute('hx-trigger', 'none');
});
var activeEl = document.getElementById('log-' + currentTab);
if (activeEl) {
activeEl.setAttribute('hx-trigger', 'every ' + refreshInterval + 's');
}
if (typeof htmx !== 'undefined') htmx.process(document.body);
}
function loadActiveTab() {
var activeEl = document.getElementById('log-' + currentTab);
if (activeEl) {
htmx.ajax('GET', activeEl);
loadTabEl(activeEl);
}
}
var origSwitchTab = switchTab;
switchTab = function(tabName) {
currentTab = tabName;
if (typeof origSwitchTab === 'function') {
origSwitchTab(tabName);
}
if (document.getElementById('auto-refresh-toggle').checked) {
setActivePolling();
}
loadActiveTab();
};
function toggleAutoRefresh() {
var toggle = document.getElementById('auto-refresh-toggle');
if (toggle.checked) {
@@ -126,7 +124,7 @@ function toggleAutoRefresh() {
loadActiveTab();
} else {
document.querySelectorAll('.log-viewer').forEach(function(el) {
htmx.abort(el);
if (typeof htmx !== 'undefined') htmx.abort(el);
el.setAttribute('hx-trigger', 'none');
});
}
@@ -135,5 +133,19 @@ function toggleAutoRefresh() {
document.addEventListener('DOMContentLoaded', function() {
loadActiveTab();
});
window.addEventListener('load', function() {
var origSwitchTab = typeof switchTab === 'function' ? switchTab : null;
switchTab = function(tabName) {
currentTab = tabName;
if (origSwitchTab) {
origSwitchTab(tabName);
}
if (document.getElementById('auto-refresh-toggle').checked) {
setActivePolling();
}
loadActiveTab();
};
});
</script>
{% endblock %}