From 2c619fb8b37bdce93c71c33a37d73861797420ea Mon Sep 17 00:00:00 2001 From: CyberMind-FR Date: Tue, 28 Jul 2026 09:14:12 +0200 Subject: [PATCH] fix(antirootkit): /status uses ExecLog.count() instead of len(recent(limit=100)) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- packages/secubox-antirootkit/api/execlog.py | 7 +++++++ packages/secubox-antirootkit/api/main.py | 9 +++++++-- packages/secubox-antirootkit/tests/test_api.py | 16 ++++++++++++++++ .../secubox-antirootkit/tests/test_execlog.py | 15 +++++++++++++++ 4 files changed, 45 insertions(+), 2 deletions(-) diff --git a/packages/secubox-antirootkit/api/execlog.py b/packages/secubox-antirootkit/api/execlog.py index 7dd8b614..a20b0d59 100644 --- a/packages/secubox-antirootkit/api/execlog.py +++ b/packages/secubox-antirootkit/api/execlog.py @@ -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. diff --git a/packages/secubox-antirootkit/api/main.py b/packages/secubox-antirootkit/api/main.py index bfba748b..2d556e09 100644 --- a/packages/secubox-antirootkit/api/main.py +++ b/packages/secubox-antirootkit/api/main.py @@ -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): diff --git a/packages/secubox-antirootkit/tests/test_api.py b/packages/secubox-antirootkit/tests/test_api.py index e76d0d99..d04c9416 100644 --- a/packages/secubox-antirootkit/tests/test_api.py +++ b/packages/secubox-antirootkit/tests/test_api.py @@ -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") diff --git a/packages/secubox-antirootkit/tests/test_execlog.py b/packages/secubox-antirootkit/tests/test_execlog.py index 245f88f5..304c1a5c 100644 --- a/packages/secubox-antirootkit/tests/test_execlog.py +++ b/packages/secubox-antirootkit/tests/test_execlog.py @@ -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