1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
|
"""Tests for the node audit store (legal compliance IP/action logging)."""
import asyncio
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
from node_source import daemon_class_source
@pytest.fixture
async def audit(tmp_path):
store = AuditStore(db_path=tmp_path / "audit.db")
await store.open()
yield store
await store.close()
@pytest.mark.asyncio
async def test_log_and_retrieve(audit):
await audit.log_event(
user_id="u1", event="handshake", ip="1.2.3.4",
username="alice", group_id="g1", detail="test")
await audit.log_event(
user_id="u2", event="file_download", ip="5.6.7.8",
username="bob", group_id="g1", detail="video.mp4")
entries = await audit.get_entries()
assert len(entries) == 2
assert entries[0].event == "file_download"
assert entries[0].ip == "5.6.7.8"
assert entries[1].event == "handshake"
@pytest.mark.asyncio
async def test_filter_by_event(audit):
await audit.log_event(user_id="u1", event="handshake", ip="1.1.1.1")
await audit.log_event(user_id="u1", event="file_download", ip="1.1.1.1")
await audit.log_event(user_id="u1", event="handshake", ip="1.1.1.1")
entries = await audit.get_entries(event="handshake")
assert len(entries) == 2
assert all(e.event == "handshake" for e in entries)
@pytest.mark.asyncio
async def test_filter_by_user(audit):
await audit.log_event(user_id="u1", event="handshake")
await audit.log_event(user_id="u2", event="handshake")
entries = await audit.get_entries(user_id="u1")
assert len(entries) == 1
assert entries[0].user_id == "u1"
@pytest.mark.asyncio
async def test_pagination_newest_first(audit):
for i in range(5):
await audit.log_event(user_id="u1", event="connect", detail=f"e{i}")
# distinct timestamps so ORDER BY is deterministic
await audit._db.execute(
"UPDATE audit_log SET timestamp = ? WHERE detail = ?",
(1000 + i, f"e{i}"))
await audit._db.commit()
page1 = await audit.get_entries(limit=2, offset=0)
page2 = await audit.get_entries(limit=2, offset=2)
page3 = await audit.get_entries(limit=2, offset=4)
assert [e.detail for e in page1] == ["e4", "e3"] # newest first
assert [e.detail for e in page2] == ["e2", "e1"]
assert [e.detail for e in page3] == ["e0"] # last, partial page
# offset past the end is empty, not an error
assert await audit.get_entries(limit=2, offset=99) == []
@pytest.mark.asyncio
async def test_entry_count(audit):
assert await audit.entry_count() == 0
await audit.log_event(user_id="u1", event="connect")
await audit.log_event(user_id="u2", event="connect")
assert await audit.entry_count() == 2
@pytest.mark.asyncio
async def test_cleanup_old_entries(audit):
await audit.log_event(user_id="u1", event="old")
# manually backdate
await audit._db.execute(
"UPDATE audit_log SET timestamp = ? WHERE event = 'old'",
(time.time() - 400 * 86400,))
await audit._db.commit()
await audit.log_event(user_id="u2", event="recent")
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 = daemon_class_source()
assert "create_task(self._purge_audit_log())" in source
|