refactor: replace Jinja templates with static frontend pages
This commit is contained in:
+50
-365
@@ -12,15 +12,14 @@ 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
|
||||
from flask import Flask, abort, request
|
||||
from werkzeug.middleware.proxy_fix import ProxyFix
|
||||
|
||||
from daemon.client import get
|
||||
from daemon.iface import GET_STATUS_ALL
|
||||
from lib.logging import setup_logging
|
||||
from lib.network import get_config
|
||||
from webui.api.certs import bp as certs_bp
|
||||
from webui.api.dhcp import bp as dhcp_bp
|
||||
from webui.api.firewall import bp as firewall_bp
|
||||
@@ -79,6 +78,15 @@ signal.signal(signal.SIGHUP, _sighup_handler)
|
||||
app = Flask(__name__)
|
||||
app.config["SECRET_KEY"] = os.urandom(32).hex()
|
||||
|
||||
STATIC_DIR = Path(__file__).resolve().parent / "static"
|
||||
|
||||
# Cache-control: short TTL in dev, aggressive caching in prod (versioned assets)
|
||||
_DEV_MODE = bool(os.environ.get("VACUUM_WALL_DEV")) or False
|
||||
app.config["SEND_FILE_MAX_AGE_DEFAULT"] = 5 if _DEV_MODE else 31536000
|
||||
|
||||
# Flask is behind nginx — trust X-Forwarded-* headers for scheme/host detection
|
||||
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_prefix=1)
|
||||
|
||||
app.register_blueprint(firewall_bp, url_prefix="/api/firewall")
|
||||
app.register_blueprint(network_bp, url_prefix="/api/network")
|
||||
app.register_blueprint(dhcp_bp, url_prefix="/api/dhcp")
|
||||
@@ -100,7 +108,6 @@ BLUEPRINTS = [
|
||||
for name, _ in BLUEPRINTS:
|
||||
logger.info("Registered blueprint '%s' at /api/%s", name, name)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Request logging
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -132,387 +139,65 @@ def _log_request_finish(response):
|
||||
response.status_code,
|
||||
elapsed_ms,
|
||||
)
|
||||
|
||||
# Set cache headers: short in dev, long with staleness tolerance in prod
|
||||
if response.content_type.startswith("text/html"):
|
||||
# index.html: always short cache so browser revalidates
|
||||
response.headers["Cache-Control"] = "no-cache"
|
||||
elif response.content_type.startswith(("text/javascript", "text/css")):
|
||||
if _DEV_MODE:
|
||||
response.headers["Cache-Control"] = "max-age=5"
|
||||
else:
|
||||
response.headers["Cache-Control"] = (
|
||||
"public, max-age=31536000, stale-while-revalidate=86400"
|
||||
)
|
||||
|
||||
return response
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Jinja2 custom filters
|
||||
# API proxy routes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@app.template_filter("timestamp")
|
||||
def timestamp_filter(value):
|
||||
"""Convert an ISO-8601 timestamp string to ``YYYY-MM-DD HH:MM:SS``.
|
||||
@app.route("/api/status/all")
|
||||
def api_status_all():
|
||||
"""Return aggregated status from all subsystems.
|
||||
|
||||
Args:
|
||||
value: ISO timestamp string (may end with ``Z``).
|
||||
Proxies the daemon's ``/status/all`` endpoint for SPA consumption.
|
||||
|
||||
Returns:
|
||||
Formatted date string, or original value on parse failure.
|
||||
"""
|
||||
if not value:
|
||||
return ""
|
||||
try:
|
||||
dt = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
|
||||
return dt.strftime("%Y-%m-%d %H:%M:%S")
|
||||
except (ValueError, TypeError):
|
||||
return str(value)
|
||||
|
||||
|
||||
@app.template_filter("bytes")
|
||||
def bytes_filter(value):
|
||||
"""Convert a byte count to a human-readable size string (B/KB/MB…).
|
||||
|
||||
Args:
|
||||
value: Numeric byte count.
|
||||
|
||||
Returns:
|
||||
Formatted size string, or original value on parse failure.
|
||||
JSON response with state data for all subsystems.
|
||||
"""
|
||||
try:
|
||||
num = float(value)
|
||||
except (ValueError, TypeError):
|
||||
return str(value)
|
||||
if num < 0:
|
||||
return "0 B"
|
||||
for unit in ("B", "KB", "MB", "GB", "TB"):
|
||||
if abs(num) < 1024:
|
||||
return f"{num:.1f} {unit}"
|
||||
num /= 1024
|
||||
return f"{num:.1f} PB"
|
||||
|
||||
|
||||
@app.template_filter("duration")
|
||||
def duration_filter(value):
|
||||
"""Convert a duration in seconds to a human-readable string.
|
||||
|
||||
Args:
|
||||
value: Duration in seconds.
|
||||
|
||||
Returns:
|
||||
Formatted string (e.g. ``3d 2h 15m 30s``), or original value on failure.
|
||||
"""
|
||||
try:
|
||||
total = int(float(value))
|
||||
except (ValueError, TypeError):
|
||||
return str(value)
|
||||
if total < 0:
|
||||
return "0s"
|
||||
parts = []
|
||||
days, remainder = divmod(total, 86400)
|
||||
hours, remainder = divmod(remainder, 3600)
|
||||
minutes, seconds = divmod(remainder, 60)
|
||||
if days:
|
||||
parts.append(f"{days}d")
|
||||
if hours:
|
||||
parts.append(f"{hours}h")
|
||||
if minutes:
|
||||
parts.append(f"{minutes}m")
|
||||
parts.append(f"{seconds}s")
|
||||
return " ".join(parts)
|
||||
|
||||
|
||||
@app.template_filter("json_pretty")
|
||||
def json_pretty_filter(value):
|
||||
"""Serialize *value* as indented JSON for template display.
|
||||
|
||||
Args:
|
||||
value: Any JSON-serializable object.
|
||||
|
||||
Returns:
|
||||
Pretty-printed JSON string with 2-space indent.
|
||||
"""
|
||||
import json
|
||||
|
||||
try:
|
||||
return json.dumps(value, indent=2, default=str)
|
||||
except (TypeError, ValueError):
|
||||
return str(value)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Page routes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _safely(fn, default=None):
|
||||
"""Call *fn* and return *default* on any exception.
|
||||
|
||||
Args:
|
||||
fn: Zero-argument callable to execute.
|
||||
default: Fallback value returned when *fn* raises.
|
||||
|
||||
Returns:
|
||||
The result of ``fn()``, or *default* if an exception occurred.
|
||||
"""
|
||||
try:
|
||||
return fn()
|
||||
return {"ok": True, "data": get(GET_STATUS_ALL)}
|
||||
except Exception as exc:
|
||||
logger.warning("WebUI data load failed: %s", exc)
|
||||
return default
|
||||
logger.warning("Status all failed: %s", exc)
|
||||
return {"ok": False, "error": str(exc)}, 500
|
||||
|
||||
|
||||
def _get_service_status(dnsmasq_info, wg_info):
|
||||
"""Build a service status dict for the dashboard template.
|
||||
# ---------------------------------------------------------------------------
|
||||
# SPA catch-all
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Args:
|
||||
dnsmasq_info: Dnsmasq status payload from the daemon.
|
||||
wg_info: WireGuard status payload from the daemon.
|
||||
|
||||
Returns:
|
||||
Dict mapping service names to ``{running: bool}``.
|
||||
"""
|
||||
services = {}
|
||||
if dnsmasq_info:
|
||||
services["Dnsmasq"] = {
|
||||
"running": dnsmasq_info.get("service_active", False),
|
||||
}
|
||||
if wg_info:
|
||||
services["WireGuard"] = {
|
||||
"running": wg_info.get("up", False),
|
||||
}
|
||||
return services
|
||||
|
||||
|
||||
def _fw_config_get() -> dict[str, Any]:
|
||||
"""Read the current firewall config from the daemon."""
|
||||
return get("/firewall/config")
|
||||
|
||||
|
||||
def _load_status_all() -> dict[str, Any]:
|
||||
"""Load all subsystem status from the daemon in a single call."""
|
||||
return get("/status/all")
|
||||
SPA_DIR = STATIC_DIR
|
||||
|
||||
|
||||
@app.route("/")
|
||||
def root_redirect():
|
||||
"""Redirect root URL to the dashboard.
|
||||
@app.route("/<path:path>")
|
||||
def spa_page(path=""):
|
||||
"""Single-page application catch-all.
|
||||
|
||||
GET /
|
||||
|
||||
Returns:
|
||||
Redirect response to the dashboard page.
|
||||
Serves ``index.html`` (rendered as a Jinja2 template) for all non-API,
|
||||
non-static paths. The client-side router handles navigation and defaults
|
||||
to ``#dashboard``.
|
||||
"""
|
||||
from flask import redirect, url_for
|
||||
|
||||
return redirect(url_for("dashboard"))
|
||||
|
||||
|
||||
@app.route("/dashboard")
|
||||
def dashboard():
|
||||
"""Render the main dashboard overview page.
|
||||
|
||||
GET /
|
||||
|
||||
Template context:
|
||||
active_zones (dict): Active firewalld zones and bound interfaces.
|
||||
interfaces (list): Available network interfaces with zone bindings.
|
||||
dnsmasq (dict): Dnsmasq status information.
|
||||
domains (list): Configured proxy domains.
|
||||
certs (list): ACME certificate inventory.
|
||||
wg_status (dict): WireGuard tunnel status.
|
||||
services (dict): Service running indicators (Dnsmasq, WireGuard).
|
||||
firewall_config (dict): Declarative firewall JSON config.
|
||||
firewall_pending (dict): Pending firewall rules awaiting apply.
|
||||
"""
|
||||
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",
|
||||
active_zones=active_zones,
|
||||
interfaces=interfaces,
|
||||
dnsmasq=dnsmasq,
|
||||
domains=domains,
|
||||
certs=certs,
|
||||
wg_status=wg,
|
||||
services=_get_service_status(dnsmasq, wg),
|
||||
firewall_config=_safely(_fw_config_get, {}),
|
||||
firewall_pending=fw_state.get("pending", {}),
|
||||
)
|
||||
|
||||
|
||||
@app.route("/interfaces")
|
||||
def interfaces_page():
|
||||
"""Render the network interfaces management page.
|
||||
|
||||
GET /interfaces
|
||||
|
||||
Template context:
|
||||
interfaces (list): Available network interfaces.
|
||||
zones (list): Zone names bound to interfaces.
|
||||
firewall_config (dict): Declarative firewall JSON config.
|
||||
firewall_pending (dict): Pending firewall rules awaiting apply.
|
||||
"""
|
||||
all_status = _safely(_load_status_all, {})
|
||||
fw_state = all_status.get("firewall", {}) or {}
|
||||
network_config = _safely(get_config, {})
|
||||
return render_template(
|
||||
"interfaces.html",
|
||||
interfaces=fw_state.get("interfaces", []),
|
||||
network_config=network_config,
|
||||
zones=fw_state.get("active_zones", {}).keys() or [],
|
||||
firewall_config=_safely(_fw_config_get, {}),
|
||||
firewall_pending=fw_state.get("pending", {}),
|
||||
)
|
||||
|
||||
|
||||
@app.route("/zones")
|
||||
def zones_page():
|
||||
"""Render the firewall zones management page.
|
||||
|
||||
GET /zones
|
||||
|
||||
Template context:
|
||||
zones (list): All zone configurations.
|
||||
services (list): Available service identifiers for zone policies.
|
||||
firewall_config (dict): Declarative firewall JSON config.
|
||||
firewall_pending (dict): Pending firewall rules awaiting apply.
|
||||
"""
|
||||
all_status = _safely(_load_status_all, {})
|
||||
fw_state = all_status.get("firewall", {}) or {}
|
||||
return render_template(
|
||||
"zones.html",
|
||||
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():
|
||||
"""Render the firewall rich-rules editor page.
|
||||
|
||||
GET /rules
|
||||
|
||||
Template context:
|
||||
zones (list): Zone names containing rich rules.
|
||||
rules (dict | None): Zone name → rich rule mappings (``None`` if empty).
|
||||
"""
|
||||
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)
|
||||
|
||||
|
||||
@app.route("/nat")
|
||||
def nat_page():
|
||||
"""Render the NAT rules management page.
|
||||
|
||||
GET /nat
|
||||
|
||||
Template context:
|
||||
zones (list): Zone configurations containing NAT rules.
|
||||
"""
|
||||
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():
|
||||
"""Render the DHCP/Dnsmasq configuration page.
|
||||
|
||||
GET /dhcp
|
||||
|
||||
Template context:
|
||||
config (dict): Dnsmasq configuration settings.
|
||||
status (dict): Dnsmasq runtime status.
|
||||
leases (list): Current DHCP lease table.
|
||||
"""
|
||||
all_status = _safely(_load_status_all, {})
|
||||
dm_state = all_status.get("dnsmasq", {}) or {}
|
||||
fw_state = all_status.get("firewall", {}) or {}
|
||||
return render_template(
|
||||
"dhcp.html",
|
||||
config=dm_state.get("config", {}),
|
||||
status=dm_state.get("status", {}),
|
||||
leases=dm_state.get("leases", []),
|
||||
interfaces=fw_state.get("interfaces", []),
|
||||
)
|
||||
|
||||
|
||||
@app.route("/proxy")
|
||||
def proxy_page():
|
||||
"""Render the reverse proxy / SSL termination management page.
|
||||
|
||||
GET /proxy
|
||||
|
||||
Template context:
|
||||
domains (list): Configured proxy domains with upstream targets.
|
||||
config (dict): Nginx configuration settings.
|
||||
"""
|
||||
all_status = _safely(_load_status_all, {})
|
||||
ng_state = all_status.get("nginx", {}) or {}
|
||||
return render_template(
|
||||
"proxy.html",
|
||||
domains=ng_state.get("domains", []),
|
||||
config=ng_state.get("config", {}),
|
||||
)
|
||||
|
||||
|
||||
@app.route("/certs")
|
||||
def certs_page():
|
||||
"""Render the SSL certificate management page.
|
||||
|
||||
GET /certs
|
||||
|
||||
Template context:
|
||||
certs (list): ACME certificate inventory.
|
||||
email (str): Configured ACME registration email.
|
||||
"""
|
||||
all_status = _safely(_load_status_all, {})
|
||||
ac_state = all_status.get("acme", {}) or {}
|
||||
return render_template(
|
||||
"certs.html",
|
||||
certs=ac_state.get("certs", []),
|
||||
email=ac_state.get("email", ""),
|
||||
)
|
||||
|
||||
|
||||
@app.route("/wireguard")
|
||||
def wireguard_page():
|
||||
"""Render the WireGuard VPN management page.
|
||||
|
||||
GET /wireguard
|
||||
|
||||
Template context:
|
||||
config (dict): WireGuard tunnel configuration.
|
||||
status (dict): WireGuard runtime status.
|
||||
"""
|
||||
all_status = _safely(_load_status_all, {})
|
||||
wg_state = all_status.get("wireguard", {}) or {}
|
||||
return render_template(
|
||||
"wireguard.html",
|
||||
config=wg_state.get("config", {}),
|
||||
status=wg_state.get("status", {}),
|
||||
)
|
||||
|
||||
|
||||
@app.route("/logs")
|
||||
def logs_page():
|
||||
"""Render the system logs viewer page.
|
||||
|
||||
GET /logs
|
||||
"""
|
||||
return render_template("logs.html")
|
||||
if path.startswith("api/") or path.startswith("static/"):
|
||||
abort(404)
|
||||
scheme = "wss" if request.is_secure else "ws"
|
||||
ws_url = f"{scheme}://{request.host}/ws"
|
||||
html = (SPA_DIR / "index.html").read_text()
|
||||
return html.replace("__WS_URL__", ws_url)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user