"""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 async def _policy(client, admin, **values): r = await client.patch("/v1/admin/settings", json={"reports": values}, headers=admin) assert r.status_code == 200, r.text return r.json()["reports"] async def _public_group(client, db_session, owner, name="commons-mod"): from datetime import UTC, datetime from meshbay_hub.db.models import Group r = await client.post("/v1/groups", json={"name": name, "visibility": "public", "join_policy": "open"}, headers=owner) assert r.status_code == 201, r.text gid = r.json()["group_id"] (await db_session.get(Group, gid)).hosted_at = datetime.now(UTC) await db_session.commit() return gid async def _members(client, gid, names): out = [] for n in names: h = await _register_and_login(client, n) assert (await client.post(f"/v1/groups/{gid}/join", headers=h)).status_code == 200 out.append(h) return out async def _report(client, headers, gid, h=FAKE_HASH, **extra): return await client.post("/v1/reports", headers=headers, json={"content_hash": h, "group_id": gid, "reason": "illegal", **extra}) @pytest.fixture async def setting(client, admin_headers): """Reports allowed from a new account, so tests need not wait a day.""" await _policy(client, admin_headers, min_account_age_hours=0) return admin_headers @pytest.mark.asyncio async def test_report_requires_auth(client): r = await client.post("/v1/reports", json={ "content_hash": FAKE_HASH, "group_id": "g", "reason": "illegal"}) assert r.status_code in (401, 422) r = await client.post("/v1/reports", json={"content_hash": FAKE_HASH, "group_id": "g", "reason": "illegal"}, headers={"Authorization": "Bearer not-a-real-token"}) assert r.status_code == 401 @pytest.mark.asyncio async def test_a_node_token_cannot_report(client, db_session, setting): owner = await _register_and_login(client, "owner_nodetok") gid = await _public_group(client, db_session, owner, "nodetok-grp") node = await _node_headers(client, "node_reporter") assert (await _report(client, node, gid)).status_code == 403 @pytest.mark.asyncio async def test_a_new_account_cannot_report_yet(client, db_session, admin_headers): owner = await _register_and_login(client, "owner_young") gid = await _public_group(client, db_session, owner, "young-grp") [young] = await _members(client, gid, ["young_member"]) r = await _report(client, young, gid) assert r.status_code == 403 and "too new" in r.json()["detail"] @pytest.mark.asyncio async def test_only_a_member_of_that_public_group_may_report(client, db_session, setting): owner = await _register_and_login(client, "owner_member") gid = await _public_group(client, db_session, owner, "member-grp") stranger = await _register_and_login(client, "stranger_one") private = (await client.post("/v1/groups", json={"name": "priv-mod"}, headers=owner)).json()["group_id"] answers = {(await _report(client, stranger, gid)).json()["detail"], (await _report(client, owner, private)).json()["detail"], (await _report(client, stranger, "no-such-group")).json()["detail"]} # One uniform refusal: it must not say which groups exist or who is in them. assert len(answers) == 1 assert (await _report(client, owner, gid)).status_code == 201 @pytest.mark.asyncio async def test_a_report_says_nothing_about_how_close_review_is(client, db_session, setting): owner = await _register_and_login(client, "owner_quiet") gid = await _public_group(client, db_session, owner, "quiet-grp") r = await _report(client, owner, gid) assert r.status_code == 201 and r.json() == {"status": "logged"} @pytest.mark.asyncio async def test_same_reporter_cannot_walk_the_threshold(client, db_session, setting): await _policy(client, setting, review_threshold=1) owner = await _register_and_login(client, "owner_walk") gid = await _public_group(client, db_session, owner, "walk-grp") h = "b" * 64 [m] = await _members(client, gid, ["walker_one"]) await _report(client, m, gid, h) for _ in range(4): r = await _report(client, m, gid, h) assert r.json()["status"] == "already_reported" @pytest.mark.asyncio async def test_reaching_the_threshold_queues_for_an_administrator(client, db_session, setting): owner = await _register_and_login(client, "owner_queue") gid = await _public_group(client, db_session, owner, "queue-grp") h = "c" * 64 for m in await _members(client, gid, ["queue_r0", "queue_r1", "queue_r2"]): assert (await _report(client, m, gid, h)).status_code == 201 assert not await _blocked(client, h), "nothing is blocked without a decision" queue = (await client.get("/v1/admin/reports", headers=setting)).json()["reports"] [item] = [q for q in queue if q["hash"] == h] assert item["reporters"] == 3 and item["reasons"] == {"illegal": 3} assert item["groups"] == [{"id": gid, "name": "queue-grp"}] notes = (await client.get("/v1/notifications", headers=setting)).json() assert any(n["kind"] == "content_review" for n in notes["notifications"]) r = await client.post(f"/v1/admin/reports/{h}/block", headers=setting) assert r.status_code == 200 assert await _blocked(client, h) queue = (await client.get("/v1/admin/reports", headers=setting)).json()["reports"] assert h not in [q["hash"] for q in queue] @pytest.mark.asyncio async def test_a_dismissed_report_stays_dismissed(client, db_session, setting): await _policy(client, setting, review_threshold=1) owner = await _register_and_login(client, "owner_dismiss") gid = await _public_group(client, db_session, owner, "dismiss-grp") h = "d" * 64 [a, b] = await _members(client, gid, ["dismiss_a", "dismiss_b"]) await _report(client, a, gid, h) assert (await client.post(f"/v1/admin/reports/{h}/dismiss", headers=setting)).status_code == 200 await _report(client, b, gid, h) queue = (await client.get("/v1/admin/reports", headers=setting)).json()["reports"] assert h not in [q["hash"] for q in queue] assert not await _blocked(client, h) @pytest.mark.asyncio async def test_automatic_blocking_is_the_instances_choice(client, db_session, setting): await _policy(client, setting, review_threshold=2, auto_block=1) owner = await _register_and_login(client, "owner_auto") gid = await _public_group(client, db_session, owner, "auto-grp") h = "e" * 64 for m in await _members(client, gid, ["autoblock_r0", "autoblock_r1"]): await _report(client, m, gid, h) assert await _blocked(client, h) @pytest.mark.asyncio async def test_a_member_has_a_daily_allowance(client, db_session, setting): await _policy(client, setting, daily_per_account=2) owner = await _register_and_login(client, "owner_daily") gid = await _public_group(client, db_session, owner, "daily-grp") for i in range(2): assert (await _report(client, owner, gid, f"{i:064x}")).status_code == 201 assert (await _report(client, owner, gid, f"{9:064x}")).status_code == 429 @pytest.mark.asyncio async def test_a_report_is_shaped(client, db_session, setting): owner = await _register_and_login(client, "owner_shape") gid = await _public_group(client, db_session, owner, "shape-grp") assert (await _report(client, owner, gid, "not-a-hash")).status_code == 422 assert (await _report(client, owner, gid, reason="because")).status_code == 422 assert (await _report(client, owner, gid, detail="x" * 257)).status_code == 422 @pytest.mark.asyncio async def test_only_an_administrator_decides(client, db_session, setting): await _policy(client, setting, review_threshold=1) owner = await _register_and_login(client, "owner_decide") gid = await _public_group(client, db_session, owner, "decide-grp") await _report(client, owner, gid, "f" * 64) r = await client.post(f"/v1/admin/reports/{'f' * 64}/block", headers=owner) assert r.status_code == 403 @pytest.mark.asyncio async def test_reports_refused_when_public_groups_disabled(client, db_session, setting): owner = await _register_and_login(client, "owner_off") gid = await _public_group(client, db_session, owner, "off-grp") await client.patch("/v1/admin/settings", json={"allow_public_groups": False}, headers=setting) assert (await _report(client, owner, gid)).status_code == 403 @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 == []