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

"""
Integration tests for the Hub API.
Uses SQLite in-memory + httpx.AsyncClient — no PostgreSQL, no network.
"""

import base64
import pytest
import pytest_asyncio
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
from cryptography.hazmat.primitives import serialization

from meshbay_common.crypto import generate_gek, pk_to_b64, wrap_gek
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,
    )


# ── 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", "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", "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", "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", "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", "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", "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", "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 decoded["pk_user"] == pk_ed
    assert "jti" in decoded   # mandatory


@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", "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", "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):
    pk_ed, pk_x, _ = _gen_user_keys()
    await client.post("/v1/users/register", json={
        "username": "frank", "email": "frank@example.com",
        "password": "frankpass99",
        "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x})
    login = await client.post("/v1/users/login", json={
        "username": "frank", "password": "frankpass99"})
    token = login.json()["access_token"]

    r = await client.get("/v1/users/frank/pubkeys",
                         headers={"Authorization": f"Bearer {token}"})
    assert r.status_code == 200
    assert r.json()["pk_ed25519"] == pk_ed
    assert r.json()["pk_x25519"]  == pk_x


# ── 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", "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", "password": "nodepass99"})
    token = login.json()["access_token"]
    hdrs = {"Authorization": f"Bearer {token}"}

    r = await client.post("/v1/nodes/announce",
                          json={"pk_node": pk_ed, "endpoint_hint": "1.2.3.4:19000"},
                          headers=hdrs)
    assert r.status_code == 201
    node_id = r.json()["node_id"]

    r2 = await client.get(f"/v1/nodes/{node_id}", headers=hdrs)
    assert r2.status_code == 200
    assert r2.json()["pk_node"] == pk_ed
    assert r2.json()["endpoint_hint"] == "1.2.3.4:19000"


# ── Groups + GEK bundles ──────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_group_gek_roundtrip(client):
    """Admin creates group, wraps GEK for member, member retrieves and can unwrap."""
    import jwt as pyjwt
    from meshbay_common.crypto import unwrap_gek

    # Register admin (alice2) and member (bob2)
    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", "a2@x.com", "alicepass99", pk_ed_a, pk_x_a),
        ("bob2",   "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})

    def _token(uname, pwd):
        async def _inner():
            r = await client.post("/v1/users/login",
                                  json={"username": uname, "password": pwd})
            return r.json()["access_token"]
        return _inner

    alice_token = (await client.post("/v1/users/login",
                   json={"username": "alice2", "password": "alicepass99"})).json()["access_token"]
    bob_token   = (await client.post("/v1/users/login",
                   json={"username": "bob2",   "password": "bobpass99"})).json()["access_token"]

    a_hdrs = {"Authorization": f"Bearer {alice_token}"}
    b_hdrs = {"Authorization": f"Bearer {bob_token}"}

    # Alice creates group
    r = await client.post("/v1/groups", json={"name": "mygroup"}, headers=a_hdrs)
    assert r.status_code == 201
    group_id = r.json()["group_id"]

    # Alice generates GEK and wraps it for bob
    gek = generate_gek()
    pk_bob_raw = base64.b64decode(pk_x_b)
    bundle = wrap_gek(gek, pk_bob_raw)

    r = await client.post(f"/v1/groups/{group_id}/members/bob2/gek",
                          json=bundle, headers=a_hdrs)
    assert r.status_code == 201

    # Bob retrieves his bundle
    r = await client.get(f"/v1/groups/{group_id}/gek", headers=b_hdrs)
    assert r.status_code == 200
    retrieved = r.json()

    # Bob unwraps — must recover original GEK
    sk_b_raw = sk_x_b.private_bytes(
        serialization.Encoding.Raw, serialization.PrivateFormat.Raw,
        serialization.NoEncryption())
    recovered = unwrap_gek(retrieved, sk_b_raw, pk_bob_raw)
    assert recovered == gek


@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", "c@x.com", "charliepass", pk_ed_a, pk_x_a),
        ("dan",     "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", "password": "charliepass"})).json()["access_token"]
    dan_token     = (await client.post("/v1/users/login",
                     json={"username": "dan", "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"]

    gek = generate_gek()
    bundle = wrap_gek(gek, base64.b64decode(pk_x_b))

    # Dan (non-admin) tries to add a member → 403
    r = await client.post(f"/v1/groups/{group_id}/members/charlie/gek",
                          json=bundle,
                          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", "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", "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"]

    gek = generate_gek()
    bundle = wrap_gek(gek, base64.b64decode(pk_x_b))
    await client.post(f"/v1/groups/{group_id}/members/grp_bob/gek",
                      json=bundle,
                      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", "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"]


# ── 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 encrypt_email, decrypt_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 current version on login."""
    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": "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
        from meshbay_hub.auth import _ARGON2_VERSIONS, _ARGON2_KEY_LEN, _ARGON2_LANES
        from cryptography.hazmat.primitives.kdf.argon2 import Argon2id
        import os
        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 rehash
    r = await client.post("/v1/users/login", json={
        "username": "rehash_user", "password": "rehashpass9"})
    assert r.status_code == 200

    # Verify pw_version is now current
    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 == current_pw_version()
        break

    # Login still works after rehash
    r = await client.post("/v1/users/login", json={
        "username": "rehash_user", "password": "rehashpass9"})
    assert r.status_code == 200


# ── 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

    class FakeWS:
        def __init__(self):
            self.sent = []

        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"}],
                })

    fake_ws = FakeWS()
    node_id = "test-node-sig"
    _connected_nodes[node_id] = fake_ws

    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)


# ── 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, timezone, 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(timezone.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 select, func
        count = (await db.execute(
            select(func.count()).where(IPLog.event.in_(["test_old", "test_recent"]))
        )).scalar_one()
        assert count == 1
        break