diff options
Diffstat (limited to 'packages/meshbay-hub/tests/test_admin_views.py')
| -rw-r--r-- | packages/meshbay-hub/tests/test_admin_views.py | 150 |
1 files changed, 150 insertions, 0 deletions
diff --git a/packages/meshbay-hub/tests/test_admin_views.py b/packages/meshbay-hub/tests/test_admin_views.py new file mode 100644 index 0000000..29f41d8 --- /dev/null +++ b/packages/meshbay-hub/tests/test_admin_views.py @@ -0,0 +1,150 @@ +""" +What the administration panel counts and lists. + +Deleting an account leaves a tombstone so the connection log stays readable +(see test_account_deletion). That row is not a user any more, and every place +that counts or lists people has to agree — a hub whose user count only ever +rises is measuring its own history rather than its population. +""" + +import base64 +import hashlib + +import pytest +from sqlalchemy import select + +from meshbay_hub.db.models import User + + +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, db_session, username="root"): + headers = await _user(client, username) + user = (await db_session.execute( + select(User).where(User.username == username))).scalar_one() + user.role = "admin" + await db_session.commit() + return headers + + +@pytest.mark.asyncio +async def test_a_deleted_account_stops_being_counted(client, db_session): + admin = await _admin(client, db_session) + leaver = await _user(client, "ghost") + + before = (await client.get("/v1/admin/stats", headers=admin)).json()["users"] + await client.request("DELETE", "/v1/users/me", headers=leaver, + json={"auth_key": _auth_key( + "a-long-enough-passphrase", "ghost")}) + after = (await client.get("/v1/admin/stats", headers=admin)).json()["users"] + + assert after == before - 1, "a tombstone is still being counted as a user" + + +@pytest.mark.asyncio +async def test_a_deleted_account_is_not_listed(client, db_session): + admin = await _admin(client, db_session, "root2") + leaver = await _user(client, "vanishing") + + await client.request("DELETE", "/v1/users/me", headers=leaver, + json={"auth_key": _auth_key( + "a-long-enough-passphrase", "vanishing")}) + + r = await client.get("/v1/admin/users", headers=admin) + names = [u["username"] for u in r.json()["users"]] + assert "vanishing" not in names + assert not any(n.startswith("deleted-") for n in names), \ + "the tombstone is showing under its placeholder name" + assert r.json()["total"] == len(names) + + +@pytest.mark.asyncio +async def test_the_member_list_of_a_group_skips_them(client, db_session): + admin = await _admin(client, db_session, "root3") + owner = await _user(client, "host3") + leaver = await _user(client, "quitter") + + g = await client.post("/v1/groups", json={"name": "party"}, headers=owner) + gid = g.json()["group_id"] + await client.post(f"/v1/groups/{gid}/members/quitter", json={}, headers=owner) + + await client.request("DELETE", "/v1/users/me", headers=leaver, + json={"auth_key": _auth_key( + "a-long-enough-passphrase", "quitter")}) + + members = (await client.get(f"/v1/groups/{gid}/members", + headers=owner)).json()["members"] + assert [m["username"] for m in members] == ["host3"] + + groups = (await client.get("/v1/admin/groups", headers=admin)).json()["groups"] + party = next(g for g in groups if g["name"] == "party") + assert party["member_count"] == 1, \ + f"the group still counts its departed member ({party['member_count']})" + + +# ── Where a node actually is ──────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_a_node_is_recorded_at_the_address_it_announced_from( + client, db_session): + """ + The hub sees the connection; the node only tells us what a STUN server told + it. The observed address is the evidence, and the announcement carries an + Ed25519 signature over a fresh timestamp, so it is the address of whoever + holds the node key. + """ + import base64 + import time + + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + from cryptography.hazmat.primitives import serialization + + from meshbay_hub.db.models import Node + + admin = await _admin(client, db_session, "root4") + owner = await _user(client, "nodeowner") + uid = (await db_session.execute( + select(User.id).where(User.username == "nodeowner"))).scalar_one() + + sk = Ed25519PrivateKey.generate() + pk_b64 = base64.b64encode(sk.public_key().public_bytes( + encoding=serialization.Encoding.Raw, + format=serialization.PublicFormat.Raw)).decode() + ts = int(time.time()) + message = f"meshbay:node_announce:{uid}:{pk_b64}:{ts}".encode() + + r = await client.post("/v1/nodes/announce", headers=owner, json={ + "pk_node": pk_b64, + # A claim, and a deliberately implausible one. + "endpoint_hint": "203.0.113.42:19001", + "timestamp": ts, + "signature": base64.b64encode(sk.sign(message)).decode(), + }) + assert r.status_code in (200, 201), r.text + + node = (await db_session.execute( + select(Node).where(Node.pk_node == pk_b64))).scalar_one() + assert node.observed_ip, "no address was recorded for the announcement" + assert node.observed_ip != "203.0.113.42", \ + "the hub recorded what the node claimed instead of what it saw" + assert node.last_seen is not None + + listed = (await client.get("/v1/admin/nodes", headers=admin)).json()["nodes"] + row = next(n for n in listed if n["pk_node"] == pk_b64) + assert row["observed_ip"] == node.observed_ip + assert row["endpoint_hint"] == "203.0.113.42:19001", \ + "the claim is still shown, just not as the answer" + assert row["username"] == "nodeowner" |