""" An admin can turn off public groups for the whole hub. The control is server-side and covers every hub-mediated path, not just creation: * `create_group` refuses `visibility=public` * `list_public_groups` (the directory) returns nothing * `join_group` refuses open joining of a public group * `group_online_nodes` hands a non-member no node to connect to * `signaling.webrtc_offer` drops the "node hosts an open group" fallback * `federation.export_directory` advertises nothing to peers Existing members of a group that predates the switch keep their membership row and their access — plan A, not a purge. """ import base64 import hashlib from datetime import UTC, datetime import pytest from meshbay_hub.api.deps import set_admin_usernames from meshbay_hub.db.models import Group def _auth_key(password: str, username: str) -> str: salt = hashlib.sha256(f"meshbay:auth:v1:{username}".encode()).digest() return base64.b64encode( hashlib.pbkdf2_hmac("sha512", password.encode(), salt, 600_000, 32)).decode() async def _user(client, username, password="a-long-enough-passphrase"): await client.post("/v1/users/register", json={ "username": username, "email": f"{username}@example.com", "auth_key": _auth_key(password, username)}) r = await client.post("/v1/users/login", json={ "username": username, "auth_key": _auth_key(password, username)}) return {"Authorization": f"Bearer {r.json()['access_token']}"} async def _admin(client, username="root_test"): await _user(client, username) set_admin_usernames([username]) # re-login so the token is minted with the admin role in context r = await client.post("/v1/users/login", json={ "username": username, "auth_key": _auth_key("a-long-enough-passphrase", username)}) return {"Authorization": f"Bearer {r.json()['access_token']}"} def _public(name): return {"name": name, "visibility": "public", "join_policy": "open"} async def _create_public(client, owner, name): r = await client.post("/v1/groups", json=_public(name), headers=owner) assert r.status_code == 201, r.text return r.json()["group_id"] async def _mark_hosted(db_session, *group_ids): """Pretend a node announced these groups, as /v1/nodes/ws would.""" for gid in group_ids: (await db_session.get(Group, gid)).hosted_at = datetime.now(UTC) await db_session.commit() async def _set_public_groups(client, admin, allowed): r = await client.patch("/v1/admin/settings", json={"allow_public_groups": allowed}, headers=admin) assert r.status_code == 200 assert r.json()["allow_public_groups"] is allowed @pytest.mark.asyncio async def test_public_groups_are_allowed_by_default(client): owner = await _user(client, "alice_test") r = await client.get("/v1/hub/info") assert r.json()["allow_public_groups"] is True r = await client.post("/v1/groups", json=_public("open-house"), headers=owner) assert r.status_code == 201, r.text @pytest.mark.asyncio async def test_a_normal_member_cannot_change_the_setting(client): member = await _user(client, "mallory_test") r = await client.patch("/v1/admin/settings", json={"allow_public_groups": False}, headers=member) assert r.status_code == 403 @pytest.mark.asyncio async def test_admin_disables_public_groups_end_to_end(client): admin = await _admin(client) owner = await _user(client, "bob_test") r = await client.patch("/v1/admin/settings", json={"allow_public_groups": False}, headers=admin) assert r.status_code == 200 assert r.json()["allow_public_groups"] is False # Reflected on both the admin read and the unauthenticated hub info. assert (await client.get("/v1/admin/settings", headers=admin) ).json()["allow_public_groups"] is False assert (await client.get("/v1/hub/info")).json()["allow_public_groups"] is False # A member is refused a public group... r = await client.post("/v1/groups", json=_public("nope"), headers=owner) assert r.status_code == 403 assert "public" in r.json()["detail"].lower() # ...and so is the admin: the way back is to re-enable it, not slip past. r = await client.post("/v1/groups", json=_public("admin-nope"), headers=admin) assert r.status_code == 403 # Private groups are unaffected. r = await client.post("/v1/groups", json={"name": "still-fine", "visibility": "private"}, headers=owner) assert r.status_code == 201, r.text @pytest.mark.asyncio async def test_re_enabling_restores_public_creation(client): admin = await _admin(client, "chief_test") owner = await _user(client, "carol_test") await client.patch("/v1/admin/settings", json={"allow_public_groups": False}, headers=admin) r = await client.post("/v1/groups", json=_public("first-try"), headers=owner) assert r.status_code == 403 await client.patch("/v1/admin/settings", json={"allow_public_groups": True}, headers=admin) r = await client.post("/v1/groups", json=_public("second-try"), headers=owner) assert r.status_code == 201, r.text @pytest.mark.asyncio async def test_a_patch_without_the_field_is_a_no_op(client): admin = await _admin(client, "keeper_test") await client.patch("/v1/admin/settings", json={"allow_public_groups": False}, headers=admin) r = await client.patch("/v1/admin/settings", json={}, headers=admin) assert r.status_code == 200 assert r.json()["allow_public_groups"] is False # ── plan A: existing public groups when the switch is off ──────────────────── @pytest.mark.asyncio async def test_disabled_empties_the_public_directory(client, db_session): admin = await _admin(client) owner = await _user(client, "dora_test") gid = await _create_public(client, owner, "town-square") await _mark_hosted(db_session, gid) listed = await client.get("/v1/groups") assert any(g["id"] == gid for g in listed.json()["groups"]) await _set_public_groups(client, admin, False) listed = await client.get("/v1/groups") body = listed.json() assert body["groups"] == [] and body["total"] == 0 @pytest.mark.asyncio async def test_disabled_refuses_open_join_of_a_public_group(client, db_session): admin = await _admin(client, "chief_test") owner = await _user(client, "erin_test") early = await _user(client, "early-bird") late = await _user(client, "late-comer") gid = await _create_public(client, owner, "commons") await _mark_hosted(db_session, gid) assert (await client.post(f"/v1/groups/{gid}/join", headers=early)).status_code == 200 await _set_public_groups(client, admin, False) r = await client.post(f"/v1/groups/{gid}/join", headers=late) assert r.status_code == 403 # The person who joined while it was allowed is still a member. mine = await client.get("/v1/groups/mine", headers=early) assert any(g["id"] == gid for g in mine.json()["groups"]) @pytest.mark.asyncio async def test_disabled_hands_a_non_member_no_node(client, db_session): admin = await _admin(client, "chief_test") owner = await _user(client, "frank_test") member = await _user(client, "grace_test") stranger = await _user(client, "heidi_test") gid = await _create_public(client, owner, "atrium") await _mark_hosted(db_session, gid) assert (await client.post(f"/v1/groups/{gid}/join", headers=member)).status_code == 200 # While allowed, anyone may ask which nodes serve a public group. assert (await client.get(f"/v1/groups/{gid}/nodes", headers=stranger) ).status_code == 200 await _set_public_groups(client, admin, False) assert (await client.get(f"/v1/groups/{gid}/nodes", headers=stranger) ).status_code == 403 # Members and the owner still get an answer (no nodes online here, but 200). assert (await client.get(f"/v1/groups/{gid}/nodes", headers=member) ).status_code == 200 assert (await client.get(f"/v1/groups/{gid}/nodes", headers=owner) ).status_code == 200 def _mhp_token(hub_id): """A peer-hub JWT — from the hub's own issuer, which is the point. This used to sign by hand, and its two comments recorded, accurately, the reasons it had to: `_issue_mhp_token` bound `_hub_sk_pem` at import, before the lifespan loads it, so it signed with `None`; and it sets an `aud` that the verifier named no audience for, which PyJWT refuses outright. Both were written down here as facts to route around rather than as the defects they were — between them MHP could not complete one authenticated request between two real hubs. Fixed at the source, so this can call it. """ from meshbay_hub.api.federation import _issue_mhp_token return _issue_mhp_token(hub_id) @pytest.mark.asyncio async def test_disabled_empties_the_federation_export(client, monkeypatch): """What the export advertises, for the day federation re-opens. MHP is switched off in the code and every route answers 503 (`federation.FEDERATION_ENABLED`, §7.6), so this opens the gate for its own duration. The subject is the public-groups switch, not federation: with public groups off, the export must advertise nothing, and that has to stay true whether the door is open today or not. """ from meshbay_hub.api import federation monkeypatch.setattr(federation, "FEDERATION_ENABLED", True) admin = await _admin(client) owner = await _user(client, "ivan_test") await _create_public(client, owner, "exported-square") info = (await client.get("/mhp/info")).json() await client.post("/mhp/peers", headers=admin, json={ "hub_id": info["hub_id"], "hub_url": "https://peer.example", "pk_hub_pem": info["pk_hub_pem"], }) tok = _mhp_token(info["hub_id"]) r = await client.get("/mhp/directory", headers={"Authorization": f"Bearer {tok}"}) assert r.status_code == 200 and len(r.json()["groups"]) == 1 await _set_public_groups(client, admin, False) r = await client.get("/mhp/directory", headers={"Authorization": f"Bearer {tok}"}) assert r.status_code == 200 and r.json()["groups"] == [] @pytest.mark.asyncio async def test_re_enabling_brings_the_directory_back(client, db_session): admin = await _admin(client, "chief_test") owner = await _user(client, "judy_test") gid = await _create_public(client, owner, "reopened") await _mark_hosted(db_session, gid) await _set_public_groups(client, admin, False) assert (await client.get("/v1/groups")).json()["groups"] == [] await _set_public_groups(client, admin, True) listed = await client.get("/v1/groups") assert any(g["id"] == gid for g in listed.json()["groups"])