# plugins/host_agent/routes.py from __future__ import annotations import hashlib from datetime import datetime, timezone from typing import Any from pathlib import Path import secrets from quart import Blueprint, current_app, jsonify, request, Response, render_template, redirect, url_for from roundtable.auth.middleware import require_role from roundtable.models.users import UserRole from sqlalchemy import select, func from datetime import timedelta from roundtable.core.settings import public_base_url from roundtable.models.hosts import Host from roundtable.models.metrics import PluginMetric from .models import HostAgentRegistration host_agent_bp = Blueprint("host_agent", __name__, template_folder="templates") SOURCE_MODULE = "host_agent" def _hash_token(raw: str) -> str: return hashlib.sha256(raw.encode("utf-8")).hexdigest() def _parse_ts(ts: str) -> datetime: if ts.endswith("Z"): ts = ts[:-1] + "+00:00" return datetime.fromisoformat(ts) def _expand_sample_to_metrics( sample: dict, host_name: str, recorded_at: datetime ) -> list[PluginMetric]: rows: list[PluginMetric] = [] def row(metric: str, resource: str, value: float) -> None: rows.append(PluginMetric( source_module=SOURCE_MODULE, resource_name=resource, metric_name=metric, value=float(value), recorded_at=recorded_at, )) if sample.get("cpu_pct") is not None: row("cpu_pct", host_name, sample["cpu_pct"]) mem = sample.get("mem") or {} if mem.get("total_bytes"): total = mem["total_bytes"] available = mem.get("available_bytes", 0) used_pct = 100.0 * (total - available) / total if total else 0.0 row("mem_used_pct", host_name, used_pct) row("mem_available_bytes", host_name, available) row("swap_used_bytes", host_name, mem.get("swap_used_bytes", 0)) load = sample.get("load") or {} for key in ("1m", "5m", "15m"): if key in load: row(f"load_{key}", host_name, load[key]) if sample.get("uptime_secs") is not None: row("uptime_secs", host_name, sample["uptime_secs"]) worst_pct = 0.0 for disk in sample.get("storage") or []: total = disk.get("total_bytes", 0) used = disk.get("used_bytes", 0) pct = 100.0 * used / total if total else 0.0 mount_res = f"{host_name}:{disk['mount']}" row("disk_used_pct", mount_res, pct) row("disk_used_bytes", mount_res, used) row("disk_total_bytes", mount_res, total) if pct > worst_pct: worst_pct = pct if sample.get("storage"): row("disk_used_pct_worst", host_name, worst_pct) return rows def _error(status: int, code: str, detail: str | None = None) -> tuple[Any, int]: body: dict = {"ok": False, "error": code} if detail: body["detail"] = detail return jsonify(body), status @host_agent_bp.post("/ingest") async def ingest(): auth = request.headers.get("Authorization", "") if not auth.startswith("Bearer "): return _error(401, "invalid_token") raw = auth[len("Bearer "):].strip() if not raw: return _error(401, "invalid_token") try: payload = await request.get_json(force=True) except Exception: return _error(400, "malformed_payload", "invalid JSON") if not isinstance(payload, dict) or "samples" not in payload: return _error(400, "malformed_payload", "missing 'samples'") samples = payload.get("samples") or [] if not isinstance(samples, list) or not samples: return _error(400, "malformed_payload", "'samples' must be a non-empty list") token_hash = _hash_token(raw) async with current_app.db_sessionmaker() as session: reg = (await session.execute( select(HostAgentRegistration).where( HostAgentRegistration.token_hash == token_hash) )).scalar_one_or_none() if reg is None: return _error(401, "invalid_token") host = (await session.execute( select(Host).where(Host.id == reg.host_id) )).scalar_one_or_none() if host is None: return _error(401, "invalid_token") accepted = 0 latest_ts: datetime | None = None for sample in samples: try: recorded_at = _parse_ts(sample["ts"]) except (KeyError, ValueError): continue metrics = _expand_sample_to_metrics(sample, host.name, recorded_at) for m in metrics: session.add(m) accepted += 1 if latest_ts is None or recorded_at > latest_ts: latest_ts = recorded_at if accepted == 0: await session.rollback() return _error(400, "malformed_payload", "no valid samples") md = payload.get("metadata") or {} changed = False if latest_ts and (reg.last_seen_at is None or latest_ts > reg.last_seen_at): reg.last_seen_at = latest_ts changed = True version = payload.get("agent_version") if version and reg.agent_version != version: reg.agent_version = version changed = True for field in ("kernel", "distro", "arch"): if field in md and getattr(reg, field) != md[field]: setattr(reg, field, md[field]) changed = True if changed: reg.updated_at = datetime.now(timezone.utc) if latest_ts: skew = abs((datetime.now(timezone.utc) - latest_ts).total_seconds()) if skew > 300: current_app.logger.warning( "host_agent ingest: clock skew %.0fs for host=%s", skew, host.name) await session.commit() return jsonify({"ok": True, "samples_accepted": accepted}), 200 AGENT_SOURCE_PATH = Path(__file__).parent / "agent.py" def _agent_version() -> str: try: for line in AGENT_SOURCE_PATH.read_text().splitlines(): if line.startswith("AGENT_VERSION"): return line.split("=", 1)[1].strip().strip('"').strip("'") except OSError: pass return "unknown" @host_agent_bp.get("/install.sh") async def install_script(): token = request.args.get("token", "") if not token: return _error(401, "invalid_token") async with current_app.db_sessionmaker() as session: reg = (await session.execute( select(HostAgentRegistration).where( HostAgentRegistration.token_hash == _hash_token(token)) )).scalar_one_or_none() if reg is None: return _error(401, "invalid_token") host = (await session.execute( select(Host).where(Host.id == reg.host_id) )).scalar_one_or_none() if host is None: return _error(401, "invalid_token") url = public_base_url(request) rendered = await render_template( "install.sh.j2", url=url, token=token, agent_version=_agent_version(), host_name=host.name, host_address=host.address, ) return Response(rendered, mimetype="text/plain") @host_agent_bp.get("/agent.py") async def agent_source(): try: body = AGENT_SOURCE_PATH.read_text(encoding="utf-8") except OSError: return _error(500, "agent_missing") return Response(body, mimetype="text/x-python") @host_agent_bp.get("/widget") async def widget_table(): """Fleet-glance table: one row per monitored host, latest metrics.""" async with current_app.db_sessionmaker() as session: regs = (await session.execute(select(HostAgentRegistration))).scalars().all() host_ids = [r.host_id for r in regs] hosts = { h.id: h for h in (await session.execute( select(Host).where(Host.id.in_(host_ids)) )).scalars().all() } if host_ids else {} subq = ( select( PluginMetric.resource_name, PluginMetric.metric_name, func.max(PluginMetric.recorded_at).label("max_ts"), ) .where(PluginMetric.source_module == SOURCE_MODULE) .group_by(PluginMetric.resource_name, PluginMetric.metric_name) ).subquery() latest_rows = (await session.execute( select(PluginMetric).join( subq, (PluginMetric.resource_name == subq.c.resource_name) & (PluginMetric.metric_name == subq.c.metric_name) & (PluginMetric.recorded_at == subq.c.max_ts), ).where(PluginMetric.source_module == SOURCE_MODULE) )).scalars().all() latest: dict[str, dict[str, float]] = {} for row in latest_rows: if ":" in row.resource_name: continue latest.setdefault(row.resource_name, {})[row.metric_name] = row.value rows = [] for reg in regs: host = hosts.get(reg.host_id) if host is None: continue m = latest.get(host.name, {}) rows.append({ "host": host, "reg": reg, "cpu_pct": m.get("cpu_pct"), "mem_used_pct": m.get("mem_used_pct"), "disk_worst": m.get("disk_used_pct_worst"), "load_1m": m.get("load_1m"), }) return await render_template("widget_table.html", rows=rows) @host_agent_bp.get("/widget/history") async def widget_history(): host_id = request.args.get("host_id", "") hours = int(request.args.get("hours", "6")) cutoff = datetime.now(timezone.utc) - timedelta(hours=hours) async with current_app.db_sessionmaker() as session: host = (await session.execute( select(Host).where(Host.id == host_id))).scalar_one_or_none() if host is None: return _error(404, "not_found") points = (await session.execute( select(PluginMetric).where( PluginMetric.source_module == SOURCE_MODULE, PluginMetric.resource_name == host.name, PluginMetric.metric_name.in_( ("cpu_pct", "mem_used_pct", "disk_used_pct_worst")), PluginMetric.recorded_at >= cutoff, ).order_by(PluginMetric.recorded_at) )).scalars().all() series: dict[str, list[dict]] = {"cpu_pct": [], "mem_used_pct": [], "disk_used_pct_worst": []} for p in points: series[p.metric_name].append({"t": p.recorded_at.isoformat(), "v": p.value}) return await render_template("widget_history.html", host=host, series=series, hours=hours) def _new_token_pair() -> tuple[str, str]: raw = secrets.token_urlsafe(32) return raw, _hash_token(raw) @host_agent_bp.get("/settings/") @require_role(UserRole.admin) async def settings_list(): async with current_app.db_sessionmaker() as session: regs = (await session.execute(select(HostAgentRegistration))).scalars().all() hosts_by_id = { h.id: h for h in (await session.execute( select(Host).where(Host.id.in_([r.host_id for r in regs])))).scalars().all() } if regs else {} new_token = request.args.get("new_token") new_host_id = request.args.get("host_id") install_url = None if new_token and new_host_id: install_url = f"{public_base_url(request)}/plugins/host_agent/install.sh?token={new_token}" return await render_template( "settings_list.html", registrations=[ {"reg": r, "host": hosts_by_id.get(r.host_id)} for r in regs ], new_token=new_token, install_url=install_url, ) @host_agent_bp.post("/settings/add-host") @require_role(UserRole.admin) async def add_host(): form = await request.form name = (form.get("name") or "").strip() address = (form.get("address") or "").strip() if not name: return _error(400, "missing_name") async with current_app.db_sessionmaker() as session: host = (await session.execute( select(Host).where(Host.name == name))).scalar_one_or_none() if host is None: host = Host(name=name, address=address) session.add(host) await session.flush() existing = (await session.execute( select(HostAgentRegistration).where( HostAgentRegistration.host_id == host.id))).scalar_one_or_none() if existing is not None: await session.rollback() return _error(400, "already_registered") raw, hashed = _new_token_pair() reg = HostAgentRegistration(host_id=host.id, token_hash=hashed) session.add(reg) await session.commit() host_id = host.id return redirect(url_for("host_agent.settings_list") + f"?new_token={raw}&host_id={host_id}") @host_agent_bp.post("/settings//rotate-token") @require_role(UserRole.admin) async def rotate_token(host_id: str): async with current_app.db_sessionmaker() as session: reg = (await session.execute( select(HostAgentRegistration).where( HostAgentRegistration.host_id == host_id))).scalar_one_or_none() if reg is None: return _error(404, "not_found") raw, hashed = _new_token_pair() reg.token_hash = hashed reg.token_created_at = datetime.now(timezone.utc) await session.commit() return redirect(url_for("host_agent.settings_list") + f"?new_token={raw}&host_id={host_id}") @host_agent_bp.post("/settings//delete") @require_role(UserRole.admin) async def delete_registration(host_id: str): async with current_app.db_sessionmaker() as session: reg = (await session.execute( select(HostAgentRegistration).where( HostAgentRegistration.host_id == host_id))).scalar_one_or_none() if reg is not None: await session.delete(reg) await session.commit() return redirect(url_for("host_agent.settings_list"))