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
137
138
139
|
"""Tests for moderation — reports + blocklist.
Reporting requires a signed-in account (it used to be anonymous, which made it a
network-wide censorship primitive), the auto-block threshold counts *distinct
reporting accounts*, and the whole flow is refused when the hub has public groups
switched off.
"""
import pytest
from meshbay_hub.api.deps import set_admin_usernames
FAKE_HASH = "a" * 64 # valid blake3 hex
async def _register_and_login(client, username: str) -> dict:
await client.post("/v1/users/register", json={
"username": username, "email": f"{username}@t.com",
"password": "reporter99pw",
})
r = await client.post("/v1/users/login",
json={"username": username, "password": "reporter99pw"})
return {"Authorization": f"Bearer {r.json()['access_token']}"}
@pytest.fixture
async def reporter(client):
return await _register_and_login(client, "reporter_one")
@pytest.fixture
async def admin_headers(client):
headers = await _register_and_login(client, "mod_admin")
set_admin_usernames(["mod_admin"])
return headers
@pytest.mark.asyncio
async def test_report_requires_auth(client):
# No credentials at all — FastAPI rejects the missing header before the body.
r = await client.post("/v1/reports", json={
"content_hash": FAKE_HASH, "reason": "illegal"})
assert r.status_code in (401, 422)
# A bogus token is a clean 401.
r = await client.post("/v1/reports",
json={"content_hash": FAKE_HASH, "reason": "illegal"},
headers={"Authorization": "Bearer not-a-real-token"})
assert r.status_code == 401
@pytest.mark.asyncio
async def test_report_content_logged(client, reporter):
r = await client.post("/v1/reports",
json={"content_hash": FAKE_HASH, "reason": "illegal"},
headers=reporter)
assert r.status_code == 201
data = r.json()
assert data["report_count"] == 1
assert data["status"] == "logged"
@pytest.mark.asyncio
async def test_same_reporter_cannot_walk_the_threshold(client, reporter):
h = "b" * 64
for _ in range(5):
r = await client.post("/v1/reports",
json={"content_hash": h, "reason": "spam"},
headers=reporter)
assert r.json()["report_count"] == 1
assert r.json()["status"] == "already_reported"
check = await client.get(f"/v1/blocklist/check?hash={h}")
assert check.json()["blocked"] is False
@pytest.mark.asyncio
async def test_auto_block_on_distinct_reporters(client):
h = "c" * 64
for i in range(3):
headers = await _register_and_login(client, f"reporter_{i}")
r = await client.post("/v1/reports",
json={"content_hash": h, "reason": "illegal"},
headers=headers)
assert r.json()["status"] == "auto_blocked"
assert r.json()["report_count"] == 3
check = await client.get(f"/v1/blocklist/check?hash={h}")
assert check.json()["blocked"] is True
@pytest.mark.asyncio
async def test_reports_refused_when_public_groups_disabled(client, reporter, admin_headers):
await client.patch("/v1/admin/settings",
json={"allow_public_groups": False},
headers=admin_headers)
r = await client.post("/v1/reports",
json={"content_hash": "d" * 64, "reason": "illegal"},
headers=reporter)
assert r.status_code == 403
@pytest.mark.asyncio
async def test_invalid_hash_rejected(client, reporter):
r = await client.post("/v1/reports",
json={"content_hash": "not-a-valid-blake3-hash",
"reason": "test"},
headers=reporter)
assert r.status_code == 422
@pytest.mark.asyncio
async def test_admin_add_remove_blocklist(client, admin_headers):
hash4 = "e" * 64
r = await client.post("/v1/admin/blocklist",
json={"content_hash": hash4, "reason": "csam"},
headers=admin_headers)
assert r.status_code == 201
r = await client.get(f"/v1/blocklist/check?hash={hash4}")
assert r.json()["blocked"] is True
r = await client.delete(f"/v1/admin/blocklist/{hash4}", headers=admin_headers)
assert r.status_code == 200
r = await client.get(f"/v1/blocklist/check?hash={hash4}")
assert r.json()["blocked"] is False
@pytest.mark.asyncio
async def test_full_blocklist(client, admin_headers):
hash5 = "f" * 64
await client.post("/v1/admin/blocklist",
json={"content_hash": hash5, "reason": "test"},
headers=admin_headers)
r = await client.get("/v1/blocklist")
assert r.status_code == 200
assert hash5 in r.json()["hashes"]
|