"""The syslog plugin's MCP tools (#6000): the sources and the lines kept from them. Registered through register_mcp_tools() in __init__.py, so they exist only while the plugin is enabled. They read what the receiver stored.""" from __future__ import annotations from datetime import datetime from sqlalchemy import select from steward.core.paging import keyset from steward.mcp.tools._shape import age_s, clamp_limit, cut_middle, db_session, iso, lookback, now_utc, pick from steward.models.hosts import Host from steward.models.users import UserRole from .models import SyslogLine, SyslogSource from .parse import facility_name, severity_name from .service import (AROUND_LINES, AROUND_SEVERITY, LineFilter, filtered, lines_between, place_for, rates, severity_level, shown_on, source_name, sources_for) async def list_syslog_sources(state: str = "") -> dict: """The devices sending syslog to Steward. state: "accepted" (added: their lines are kept and searchable), "offered" (sending but not added: counted, not kept), "ignored"; empty for all. Each gives its address, the host it is shown on, lines received in all (`lines_total`) and, for an added source, in the last hour, when it was last heard and its newest line. search_syslog searches the kept lines. """ if state and state not in ("accepted", "offered", "ignored"): raise ValueError("state must be accepted, offered or ignored, or empty for all.") now = now_utc() async with db_session() as db: stmt = select(SyslogSource).order_by(SyslogSource.last_seen_at.desc()) if state: stmt = stmt.where(SyslogSource.state == state) sources = list((await db.execute(stmt)).scalars()) per_hour = await rates(db, [s.id for s in sources if s.state == "accepted"], now) hosts = {h.id: h.name for h in (await db.execute(select(Host))).scalars()} return {"sources": [{ "source_id": s.id, "name": source_name(s), "address": s.address, "state": s.state, "host_id": s.host_id, "host_name": hosts.get(s.host_id), "calls_itself": s.hostname_seen or None, "lines_total": s.lines_total, "lines_last_hour": per_hour.get(s.id, 0) if s.state == "accepted" else None, "last_heard": iso(s.last_seen_at), "last_heard_age_s": age_s(s.last_seen_at, now), "newest_line": {"severity": severity_name(s.last_severity) if s.last_severity is not None else None, "text": cut_middle(s.last_line)}, } for s in sources]} async def search_syslog(source: str = "", host: str = "", text: str = "", severity: str = "", app: str = "", range: str = "1h", before: str = "", limit: int = 50) -> dict: """Search the syslog lines Steward has kept from added sources, newest first: a NAS's, a router's or another device's own log. source: a source's name, address or id (list_syslog_sources lists them); empty searches every added source. host: a host's name or id: the sources shown on that host (bound to it, or at its address). Empty for all. text: case-insensitive substring of the message; empty matches every line. severity: this level and worse: "err" (errors, crit, alert, emerg), "warning", "notice", "info", "debug", or 0-7. Empty for all. app: the app or tag that wrote the line (e.g. "kernel", "sshd"), exact. range: how far back — "1h" (default), "6h", "24h", "7d" or "30d". Each source keeps its lines up to a size and age limit, so older ones may be gone. before: an ISO time; only lines received before it. Pass a response's `next_before` to page further back. limit: lines to return, default 50, max 200. `at` is when Steward received the line, the time to trust; `sent_at` is the device's own clock, given only when the line carried its zone. """ _, key = lookback(range) limit = clamp_limit(limit, 50, 200) f = LineFilter(severity=severity_level(severity), app=app.strip(), text=text.strip(), range=key) before_dt = None if before: try: before_dt = datetime.fromisoformat(before.replace("Z", "+00:00")) except ValueError as exc: raise ValueError(f"before {before!r} is not an ISO-8601 time; pass a response's next_before.") from exc async with db_session() as db: accepted = list((await db.execute( select(SyslogSource).where(SyslogSource.state == "accepted"))).scalars()) if not accepted: return {"range": key, "matches": [], "returned": 0, "truncated": False, "note": "No syslog source is added yet; list_syslog_sources shows who is sending."} chosen = accepted if source: want = source.strip() by_address = [s for s in accepted if s.address == want] chosen = by_address or [pick(want, accepted, kind="syslog source", list_tool="list_syslog_sources", name=source_name)] if host: h = pick(host, list((await db.execute(select(Host))).scalars()), kind="host", list_tool="list_hosts") chosen = sources_for(chosen, h.id, h.address) if not chosen: return {"range": key, "matches": [], "returned": 0, "truncated": False, "note": f"No added syslog source is shown on {h.name}."} names = {s.id: source_name(s) for s in chosen} stmt = filtered(select(SyslogLine).where(SyslogLine.source_id.in_(list(names))), f) if before_dt is not None: stmt = stmt.where(SyslogLine.received_at < before_dt) rows = list((await db.execute( keyset(stmt, SyslogLine.received_at, SyslogLine.id, None, size=limit))).scalars()) more = len(rows) > limit rows = rows[:limit] out = {"range": key, "returned": len(rows), "truncated": more, "matches": [{ "at": iso(r.received_at), "sent_at": iso(r.sent_at), "source": names[r.source_id], "severity": severity_name(r.severity), "facility": facility_name(r.facility), "app": r.app or None, "hostname": r.hostname or None, "message": cut_middle(r.message), } for r in rows]} if more: # Full precision: `at` is cut to the second, and a cursor cut the same # way would skip lines received earlier in that second. out["next_before"] = rows[-1].received_at.isoformat() out["more"] = "More lines are older; pass next_before as `before`, or narrow with `text` or `severity`." return out def _as_events(lines: list[SyslogLine], sources: list[SyslogSource]) -> list[dict]: from steward.mcp.tools._events import event names = {s.id: source_name(s) for s in sources} return [event(ln.received_at, "syslog", f"{severity_name(ln.severity)}{' ' + ln.app if ln.app else ''}: {cut_middle(ln.message, 300)}", subject=names[ln.source_id]) for ln in reversed(lines)] async def mcp_events(db, host_id: str, host_name: str, since, until) -> list[dict]: """Warnings and worse in a window (get_anomaly) from the syslog sources shown on a host: bound to it, or at its address.""" sources = await shown_on(db, host_id, None) return _as_events(await lines_between(db, sources, since, until, AROUND_SEVERITY, AROUND_LINES), sources) async def mcp_resource_events(db, source: str, resource: str, since, until) -> list[dict]: """The same for an SNMP device, which has no host to read events by: the sources at its address, or bound to its host.""" if source != "snmp": return [] host_id, address = await place_for(db, source, resource) if not (host_id or address): return [] sources = await shown_on(db, host_id, address) return _as_events(await lines_between(db, sources, since, until, AROUND_SEVERITY, AROUND_LINES), sources) def register_tools(register) -> None: for fn in (list_syslog_sources, search_syslog): register(fn, role=UserRole.viewer, read_only=True)