"""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']}"} async def _node_headers(client, username: str) -> dict: """A node daemon's token for a fresh account — what a node syncs with.""" from meshbay_hub.auth import issue_access_token user = await _register_and_login(client, username) me = (await client.get("/v1/users/me", headers=user)).json() tok = issue_access_token(me["user_id"], ttl=3600, groups=[], scope="node") return {"Authorization": f"Bearer {tok}"} async def _blocked(client, h: str) -> bool: node = await _node_headers(client, f"node_{h[:6]}_{len(h)}") return h in (await client.get("/v1/blocklist", headers=node)).json()["hashes"] @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" assert not await _blocked(client, h) @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 assert await _blocked(client, h) @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 assert await _blocked(client, hash4) r = await client.delete(f"/v1/admin/blocklist/{hash4}", headers=admin_headers) assert r.status_code == 200 node = await _node_headers(client, "node_after_unblock") assert hash4 not in (await client.get("/v1/blocklist", headers=node)).json()["hashes"] @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) node = await _node_headers(client, "node_full") r = await client.get("/v1/blocklist", headers=node) assert r.status_code == 200 assert hash5 in r.json()["hashes"] @pytest.mark.asyncio async def test_only_a_node_token_reads_the_blocklist(client, reporter): assert (await client.get("/v1/blocklist")).status_code in (401, 422) assert (await client.get("/v1/blocklist", headers=reporter)).status_code == 403 @pytest.mark.asyncio async def test_the_blocklist_pages_past_its_limit(client, admin_headers): hashes = sorted(f"{i:064x}" for i in range(5)) for h in hashes: await client.post("/v1/admin/blocklist", json={"content_hash": h, "reason": "test"}, headers=admin_headers) node = await _node_headers(client, "node_pager") seen, after = [], "" while True: page = (await client.get(f"/v1/blocklist?limit=2&after={after}", headers=node)).json() seen += page["hashes"] if not page["next"]: break after = page["next"] assert [h for h in seen if h in hashes] == hashes @pytest.mark.asyncio async def test_an_admin_cannot_block_something_that_is_not_a_hash(client, admin_headers): r = await client.post("/v1/admin/blocklist", json={"content_hash": "x" * 5000, "reason": "test"}, headers=admin_headers) assert r.status_code == 422 @pytest.mark.asyncio async def test_a_change_is_pushed_to_nodes_hosting_a_public_group_only(db_session): """A node hosting only private groups has nothing to apply the list to.""" import json import meshbay_hub.api.revocation as rev from meshbay_hub.db.models import Group, User db_session.add(User(id="owner-bl", username="owner_bl", email="x", hub_id="h", pw_hash=b"x", pw_salt=b"x")) db_session.add(Group(id="pub-bl", name="pub", admin_id="owner-bl", visibility="public")) db_session.add(Group(id="priv-bl", name="priv", admin_id="owner-bl", visibility="private")) await db_session.commit() class _WS: def __init__(self): self.sent = [] async def send_text(self, text): self.sent.append(json.loads(text)) public_node, private_node = _WS(), _WS() saved = dict(rev._connected_nodes), dict(rev._node_groups) rev._connected_nodes.update({"n-pub": public_node, "n-priv": private_node}) rev._node_groups.update({"n-pub": ["pub-bl"], "n-priv": ["priv-bl"]}) try: sent = await rev.broadcast_blocklist_update(db_session, add=["a" * 64]) finally: rev._connected_nodes.clear() rev._connected_nodes.update(saved[0]) rev._node_groups.clear() rev._node_groups.update(saved[1]) assert sent == 1 assert public_node.sent == [{"type": "blocklist_update", "add": ["a" * 64], "remove": []}] assert private_node.sent == []