"""aiohttp server for vacuum-walld.
Listens on a Unix socket, serves the daemon API to the web UI.
Handles routing, batching, and request/response lifecycle.
"""
import asyncio
import hashlib
import json
import logging
import os
import re
import signal
import time
from collections.abc import Callable
from pathlib import Path
from typing import Any
from aiohttp import web
from daemon.iface import PathLike
from lib.auth import blacklist_expired
from lib.state import _DEFAULT_POLL_INTERVALS
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"
_WS_PORT = int(os.environ.get("VACUUM_WALLD_WS_PORT", "9091"))
# Polling intervals per subsystem (seconds). VACUUM_WALL_POLL_INTERVALS overrides.
_RAW_POLL = os.environ.get("VACUUM_WALL_POLL_INTERVALS", "")
if _RAW_POLL:
_POLL_OVERRIDE: dict[str, int] = {}
for pair in _RAW_POLL.split(","):
if ":" in pair:
name, _, val = pair.partition(":")
try:
parsed = int(val.strip())
if parsed <= 0:
logger.warning(
"Invalid poll interval value %r for %r (must be > 0), skipping",
val,
name,
)
continue
_POLL_OVERRIDE[name.strip()] = parsed
except ValueError:
logger.warning(
"Invalid poll interval value %r for %r, skipping", val, name
)
_POLL_INTERVALS = {**_DEFAULT_POLL_INTERVALS, **_POLL_OVERRIDE}
else:
_POLL_INTERVALS = dict(_DEFAULT_POLL_INTERVALS)
class Handler:
"""Wrapper for a daemon handler function.
Attributes:
method: HTTP method (e.g. "GET", "POST").
path: URL path pattern.
"""
def __init__(self, method: str, path: str) -> None:
"""Initialize the handler metadata.
Args:
method: HTTP method.
path: URL path pattern.
"""
self.method = method.upper()
self.path = path
class Registry:
"""Route registry for daemon handlers.
Attributes:
_routes: Mapping of (method, path) tuples to handler callables.
"""
def __init__(self) -> None:
"""Initialize an empty route registry."""
self._routes: dict[tuple[str, str], Callable] = {}
def register(self, method: PathLike, path: str | None = None):
"""Decorator that registers a handler for the given method and path.
Accepts either two separate arguments (``method``, ``path``) or a
single :class:`daemon.iface.Endpoint` tuple.
Args:
method: HTTP method string, or an :class:`Endpoint` tuple.
path: URL path (omit when passing an :class:`Endpoint`).
Returns:
Decorator function wrapping the handler.
"""
if isinstance(method, tuple):
ep_method, ep_path = method
path = ep_path
method = ep_method
def decorator(fn: Callable) -> Callable:
"""Inner decorator that stores *fn* in the registry and attaches Handler metadata."""
self._routes[(method.upper(), path)] = fn # type: ignore[arg-type]
fn._handler = Handler(method, path) # type: ignore[attr-defined,reportArgumentType]
return fn
return decorator
def get(self, method: str, path: str) -> Callable | None:
"""Look up a registered handler by method and path.
Args:
method: HTTP method to match.
path: URL path to match.
Returns:
The handler callable, or None if not registered.
"""
return self._routes.get((method.upper(), path))
def match(self, method: str, path: str):
"""Match *path* against registered patterns, returning handler + params.
Patterns may contain ```` segments (e.g. ``/foo/``).
Matching segments are captured into a dict and merged into *body*.
Returns:
Tuple of (handler_fn, params_dict) or (None, None) if no match.
"""
for (reg_method, reg_path), fn in self._routes.items():
if reg_method != method.upper():
continue
pat_params = _match_path(reg_path, path)
if pat_params is not None:
return fn, pat_params
return None, None
registry = Registry()
def refresh_state(subsystems: list[str] | None = None) -> None:
"""Refresh the pre-computed state for the given subsystems (or all).
Args:
subsystems: List of subsystem names to refresh. If None, all subsystems are refreshed.
"""
state_store.populate(subsystems)
targets = subsystems or state_store.SUBSYSTEMS
for name in targets:
state_store.bump(name)
try:
asyncio.get_running_loop()
except RuntimeError:
pass
else:
# Broadcast each refreshed subsystem individually (gather for parallelism)
async def _broadcast_all():
await asyncio.gather(
*[broadcast_versions(name) for name in targets],
return_exceptions=True,
)
task = asyncio.create_task(_broadcast_all())
task.add_done_callback(_ws_tasks.discard)
_ws_tasks.add(task)
class NotFoundError(Exception):
"""Raised when a requested resource is not found."""
pass
class ConflictError(Exception):
"""Raised when a request conflicts with an existing resource."""
pass
def ok(data: Any = None) -> web.Response:
"""Create a success JSON response.
Args:
data: Response payload. Defaults to None.
Returns:
A JSON Response with `ok` set to True.
"""
return web.json_response({"ok": True, "data": data})
def error(msg: str, code: int = 400) -> web.Response:
"""Create an error JSON response.
Args:
msg: Error message.
code: HTTP status code. Defaults to 400.
Returns:
A JSON Response with `ok` set to False.
"""
return web.json_response({"ok": False, "error": msg}, status=code)
def _match_path(pattern: str, path: str) -> dict[str, str] | None:
"""Match *path* against a URL pattern containing ```` segments.
Args:
pattern: URL pattern like ``/network/interfaces/``.
path: Actual request path like ``/network/interfaces/eth1``.
Returns:
Dict mapping param names to their matched values, or ``None`` if no match.
"""
p_parts = pattern.strip("/").split("/")
r_parts = path.strip("/").split("/")
if len(p_parts) != len(r_parts):
return None
params: dict[str, str] = {}
for p_seg, r_seg in zip(p_parts, r_parts, strict=True):
if p_seg.startswith("<") and p_seg.endswith(">"):
params[p_seg[1:-1]] = r_seg
elif p_seg != r_seg:
return None
return params
async def _handle_request(request: web.Request) -> web.Response:
"""Dispatch a request to the appropriate handler.
Args:
request: The incoming HTTP request.
Returns:
The handler's response.
"""
handler_fn, pat_params = registry.match(request.method, request.path)
if handler_fn is None:
return error(f"Method {request.method} not allowed for {request.path}", 404)
# Build body — merge order (highest wins): path params > JSON body > query params.
# Path params come from the URL path (e.g. /interfaces/eth0) and should not
# be overridable by body or query parameters. This prevents callers from
# spoofing path-scoped parameters via request body.
body: dict[str, Any] | None = pat_params if pat_params else None
if request.content_type == "application/json":
try:
json_body = await request.json()
body = {**json_body, **body} if body is not None else json_body
except json.JSONDecodeError:
return error("Invalid JSON body", 400)
query_dict = dict(request.query)
if query_dict:
query_body = {k: v[0] if len(v) == 1 else v for k, v in query_dict.items()}
body = {**query_body, **body} if body is not None else query_body
try:
if body is not None:
result = handler_fn(request, body)
if asyncio.iscoroutine(result):
result = await result
else:
result = handler_fn(request, None)
if asyncio.iscoroutine(result):
result = await result
except NotFoundError as exc:
return error(str(exc), 404)
except ConflictError as exc:
return error(str(exc), 409)
except ValueError as exc:
return error(str(exc), 400)
except RuntimeError as exc:
logger.error("Handler error: %s", exc)
return error(str(exc), 500)
except Exception as exc:
logger.exception(
"Unexpected handler error in %s %s", request.method, request.path
)
return error(f"Internal error: {exc}", 500)
# Convert result to response if not already
if isinstance(result, web.Response):
return result
if isinstance(result, dict) and result.get("ok") is False:
return error(result["error"], result.get("code", 400))
return ok(result)
async def _handle_batch(request: web.Request) -> web.Response:
"""Handle batch requests: execute operations in order, return keyed results.
Args:
request: The incoming HTTP request containing operations.
Returns:
A JSON response keyed by operation IDs.
"""
try:
body = await request.json()
except json.JSONDecodeError:
return error("Invalid JSON body", 400)
ops = body.get("ops", [])
if not isinstance(ops, list):
return error("'ops' must be a list", 400)
results: dict[str, Any] = {}
for op in ops:
op_id = op.get("id")
method = op.get("method", "GET").upper()
path = op.get("path", "")
if not op_id or not path:
results[op_id] = {"ok": False, "error": "'id' and 'path' are required"}
continue
handler_fn, pat_params = registry.match(method, path)
if handler_fn is None:
results[op_id] = {
"ok": False,
"error": f"Endpoint not found: {method} {path}",
}
continue
op_body = op.get("body")
if pat_params:
op_body = {**(op_body or {}), **pat_params}
try:
result = handler_fn(None, op_body)
if asyncio.iscoroutine(result):
result = await result
except Exception as exc:
logger.error("Batch handler error for %s: %s", op_id, exc)
results[op_id] = {"ok": False, "error": str(exc)}
continue
# Strip ok/data wrapper for batch results
if isinstance(result, dict) and result.get("ok") is not None:
results[op_id] = result
else:
results[op_id] = {"ok": True, "data": result}
return ok(results)
def create_app() -> web.Application:
"""Create and configure the aiohttp application.
Returns:
A configured web.Application instance.
"""
app = web.Application()
app.router.add_route("GET", "/health", _health)
app.router.add_route("POST", "/status/refresh", refresh_status)
app.router.add_route("POST", "/batch", _handle_batch)
app.router.add_route("GET", "/ws", _handle_ws)
app.router.add_route("*", "/{tail:.*}", _catch_all)
return app
# JWTs are dot-joined base64url segments — a subset of the RFC 6455 token
# character set. Browsers therefore carry the access token as the
# Sec-WebSocket-Protocol subprotocol name itself (see websocket.js); the
# "Bearer " form is not a valid subprotocol (space is not a token
# character) and is rejected by the WebSocket constructor.
_JWT_SUBPROTOCOL_RE = re.compile(r"^[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+$")
def _extract_ws_token(subprotocols: list[str]) -> tuple[str | None, str | None]:
"""Extract the auth JWT from parsed Sec-WebSocket-Protocol names.
Accepts the raw JWT as a subprotocol name (bundled client) or the
legacy "Bearer " form (non-browser clients that can send spaces).
Args:
subprotocols: Parsed subprotocol names (already stripped/split).
Returns:
``(token, matched_subprotocol)`` or ``(None, None)`` if no usable
auth subprotocol was found.
"""
for proto in subprotocols:
if _JWT_SUBPROTOCOL_RE.fullmatch(proto):
return proto, proto
for proto in subprotocols:
if proto.startswith("Bearer "):
token = proto[7:].strip()
if token:
return token, proto
return None, None
# WebSocket subscribers
_ws_subscribers: set[web.WebSocketResponse] = set()
_ws_tasks: set[asyncio.Task[None]] = set()
_poll_tasks: set[asyncio.Task[None]] = set()
_last_blacklist_cleanup: float = 0
_last_blacklist_cleanup_lock: asyncio.Lock | None = None
async def _handle_ws(request: web.Request) -> web.Response:
"""WebSocket endpoint for real-time state streaming.
On connect: sends a full state snapshot of every subsystem
({type: snapshot, data: {subsystem: state, …}}). On state change:
broadcasts a data-carrying per-subsystem delta (versions or tick).
Clients disconnect to unsubscribe.
Authentication: JWT access token passed via:
1. WebSocket subprotocol header — the bundled client sends the raw JWT
as the subprotocol name (Sec-WebSocket-Protocol must carry a valid
RFC 6455 token; a JWT is one, "Bearer " is not).
2. "Bearer " subprotocol — legacy form for non-browser clients.
3. X-Auth-Token header — fallback for custom nginx setups that inject
it (not set by the bundled nginx config).
"""
from aiohttp import hdrs
from lib.auth import validate_token
token_param = None
matched_proto = None
# Prefer the subprotocol header. Sec-WebSocket-Protocol is a
# comma-separated list; parse it the same way aiohttp's own handshake
# does (Request has no subprotocol helper).
protocol_header = request.headers.get(hdrs.SEC_WEBSOCKET_PROTOCOL, "")
subprotocols = [p.strip() for p in protocol_header.split(",") if p.strip()]
token_param, matched_proto = _extract_ws_token(subprotocols)
if token_param is None:
token_param = request.headers.get("X-Auth-Token")
if token_param is None:
return web.json_response(
{"ok": False, "error": "authentication required"}, status=401
)
# Validate token. Session binding is skipped because browsers cannot send
# custom headers on WebSocket connections (no X-Session-Id available).
payload = validate_token(
token_param,
token_type="access",
)
if payload is None:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
# Negotiate only the matched auth subprotocol (or all if token came from header)
ws = web.WebSocketResponse(
protocols=[matched_proto] if matched_proto else subprotocols
)
await ws.prepare(request)
_ws_subscribers.add(ws)
snapshot = state_store.get_snapshot()
await ws.send_json({"type": "snapshot", "data": snapshot})
try:
async for msg in ws:
if msg.type == web.WSMsgType.ERROR:
break
if msg.type == web.WSMsgType.CLOSE:
break
finally:
_ws_subscribers.discard(ws)
return ws
async def _send_all(message: str) -> None:
"""Send a message string to all WS subscribers, removing dead ones."""
dead: set[web.WebSocketResponse] = set()
for ws in _ws_subscribers:
try:
await ws.send_str(message)
except Exception:
dead.add(ws)
_ws_subscribers.difference_update(dead)
if dead:
logger.warning("Removed %d dead WS subscribers", len(dead))
async def broadcast_versions(subsystem: str) -> None:
"""Send {type: versions} + full subsystem data to WS clients.
NOTE: Does NOT call state_store.bump(). Callers who need version bumps
(e.g. refresh_state) call bump themselves. poll_loop bumps before calling.
No legacy `updated` field is emitted (no backward compat) — version
counters still advance but are not sent over the wire.
"""
data = state_store.get(subsystem)
if data is None:
# Collector failed during the triggering populate/refresh — populate()
# cleared this subsystem to None (state.py). Skip the broadcast:
# a null payload would overwrite good client data. The next successful
# poll or mutation broadcasts the real value.
return
message = json.dumps(
{
"type": "versions",
"subsystem": subsystem,
"data": data,
}
)
await _send_all(message)
async def broadcast_tick(subsystem: str) -> None:
"""Send {type: tick} with the changed subsystem data.
No bump — tick is volatile-only; the version counter only bumps on
structural changes.
"""
data = state_store.get(subsystem)
message = json.dumps(
{
"type": "tick",
"subsystem": subsystem,
"data": data,
}
)
await _send_all(message)
async def _poll_loop(subsystem: str, interval: int) -> None:
"""Periodically poll a subsystem for state changes and broadcast as needed."""
global _last_blacklist_cleanup, _last_blacklist_cleanup_lock
# Lazy-init lock (requires running event loop)
if _last_blacklist_cleanup_lock is None:
_last_blacklist_cleanup_lock = asyncio.Lock()
offset = int(hashlib.md5(subsystem.encode()).hexdigest(), 16) % interval
await asyncio.sleep(offset)
while True:
try:
structural, volatile = state_store.poll(subsystem)
if structural:
state_store.bump(subsystem)
await broadcast_versions(subsystem) # now per-subsystem, carries data
elif volatile:
await broadcast_tick(subsystem) # now per-subsystem, carries data
# Periodic blacklist cleanup — coordinated across all poll loops
async with _last_blacklist_cleanup_lock:
now = time.time()
if now - _last_blacklist_cleanup >= 60:
blacklist_expired()
_last_blacklist_cleanup = now
except asyncio.CancelledError:
raise
except Exception:
logger.error("Poll loop error for %s", subsystem, exc_info=True)
await asyncio.sleep(interval)
def start_polling(loop: asyncio.AbstractEventLoop) -> None:
"""Start one poll loop task per subsystem."""
for subsystem, interval in _POLL_INTERVALS.items():
task = loop.create_task(_poll_loop(subsystem, interval))
task.add_done_callback(_poll_tasks.discard)
_poll_tasks.add(task)
def _stop_polling() -> None:
"""Cancel all polling tasks."""
for task in _poll_tasks:
task.cancel()
_poll_tasks.clear()
async def _health(_request: web.Request) -> web.Response:
"""Return the health check response.
Returns:
JSON response with process ID and socket path.
"""
return ok({"pid": os.getpid(), "socket": str(SOCKET_PATH)})
async def refresh_status(_request: web.Request) -> web.Response:
"""Re-collect all state from system.
Args:
_request: The incoming request. JSON body may contain a "subsystems" list.
Returns:
JSON response with the updated state snapshot.
"""
try:
body = await _request.json()
except (json.JSONDecodeError, ValueError):
body = None
subsystems = body.get("subsystems") if body else None
state_store.populate(subsystems)
targets = subsystems or state_store.SUBSYSTEMS
snapshot = {name: state_store.get(name) for name in targets}
# Broadcast to all WS clients (fire-and-forget, gather for parallelism).
# Deliberately no version bump — versions advance on structural poll
# diffs and on refresh_state() only.
async def _broadcast_all():
await asyncio.gather(
*[broadcast_versions(name) for name in targets],
return_exceptions=True,
)
task = asyncio.create_task(_broadcast_all())
task.add_done_callback(_ws_tasks.discard)
_ws_tasks.add(task)
return ok(snapshot)
async def _catch_all(request: web.Request) -> web.Response:
"""Catch-all for registered routes.
Args:
request: The incoming HTTP request.
Returns:
The dispatched handler's response.
"""
return await _handle_request(request)
def _register_routes() -> None:
"""Import all handler modules to register routes.
Side effect: each imported module's `@registry.register` calls populate the
route registry with their handler endpoints.
"""
from daemon.handlers import (
acme, # noqa: F401
auth, # noqa: F401
dnsmasq, # noqa: F401
firewall, # noqa: F401
logs, # noqa: F401
network, # noqa: F401
nginx, # noqa: F401
status, # noqa: F401
system, # noqa: F401
wireguard, # noqa: F401
)
def main() -> None:
"""Entry point for vacuum-walld.
Sets up logging, registers routes, creates the aiohttp application,
and starts listening on the Unix socket.
"""
from lib.logging import setup_logging
setup_logging()
# Startup checks
try:
from lib import webauthn as lib_webauthn
lib_webauthn.check_webauthn_config()
except Exception:
pass # ignore if webauthn module import failed
_register_routes()
app = create_app()
socket_path = os.environ.get("VACUUM_WALLD_SOCKET", str(SOCKET_PATH))
socket_dir = Path(socket_path).parent
socket_dir.mkdir(parents=True, exist_ok=True)
if Path(socket_path).exists():
os.unlink(socket_path)
loop = asyncio.new_event_loop()
def _teardown_exception_handler(_loop, context) -> None:
# Swallow teardown noise ("Task was destroyed but it is pending",
# in-flight task exceptions on SIGTERM) instead of the default
# logging-error tracebacks.
logger.debug("Suppressed teardown exception: %s", context)
async def _shutdown() -> None:
"""Graceful shutdown: stop accepting, drain in-flight work, teardown.
Bounded grace periods + a suppressed exception handler during the
teardown window avoid the "Task was destroyed but it is pending" and
logging-error tracebacks that otherwise appear on SIGTERM.
"""
logger.info("Shutting down daemon...")
_stop_polling()
# Suppress the default exception handler during teardown so that
# cancelling in-flight tasks does not spew tracebacks on SIGTERM.
prev_handler = loop.get_exception_handler()
loop.set_exception_handler(_teardown_exception_handler)
try:
# Stop accepting new connections (also waits for open sockets,
# bounded so a stuck WebSocket can't hang shutdown).
try:
await asyncio.wait_for(runner.cleanup(), timeout=5)
except TimeoutError:
logger.warning("Runner cleanup timed out, abandoning")
# Give in-flight request/WS tasks a bounded grace period to
# finish; cancel anything still pending so they are not
# "destroyed but pending" when the loop closes.
pending = [
t
for t in asyncio.all_tasks()
if t is not asyncio.current_task() and not t.done()
]
if pending:
_, still_pending = await asyncio.wait(pending, timeout=3)
for t in still_pending:
t.cancel()
if still_pending:
await asyncio.wait(still_pending, timeout=1)
finally:
loop.set_exception_handler(prev_handler)
if Path(socket_path).exists():
os.unlink(socket_path)
logger.info("vacuum-walld stopped")
loop.call_soon(loop.stop)
for sig in (signal.SIGTERM, signal.SIGINT):
loop.add_signal_handler(sig, lambda: loop.create_task(_shutdown()))
runner = web.AppRunner(app)
loop.run_until_complete(runner.setup())
site = web.UnixSite(runner, socket_path)
loop.run_until_complete(site.start())
tcp_site = web.TCPSite(runner, "127.0.0.1", _WS_PORT)
loop.run_until_complete(tcp_site.start())
os.chmod(socket_path, 0o660)
# Import system configs → JSON (blocking — OK at startup)
from lib.system_import import import_all
reconciled = import_all()
if reconciled:
logger.info("Reconciled subsystems: %s", ", ".join(reconciled))
# Reopen group access on the ACME home before the first acme.sh
# collection: a tree left owner-only by a prior run (e.g. a manual
# run as the WebUI user) would otherwise fail every daemon acme.sh
# call until the next issue/renew. Never fatal at startup.
try:
from daemon.handlers.acme import normalize_acme_home
normalize_acme_home()
except Exception:
logger.warning("ACME home normalization failed at startup", exc_info=True)
# Populate state from system (blocking — OK at startup)
logger.info("Populating system state...")
state_store.populate()
for subsystem in state_store.SUBSYSTEMS:
if state_store.get(subsystem) is not None:
state_store.bump(subsystem)
start_polling(loop)
logger.info("vacuum-walld listening on %s", socket_path)
logger.info("WebSocket on 127.0.0.1:%d", _WS_PORT)
try:
loop.run_forever()
finally:
# _shutdown() handles cleanup when invoked via signal handler;
# this block is only reached if shutdown didn't happen cleanly
# (e.g., unexpected exit), in which case we unlink the socket.
if Path(socket_path).exists():
os.unlink(socket_path)
if __name__ == "__main__":
main()