selfhostedops/selfhostedops-stack
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
selfhostedops-stack/ops-readonly/reader.py
Maximilian Grimm 1655d04d7d Episode 16: Kuma alerts sorted by rules, a local agent and a script 
ops-readonly gets kuma_alerts (host export kuma-export.py, one more read-only
mount). kuma-triage.py sorts the same data without a model. Flow H runs the
agent with skill alert-triage.md. Flow B: one open incident per monitor, an
UP closes all of them, titles name start and end and stay under Outline's
limit. B and D accept Kuma only with a header token. Monitoring notes: ntfy
rejects Kuma messages for monitors without a URL, resend for long outages.
2026-09-29 22:14:57 +02:00

385 lines
14 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
ops-readonly: the read-only twin of ops-actions (episode 11).
The security report agents never get a shell. They get this: a fixed list of
questions, each answered by a few lines of code over three read-only mounts.
/logs/caddy/access.log Caddy's JSON access log (episode 10)
/var/log/journal the host journal, sshd lines (read via libsystemd)
/run/fail2ban fail2ban's control socket (status queries only)
/waf WAF alerts as JSON, written by waf-export.py on the
host (episode 14); cscli and its credentials stay there
/kuma Uptime Kuma state per monitor as JSON, written by
kuma-export.py on the host (episode 16); the Kuma
database stays there
No docker socket, no write access, no free-form arguments. An action name that
is not in ACTIONS is refused, an argument outside its range is refused, and
every call is written to stdout so the container log shows what the agents
asked for.
The fail2ban socket speaks pickle. Only two verbs are ever sent through it,
"status" and "banned"; the same protocol could also send "set ... unbanip", so
the verb list below is the allowlist for that socket.
"""
import calendar, collections, datetime, json, os, pickle, re, socket, time
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlparse, parse_qs
TOKEN = os.environ.get("OPS_READONLY_TOKEN", "")
CADDY_LOG = "/logs/caddy/access.log"
JOURNAL_DIR = "/var/log/journal"
F2B_SOCK = "/run/fail2ban/fail2ban.sock"
WINDOW_H = 24
JAILS = ("sshd", "caddy-probes")
N_MIN, N_MAX, N_DEFAULT = 1, 20, 5
WAF_DIR = os.environ.get("WAF_DIR", "/waf")
H_MIN, H_MAX, H_DEFAULT = 1, 48, 24
# ---------------------------------------------------------------- fail2ban --
F2B_END = b"<F2B_END_COMMAND>"
F2B_VERBS = ("status", "banned")
def f2b(cmd):
if cmd[0] not in F2B_VERBS:
raise ValueError("verb not allowed: %s" % cmd[0])
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
s.settimeout(5)
s.connect(F2B_SOCK)
s.sendall(pickle.dumps(cmd, 2) + F2B_END)
buf = b""
while not buf.endswith(F2B_END):
chunk = s.recv(65536)
if not chunk:
break
buf += chunk
s.close()
code, payload = pickle.loads(buf[: -len(F2B_END)])
if code != 0:
raise RuntimeError("fail2ban answered %s: %s" % (code, payload))
return payload
def f2b_status(jail):
if jail not in JAILS:
raise ValueError("jail must be one of %s" % ", ".join(JAILS))
raw = f2b(["status", jail])
flat = {}
for section, rows in raw:
for key, value in rows:
flat[key] = value
return {
"jail": jail,
"currently_failed": int(flat.get("Currently failed", 0)),
"total_failed": int(flat.get("Total failed", 0)),
"currently_banned": int(flat.get("Currently banned", 0)),
"total_banned": int(flat.get("Total banned", 0)),
"banned_ips": list(flat.get("Banned IP list", []) or []),
}
def f2b_banned():
raw = f2b(["banned"])
out = {}
for entry in raw:
for jail, ips in entry.items():
out[jail] = list(ips or [])
return {"jails": out, "banned_total_now": sum(len(v) for v in out.values())}
# -------------------------------------------------------------- sshd journal --
RE_INVALID = re.compile(r"^Invalid user (.*?) from ([0-9a-f.:]+) port \d+")
RE_FAILED = re.compile(r"^Failed password for (?:invalid user )?(.+?) from ([0-9a-f.:]+) port \d+")
RE_PREAUTH = re.compile(r"^Connection closed by (?:authenticating|invalid) user (.+?) ([0-9a-f.:]+) port \d+")
RE_ACCEPTED = re.compile(r"^Accepted (?:publickey|password) for (\S+) from ([0-9a-f.:]+)")
def sshd_failed_summary(n=N_DEFAULT):
from systemd import journal # libsystemd, reads the mounted journal files
since = datetime.datetime.now() - datetime.timedelta(hours=WINDOW_H)
j = journal.Reader(path=JOURNAL_DIR)
j.add_match(_SYSTEMD_UNIT="ssh.service")
j.seek_realtime(since)
lines = invalid = failed = preauth = accepted = 0
users = collections.Counter()
ips = collections.Counter()
for entry in j:
msg = str(entry.get("MESSAGE", ""))
lines += 1
m = RE_INVALID.match(msg)
if m:
invalid += 1
users[m.group(1)] += 1
ips[m.group(2)] += 1
continue
m = RE_FAILED.match(msg)
if m:
failed += 1
users[m.group(1)] += 1
ips[m.group(2)] += 1
continue
m = RE_PREAUTH.match(msg)
if m:
preauth += 1
ips[m.group(2)] += 1
continue
if RE_ACCEPTED.match(msg):
accepted += 1
return {
"window_hours": WINDOW_H,
"journal_lines": lines,
"login_failures": invalid + failed,
"invalid_user": invalid,
"failed_password": failed,
"preauth_closed": preauth,
"accepted_logins": accepted,
"distinct_source_ips": len(ips),
"top_usernames": [{"user": u[:120], "count": c} for u, c in users.most_common(n)],
"top_source_ips": [{"ip": i, "count": c} for i, c in ips.most_common(n)],
}
# ---------------------------------------------------------------- caddy log --
def caddy_rows():
"""Yield (status, client_ip, uri) for every request inside the window."""
cutoff = time.time() - WINDOW_H * 3600
with open(CADDY_LOG, "rb") as fh:
for line in fh:
try:
e = json.loads(line)
except ValueError:
continue
if e.get("ts", 0) < cutoff:
continue
req = e.get("request", {})
yield int(e.get("status", 0)), str(req.get("client_ip", "")), str(req.get("uri", ""))[:200]
def caddy_status_counts():
by_status = collections.Counter()
ips = set()
total = 0
for status, ip, _ in caddy_rows():
total += 1
by_status[str(status)] += 1
ips.add(ip)
by_class = collections.Counter()
for s, c in by_status.items():
by_class[s[0] + "xx"] += c
return {
"window_hours": WINDOW_H,
"requests_total": total,
"by_class": dict(sorted(by_class.items())),
"by_status": dict(sorted(by_status.items())),
"rejected_4xx": by_class.get("4xx", 0),
"rate_limited_429": by_status.get("429", 0),
"server_errors_5xx": by_class.get("5xx", 0),
"distinct_client_ips": len(ips),
}
def caddy_top_paths(n=N_DEFAULT):
paths = collections.Counter()
statuses = collections.defaultdict(set)
for status, _, uri in caddy_rows():
if status >= 400:
paths[uri] += 1
statuses[uri].add(status)
return {
"window_hours": WINDOW_H,
"filter": "responses with status 400 or higher",
"paths": [{"path": p, "count": c, "statuses": sorted(statuses[p])} for p, c in paths.most_common(n)],
}
def caddy_top_ips(n=N_DEFAULT):
ips = collections.Counter()
for status, ip, _ in caddy_rows():
if status >= 400:
ips[ip] += 1
return {
"window_hours": WINDOW_H,
"filter": "responses with status 400 or higher",
"ips": [{"ip": i, "count": c} for i, c in ips.most_common(n)],
}
# ---------------------------------------------------------------- WAF alerts --
# Episode 14. Everything in here except rule ids, counts and the source IP was
# written by whoever sent the request. The reader truncates, it does not clean.
def _waf_json(name):
with open(os.path.join(WAF_DIR, name), "rb") as fh:
return json.load(fh)
def _auth_index():
"""(client_ip, host, uri) -> times of every 403 that carried an Authorization
or Cookie header. Presence only: the WAF blocks before the app checks the
credential, so anyone can set this. It is a hint, never proof."""
seen = collections.defaultdict(list)
with open(CADDY_LOG, "rb") as fh:
for line in fh:
if b'"status":403' not in line:
continue
try:
e = json.loads(line)
except ValueError:
continue
req = e.get("request", {})
h = req.get("headers", {})
if "Authorization" in h or "Cookie" in h:
key = (str(req.get("client_ip", "")), str(req.get("host", "")), str(req.get("uri", ""))[:300])
seen[key].append(float(e.get("ts", 0)))
return seen
def _had_auth(index, alert):
"""Same request, not just the same URI: the log line must sit within 3 s of the alert."""
t = calendar.timegm(time.strptime(alert["ts"], "%Y-%m-%dT%H:%M:%SZ"))
return any(abs(ts - t) <= 3 for ts in index.get((alert["source_ip"], alert["host"], alert["uri"]), ()))
def waf_blocked(hours=H_DEFAULT):
data = _waf_json("alerts.json")
catalog = _waf_json("crs-rules.json")
cutoff = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() - hours * 3600))
auth = _auth_index()
groups = collections.OrderedDict()
total = 0
for a in data.get("alerts", []):
if a.get("ts", "") < cutoff:
continue
total += 1
ids = [r["id"] for r in a.get("rules", [])]
key = (a["source_ip"], a["host"], a["method"], a["uri"], tuple(ids))
g = groups.get(key)
if g is None:
g = groups[key] = {
"source_ip": a["source_ip"], "host": a["host"], "method": a["method"],
"uri": a["uri"][:160], "count": 0, "first_seen": a["ts"], "last_seen": a["ts"],
"auth_header_present": _had_auth(auth, a),
"rule_ids": ids,
"attack_classes": sorted({catalog.get(i, {}).get("attack_class", "") for i in ids} - {""}),
}
g["count"] += 1
g["last_seen"] = max(g["last_seen"], a["ts"])
return {"window_hours": hours, "exported_at": data.get("exported_at"), "blocked_total": total,
"distinct_requests": len(groups), "requests": list(groups.values())}
def waf_rule_info(rule_id):
if not re.fullmatch(r"\d{6}", rule_id or ""):
raise ValueError("id must be a six digit CRS rule id")
rule = _waf_json("crs-rules.json").get(rule_id)
if rule is None:
return {"id": rule_id, "known": False, "note": "no such rule in the installed CRS"}
return dict(rule, id=rule_id, known=True)
# ---------------------------------------------------------------- Kuma --
# Episode 16. Monitor names and messages (which quote the monitored page) are
# written by whoever controls the monitor or the page. Served as exported.
KUMA_DIR = os.environ.get("KUMA_DIR", "/kuma")
def kuma_alerts():
with open(os.path.join(KUMA_DIR, "alerts.json"), "rb") as fh:
return json.load(fh)
# ---------------------------------------------------------------- the list --
def arg_n(q):
raw = q.get("n", [str(N_DEFAULT)])[0]
try:
n = int(str(raw).strip())
except ValueError:
raise ValueError("n must be a whole number")
if not N_MIN <= n <= N_MAX:
raise ValueError("n must be between %d and %d" % (N_MIN, N_MAX))
return n
def arg_hours(q):
raw = q.get("hours", [str(H_DEFAULT)])[0]
try:
h = int(str(raw).strip())
except ValueError:
raise ValueError("hours must be a whole number")
if not H_MIN <= h <= H_MAX:
raise ValueError("hours must be between %d and %d" % (H_MIN, H_MAX))
return h
ACTIONS = {
"f2b_status": lambda q: f2b_status(q.get("jail", [""])[0]),
"f2b_banned": lambda q: f2b_banned(),
"sshd_failed_summary": lambda q: sshd_failed_summary(arg_n(q)),
"caddy_status_counts": lambda q: caddy_status_counts(),
"caddy_top_paths": lambda q: caddy_top_paths(arg_n(q)),
"caddy_top_ips": lambda q: caddy_top_ips(arg_n(q)),
"waf_blocked": lambda q: waf_blocked(arg_hours(q)),
"waf_rule_info": lambda q: waf_rule_info(q.get("id", [""])[0]),
"kuma_alerts": lambda q: kuma_alerts(),
}
def log(entry):
entry["ts"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
print(json.dumps(entry), flush=True)
class Handler(BaseHTTPRequestHandler):
def _reply(self, code, obj):
body = json.dumps(obj).encode()
self.send_response(code)
self.send_header("content-type", "application/json")
self.send_header("content-length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *a):
pass
def do_GET(self):
u = urlparse(self.path)
if u.path == "/healthz":
return self._reply(200, {"status": "ok", "actions": sorted(ACTIONS), "window_hours": WINDOW_H})
if not u.path.startswith("/read/"):
return self._reply(404, {"error": "not found"})
if TOKEN and self.headers.get("x-ops-token") != TOKEN:
log({"event": "denied", "why": "bad token", "path": u.path})
return self._reply(401, {"error": "bad token"})
action = u.path[len("/read/"):]
q = parse_qs(u.query)
if action not in ACTIONS:
log({"event": "denied", "why": "action not on allowlist", "action": action})
return self._reply(403, {"error": "action not on allowlist", "allowed": sorted(ACTIONS)})
t0 = time.time()
try:
result = ACTIONS[action](q)
except ValueError as e:
log({"event": "denied", "why": "bad argument", "action": action, "detail": str(e)})
return self._reply(400, {"error": "bad argument", "detail": str(e)})
except Exception as e:
log({"event": "failed", "action": action, "detail": str(e)})
return self._reply(500, {"error": "read failed", "detail": str(e)})
log({"event": "served", "action": action, "args": {k: v[0] for k, v in q.items()},
"seconds": round(time.time() - t0, 3)})
return self._reply(200, result)
def do_POST(self):
self._reply(405, {"error": "read only"})
if __name__ == "__main__":
log({"event": "start", "actions": sorted(ACTIONS), "token_set": bool(TOKEN)})
HTTPServer(("0.0.0.0", 8098), Handler).serve_forever()
Reference in a new issue View git blame Copy permalink