Add HTMX-powered Web UI server over stdlib HTTP

JE-Chen · JE-Chen · commit 6325a69017bc · 2026-04-22T11:25:56.000+08:00
diff --git a/automation_file/__init__.py b/automation_file/__init__.py
@@ -217,6 +217,7 @@
     TCPActionServer,
     start_autocontrol_socket_server,
 )
+from automation_file.server.web_ui import WebUIServer, start_web_ui
 from automation_file.trigger import (
     FileWatcher,
     TriggerManager,
@@ -426,6 +427,8 @@ def __getattr__(name: str) -> Any:
     "render_metrics",
     "MetricsServer",
     "start_metrics_server",
+    "WebUIServer",
+    "start_web_ui",
     # Triggers
     "FileWatcher",
     "TriggerManager",
diff --git a/automation_file/server/web_ui.py b/automation_file/server/web_ui.py
@@ -0,0 +1,215 @@
+"""Read-only observability Web UI (stdlib + HTMX).
+
+Serves a single HTML page that polls three HTML fragments — registered
+actions, live progress, and health summary — using HTMX (loaded from a
+pinned CDN URL). Write operations are deliberately out of scope; trigger
+actions through :mod:`http_server` / :mod:`tcp_server` with their auth
+story intact.
+
+Loopback-only by default; ``allow_non_loopback=True`` is required to bind
+elsewhere. When ``shared_secret`` is supplied every request must carry
+``Authorization: Bearer <secret>`` — the rendered HTML includes a
+``hx-headers`` attribute so HTMX's polled requests carry the token.
+"""
+
+from __future__ import annotations
+
+import hmac
+import html as html_lib
+import json
+import threading
+import time
+from http import HTTPStatus
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+from automation_file.core.action_executor import executor
+from automation_file.core.progress import progress_registry
+from automation_file.logging_config import file_automation_logger
+from automation_file.server.network_guards import ensure_loopback
+
+_DEFAULT_HOST = "127.0.0.1"
+_DEFAULT_PORT = 9955
+_HTMX_CDN = "https://unpkg.com/htmx.org@1.9.12/dist/htmx.min.js"
+_HTMX_SRI = "sha384-ujb1lZYygJmzgSwoxRggbCHcjc0rB2XoQrxeTUQyRjrOnlCoYta87iKBWq3EsdM2"
+
+_INDEX_TEMPLATE = """<!doctype html>
+<html lang="en">
+<head>
+<meta charset="utf-8">
+<title>automation_file</title>
+<script src="{htmx_src}" integrity="{htmx_sri}" crossorigin="anonymous"></script>
+<style>
+  body {{ font-family: system-ui, sans-serif; margin: 2rem; color: #1d1f21; }}
+  h1 {{ font-size: 1.4rem; margin-bottom: 0.2rem; }}
+  h2 {{ font-size: 1.05rem; margin-top: 1.5rem; color: #555; }}
+  table {{ border-collapse: collapse; width: 100%; font-size: 0.9rem; }}
+  th, td {{ padding: 0.35rem 0.6rem; border-bottom: 1px solid #eee; text-align: left; }}
+  th {{ background: #f5f5f5; }}
+  .muted {{ color: #888; }}
+  code {{ background: #f3f3f3; padding: 0.1rem 0.3rem; border-radius: 3px; }}
+</style>
+</head>
+<body hx-headers='{auth_headers}'>
+<h1>automation_file</h1>
+<p class="muted">Read-only dashboard. Write operations live on the action server.</p>
+
+<h2>Health</h2>
+<div id="health" hx-get="/ui/health" hx-trigger="load, every 5s" hx-swap="innerHTML">
+  <em class="muted">loading…</em>
+</div>
+
+<h2>Progress</h2>
+<div id="progress" hx-get="/ui/progress" hx-trigger="load, every 2s" hx-swap="innerHTML">
+  <em class="muted">loading…</em>
+</div>
+
+<h2>Registered actions</h2>
+<div id="registry" hx-get="/ui/registry" hx-trigger="load, every 30s" hx-swap="innerHTML">
+  <em class="muted">loading…</em>
+</div>
+</body>
+</html>
+"""
+
+
+class _WebUIHandler(BaseHTTPRequestHandler):
+    """Serves the dashboard page plus its three HTMX fragment endpoints."""
+
+    def log_message(  # pylint: disable=arguments-differ
+        self, format_str: str, *args: object
+    ) -> None:
+        file_automation_logger.info("web_ui: " + format_str, *args)
+
+    def do_GET(self) -> None:  # pylint: disable=invalid-name
+        if not self._authorized():
+            self._send_html(HTTPStatus.UNAUTHORIZED, "<p>unauthorized</p>")
+            return
+        path = self.path.split("?", 1)[0]
+        if path in ("/", "/index.html"):
+            self._send_html(HTTPStatus.OK, self._render_index())
+            return
+        if path == "/ui/health":
+            self._send_html(HTTPStatus.OK, _render_health())
+            return
+        if path == "/ui/progress":
+            self._send_html(HTTPStatus.OK, _render_progress())
+            return
+        if path == "/ui/registry":
+            self._send_html(HTTPStatus.OK, _render_registry())
+            return
+        self._send_html(HTTPStatus.NOT_FOUND, "<p>not found</p>")
+
+    def _authorized(self) -> bool:
+        secret: str | None = getattr(self.server, "shared_secret", None)
+        if not secret:
+            return True
+        header = self.headers.get("Authorization", "")
+        if not header.startswith("Bearer "):
+            return False
+        return hmac.compare_digest(header[len("Bearer ") :], secret)
+
+    def _send_html(self, status: HTTPStatus, body: str) -> None:
+        payload = body.encode("utf-8")
+        self.send_response(status)
+        self.send_header("Content-Type", "text/html; charset=utf-8")
+        self.send_header("Content-Length", str(len(payload)))
+        self.send_header("Cache-Control", "no-store")
+        self.end_headers()
+        self.wfile.write(payload)
+
+    def _render_index(self) -> str:
+        secret: str | None = getattr(self.server, "shared_secret", None)
+        auth_headers_obj = {"Authorization": f"Bearer {secret}"} if secret else {}
+        auth_headers = html_lib.escape(json.dumps(auth_headers_obj), quote=True)
+        return _INDEX_TEMPLATE.format(
+            htmx_src=_HTMX_CDN,
+            htmx_sri=_HTMX_SRI,
+            auth_headers=auth_headers,
+        )
+
+
+def _render_health() -> str:
+    names = list(executor.registry.event_dict.keys())
+    return (
+        "<table>"
+        "<tr><th>process</th><td>alive</td></tr>"
+        f"<tr><th>registry size</th><td>{len(names)}</td></tr>"
+        f"<tr><th>time</th><td>{html_lib.escape(time.strftime('%Y-%m-%d %H:%M:%S'))}</td></tr>"
+        "</table>"
+    )
+
+
+def _render_progress() -> str:
+    snapshots = progress_registry.list()
+    if not snapshots:
+        return "<p class='muted'>no active transfers</p>"
+    rows = []
+    for item in snapshots:
+        name = html_lib.escape(str(item.get("name", "")))
+        status = html_lib.escape(str(item.get("status", "")))
+        transferred = int(item.get("transferred", 0) or 0)
+        total = item.get("total")
+        total_cell = "—" if total in (None, 0) else str(total)
+        pct = ""
+        if isinstance(total, int) and total > 0:
+            pct = f" ({(transferred / total) * 100:.1f}%)"
+        rows.append(
+            "<tr>"
+            f"<td><code>{name}</code></td>"
+            f"<td>{status}</td>"
+            f"<td>{transferred}{pct}</td>"
+            f"<td>{total_cell}</td>"
+            "</tr>"
+        )
+    return (
+        "<table>"
+        "<tr><th>name</th><th>status</th><th>transferred</th><th>total</th></tr>"
+        + "".join(rows)
+        + "</table>"
+    )
+
+
+def _render_registry() -> str:
+    names = sorted(executor.registry.event_dict.keys())
+    if not names:
+        return "<p class='muted'>registry empty</p>"
+    items = "".join(f"<li><code>{html_lib.escape(name)}</code></li>" for name in names)
+    return f"<ul>{items}</ul>"
+
+
+class WebUIServer(ThreadingHTTPServer):
+    """Threaded HTTP server for the HTMX dashboard."""
+
+    def __init__(
+        self,
+        server_address: tuple[str, int],
+        handler_class: type = _WebUIHandler,
+        shared_secret: str | None = None,
+    ) -> None:
+        super().__init__(server_address, handler_class)
+        self.shared_secret: str | None = shared_secret
+
+
+def start_web_ui(
+    host: str = _DEFAULT_HOST,
+    port: int = _DEFAULT_PORT,
+    allow_non_loopback: bool = False,
+    shared_secret: str | None = None,
+) -> WebUIServer:
+    """Start the Web UI server on a background thread."""
+    if not allow_non_loopback:
+        ensure_loopback(host)
+    if allow_non_loopback and not shared_secret:
+        file_automation_logger.warning(
+            "web_ui: non-loopback bind without shared_secret is insecure",
+        )
+    server = WebUIServer((host, port), shared_secret=shared_secret)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    file_automation_logger.info(
+        "web_ui: listening on %s:%d (auth=%s)",
+        host,
+        port,
+        "on" if shared_secret else "off",
+    )
+    return server
diff --git a/tests/test_web_ui.py b/tests/test_web_ui.py
@@ -0,0 +1,114 @@
+"""Tests for the HTMX Web UI server."""
+# pylint: disable=cyclic-import
+
+from __future__ import annotations
+
+import urllib.request
+
+import pytest
+
+from automation_file.core.action_executor import executor
+from automation_file.core.progress import progress_registry
+from automation_file.server.web_ui import start_web_ui
+from tests._insecure_fixtures import ipv4
+
+
+def _ensure_echo_registered() -> None:
+    if "test_webui_echo" not in executor.registry:
+        executor.registry.register("test_webui_echo", lambda value: value)
+
+
+def _get(url: str, headers: dict[str, str] | None = None) -> tuple[int, str]:
+    request = urllib.request.Request(url, headers=headers or {}, method="GET")
+    try:
+        with urllib.request.urlopen(request, timeout=3) as resp:  # nosec B310
+            return resp.status, resp.read().decode("utf-8")
+    except urllib.error.HTTPError as error:
+        return error.code, error.read().decode("utf-8")
+
+
+def test_index_returns_html_with_htmx_script() -> None:
+    _ensure_echo_registered()
+    server = start_web_ui(host="127.0.0.1", port=0)
+    host, port = server.server_address
+    try:
+        status, body = _get(f"http://{host}:{port}/")
+        assert status == 200
+        assert "htmx.org" in body
+        assert 'hx-get="/ui/progress"' in body
+        assert 'hx-get="/ui/registry"' in body
+        assert 'hx-get="/ui/health"' in body
+    finally:
+        server.shutdown()
+
+
+def test_registry_fragment_lists_actions() -> None:
+    _ensure_echo_registered()
+    server = start_web_ui(host="127.0.0.1", port=0)
+    host, port = server.server_address
+    try:
+        status, body = _get(f"http://{host}:{port}/ui/registry")
+        assert status == 200
+        assert "<ul>" in body or "registry empty" in body
+        assert "test_webui_echo" in body
+    finally:
+        server.shutdown()
+
+
+def test_progress_fragment_reflects_registry() -> None:
+    _ensure_echo_registered()
+    reporter, _ = progress_registry.create("webui_probe", total=100)
+    reporter.update(50)
+    server = start_web_ui(host="127.0.0.1", port=0)
+    host, port = server.server_address
+    try:
+        status, body = _get(f"http://{host}:{port}/ui/progress")
+        assert status == 200
+        assert "webui_probe" in body
+        assert "50.0%" in body
+    finally:
+        progress_registry.forget("webui_probe")
+        server.shutdown()
+
+
+def test_health_fragment_contains_registry_size() -> None:
+    _ensure_echo_registered()
+    server = start_web_ui(host="127.0.0.1", port=0)
+    host, port = server.server_address
+    try:
+        status, body = _get(f"http://{host}:{port}/ui/health")
+        assert status == 200
+        assert "registry size" in body
+        assert "alive" in body
+    finally:
+        server.shutdown()
+
+
+def test_rejects_non_loopback() -> None:
+    with pytest.raises(ValueError):
+        start_web_ui(host=ipv4(8, 8, 8, 8), port=0)
+
+
+def test_shared_secret_required_when_set() -> None:
+    _ensure_echo_registered()
+    server = start_web_ui(host="127.0.0.1", port=0, shared_secret="s3cr3t")
+    host, port = server.server_address
+    try:
+        status, _ = _get(f"http://{host}:{port}/")
+        assert status == 401
+        status, body = _get(f"http://{host}:{port}/", headers={"Authorization": "Bearer s3cr3t"})
+        assert status == 200
+        assert "Bearer s3cr3t" in body  # echoed into hx-headers for polling
+    finally:
+        server.shutdown()
+
+
+def test_unknown_path_returns_404() -> None:
+    _ensure_echo_registered()
+    server = start_web_ui(host="127.0.0.1", port=0)
+    host, port = server.server_address
+    try:
+        status, _ = _get(f"http://{host}:{port}/nope")
+        assert status == 404
+    finally:
+        server.shutdown()
