Episode 11: a read-only agent that reads the security logs, one agent vs a crew of three

New: ops-readonly, the read-only twin of the episode 9 ops-actions runner
(fixed question list, three read-only mounts, no docker socket, read_only,
cap_drop ALL). Four role skills in agents/skills/, loaded at run time from
this repo's Forgejo. Two workflows: E (one agent, seven tools, one skill) and
F (three agents, one skill each, the workflow is the orchestrator).

Measured over 20 runs: no run of either variant produced all numbers and
lists correct. The solo agent confuses similar counts and skips a tool; the
crew calls every tool but drops fields at each role boundary. Sharpening the
skill files did not help the crew, it made it worse. Logs are attacker text:
an instruction in a request path is carried into the report as data, not
obeyed, but carried.

Claude-Session: https://claude.ai/code/session_017iP581CYtPfxdZYPPAyWZA

ops-readonly/Dockerfile

@@ -0,0 +1,10 @@
+# ops-readonly (episode 11): the read-only executor behind the security report
+# agents. python3-systemd is the only extra, it reads the host journal through
+# libsystemd from a read-only bind mount. No docker client in here on purpose.
+FROM debian:trixie-slim
+RUN apt-get update \
+ && apt-get install -y --no-install-recommends python3 python3-systemd \
+ && rm -rf /var/lib/apt/lists/*
+COPY reader.py /reader.py
+EXPOSE 8098
+CMD ["python3", "-u", "/reader.py"]

ops-readonly/reader.py

@@ -0,0 +1,279 @@
+#!/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)
+
+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 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
+
+
+# ---------------------------------------------------------------- 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)],
+    }
+
+
+# ---------------------------------------------------------------- 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
+
+
+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)),
+}
+
+
+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()

ops-readonly/token.env.example

@@ -0,0 +1,4 @@
+# Copy to ops-readonly/token.env (git-ignored). The same header the agents'
+# tool nodes send. A leaked token reads counters, nothing more, but rotate it
+# anyway:   openssl rand -hex 24
+OPS_READONLY_TOKEN=