diff options
Diffstat (limited to 'packages/meshbay-hub/tests/test_public_groups_toggle.py')
| -rw-r--r-- | packages/meshbay-hub/tests/test_public_groups_toggle.py | 269 |
1 files changed, 269 insertions, 0 deletions
diff --git a/packages/meshbay-hub/tests/test_public_groups_toggle.py b/packages/meshbay-hub/tests/test_public_groups_toggle.py new file mode 100644 index 0000000..0d36b99 --- /dev/null +++ b/packages/meshbay-hub/tests/test_public_groups_toggle.py @@ -0,0 +1,269 @@ +""" +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 datetime, timezone + +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"): + 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(timezone.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") + 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") + 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") + + 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") + owner = await _user(client, "carol") + + 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") + 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") + 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") + owner = await _user(client, "erin") + 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") + owner = await _user(client, "frank") + member = await _user(client, "grace") + stranger = await _user(client, "heidi") + 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, signed with the running hub's own key. + + `federation._issue_mhp_token` binds `_hub_sk_pem` at import time, before the + lifespan loads it, so it cannot be used from a test. This signs directly. + """ + import time + import uuid + + import jwt + from meshbay_hub import auth as hub_auth + + now = int(time.time()) + # No `aud`: export_directory verifies without an expected audience, and PyJWT + # rejects a token that carries `aud` when decode() is given none. + return jwt.encode( + {"iss": hub_id, "sub": hub_id, + "jti": str(uuid.uuid4()), "iat": now, "exp": now + 300}, + hub_auth._hub_sk_pem, algorithm="EdDSA") + + +@pytest.mark.asyncio +async def test_disabled_empties_the_federation_export(client): + admin = await _admin(client) + owner = await _user(client, "ivan") + 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") + owner = await _user(client, "judy") + 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"]) |