path: root/packages/meshbay-hub/tests/test_moderation.py

"""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 == []