fix(antirootkit): /status uses ExecLog.count() instead of len(recent(limit=100))

len(recent(limit=100)) caps the "Exec Rows" badge at 100 forever once a
live host records more than that — a monitoring blind spot. Add
ExecLog.count() (true SELECT COUNT(*) over the table, unbounded by any
recent() limit) and use it in the /status handler.

Co-Authored-By: Gerald KERMA <devel@cybermind.fr>
This commit is contained in:
CyberMind-FR 2026-07-28 09:14:12 +02:00
parent 29ec46b338
commit 2c619fb8b3
4 changed files with 45 additions and 2 deletions

View File

@ -79,6 +79,13 @@ class ExecLog:
)
return [dict(r) for r in cur.fetchall()]
def count(self) -> int:
"""Return the true total number of execlog rows (unbounded by any
`limit`, unlike `recent()`) used for the /status monitoring badge
so it doesn't freeze at 100 once the log grows past that size."""
cur = self.db.execute("SELECT COUNT(*) c FROM execlog")
return cur.fetchone()["c"]
def failed_exec_count(self, exe: str, window_s: int, now=time.time) -> int:
"""Count failed exec attempts for exe within time window.

View File

@ -92,8 +92,13 @@ def create_app(execlog: Optional[ExecLog] = None) -> FastAPI:
@app.get("/status")
def get_status():
"""Small status blob for the panel header/sidebar badge."""
return {"execlog_rows": len(log.recent(limit=100))}
"""Small status blob for the panel header/sidebar badge.
Uses log.count() (the true table size), not len(recent()) the
latter is capped at its `limit` and would freeze the badge at 100
once the log grows past that on a live host.
"""
return {"execlog_rows": log.count()}
@app.get("/execlog")
def get_execlog(limit: int = 100):

View File

@ -67,6 +67,22 @@ def test_execlog_respects_limit_param(env):
assert len(r.json()) == 2
def test_status_execlog_rows_is_true_count_not_capped_by_recent_limit(env):
# /status must report the real table size (ExecLog.count()), not
# len(recent(limit=100)) — otherwise the dashboard badge would freeze
# once the log grows past 100 rows on a live host.
client, log = env
for i in range(5):
log.record(
ExecEvent(pid=i, ppid=1, uid=0, exe=f"/tmp/y{i}", argv=[], success=True),
"allow",
"somepkg",
)
# /execlog queried with a small limit must not influence /status
assert len(client.get("/execlog?limit=2").json()) == 2
assert client.get("/status").json()["execlog_rows"] == 5
def test_alerts_returns_200_list(env):
client, _log = env
r = client.get("/alerts")

View File

@ -67,3 +67,18 @@ def test_check_same_thread_false_usable_cross_thread(tmp_path):
t.join()
assert not errors
assert lg.recent()[0]["exe"] == "/tmp/y"
def test_count_not_capped_by_recent_limit(tmp_path):
# count() must reflect the true table size, unlike recent(limit=...)
# which truncates — the /status badge uses count() so it doesn't freeze
# at the limit once the log grows past it on a live host.
lg = ExecLog(str(tmp_path / "e3.db"))
for i in range(5):
lg.record(
ExecEvent(pid=i, ppid=0, uid=0, exe=f"/tmp/z{i}", argv=[], success=True),
"allow",
"somepkg",
)
assert lg.count() == 5
assert len(lg.recent(limit=2)) == 2