diff options
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/daemon.py | 22 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_audit.py | 41 |
2 files changed, 61 insertions, 2 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 7646c5d..666a5cc 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -41,7 +41,7 @@ from meshbay_common.background import spawn from meshbay_common.paths import fold from meshbay_common import MNP_VERSION from meshbay_common.protocol import MNP -from meshbay_node.audit import AuditStore +from meshbay_node.audit import RETENTION_DAYS as AUDIT_RETENTION_DAYS, AuditStore from meshbay_node.bundle_store import BundleStore from meshbay_node.chat.store import ChatStore from meshbay_node.config import Config, DEFAULT_CONFIG_PATH, load_config, write_example_config @@ -469,6 +469,7 @@ class NodeDaemon: self._audit_store = AuditStore(db_path=audit_db) await self._audit_store.open() log.info("Audit store opened: %s", audit_db) + self._tasks.append(asyncio.create_task(self._purge_audit_log())) # 6b. Media cache (Videos app — TMDB metadata + thumbnails). # Node-wide like audit.db, not per-group: a thumbnail is the same @@ -1128,6 +1129,25 @@ class NodeDaemon: log.warning("Reaping partial uploads failed: %s", exc) await asyncio.sleep(interval) + async def _purge_audit_log(self, interval: float = 86400.0) -> None: + """ + Delete audit entries past the retention period, at start then daily. + + `AuditStore.cleanup` leaves scheduling to its caller; until this loop + nobody called it and a node kept every entry for ever. `interval` is a + parameter so a test can drive this without waiting a day. + """ + while True: + try: + if self._audit_store: + deleted = await self._audit_store.cleanup() + if deleted: + log.info("Purged %d audit entries older than %d days", + deleted, AUDIT_RETENTION_DAYS) + except Exception as exc: # never let the janitor kill the node + log.warning("Purging the audit log failed: %s", exc) + await asyncio.sleep(interval) + def _reap_once(self, now: float | None = None) -> int: """One pass over every group. Returns how many files were deleted.""" groups = (self._webrtc._ctx.get("groups") or {}) if self._webrtc else {} diff --git a/packages/meshbay-node/tests/test_audit.py b/packages/meshbay-node/tests/test_audit.py index 8755be2..b5b98b5 100644 --- a/packages/meshbay-node/tests/test_audit.py +++ b/packages/meshbay-node/tests/test_audit.py @@ -1,10 +1,13 @@ """Tests for the node audit store (legal compliance IP/action logging).""" +import asyncio +import inspect import time import pytest - from meshbay_node.audit import AuditStore +from meshbay_node.config import Config, HubConfig, KeystoreConfig, NodeConfig +from meshbay_node.daemon import NodeDaemon @pytest.fixture @@ -95,3 +98,39 @@ async def test_cleanup_old_entries(audit): deleted = await audit.cleanup(retention_days=365) assert deleted == 1 assert await audit.entry_count() == 1 + + +@pytest.mark.asyncio +async def test_daemon_purges_the_audit_log(audit, tmp_path): + """The daemon runs the retention itself: at start, then on its interval.""" + daemon = NodeDaemon(Config( + hub=HubConfig(url="http://localhost:9999", username="testuser"), + node=NodeConfig(), groups=[], + keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), + data_dir=tmp_path / "data")) + daemon._audit_store = audit + + async def backdated(event): + await audit.log_event(user_id="u1", event=event) + await audit._db.execute( + "UPDATE audit_log SET timestamp = ? WHERE event = ?", + (time.time() - 400 * 86400, event)) + await audit._db.commit() + + await backdated("old") + await audit.log_event(user_id="u2", event="recent") + task = asyncio.create_task(daemon._purge_audit_log(interval=0.05)) + try: + await asyncio.sleep(0.02) + assert [e.event for e in await audit.get_entries()] == ["recent"] + await backdated("later") + await asyncio.sleep(0.1) + assert [e.event for e in await audit.get_entries()] == ["recent"] + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +def test_daemon_starts_the_purge(): + source = inspect.getsource(NodeDaemon) + assert "create_task(self._purge_audit_log())" in source |