diff options
Diffstat (limited to 'packages/meshbay-hub/tests')
| -rw-r--r-- | packages/meshbay-hub/tests/conftest.py | 8 | ||||
| -rw-r--r-- | packages/meshbay-hub/tests/test_hub_api.py | 197 | ||||
| -rw-r--r-- | packages/meshbay-hub/tests/test_moderation.py | 2 | ||||
| -rw-r--r-- | packages/meshbay-hub/tests/test_revocation.py | 4 |
4 files changed, 211 insertions, 0 deletions
diff --git a/packages/meshbay-hub/tests/conftest.py b/packages/meshbay-hub/tests/conftest.py index b42aba1..ffd6c2e 100644 --- a/packages/meshbay-hub/tests/conftest.py +++ b/packages/meshbay-hub/tests/conftest.py @@ -48,10 +48,18 @@ async def app(hub_config): from meshbay_hub.app import create_app application = create_app(hub_config) + # Disable rate limiting in tests + from meshbay_hub.api.middleware import limiter + limiter.enabled = False + # Run lifespan startup manually async with application.router.lifespan_context(application): yield application + # Reset admin usernames after each test + from meshbay_hub.api.deps import set_admin_usernames + set_admin_usernames([]) + @pytest_asyncio.fixture async def client(app): diff --git a/packages/meshbay-hub/tests/test_hub_api.py b/packages/meshbay-hub/tests/test_hub_api.py index 568d5d9..327158f 100644 --- a/packages/meshbay-hub/tests/test_hub_api.py +++ b/packages/meshbay-hub/tests/test_hub_api.py @@ -11,6 +11,7 @@ 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(): @@ -35,6 +36,16 @@ async def test_hub_info(client): @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 @@ -123,6 +134,33 @@ async def test_token_refresh(client): 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 @@ -311,3 +349,162 @@ async def test_jwt_contains_groups_claim(client): "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 + + +# ── 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 diff --git a/packages/meshbay-hub/tests/test_moderation.py b/packages/meshbay-hub/tests/test_moderation.py index 68e08c9..93d23cd 100644 --- a/packages/meshbay-hub/tests/test_moderation.py +++ b/packages/meshbay-hub/tests/test_moderation.py @@ -4,6 +4,7 @@ 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 FAKE_HASH = "a" * 64 # valid blake3 hex @@ -20,6 +21,7 @@ async def auth_headers(client): }) r = await client.post("/v1/users/login", json={"username": "mod_admin", "password": "modpass99"}) + set_admin_usernames(["mod_admin"]) return {"Authorization": f"Bearer {r.json()['access_token']}"} diff --git a/packages/meshbay-hub/tests/test_revocation.py b/packages/meshbay-hub/tests/test_revocation.py index f5d6050..494d77d 100644 --- a/packages/meshbay-hub/tests/test_revocation.py +++ b/packages/meshbay-hub/tests/test_revocation.py @@ -7,6 +7,7 @@ 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 async def _register_and_login(client, username, pk_ed, pk_x): @@ -45,6 +46,7 @@ async def test_revoke_user_marks_db(client): pk_to_b64(sk_admin_ed.public_key()), pk_to_b64(sk_admin_x.public_key()), ) + set_admin_usernames(["admin1"]) r = await client.post("/v1/admin/revoke", json={ "target": "user", "target_id": victim_id, "reason": "spam", @@ -80,6 +82,7 @@ async def test_revocation_token_verifiable_offline(client): pk_to_b64(sk_admin_ed.public_key()), pk_to_b64(sk_admin_x.public_key()), ) + set_admin_usernames(["admin2"]) r = await client.post("/v1/admin/revoke", json={ "target": "user", "target_id": victim_id, "reason": "test", @@ -107,6 +110,7 @@ async def test_revoke_group(client): pk_to_b64(sk_ed.public_key()), pk_to_b64(sk_x.public_key()), ) + set_admin_usernames(["admin3"]) hdrs = {"Authorization": f"Bearer {admin_token}"} r = await client.post("/v1/groups", json={"name": "grp-to-revoke"}, headers=hdrs) |