"""Infrastructure health: CheckMK and every OPNsense firewall, polled on a timer. WHY THIS LIVES IN `workshop` AND NOT IN `digest-engine` -------------------------------------------------------- Both consume it, and it was tempting to put the polling in the digest since the digest already reads one firewall. That would have been wrong in one specific way: the digest runs **four times a day**, and "is the NAS disk failing right now" is not a question with a six-hour answer. So the poller lives in the always-on service, keeps a small history, and the digest reads *that* — one poller, two consumers, and the digest gets trend ("this has been critical since Tuesday") instead of a snapshot it cannot compare to anything. WHAT IT POLLS, AND WHAT IT WILL NEVER DO ----------------------------------------- - **CheckMK**: `GET /check_mk/api/1.0/domain-types/{host,service}/collections/all`, read-only, with a Guest-role automation user. Never acknowledges, never downtimes, never reschedules. Those are the endpoints a monitoring integration is *expected* to call, and the reason not to is that an assistant silencing an alert is indistinguishable from the alert being fixed. - **OPNsense**: `GET /api/ids/service/status` only — is Suricata running. The alert *query* stays in digest-engine, which already does it properly; duplicating the paging logic here would give the household two different answers about the same log. MULTIPLE FIREWALLS ARE THE POINT --------------------------------- `OPNSENSE_JSON` carries a list, and every row this module writes carries the firewall's `name`. A household with a main and a DMZ firewall must never be told "the IDS is running" — the honest sentence is "main is running, dmz has not answered since 14:20", and that is only possible if the name survives all the way to the display. EVERY FAILURE IS A ROW, NOT AN EXCEPTION ----------------------------------------- A target that cannot be reached is recorded as `unreachable` with the error attached. That is the whole point of a health poller: silence must be visible. A poll that raises and logs would leave the last good row in place and the display would keep showing green for a machine that has been off for a day. """ from __future__ import annotations import base64 import json import logging import os import sqlite3 import threading import time import urllib.error import urllib.parse import urllib.request from datetime import datetime, timedelta, timezone from pathlib import Path LOG = logging.getLogger("workshop.health") # Same database as the rest of the knowledge store: this is a table about the house's # machines, which is exactly the "durable facts" category knowledge.py holds. DB_PATH = Path(os.environ.get("WORKSHOP_KNOWLEDGE_DB_PATH", "/data/workshop-knowledge.db")) CHECKMK_URL = os.environ.get("CHECKMK_BASE_URL", "").rstrip("/") CHECKMK_SITE = os.environ.get("CHECKMK_SITE", "cmk").strip() CHECKMK_USERNAME = os.environ.get("CHECKMK_USERNAME", "") CHECKMK_SECRET = os.environ.get("CHECKMK_SECRET", "") CHECKMK_ONLY_PROBLEMS = os.environ.get("CHECKMK_ONLY_PROBLEMS", "true").lower() != "false" CHECKMK_MAX_ROWS = int(os.environ.get("CHECKMK_MAX_ROWS", "200")) POLL_INTERVAL = int(os.environ.get("WORKSHOP_HEALTH_INTERVAL_SECONDS", "300")) HTTP_TIMEOUT = float(os.environ.get("WORKSHOP_HEALTH_TIMEOUT", "20")) # Long enough to see a pattern ("this flaps every night"), short enough that the table # stays small. Unlike knowledge.py's own tables, samples DO expire: a service state # from three weeks ago is an observation, not a fact, and observations go stale — the # same distinction doorway.py draws. RETENTION_DAYS = int(os.environ.get("WORKSHOP_HEALTH_RETENTION_DAYS", "30")) # CheckMK's numeric states, which appear in the API as integers with no labels. HOST_STATES = {0: "up", 1: "down", 2: "unreachable"} SERVICE_STATES = {0: "ok", 1: "warning", 2: "critical", 3: "unknown"} # What counts as "needs a human". `unknown` is included deliberately: a check that # cannot report is not a check that passed. BAD_SERVICE_STATES = {1, 2, 3} def _now() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") def firewalls() -> list[dict]: """The firewall list, from the JSON blob the config export writes.""" raw = os.environ.get("OPNSENSE_JSON", "").strip() if not raw: return [] try: parsed = json.loads(raw) except ValueError: LOG.warning("workshop: OPNSENSE_JSON is not valid JSON — no firewalls will be polled") return [] entries = parsed.get("firewalls") if isinstance(parsed, dict) else parsed return [f for f in (entries or []) if isinstance(f, dict) and f.get("base_url")] def init_db() -> None: with sqlite3.connect(DB_PATH, timeout=10) as conn: conn.executescript( """ -- One row per target per poll. A time series, not a current-state table: -- "the NAS has been critical since Tuesday" is the sentence worth being -- able to say, and a single-row-per-target design cannot say it. CREATE TABLE IF NOT EXISTS infra_status ( id INTEGER PRIMARY KEY, -- 'checkmk' | 'opnsense' source TEXT NOT NULL, -- Which instance. The firewall's name, or the CheckMK site. NEVER empty: -- a household with two firewalls must never be told "the IDS is running". target TEXT NOT NULL, -- ok | problem | unreachable. Three, not two: "I could not ask" is a -- different fact from "I asked and it is broken", and collapsing them is -- how a display shows green for a machine that has been off all day. state TEXT NOT NULL, summary TEXT NOT NULL, -- The problem rows themselves, as JSON, so the display can list them -- without a second query and the digest can quote them. detail TEXT, problem_count INTEGER NOT NULL DEFAULT 0, checked_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS infra_by_target ON infra_status (source, target, checked_at); """ ) def _record(source: str, target: str, state: str, summary: str, detail: list | None = None, problem_count: int = 0) -> None: try: with sqlite3.connect(DB_PATH, timeout=10) as conn: conn.execute( "INSERT INTO infra_status (source, target, state, summary, detail, problem_count, " "checked_at) VALUES (?, ?, ?, ?, ?, ?, ?)", (source, target, state, summary, json.dumps(detail or []), problem_count, _now()), ) cutoff = (datetime.now(timezone.utc) - timedelta(days=RETENTION_DAYS)).isoformat() conn.execute("DELETE FROM infra_status WHERE checked_at < ?", (cutoff.replace("+00:00", "Z"),)) except sqlite3.Error: LOG.warning("workshop: could not record health for %s/%s", source, target, exc_info=True) # --- CheckMK ---------------------------------------------------------------------- def _checkmk_get(path: str) -> tuple[int, dict]: url = f"{CHECKMK_URL}/{CHECKMK_SITE}/check_mk/api/1.0{path}" req = urllib.request.Request(url) # CheckMK's own scheme: `Authorization: Bearer `. It is # not an OAuth bearer despite the word, which is why this is built by hand. req.add_header("Authorization", f"Bearer {CHECKMK_USERNAME} {CHECKMK_SECRET}") req.add_header("Accept", "application/json") try: with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp: return resp.status, json.loads(resp.read() or b"{}") except urllib.error.HTTPError as exc: return exc.code, {"detail": exc.read().decode("utf-8", "replace")[:200]} except (urllib.error.URLError, TimeoutError, OSError, ValueError) as exc: return 0, {"detail": str(exc)} def poll_checkmk() -> dict | None: """Hosts and services that are not OK. None when CheckMK isn't configured.""" if not (CHECKMK_URL and CHECKMK_USERNAME and CHECKMK_SECRET): return None target = CHECKMK_SITE or "checkmk" columns = "&columns=name&columns=state&columns=description&columns=plugin_output&columns=host_name" query = "?query=" + urllib.parse.quote(json.dumps( {"op": "!=", "left": "state", "right": "0"} )) if CHECKMK_ONLY_PROBLEMS else "" status, hosts = _checkmk_get(f"/domain-types/host/collections/all{query}{columns}") if status != 200: message = hosts.get("detail") or f"HTTP {status}" _record("checkmk", target, "unreachable", f"CheckMK did not answer: {message}") return {"source": "checkmk", "target": target, "state": "unreachable", "summary": message} status, services = _checkmk_get(f"/domain-types/service/collections/all{query}{columns}") service_rows = services.get("value", []) if status == 200 else [] problems = [] for row in (hosts.get("value") or [])[:CHECKMK_MAX_ROWS]: extensions = row.get("extensions", {}) state = extensions.get("state") if state: problems.append({ "kind": "host", "host": extensions.get("name") or row.get("title"), "state": HOST_STATES.get(state, str(state)), }) for row in service_rows[:CHECKMK_MAX_ROWS]: extensions = row.get("extensions", {}) state = extensions.get("state") if state in BAD_SERVICE_STATES: problems.append({ "kind": "service", "host": extensions.get("host_name"), "service": extensions.get("description"), "state": SERVICE_STATES.get(state, str(state)), "output": (extensions.get("plugin_output") or "")[:200], }) state = "problem" if problems else "ok" summary = ( f"{len(problems)} problem(s): " + ", ".join(f"{p.get('host')}{'/' + p['service'] if p.get('service') else ''} {p['state']}" for p in problems[:5]) if problems else "everything CheckMK watches is OK" ) _record("checkmk", target, state, summary, problems, len(problems)) return {"source": "checkmk", "target": target, "state": state, "summary": summary, "problems": problems} # --- OPNsense --------------------------------------------------------------------- def poll_firewall(firewall: dict) -> dict: """Is Suricata running on this firewall? Read-only, one endpoint. Deliberately NOT the alert query — digest-engine owns that, including the paging and the window logic, and two implementations of the same read would eventually give the household two different answers about one log file. """ name = str(firewall.get("name") or "opnsense") base = str(firewall.get("base_url") or "").rstrip("/") auth = base64.b64encode( f"{firewall.get('api_key', '')}:{firewall.get('api_secret', '')}".encode() ).decode() req = urllib.request.Request(f"{base}/api/ids/service/status") req.add_header("Authorization", f"Basic {auth}") context = None if firewall.get("verify_tls") is False: # OPNsense ships a self-signed certificate. Off is a real choice a household # makes; it is recorded in the row's summary so nobody later mistakes a # working poll for a verified one. import ssl context = ssl._create_unverified_context() # noqa: S323 try: with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT, context=context) as resp: body = json.loads(resp.read() or b"{}") except Exception as exc: # noqa: BLE001 — every failure is a row, see the docstring _record("opnsense", name, "unreachable", f"{name} did not answer: {exc}") return {"source": "opnsense", "target": name, "state": "unreachable", "summary": str(exc)} running = str(body.get("status", "")).lower() == "running" state = "ok" if running else "problem" summary = ( f"Suricata is running on {name}" if running else f"Suricata is NOT running on {name} (status: {body.get('status', 'unknown')}) — " f"this firewall's IDS section of the digest will be empty, which is not the same " f"as quiet" ) _record("opnsense", name, state, summary, [body], 0 if running else 1) return {"source": "opnsense", "target": name, "state": state, "summary": summary} # --- the round --------------------------------------------------------------------- def poll_all() -> dict: results = [] checkmk = poll_checkmk() if checkmk: results.append(checkmk) for firewall in firewalls(): results.append(poll_firewall(firewall)) return {"checked_at": _now(), "results": results} def latest() -> dict: """The most recent sample per target, plus how long each state has held. `since` is what makes this worth reading twice: "critical" tells you to look, "critical since Tuesday 06:00" tells you whether it is new. Computed by walking back through the samples until the state changes. """ try: with sqlite3.connect(DB_PATH, timeout=10) as conn: conn.row_factory = sqlite3.Row rows = conn.execute( "SELECT source, target, state, summary, detail, problem_count, MAX(checked_at) " "AS checked_at FROM infra_status GROUP BY source, target ORDER BY source, target" ).fetchall() results = [] for row in rows: entry = {k: row[k] for k in row.keys()} entry["detail"] = json.loads(entry.get("detail") or "[]") changed = conn.execute( "SELECT MAX(checked_at) AS t FROM infra_status WHERE source = ? AND target = ? " "AND state != ?", (row["source"], row["target"], row["state"]), ).fetchone() entry["since"] = None if changed and changed["t"]: later = conn.execute( "SELECT MIN(checked_at) AS t FROM infra_status WHERE source = ? AND " "target = ? AND state = ? AND checked_at > ?", (row["source"], row["target"], row["state"], changed["t"]), ).fetchone() entry["since"] = later["t"] if later else None results.append(entry) except sqlite3.Error: LOG.warning("workshop: could not read infra_status", exc_info=True) return {"results": [], "available": False} worst = "ok" for entry in results: if entry["state"] == "unreachable": worst = "unreachable" elif entry["state"] == "problem" and worst == "ok": worst = "problem" return { "results": results, "available": True, # One word for the display's overlay to key off, so it does not re-derive # "is anything wrong" from a list in four different places. "overall": worst, "configured": bool(CHECKMK_URL) or bool(firewalls()), } def start_poller() -> None: """Poll in the background for as long as the service runs. A daemon thread rather than a systemd timer or a cron container: this service is already always-on, the work is two HTTP calls, and a separate scheduler would be a second thing to deploy and a second place for the credentials to live. """ if not (CHECKMK_URL or firewalls()): LOG.info("workshop: no CheckMK and no firewalls configured — health polling is off") return def loop(): while True: try: poll_all() except Exception: # noqa: BLE001 — a poller that dies is worse than a bad poll LOG.warning("workshop: health poll round failed", exc_info=True) time.sleep(max(60, POLL_INTERVAL)) threading.Thread(target=loop, name="health-poller", daemon=True).start() LOG.info("workshop: health polling every %ds (checkmk=%s, firewalls=%d)", max(60, POLL_INTERVAL), bool(CHECKMK_URL), len(firewalls()))