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.
"""

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/<hash>/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 "<!DOCTYPE html>" in html
    assert '<div id="app">' 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"]