vacuum-wall/webui/server.py

"""
server.py - Vacuum Wall management WebUI entry point.

Serves the Flask application on 127.0.0.1:9090.  Nginx terminates SSL
and enforces basic authentication before proxying to this port.
"""

import logging
import os
import sys
import time
from datetime import datetime
from pathlib import Path

from flask import Flask, render_template, request

from lib.acme import get_email, list_certs
from lib.dnsmasq import get_config as dnsmasq_config
from lib.dnsmasq import get_lease_table
from lib.dnsmasq import get_status as dnsmasq_status
from lib.firewall import (
    config_pending,
    get_active_zones,
    get_interfaces,
    get_services,
    get_zone_info,
)
from lib.firewall import (
    get_config as fw_config_get,
)
from lib.logging import setup_logging
from lib.nginx import get_config as nginx_config
from lib.nginx import get_domains
from lib.wireguard import get_config as wg_config
from lib.wireguard import status as wg_status
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
from webui.api.logs import bp as logs_bp
from webui.api.proxy import bp as proxy_bp
from webui.api.wireguard import bp as wireguard_bp

# ---------------------------------------------------------------------------
# Logging — must be first so subsequent modules inherit the config
# ---------------------------------------------------------------------------

PROJECT_DIR = Path(__file__).resolve().parent.parent

setup_logging()
logger = logging.getLogger(__name__)

logger.info(
    "Python %s.%s.%s",
    sys.version_info.major,
    sys.version_info.minor,
    sys.version_info.micro,
)
logger.info("Project directory: %s", PROJECT_DIR)
logger.info("Process ID: %d", os.getpid())

# ---------------------------------------------------------------------------
# App factory
# ---------------------------------------------------------------------------

app = Flask(__name__)
app.config["SECRET_KEY"] = os.urandom(32).hex()

app.register_blueprint(firewall_bp, url_prefix="/api/firewall")
app.register_blueprint(dhcp_bp, url_prefix="/api/dhcp")
app.register_blueprint(proxy_bp, url_prefix="/api/proxy")
app.register_blueprint(certs_bp, url_prefix="/api/certs")
app.register_blueprint(wireguard_bp, url_prefix="/api/wireguard")
app.register_blueprint(logs_bp, url_prefix="/api/logs")

BLUEPRINTS = [
    ("firewall", firewall_bp),
    ("dhcp", dhcp_bp),
    ("proxy", proxy_bp),
    ("certs", certs_bp),
    ("wireguard", wireguard_bp),
    ("logs", logs_bp),
]

for name, _ in BLUEPRINTS:
    logger.info("Registered blueprint '%s' at /api/%s", name, name)


# ---------------------------------------------------------------------------
# Request logging
# ---------------------------------------------------------------------------


@app.before_request
def _log_request_start():
    request._start_time = time.monotonic()


@app.after_request
def _log_request_finish(response):
    elapsed_ms = (
        time.monotonic() - getattr(request, "_start_time", time.monotonic())
    ) * 1000
    logger.info(
        "%s %s -> %d (%.1f ms)",
        request.method,
        request.path,
        response.status_code,
        elapsed_ms,
    )
    return response


# ---------------------------------------------------------------------------
# Jinja2 custom filters
# ---------------------------------------------------------------------------


@app.template_filter("timestamp")
def timestamp_filter(value):
    """Convert an ISO timestamp string to a human-readable date."""
    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):
    """Format a byte count to a human-readable string (KB / MB / GB)."""
    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):
    """Format a duration in seconds to a human-readable string."""
    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):
    """Pretty-print a JSON-serialisable value for debug displays."""
    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."""
    try:
        return fn()
    except Exception as exc:
        logger.warning("WebUI data load failed: %s", exc)
        return default


def _get_service_status(dnsmasq_info, wg_info):
    """Build a service status dict for the dashboard template."""
    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


@app.route("/")
def dashboard():
    active_zones = _safely(get_active_zones, {})
    interfaces = _safely(get_interfaces, [])
    dnsmasq = _safely(dnsmasq_status, {})
    domains = _safely(get_domains, [])
    certs = _safely(list_certs, [])
    wg = _safely(wg_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=_safely(config_pending, {}),
    )


@app.route("/interfaces")
def interfaces_page():
    firewall_config = _safely(fw_config_get, {})
    firewall_pending = _safely(config_pending, {})
    return render_template(
        "interfaces.html",
        interfaces=_safely(get_interfaces, []),
        active_zones=_safely(get_active_zones, {}),
        firewall_config=firewall_config,
        firewall_pending=firewall_pending,
    )


@app.route("/zones")
def zones_page():
    firewall_config = _safely(fw_config_get, {})
    firewall_pending = _safely(config_pending, {})
    zones_data = {}
    for name in _safely(get_active_zones, {}):
        zones_data[name] = _safely(lambda n=name: get_zone_info(n), {})
    return render_template(
        "zones.html",
        zones=zones_data,
        interfaces=_safely(get_interfaces, []),
        services=_safely(get_services, []),
        firewall_config=firewall_config,
        firewall_pending=firewall_pending,
    )


@app.route("/rules")
def rules_page():
    zones = list(_safely(get_active_zones, {}).keys())
    raw = _safely(fw_config_get, {})
    rules = {}
    for zname, zcfg in raw.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():
    zones = {}
    for name in _safely(get_active_zones, {}):
        zones[name] = _safely(lambda n=name: get_zone_info(n), {})
    return render_template("nat.html", zones=zones)


@app.route("/dhcp")
def dhcp_page():
    return render_template(
        "dhcp.html",
        config=_safely(dnsmasq_config, {}),
        status=_safely(dnsmasq_status, {}),
        leases=_safely(get_lease_table, []),
    )


@app.route("/proxy")
def proxy_page():
    return render_template(
        "proxy.html", domains=_safely(get_domains, []), config=_safely(nginx_config, {})
    )


@app.route("/certs")
def certs_page():
    return render_template(
        "certs.html", certs=_safely(list_certs, []), email=_safely(get_email, "")
    )


@app.route("/wireguard")
def wireguard_page():
    return render_template(
        "wireguard.html", config=_safely(wg_config, {}), status=_safely(wg_status, {})
    )


@app.route("/logs")
def logs_page():
    return render_template("logs.html")


if __name__ == "__main__":
    logger.info("Starting Flask on 127.0.0.1:9090")
    app.run(host="127.0.0.1", port=9090)