diff --git a/docs/core/mcp.md b/docs/core/mcp.md index 80ff24a..e3ec549 100644 --- a/docs/core/mcp.md +++ b/docs/core/mcp.md @@ -81,6 +81,7 @@ Each role includes the tools of the roles above it in this list. | Docker plugin | `list_containers`, `get_container`, `list_swarm_services`, `search_container_logs`, `get_docker_disk_usage` | | Host agent (built in, always on) | `list_agents`, `list_maintenance_plans` (each plan's group, weekly slot, next run and latest run: how far it got, who it skipped and why), `get_maintenance_job` (a host's update jobs or one job: state, packages updated, whether it rebooted, why it failed) | | Traefik plugin | `list_routers` (each router graded as the Traefik page grades it, with the status codes behind a problem, what they mean and the swarm service behind it), `get_router` (one router's history, codes, normal and anomalies), `list_certificates` (served certificates, days left, the routers they cover), `get_traffic` (the access log's clients, countries and routers, and Traefik's own figures) | +| UniFi plugin | `get_network` (the WAN, clients, Wi-Fi networks, speed tests and devices by status), `list_devices`, `list_clients` (everyone seen, or the newest poll's top talkers), `list_network_events`, `list_alarms`, `get_network_config` (networks and VLANs, port forwards, firewall rules) | ### Operator: run playbooks, change monitoring diff --git a/plugins/unifi/__init__.py b/plugins/unifi/__init__.py index 80b7337..53219d2 100644 --- a/plugins/unifi/__init__.py +++ b/plugins/unifi/__init__.py @@ -45,3 +45,13 @@ def get_alert_metrics() -> dict[str, list[str]]: """What scheduler.py records: the WAN (resource "wan") and the client count (resource "clients").""" return {"unifi": ["is_up", "latency_ms", "total_clients"]} + + +def register_mcp_tools(register) -> None: + from .mcp_tools import register_tools + register_tools(register) + + +async def mcp_overview(db) -> list[dict]: + from .mcp_tools import mcp_overview as findings + return await findings(db) diff --git a/plugins/unifi/mcp_tools.py b/plugins/unifi/mcp_tools.py new file mode 100644 index 0000000..cefd6ae --- /dev/null +++ b/plugins/unifi/mcp_tools.py @@ -0,0 +1,268 @@ +"""The unifi plugin's MCP tools and its get_overview findings (#5839). + +Registered through register_mcp_tools() in __init__.py, so they exist only +while the plugin is enabled. Everything here is read from what the poller +stored; nothing reaches the UniFi controller. +""" +from __future__ import annotations + +import json +from datetime import timedelta + +from sqlalchemy import or_, select + +from steward.mcp._context import mcp_app +from steward.mcp.tools._shape import ( + age_s, bounded, clamp_limit, db_session, iso, lookback, now_utc, rnd, summarize, +) +from steward.mcp.tools.overview import finding +from steward.models.users import UserRole + +from .models import ( + UnifiAlarm, UnifiClientSnapshot, UnifiDevice, UnifiEvent, UnifiFwRule, UnifiKnownClient, + UnifiNetwork, UnifiPortForward, UnifiSpeedtestResult, UnifiWanStat, UnifiWlanStat, +) +from .routes import device_state, devices_in_order, history + +# The poller's readings are trusted while newer than this many poll intervals: +# an older "WAN down" is a poller that stopped, not a WAN that went down. +FRESH_POLLS = 5 +DEVICE_STATES = ("down", "degraded", "pending", "up") + + +def _poll_seconds() -> int: + cfg = mcp_app().config.get("PLUGINS", {}).get("unifi", {}) + return int(cfg.get("poll_interval_seconds", 60) or 60) + + +def _device_row(d, now) -> dict: + status, label = device_state(d.state) + return {"name": d.name or d.mac, "mac": d.mac, "model": d.model, "type": d.device_type, + "ip": d.ip, "status": status, "state": label, "version": d.version, + "uptime_s": d.uptime_seconds, "last_seen_at": iso(d.last_seen), + "last_seen_age_s": age_s(d.last_seen, now)} + + +async def get_network(range: str = "24h") -> dict: + """The UniFi network at a glance: the WAN, clients, Wi-Fi networks, the + last speed test, and devices by status. + + range: for the WAN latency and client-count series: "1h", "6h", "24h" + (default), "7d" or "30d". + + `wan`: up or down, its IP, latency, rx/tx bytes per second and uptime, + from the newest poll (`polled_at`); `stale: true` when that poll is old, + so the figures are not current. `clients`: connected now (wireless, + wired, guest) and the series. `wlans`: per SSID and radio (ng 2.4 GHz, + na 5 GHz, 6e 6 GHz) the clients, channel and UniFi's satisfaction score + (0-100). `speedtests`: the newest few. `devices`: counts by status + (list_devices names them). + """ + since, key = lookback(range) + now = now_utc() + async with db_session() as db: + wan_history, client_history = await history(db, since, max(60, int((now - since).total_seconds() // 60))) + wan = (await db.execute(select(UnifiWanStat).order_by(UnifiWanStat.scraped_at.desc()).limit(1))).scalar_one_or_none() + clients = (await db.execute(select(UnifiClientSnapshot) + .order_by(UnifiClientSnapshot.scraped_at.desc()).limit(1))).scalar_one_or_none() + wlans = (await db.execute(select(UnifiWlanStat).distinct(UnifiWlanStat.ssid, UnifiWlanStat.radio) + .order_by(UnifiWlanStat.ssid, UnifiWlanStat.radio, + UnifiWlanStat.scraped_at.desc()))).scalars().all() + tests = (await db.execute(select(UnifiSpeedtestResult) + .order_by(UnifiSpeedtestResult.run_at.desc()).limit(5))).scalars().all() + devices = (await db.execute(select(UnifiDevice))).scalars().all() + fresh = timedelta(seconds=FRESH_POLLS * _poll_seconds()) + counts: dict[str, int] = {} + for d in devices: + counts[device_state(d.state)[0]] = counts.get(device_state(d.state)[0], 0) + 1 + return { + "range": key, + "wan": ({"up": wan.is_up, "ip": wan.wan_ip, "latency_ms": rnd(wan.latency_ms, 1), + "rx_bytes_per_s": rnd(wan.rx_bytes_rate, 0), "tx_bytes_per_s": rnd(wan.tx_bytes_rate, 0), + "uptime_s": wan.uptime_seconds, "polled_at": iso(wan.scraped_at), + "stale": now - wan.scraped_at > fresh, + "latency_ms_series": summarize([(r.bucket, float(r.latency_ms)) for r in wan_history + if r.latency_ms is not None])} if wan else None), + "clients": ({"total": clients.total_clients, "wireless": clients.wireless_clients, + "wired": clients.wired_clients, "guest": clients.guest_clients, + "polled_at": iso(clients.scraped_at), + "series": summarize([(r.bucket, float(r.total_clients)) for r in client_history])} + if clients else None), + "wlans": [{"ssid": w.ssid, "radio": w.radio, "clients": w.num_sta, "channel": w.channel, + "satisfaction": w.satisfaction, "polled_at": iso(w.scraped_at)} for w in wlans], + "speedtests": [{"run_at": iso(t.run_at), "download_mbps": rnd(t.download_mbps, 1), + "upload_mbps": rnd(t.upload_mbps, 1), "latency_ms": rnd(t.latency_ms, 1), + "server": t.server_name} for t in tests], + "devices": counts, + } + + +async def list_devices(status: str = "") -> dict: + """UniFi's network devices (gateways, switches, access points): offline + and failing first, then those in a transition, then working ones. + + status: "down" (offline, inform error, adoption failed), "degraded" + (firmware mismatch, heartbeat missed, isolated), "pending" (adopting, + upgrading, provisioning), "up"; empty for all. + """ + if status and status not in DEVICE_STATES: + raise ValueError(f"status must be one of {', '.join(DEVICE_STATES)}, or empty for all.") + now = now_utc() + async with db_session() as db: + devices = devices_in_order((await db.execute(select(UnifiDevice))).scalars().all()) + rows = [_device_row(d, now) for d in devices] + return {"devices": [r for r in rows if not status or r["status"] == status], "total": len(rows)} + + +async def list_clients(search: str = "", active: bool = False, limit: int = 50) -> dict: + """Network clients UniFi has seen, most recently seen first. + + search: part of a hostname, alias, IP, MAC or vendor; empty for all. + active: true for the top talkers of the newest poll instead (with their + current tx/rx rates, signal and SSID). + limit: default 50, max 500. + + A known client carries `first_seen_at` / `last_seen_at`, its vendor (from + the MAC), `blocked`, and the alias and note set in UniFi. + """ + limit = clamp_limit(limit, 50, 500) + now = now_utc() + async with db_session() as db: + if active: + snap = (await db.execute(select(UnifiClientSnapshot) + .order_by(UnifiClientSnapshot.scraped_at.desc()).limit(1))).scalar_one_or_none() + top = json.loads(snap.top_clients_json) if snap else [] + if search: + needle = search.lower() + top = [c for c in top if any(needle in str(c.get(k) or "").lower() + for k in ("hostname", "ip", "mac", "essid"))] + out = bounded(top, limit, key="clients") + out["polled_at"] = iso(snap.scraped_at) if snap else None + return out + q = select(UnifiKnownClient) + if search: + like = f"%{search.strip()}%" + q = q.where(or_(*(col.ilike(like) for col in ( + UnifiKnownClient.hostname, UnifiKnownClient.name, UnifiKnownClient.ip, + UnifiKnownClient.mac, UnifiKnownClient.mac_vendor)))) + known = (await db.execute(q.order_by(UnifiKnownClient.last_seen.desc().nullslast()) + .limit(limit + 1))).scalars().all() + rows = [{"mac": c.mac, "hostname": c.hostname, "alias": c.name, "ip": c.ip, "vendor": c.mac_vendor, + "blocked": c.is_blocked, "note": c.note, "first_seen_at": iso(c.first_seen), + "last_seen_at": iso(c.last_seen), "last_seen_age_s": age_s(c.last_seen, now)} for c in known] + out = {"clients": rows[:limit], "truncated": len(rows) > limit} + if out["truncated"]: + out["more"] = f"More than {limit}. Pass a larger `limit`, or a `search`." + return out + + +async def list_network_events(range: str = "24h", category: str = "", search: str = "", + limit: int = 100) -> dict: + """UniFi's site events, newest first: clients joining and leaving, devices + restarting or losing contact, WAN changes, IDS hits. + + range: "1h", "6h", "24h" (default), "7d" or "30d". + category: UniFi's own, e.g. "wlan", "wired", "ap", "wan", "ids", + "system"; empty for all. + search: part of the message or the event key (e.g. "EVT_AP_Lost_Contact"). + limit: default 100, max 500. + """ + since, key = lookback(range) + limit = clamp_limit(limit, 100, 500) + q = select(UnifiEvent).where(UnifiEvent.event_time >= since) + if category: + q = q.where(UnifiEvent.category == category) + if search: + like = f"%{search.strip()}%" + q = q.where(or_(UnifiEvent.message.ilike(like), UnifiEvent.event_key.ilike(like))) + async with db_session() as db: + rows = (await db.execute(q.order_by(UnifiEvent.event_time.desc()).limit(limit + 1))).scalars().all() + events = [{"at": iso(e.event_time), "key": e.event_key, "category": e.category, "message": e.message, + "source_mac": e.source_mac, "source_ip": e.source_ip} for e in rows] + out = {"range": key, "events": events[:limit], "truncated": len(events) > limit} + if out["truncated"]: + out["more"] = f"More than {limit}. Pass a `category`, a `search`, a shorter `range` or a larger `limit`." + return out + + +async def list_alarms(include_archived: bool = False) -> dict: + """UniFi's alarms, newest first: those still active, and with + include_archived the last 20 that were archived (dealt with in UniFi).""" + async with db_session() as db: + active = (await db.execute(select(UnifiAlarm).where(UnifiAlarm.archived.is_(False)) + .order_by(UnifiAlarm.alarm_time.desc()))).scalars().all() + archived = ((await db.execute(select(UnifiAlarm).where(UnifiAlarm.archived.is_(True)) + .order_by(UnifiAlarm.alarm_time.desc()).limit(20))).scalars().all() + if include_archived else []) + + def row(a): + return {"at": iso(a.alarm_time), "key": a.alarm_key, "message": a.message} + + out = {"active": [row(a) for a in active]} + if include_archived: + out["archived"] = [row(a) for a in archived] + return out + + +async def get_network_config() -> dict: + """How the UniFi network is set up, as the controller last reported it: + networks and VLANs (subnet, DHCP, guest), port forwards (what is open + from the internet, to which internal address), and firewall rules by + ruleset in order. Read-only: Steward never changes UniFi.""" + async with db_session() as db: + networks = (await db.execute(select(UnifiNetwork).order_by(UnifiNetwork.name))).scalars().all() + forwards = (await db.execute(select(UnifiPortForward).order_by(UnifiPortForward.name))).scalars().all() + rules = (await db.execute(select(UnifiFwRule) + .order_by(UnifiFwRule.ruleset, UnifiFwRule.rule_index))).scalars().all() + return { + "networks": [{"name": n.name, "purpose": n.purpose, "vlan": n.vlan, "subnet": n.ip_subnet, + "dhcp": n.dhcp_enabled, "guest": n.is_guest} for n in networks], + "port_forwards": [{"name": f.name, "enabled": f.enabled, "proto": f.proto, "port": f.dst_port, + "to": f"{f.fwd_ip}:{f.fwd_port}", "from": f.src_filter or "any"} for f in forwards], + "firewall_rules": [{"ruleset": r.ruleset, "index": r.rule_index, "name": r.name, "action": r.action, + "protocol": r.protocol, "enabled": r.enabled, "src": r.src_address, + "dst": r.dst_address} for r in rules], + } + + +# ── get_overview ───────────────────────────────────────────────────────────── + +def network_findings(wan, devices, alarms, now, fresh: timedelta) -> list[dict]: + """The WAN down (critical, while the poll is current), devices not + working (offline critical, degraded warning; a transition such as an + upgrade is not a problem; a device the controller no longer reports is + left out), and active alarms (one warning naming the newest).""" + out = [] + if wan is not None and not wan.is_up and now - wan.scraped_at <= fresh: + out.append(finding("critical", "unifi_wan_down", "WAN", "UniFi reports the WAN down", + since=wan.scraped_at, next="get_network()")) + for d in devices: + status, label = device_state(d.state) + if status not in ("down", "degraded") or now - d.scraped_at > fresh: + continue + name = d.name or d.mac + out.append(finding("critical" if status == "down" else "warning", "unifi_device_" + status, name, + f"UniFi {d.device_type or 'device'} {name} ({d.model or 'unknown model'}): {label}", + since=d.last_seen, ids={"mac": d.mac}, next='list_devices(status="%s")' % status)) + if alarms: + newest = alarms[0] + more = f" and {len(alarms) - 1} more" if len(alarms) > 1 else "" + out.append(finding("warning", "unifi_alarms", "UniFi alarms", + f"UniFi alarm: {newest.message}{more}", since=newest.alarm_time, + next="list_alarms()")) + return out + + +async def mcp_overview(db) -> list[dict]: + now = now_utc() + wan = (await db.execute(select(UnifiWanStat).order_by(UnifiWanStat.scraped_at.desc()).limit(1))).scalar_one_or_none() + devices = (await db.execute(select(UnifiDevice))).scalars().all() + alarms = (await db.execute(select(UnifiAlarm).where(UnifiAlarm.archived.is_(False)) + .order_by(UnifiAlarm.alarm_time.desc()))).scalars().all() + return network_findings(wan, devices, alarms, now, timedelta(seconds=FRESH_POLLS * _poll_seconds())) + + +def register_tools(register) -> None: + for fn in (get_network, list_devices, list_clients, list_network_events, list_alarms, + get_network_config): + register(fn, role=UserRole.viewer, read_only=True) diff --git a/tests/plugins/unifi/test_mcp_tools.py b/tests/plugins/unifi/test_mcp_tools.py new file mode 100644 index 0000000..634927d --- /dev/null +++ b/tests/plugins/unifi/test_mcp_tools.py @@ -0,0 +1,57 @@ +"""The unifi plugin's MCP tools (#5839): what the overview says about the WAN, +devices and alarms, and the tools it registers.""" +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace as NS + +import pytest + +import plugins.unifi as unifi_plugin +from plugins.unifi import mcp_tools as t +from steward.models.users import UserRole + +NOW = datetime(2026, 10, 15, 12, 0, tzinfo=timezone.utc) +FRESH = timedelta(minutes=5) + + +def _device(name, state, ago=timedelta(minutes=1)): + return NS(name=name, mac=f"aa:{name}", model="U6-Pro", device_type="uap", state=state, + last_seen=NOW - ago, scraped_at=NOW - ago) + + +def test_a_wan_down_is_critical_only_while_the_poll_is_current(): + down = NS(is_up=False, scraped_at=NOW - timedelta(minutes=1)) + [f] = t.network_findings(down, [], [], NOW, FRESH) + assert f["kind"] == "unifi_wan_down" and f["severity"] == "critical" + stale = NS(is_up=False, scraped_at=NOW - timedelta(hours=1)) + assert t.network_findings(stale, [], [], NOW, FRESH) == [] + assert t.network_findings(NS(is_up=True, scraped_at=NOW), [], [], NOW, FRESH) == [] + + +def test_devices_offline_are_critical_degraded_warn_and_transitions_say_nothing(): + found = {f["subject"]: f for f in t.network_findings(None, [ + _device("ap-hall", 0), _device("ap-loft", 6), _device("ap-shed", 4), _device("gw", 1), + _device("ap-old", 0, ago=timedelta(days=30))], [], NOW, FRESH)} + assert set(found) == {"ap-hall", "ap-loft"} + assert found["ap-hall"]["severity"] == "critical" and "Offline" in found["ap-hall"]["summary"] + assert found["ap-loft"]["severity"] == "warning" and "Heartbeat missed" in found["ap-loft"]["summary"] + + +def test_active_alarms_are_one_finding_naming_the_newest(): + alarms = [NS(message="IPS alert: scan from 1.2.3.4", alarm_time=NOW), + NS(message="older", alarm_time=NOW - timedelta(hours=1))] + [f] = t.network_findings(None, [], alarms, NOW, FRESH) + assert f["kind"] == "unifi_alarms" and f["summary"] == "UniFi alarm: IPS alert: scan from 1.2.3.4 and 1 more" + + +def test_the_plugin_registers_its_tools_for_viewers(): + got = {} + unifi_plugin.register_mcp_tools(lambda fn, role, read_only: got.update({fn.__name__: (role, read_only)})) + assert got == {name: (UserRole.viewer, True) for name in ( + "get_network", "list_devices", "list_clients", "list_network_events", "list_alarms", + "get_network_config")} + assert callable(unifi_plugin.mcp_overview) + + +async def test_list_devices_refuses_an_unknown_status(): + with pytest.raises(ValueError, match="status must be one of"): + await t.list_devices(status="broken")