""" Integration tests for the Hub API. Uses SQLite in-memory + httpx.AsyncClient — no PostgreSQL, no network. """ from datetime import UTC import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey from meshbay_common.crypto import pk_to_b64 from meshbay_hub.api.deps import set_admin_usernames def _gen_user_keys(): sk_ed = Ed25519PrivateKey.generate() sk_x = X25519PrivateKey.generate() return ( pk_to_b64(sk_ed.public_key()), pk_to_b64(sk_x.public_key()), sk_x, ) async def _announce_signed(client, token: str) -> tuple[str, str]: """ Announce a node with proof of possession (M8). The node key is independent of the user's identity key, so this mints a fresh one and signs the domain-separated announce message with it. """ import base64 as _b64 import time as _t me = await client.get("/v1/users/me", headers={"Authorization": f"Bearer {token}"}) user_id = me.json()["user_id"] sk_node = Ed25519PrivateKey.generate() pk_node = pk_to_b64(sk_node.public_key()) ts = _t.time().__trunc__() msg = f"meshbay:node_announce:{user_id}:{pk_node}:{ts}".encode() r = await client.post("/v1/nodes/announce", json={ "pk_node": pk_node, "endpoint_hint": "1.2.3.4:19000", "timestamp": ts, "signature": _b64.b64encode(sk_node.sign(msg)).decode(), }, headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 201, r.text return r.json()["node_id"], pk_node # ── Hub info ────────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_hub_info(client): r = await client.get("/v1/hub/info") assert r.status_code == 200 data = r.json() assert "mnp_version" in data assert "mhp_version" in data @pytest.mark.asyncio async def test_health(client): r = await client.get("/v1/health") assert r.status_code == 200 data = r.json() assert data["status"] == "ok" assert "version" in data assert "connected_nodes" in data @pytest.mark.asyncio async def test_hub_pubkey(client): r = await client.get("/v1/hub/pubkey") assert r.status_code == 200 pem = r.json()["pk_hub_pem"] assert pem.startswith("-----BEGIN PUBLIC KEY-----") # ── Users ───────────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_register_and_login(client): pk_ed, pk_x, _ = _gen_user_keys() r = await client.post("/v1/users/register", json={ "username": "alice_test", "email": "alice@example.com", "password": "alicepass99", "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x, }) assert r.status_code == 201 assert "user_id" in r.json() r = await client.post("/v1/users/login", json={ "username": "alice_test", "password": "alicepass99"}) assert r.status_code == 200 data = r.json() assert "access_token" in data assert "refresh_token" in data assert data["token_type"] == "bearer" @pytest.mark.asyncio async def test_register_duplicate_rejected(client): pk_ed, pk_x, _ = _gen_user_keys() body = {"username": "bob_test", "email": "bob@example.com", "password": "bobpass99", "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x} await client.post("/v1/users/register", json=body) r = await client.post("/v1/users/register", json=body) assert r.status_code == 409 @pytest.mark.asyncio async def test_wrong_password_rejected(client): pk_ed, pk_x, _ = _gen_user_keys() await client.post("/v1/users/register", json={ "username": "carol_test", "email": "carol@example.com", "password": "carolpass99", "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x}) r = await client.post("/v1/users/login", json={ "username": "carol_test", "password": "wrongpass"}) assert r.status_code == 401 @pytest.mark.asyncio async def test_jwt_offline_verify(client, hub_key_path): """JWT returned by login must be verifiable offline with hub's public key.""" import jwt as pyjwt pk_ed, pk_x, _ = _gen_user_keys() await client.post("/v1/users/register", json={ "username": "dave_test", "email": "dave@example.com", "password": "davepass99", "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x}) r = await client.post("/v1/users/login", json={ "username": "dave_test", "password": "davepass99"}) token = r.json()["access_token"] r_pk = await client.get("/v1/hub/pubkey") hub_pk_pem = r_pk.json()["pk_hub_pem"].encode() decoded = pyjwt.decode(token, hub_pk_pem, algorithms=["EdDSA"]) assert "jti" in decoded # mandatory # The token carries no user key. It used to, and the node recorded it as the # uploader's identity — so whoever issued tokens decided who could delete a # file. The hub certifies accounts; nodes pin keys. assert "pk_user" not in decoded @pytest.mark.asyncio async def test_token_refresh(client): pk_ed, pk_x, _ = _gen_user_keys() await client.post("/v1/users/register", json={ "username": "eve_test", "email": "eve@example.com", "password": "evepass99", "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x}) r = await client.post("/v1/users/login", json={ "username": "eve_test", "password": "evepass99"}) rt = r.json()["refresh_token"] at = r.json()["access_token"] r2 = await client.post("/v1/users/token/refresh", json={"refresh_token": rt}) assert r2.status_code == 200 assert r2.json()["access_token"] != at # new token (different jti) assert "refresh_token" in r2.json() # rotated refresh token returned @pytest.mark.asyncio async def test_refresh_token_rotation_old_rejected(client): """After rotation, old refresh token is rejected.""" pk_ed, pk_x, _ = _gen_user_keys() await client.post("/v1/users/register", json={ "username": "rot_user", "email": "rot@x.com", "password": "rotpass99", "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x}) r = await client.post("/v1/users/login", json={ "username": "rot_user", "password": "rotpass99"}) rt1 = r.json()["refresh_token"] r2 = await client.post("/v1/users/token/refresh", json={"refresh_token": rt1}) assert r2.status_code == 200 rt2 = r2.json()["refresh_token"] assert rt2 != rt1 # Old token reuse → detected and family revoked r3 = await client.post("/v1/users/token/refresh", json={"refresh_token": rt1}) assert r3.status_code == 401 assert "reuse" in r3.json()["detail"].lower() # New token also revoked (entire family) r4 = await client.post("/v1/users/token/refresh", json={"refresh_token": rt2}) assert r4.status_code == 401 @pytest.mark.asyncio async def test_get_user_pubkeys(client): """ The endpoint resolves an account; it is not a key directory any more. Publishing user identity keys is what finding H3 exploited — the invite flow wrapped the group key for whatever came back. Keys are now generated per node and pinned there, so there is nothing here to substitute. """ pk_ed, pk_x, _ = _gen_user_keys() await client.post("/v1/users/register", json={ "username": "frank_test", "email": "frank@example.com", "password": "frankpass99"}) login = await client.post("/v1/users/login", json={ "username": "frank_test", "password": "frankpass99"}) token = login.json()["access_token"] r = await client.get("/v1/users/frank_test/pubkeys", headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 200 body = r.json() assert body["user_id"] and body["username"] == "frank_test" assert "pk_ed25519" not in body, "user identity keys must not be published (H3)" assert "pk_x25519" not in body, "user identity keys must not be published (H3)" # ── Nodes ───────────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_announce_and_get_node(client): pk_ed, pk_x, _ = _gen_user_keys() await client.post("/v1/users/register", json={ "username": "node1_test", "email": "n@example.com", "password": "nodepass99", "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x}) login = await client.post("/v1/users/login", json={ "username": "node1_test", "password": "nodepass99"}) token = login.json()["access_token"] hdrs = {"Authorization": f"Bearer {token}"} node_id, pk_node = await _announce_signed(client, token) r2 = await client.get(f"/v1/nodes/{node_id}", headers=hdrs) assert r2.status_code == 200 # The node key is independent of the user identity key (M8). assert r2.json()["pk_node"] == pk_node assert r2.json()["endpoint_hint"] == "1.2.3.4:19000" # ── Groups + GEK bundles ────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_group_member_add(client): """Admin creates group and adds member (GEK exchange happens P2P on node).""" pk_ed_a, pk_x_a, _ = _gen_user_keys() pk_ed_b, pk_x_b, sk_x_b = _gen_user_keys() for uname, email, pwd, pk_ed, pk_x in [ ("alice2_test", "a2@x.com", "alicepass99", pk_ed_a, pk_x_a), ("bob2_test", "b2@x.com", "bobpass99", pk_ed_b, pk_x_b), ]: await client.post("/v1/users/register", json={ "username": uname, "email": email, "password": pwd, "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x}) alice_token = (await client.post("/v1/users/login", json={"username": "alice2_test", "password": "alicepass99"}) ).json()["access_token"] a_hdrs = {"Authorization": f"Bearer {alice_token}"} r = await client.post("/v1/groups", json={"name": "mygroup"}, headers=a_hdrs) assert r.status_code == 201 group_id = r.json()["group_id"] # Add bob as member (hub handles membership only, GEK exchange is P2P) r = await client.post(f"/v1/groups/{group_id}/members/bob2_test", json={}, headers=a_hdrs) assert r.status_code == 201 # Verify bob is in the group bob_token = (await client.post("/v1/users/login", json={"username": "bob2_test", "password": "bobpass99"})).json()["access_token"] b_hdrs = {"Authorization": f"Bearer {bob_token}"} r = await client.get(f"/v1/groups/{group_id}/members", headers=b_hdrs) assert r.status_code == 200 members = [m["username"] for m in r.json()["members"]] assert "alice2_test" in members assert "bob2_test" in members @pytest.mark.asyncio async def test_non_admin_cannot_add_member(client): pk_ed_a, pk_x_a, _ = _gen_user_keys() pk_ed_b, pk_x_b, _ = _gen_user_keys() for uname, email, pwd, pk_ed, pk_x in [ ("charlie_test", "c@x.com", "charliepass", pk_ed_a, pk_x_a), ("dan_test", "d@x.com", "danpass1234", pk_ed_b, pk_x_b), ]: await client.post("/v1/users/register", json={ "username": uname, "email": email, "password": pwd, "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x}) charlie_token = (await client.post("/v1/users/login", json={"username": "charlie_test", "password": "charliepass"}) ).json()["access_token"] dan_token = (await client.post("/v1/users/login", json={"username": "dan_test", "password": "danpass1234"}) ).json()["access_token"] r = await client.post("/v1/groups", json={"name": "charlies-group"}, headers={"Authorization": f"Bearer {charlie_token}"}) group_id = r.json()["group_id"] # Dan (non-admin) tries to add a member → 403 r = await client.post(f"/v1/groups/{group_id}/members/charlie_test", json={}, headers={"Authorization": f"Bearer {dan_token}"}) assert r.status_code == 403 @pytest.mark.asyncio async def test_jwt_contains_groups_claim(client): """JWT must contain a 'groups' list with group_ids the user is a member of.""" import jwt as pyjwt pk_ed_a, pk_x_a, _ = _gen_user_keys() pk_ed_b, pk_x_b, sk_x_b = _gen_user_keys() await client.post("/v1/users/register", json={ "username": "grp_alice", "email": "ga@x.com", "password": "alicepass99", "pk_user_ed25519": pk_ed_a, "pk_user_x25519": pk_x_a}) await client.post("/v1/users/register", json={ "username": "grp_bob_test", "email": "gb@x.com", "password": "bobpass99", "pk_user_ed25519": pk_ed_b, "pk_user_x25519": pk_x_b}) # Login before joining any group — groups should be empty r = await client.post("/v1/users/login", json={ "username": "grp_bob_test", "password": "bobpass99"}) token_pre = r.json()["access_token"] r_pk = await client.get("/v1/hub/pubkey") hub_pk = r_pk.json()["pk_hub_pem"].encode() decoded_pre = pyjwt.decode(token_pre, hub_pk, algorithms=["EdDSA"]) assert decoded_pre["groups"] == [] # Alice creates a group and adds Bob alice_token = (await client.post("/v1/users/login", json={"username": "grp_alice", "password": "alicepass99"}) ).json()["access_token"] r = await client.post("/v1/groups", json={"name": "testgroup"}, headers={"Authorization": f"Bearer {alice_token}"}) group_id = r.json()["group_id"] await client.post(f"/v1/groups/{group_id}/members/grp_bob_test", json={}, headers={"Authorization": f"Bearer {alice_token}"}) # Login again — groups should contain the new group r = await client.post("/v1/users/login", json={ "username": "grp_bob_test", "password": "bobpass99"}) token_post = r.json()["access_token"] decoded_post = pyjwt.decode(token_post, hub_pk, algorithms=["EdDSA"]) assert group_id in decoded_post["groups"] # Alice (admin) should also have the group in her JWT r = await client.post("/v1/users/login", json={ "username": "grp_alice", "password": "alicepass99"}) decoded_alice = pyjwt.decode(r.json()["access_token"], hub_pk, algorithms=["EdDSA"]) assert group_id in decoded_alice["groups"] # ── My groups (9.6) ───────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_my_groups(client, db_session): """GET /v1/groups/mine returns groups the user belongs to.""" pk_ed_a, pk_x_a, _ = _gen_user_keys() pk_ed_b, pk_x_b, _ = _gen_user_keys() await client.post("/v1/users/register", json={ "username": "mg_alice", "email": "mga@x.com", "password": "alicepass99", "pk_user_ed25519": pk_ed_a, "pk_user_x25519": pk_x_a}) await client.post("/v1/users/register", json={ "username": "mg_bob_test", "email": "mgb@x.com", "password": "bobpass99", "pk_user_ed25519": pk_ed_b, "pk_user_x25519": pk_x_b}) alice_token = (await client.post("/v1/users/login", json={"username": "mg_alice", "password": "alicepass99"})).json()["access_token"] bob_token = (await client.post("/v1/users/login", json={"username": "mg_bob_test", "password": "bobpass99"})).json()["access_token"] # Bob has no groups initially r = await client.get("/v1/groups/mine", headers={"Authorization": f"Bearer {bob_token}"}) assert r.status_code == 200 assert r.json()["groups"] == [] # Alice creates a group and adds Bob r = await client.post("/v1/groups", json={"name": "mg-group"}, headers={"Authorization": f"Bearer {alice_token}"}) group_id = r.json()["group_id"] await client.post(f"/v1/groups/{group_id}/members/mg_bob_test", json={}, headers={"Authorization": f"Bearer {alice_token}"}) # A group no node has announced is shown to its owner only — a member would # otherwise see a name they cannot open. Stamped here so the rest of this # test is about membership, which is what it was written for. from datetime import datetime from meshbay_hub.db.models import Group (await db_session.get(Group, group_id)).hosted_at = datetime.now(UTC) await db_session.commit() # Re-login to get fresh token with group claims bob_token = (await client.post("/v1/users/login", json={"username": "mg_bob_test", "password": "bobpass99"})).json()["access_token"] # Now Bob should see the group r = await client.get("/v1/groups/mine", headers={"Authorization": f"Bearer {bob_token}"}) assert r.status_code == 200 groups = r.json()["groups"] assert len(groups) == 1 assert groups[0]["id"] == group_id assert groups[0]["name"] == "mg-group" assert groups[0]["is_admin"] is False # Alice should see it too, with is_admin=True alice_token = (await client.post("/v1/users/login", json={"username": "mg_alice", "password": "alicepass99"})).json()["access_token"] r = await client.get("/v1/groups/mine", headers={"Authorization": f"Bearer {alice_token}"}) groups = r.json()["groups"] assert any(g["id"] == group_id and g["is_admin"] for g in groups) # Unauthenticated → rejected r = await client.get("/v1/groups/mine") assert r.status_code >= 400 @pytest.mark.asyncio async def test_group_online_nodes(client): """GET /v1/groups/{id}/nodes returns online nodes serving the group.""" from meshbay_hub.api.revocation import _connected_nodes, _node_groups pk_ed, pk_x, _ = _gen_user_keys() await client.post("/v1/users/register", json={ "username": "gn_user_test", "email": "gn@x.com", "password": "gnpass999", "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x}) r = await client.post("/v1/users/login", json={"username": "gn_user_test", "password": "gnpass999"}) token = r.json()["access_token"] r = await client.post("/v1/groups", json={"name": "gn-group"}, headers={"Authorization": f"Bearer {token}"}) group_id = r.json()["group_id"] # Re-login to get fresh token with group claims token = (await client.post("/v1/users/login", json={"username": "gn_user_test", "password": "gnpass999"})).json()["access_token"] # Announce a node node_id, pk_node = await _announce_signed(client, token) # No nodes online yet r = await client.get(f"/v1/groups/{group_id}/nodes", headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 200 assert r.json()["nodes"] == [] # Simulate node connecting via WS with group_ids class FakeWS: async def send_text(self, text): pass _connected_nodes[node_id] = FakeWS() _node_groups[node_id] = [group_id] try: r = await client.get(f"/v1/groups/{group_id}/nodes", headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 200 nodes = r.json()["nodes"] assert len(nodes) == 1 assert nodes[0]["node_id"] == node_id assert nodes[0]["pk_node"] == pk_node finally: _connected_nodes.pop(node_id, None) _node_groups.pop(node_id, None) # 404 for nonexistent group r = await client.get("/v1/groups/fake-id/nodes", headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 404 # Unauthenticated → rejected r = await client.get(f"/v1/groups/{group_id}/nodes") assert r.status_code >= 400 # ── Admin authz (8.1) ─────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_non_admin_cannot_revoke(client): """Non-admin user gets 403 on admin endpoints.""" pk_ed, pk_x, _ = _gen_user_keys() await client.post("/v1/users/register", json={ "username": "regular_user", "email": "ru@x.com", "password": "regularpass", "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x}) r = await client.post("/v1/users/login", json={ "username": "regular_user", "password": "regularpass"}) token = r.json()["access_token"] set_admin_usernames(["someone_else"]) r = await client.post("/v1/admin/revoke", json={ "target": "user", "target_id": "fake-id", "reason": "test", }, headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 403 assert "Admin" in r.json()["detail"] @pytest.mark.asyncio async def test_admin_can_revoke(client): """Admin user (in config) can access admin endpoints.""" pk_ed_v, pk_x_v, _ = _gen_user_keys() r = await client.post("/v1/users/register", json={ "username": "victim_a", "email": "va@x.com", "password": "victimpass9", "pk_user_ed25519": pk_ed_v, "pk_user_x25519": pk_x_v}) victim_id = r.json()["user_id"] pk_ed_a, pk_x_a, _ = _gen_user_keys() await client.post("/v1/users/register", json={ "username": "the_admin", "email": "ta@x.com", "password": "adminpass99", "pk_user_ed25519": pk_ed_a, "pk_user_x25519": pk_x_a}) r = await client.post("/v1/users/login", json={ "username": "the_admin", "password": "adminpass99"}) admin_token = r.json()["access_token"] set_admin_usernames(["the_admin"]) r = await client.post("/v1/admin/revoke", json={ "target": "user", "target_id": victim_id, "reason": "test", }, headers={"Authorization": f"Bearer {admin_token}"}) assert r.status_code == 200 assert r.json()["status"] == "revoked" # ── Email encryption (8.2) ────────────────────────────────────────────────── @pytest.mark.asyncio async def test_email_encrypted_at_rest(client): """Email stored in DB must not contain plaintext address.""" from meshbay_hub.auth import decrypt_email, encrypt_email encrypted = encrypt_email("test@example.com") assert "@" not in encrypted assert decrypt_email(encrypted) == "test@example.com" @pytest.mark.asyncio async def test_registered_email_not_plaintext(client, app): """Registration stores encrypted email, not plaintext.""" pk_ed, pk_x, _ = _gen_user_keys() await client.post("/v1/users/register", json={ "username": "email_test", "email": "secret@example.com", "password": "emailpass9", "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x}) from meshbay_hub.db.engine import get_db from meshbay_hub.db.models import User from sqlalchemy import select async for db in get_db(): result = await db.execute(select(User).where(User.username == "email_test")) user = result.scalar_one() assert "@" not in user.email from meshbay_hub.auth import decrypt_email assert decrypt_email(user.email) == "secret@example.com" break # ── Argon2id rehash (8.10) ────────────────────────────────────────────────── @pytest.mark.asyncio async def test_password_rehash_on_login(client, app): """Users with pw_version=1 get rehashed to v2 on legacy password login.""" from meshbay_hub.db.engine import get_db from meshbay_hub.db.models import User from sqlalchemy import select pk_ed, pk_x, _ = _gen_user_keys() await client.post("/v1/users/register", json={ "username": "rehash_user", "email": "rh@x.com", "password": "rehashpass9", "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x}) # Force pw_version to 1 (simulating pre-upgrade user) async for db in get_db(): result = await db.execute(select(User).where(User.username == "rehash_user")) user = result.scalar_one() user.pw_version = 1 # Re-hash with v1 params so verify_password(version=1) succeeds import os from cryptography.hazmat.primitives.kdf.argon2 import Argon2id from meshbay_hub.auth import _ARGON2_KEY_LEN, _ARGON2_LANES, _ARGON2_VERSIONS salt = os.urandom(16) params = _ARGON2_VERSIONS[1] pw_hash = Argon2id( salt=salt, length=_ARGON2_KEY_LEN, iterations=params["iterations"], lanes=_ARGON2_LANES, memory_cost=params["memory_cost"], ).derive(b"rehashpass9") user.pw_hash = pw_hash user.pw_salt = salt await db.commit() break # Login should succeed and trigger legacy rehash (v1 -> v2) r = await client.post("/v1/users/login", json={ "username": "rehash_user", "password": "rehashpass9"}) assert r.status_code == 200 # Verify pw_version is now 2 (legacy rehash stays within password scheme) async for db in get_db(): result = await db.execute(select(User).where(User.username == "rehash_user")) user = result.scalar_one() assert user.pw_version == 2 break # Login still works after rehash r = await client.post("/v1/users/login", json={ "username": "rehash_user", "password": "rehashpass9"}) assert r.status_code == 200 @pytest.mark.asyncio async def test_a_v3_auth_key_hash_is_rewritten_at_the_current_parameters(client, db_session): """Every account registered before the 64 MiB change holds a v3 (256 MB) hash. It must keep signing in, and be rewritten on the way through — otherwise the old cost is paid at every sign-in for ever.""" import os from cryptography.hazmat.primitives.kdf.argon2 import Argon2id from meshbay_hub.auth import ( _ARGON2_KEY_LEN, _ARGON2_LANES, _ARGON2_VERSIONS, current_pw_version, ) from meshbay_hub.db.models import User from sqlalchemy import select assert current_pw_version() > 3 assert _ARGON2_VERSIONS[current_pw_version()]["memory_cost"] == 65536 key = "dGVzdGF1dGhrZXl0ZXN0YXV0aGtleXRlc3RhdXRo" r = await client.post("/v1/users/register", json={ "username": "v3holder", "email": "v3holder@test.com", "auth_key": key}) assert r.status_code == 201, r.text salt = os.urandom(16) v3 = _ARGON2_VERSIONS[3] user = (await db_session.execute( select(User).where(User.username == "v3holder"))).scalar_one() user.pw_hash = Argon2id(salt=salt, length=_ARGON2_KEY_LEN, iterations=v3["iterations"], lanes=_ARGON2_LANES, memory_cost=v3["memory_cost"]).derive(key.encode()) user.pw_salt, user.pw_version = salt, 3 await db_session.commit() r = await client.post("/v1/users/login", json={"username": "v3holder", "auth_key": key}) assert r.status_code == 200, r.text db_session.expire_all() user = (await db_session.execute( select(User).where(User.username == "v3holder"))).scalar_one() assert user.pw_version == current_pw_version() assert user.pw_hash != Argon2id(salt=salt, length=_ARGON2_KEY_LEN, iterations=3, lanes=_ARGON2_LANES, memory_cost=65536).derive(key.encode()) r = await client.post("/v1/users/login", json={"username": "v3holder", "auth_key": key}) assert r.status_code == 200, r.text # ── WebRTC signaling (9.2) ────────────────────────────────────────────────── @pytest.mark.asyncio async def test_webrtc_offer_no_node(client): """WebRTC offer to a non-connected node returns 404.""" pk_ed, pk_x, _ = _gen_user_keys() await client.post("/v1/users/register", json={ "username": "sig_user", "email": "sig@x.com", "password": "sigpass99", "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x}) r = await client.post("/v1/users/login", json={ "username": "sig_user", "password": "sigpass99"}) token = r.json()["access_token"] r = await client.post("/v1/nodes/fake-node-id/webrtc/offer", json={"sdp": "v=0\r\n...", "ice_candidates": []}, headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 404 assert "not connected" in r.json()["detail"].lower() @pytest.mark.asyncio async def test_webrtc_signaling_roundtrip(client, app): """WebRTC signaling: offer relayed to node via WS, answer returned to browser.""" import asyncio import json pk_ed, pk_x, _ = _gen_user_keys() await client.post("/v1/users/register", json={ "username": "sig_user2", "email": "sig2@x.com", "password": "sigpass99", "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x}) r = await client.post("/v1/users/login", json={ "username": "sig_user2", "password": "sigpass99"}) token = r.json()["access_token"] from meshbay_hub.api.revocation import _connected_nodes from meshbay_hub.api.signaling import handle_webrtc_answer node_id = "test-node-sig" class FakeWS: def __init__(self, answering_as): self.sent = [] # An answer is accepted only from the node the offer was relayed # to, so the stand-in has to say which node it is. self._node_id = answering_as async def send_text(self, text): self.sent.append(json.loads(text)) msg = self.sent[-1] if msg.get("type") == "webrtc_offer": await asyncio.sleep(0.01) handle_webrtc_answer({ "type": "webrtc_answer", "peer_id": msg["peer_id"], "sdp": "v=0\r\nanswer-sdp", "ice_candidates": [{"candidate": "test"}], }, self._node_id) fake_ws = FakeWS(node_id) _connected_nodes[node_id] = fake_ws # The caller must share an active group with the node, which is what this # test used to get away without: a node registered for no group skipped the # membership check entirely, so this exercised the relay through the hole # rather than through the door. Registering the group is what a real node # does on its socket. from meshbay_hub.api.revocation import _node_groups r = await client.post("/v1/groups", json={"name": "sig-group"}, headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 201, r.text _node_groups[node_id] = [r.json()["group_id"]] try: r = await client.post(f"/v1/nodes/{node_id}/webrtc/offer", json={"sdp": "v=0\r\noffer-sdp", "ice_candidates": []}, headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 200 data = r.json() assert "answer-sdp" in data["sdp"] assert len(data["ice_candidates"]) == 1 assert "peer_id" in data finally: _connected_nodes.pop(node_id, None) _node_groups.pop(node_id, None) # ── IP log cleanup (8.9) ──────────────────────────────────────────────────── @pytest.mark.asyncio async def test_ip_log_cleanup(app): """Old IP log entries are purged by cleanup task.""" from datetime import datetime, timedelta from meshbay_hub.db.engine import get_db from meshbay_hub.db.models import IPLog from meshbay_hub.tasks.cleanup import purge_old_ip_logs async for db in get_db(): old_ts = datetime.now(UTC) - timedelta(days=400) db.add(IPLog(event="test_old", ip_address="1.2.3.4", timestamp=old_ts)) db.add(IPLog(event="test_recent", ip_address="5.6.7.8")) await db.commit() deleted = await purge_old_ip_logs(db, retention_days=365) assert deleted == 1 from sqlalchemy import func, select count = (await db.execute( select(func.count()).where(IPLog.event.in_(["test_old", "test_recent"])) )).scalar_one() assert count == 1 break # ── Webapp HTML shell (9.13) ───────────────────────────────────────────────── @pytest.mark.asyncio async def test_webapp_html_includes_scripts(client): """SPA HTML shell includes all required script tags. The paths carry a build fingerprint (`/a//app.js`) so that a browser cannot serve an older build out of its own cache — see test_asset_versioning.py. What this test still owns is that every piece is referenced at all, and in an order where each one's dependencies are already loaded. """ from meshbay_hub.api.webapp import ASSET_V r = await client.get("/") assert r.status_code == 200 html = r.text assert "" in html assert '
' in html for asset in ("keyderive.js", "crypto.js", "transport.js", "app.js"): assert f'src="/a/{ASSET_V}/{asset}"' in html, f"{asset} is not loaded" assert 'type="module"' in html assert 'rel="stylesheet"' in html assert f'href="/a/{ASSET_V}/style.css"' in html # ── Password split (T1 fix) ───────────────────────────────────────────────── @pytest.mark.asyncio async def test_register_with_auth_key(client, app): """Registration with auth_key sets the current auth_key pw_version.""" from meshbay_hub.auth import current_pw_version from meshbay_hub.db.engine import get_db from meshbay_hub.db.models import User from sqlalchemy import select pk_ed, pk_x, _ = _gen_user_keys() r = await client.post("/v1/users/register", json={ "username": "authuser", "email": "auth@test.com", "auth_key": "dGVzdGF1dGhrZXl0ZXN0YXV0aGtleXRlc3RhdXRo", "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x, }) assert r.status_code == 201 assert "user_id" in r.json() async for db in get_db(): result = await db.execute(select(User).where(User.username == "authuser")) user = result.scalar_one() assert user.pw_version == current_pw_version() break @pytest.mark.asyncio async def test_register_with_password_sets_v2(client, app): """Registration with raw password (legacy) sets pw_version 2.""" from meshbay_hub.db.engine import get_db from meshbay_hub.db.models import User from sqlalchemy import select pk_ed, pk_x, _ = _gen_user_keys() r = await client.post("/v1/users/register", json={ "username": "legacyreg", "email": "legacy@test.com", "password": "legacypass99", "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x, }) assert r.status_code == 201 async for db in get_db(): result = await db.execute(select(User).where(User.username == "legacyreg")) user = result.scalar_one() assert user.pw_version == 2 break @pytest.mark.asyncio async def test_register_no_credentials_rejected(client): """Registration without auth_key or password returns 400.""" pk_ed, pk_x, _ = _gen_user_keys() r = await client.post("/v1/users/register", json={ "username": "nocred_test", "email": "nocred@test.com", "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x, }) assert r.status_code == 400 assert "auth_key or password required" in r.json()["detail"] @pytest.mark.asyncio async def test_login_with_auth_key(client, app): """Login with auth_key for pw_version 3 account succeeds.""" pk_ed, pk_x, _ = _gen_user_keys() auth_key = "dGVzdGF1dGhrZXl0ZXN0YXV0aGtleXRlc3RhdXRo" await client.post("/v1/users/register", json={ "username": "authlogin", "email": "authlogin@test.com", "auth_key": auth_key, "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x, }) r = await client.post("/v1/users/login", json={ "username": "authlogin", "auth_key": auth_key}) assert r.status_code == 200 data = r.json() assert "access_token" in data assert "refresh_token" in data @pytest.mark.asyncio async def test_login_auth_key_wrong_rejected(client): """Login with wrong auth_key returns 401.""" pk_ed, pk_x, _ = _gen_user_keys() await client.post("/v1/users/register", json={ "username": "authwrong", "email": "authwrong@test.com", "auth_key": "dGVzdGF1dGhrZXl0ZXN0YXV0aGtleXRlc3RhdXRo", "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x, }) r = await client.post("/v1/users/login", json={ "username": "authwrong", "auth_key": "d3JvbmdrZXl3cm9uZ2tleXdyb25na2V5d3Jvbmc="}) assert r.status_code == 401 @pytest.mark.asyncio async def test_login_v3_account_password_only_rejected(client): """Login with raw password to a v3 (auth_key) account returns 401.""" pk_ed, pk_x, _ = _gen_user_keys() await client.post("/v1/users/register", json={ "username": "v3nopw_test", "email": "v3nopw@test.com", "auth_key": "dGVzdGF1dGhrZXl0ZXN0YXV0aGtleXRlc3RhdXRo", "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x, }) r = await client.post("/v1/users/login", json={ "username": "v3nopw_test", "password": "somepassword"}) assert r.status_code == 401 @pytest.mark.asyncio async def test_login_legacy_upgrade_required(client): """Legacy account (pw_version 2) with auth_key only returns auth_upgrade_required.""" pk_ed, pk_x, _ = _gen_user_keys() await client.post("/v1/users/register", json={ "username": "legacyupg", "email": "legacyupg@test.com", "password": "legacypass99", "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x, }) r = await client.post("/v1/users/login", json={ "username": "legacyupg", "auth_key": "dGVzdGF1dGhrZXl0ZXN0YXV0aGtleXRlc3RhdXRo"}) assert r.status_code == 401 assert r.json()["detail"] == "auth_upgrade_required" @pytest.mark.asyncio async def test_login_legacy_migration(client, app): """Legacy account migrates to auth_key on login with both fields.""" from meshbay_hub.auth import current_pw_version from meshbay_hub.db.engine import get_db from meshbay_hub.db.models import User from sqlalchemy import select pk_ed, pk_x, _ = _gen_user_keys() await client.post("/v1/users/register", json={ "username": "migrateuser", "email": "migrate@test.com", "password": "migratepass9", "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x, }) # Verify starts at pw_version 2 async for db in get_db(): result = await db.execute(select(User).where(User.username == "migrateuser")) user = result.scalar_one() assert user.pw_version == 2 break auth_key = "bWlncmF0ZWF1dGhrZXltaWdyYXRlYXV0aGtleW1p" # Login with password + auth_key → should succeed and migrate r = await client.post("/v1/users/login", json={ "username": "migrateuser", "password": "migratepass9", "auth_key": auth_key, }) assert r.status_code == 200 # Verify pw_version is now the current auth_key version (migrated) async for db in get_db(): result = await db.execute(select(User).where(User.username == "migrateuser")) user = result.scalar_one() assert user.pw_version == current_pw_version() break # Login again with auth_key only → should succeed (migrated account) r = await client.post("/v1/users/login", json={ "username": "migrateuser", "auth_key": auth_key}) assert r.status_code == 200 assert "access_token" in r.json() # Old password no longer works (hash was replaced with auth_key hash) r = await client.post("/v1/users/login", json={ "username": "migrateuser", "password": "migratepass9"}) assert r.status_code == 401 @pytest.mark.asyncio async def test_login_no_credentials_rejected(client): """Login without auth_key or password returns 401.""" r = await client.post("/v1/users/login", json={"username": "nobody_test"}) assert r.status_code == 401 assert "No credentials" in r.json()["detail"]