345 lines
16 KiB
Python
345 lines
16 KiB
Python
"""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 <user> <automation secret>`. 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()))
|