1 files changed, 119 insertions, 0 deletions
diff --git a/packages/meshbay-hub/tests/test_revocation.py b/packages/meshbay-hub/tests/test_revocation.py
new file mode 100644
index 0000000..f5d6050
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_revocation.py
@@ -0,0 +1,119 @@
+"""Tests for revocation — admin endpoint + token signing."""
+
+import time
+import pytest
+import jwt
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
+
+from meshbay_common.crypto import pk_to_b64
+
+
+async def _register_and_login(client, username, pk_ed, pk_x):
+    await client.post("/v1/users/register", json={
+        "username": username, "email": f"{username}@test.com",
+        "password": "testpass99",
+        "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x,
+    })
+    r = await client.post("/v1/users/login",
+                          json={"username": username, "password": "testpass99"})
+    return r.json()["access_token"], r.json()["user_id"] if "user_id" in r.json() else None
+
+
+@pytest.mark.asyncio
+async def test_revoke_user_marks_db(client):
+    """POST /v1/admin/revoke marks user as revoked in DB."""
+    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+    from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
+
+    sk_ed = Ed25519PrivateKey.generate()
+    sk_x  = X25519PrivateKey.generate()
+    pk_ed = pk_to_b64(sk_ed.public_key())
+    pk_x  = pk_to_b64(sk_x.public_key())
+
+    r = await client.post("/v1/users/register", json={
+        "username": "vic1", "email": "v@t.com", "password": "vicpass99",
+        "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x,
+    })
+    victim_id = r.json()["user_id"]
+
+    # Admin (any registered user for now — production would add role check)
+    sk_admin_ed = Ed25519PrivateKey.generate()
+    sk_admin_x  = X25519PrivateKey.generate()
+    admin_token, _ = await _register_and_login(
+        client, "admin1",
+        pk_to_b64(sk_admin_ed.public_key()),
+        pk_to_b64(sk_admin_x.public_key()),
+    )
+
+    r = await client.post("/v1/admin/revoke", json={
+        "target": "user", "target_id": victim_id, "reason": "spam",
+    }, headers={"Authorization": f"Bearer {admin_token}"})
+    assert r.status_code == 200
+    data = r.json()
+    assert data["status"] == "revoked"
+    assert "token" in data
+
+    # Victim can no longer login
+    r = await client.post("/v1/users/login", json={
+        "username": "vic1", "password": "vicpass99"})
+    assert r.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_revocation_token_verifiable_offline(client):
+    """Revocation token is a valid JWT signed by hub Ed25519 key."""
+    sk_ed = Ed25519PrivateKey.generate()
+    sk_x  = X25519PrivateKey.generate()
+
+    r = await client.post("/v1/users/register", json={
+        "username": "vic2", "email": "v2@t.com", "password": "vicpass99",
+        "pk_user_ed25519": pk_to_b64(sk_ed.public_key()),
+        "pk_user_x25519":  pk_to_b64(sk_x.public_key()),
+    })
+    victim_id = r.json()["user_id"]
+
+    sk_admin_ed = Ed25519PrivateKey.generate()
+    sk_admin_x  = X25519PrivateKey.generate()
+    admin_token, _ = await _register_and_login(
+        client, "admin2",
+        pk_to_b64(sk_admin_ed.public_key()),
+        pk_to_b64(sk_admin_x.public_key()),
+    )
+
+    r = await client.post("/v1/admin/revoke", json={
+        "target": "user", "target_id": victim_id, "reason": "test",
+    }, headers={"Authorization": f"Bearer {admin_token}"})
+    rev_token = r.json()["token"]
+
+    # Verify offline with hub's public key
+    r_pk = await client.get("/v1/hub/pubkey")
+    hub_pk_pem = r_pk.json()["pk_hub_pem"].encode()
+
+    decoded = jwt.decode(rev_token, hub_pk_pem, algorithms=["EdDSA"],
+                         options={"verify_exp": False})
+    assert decoded["type"] == "revocation"
+    assert decoded["target"] == "user"
+    assert decoded["target_id"] == victim_id
+    assert "jti" in decoded
+
+
+@pytest.mark.asyncio
+async def test_revoke_group(client):
+    sk_ed = Ed25519PrivateKey.generate()
+    sk_x  = X25519PrivateKey.generate()
+    admin_token, _ = await _register_and_login(
+        client, "admin3",
+        pk_to_b64(sk_ed.public_key()),
+        pk_to_b64(sk_x.public_key()),
+    )
+    hdrs = {"Authorization": f"Bearer {admin_token}"}
+
+    r = await client.post("/v1/groups", json={"name": "grp-to-revoke"}, headers=hdrs)
+    group_id = r.json()["group_id"]
+
+    r = await client.post("/v1/admin/revoke", json={
+        "target": "group", "target_id": group_id, "reason": "tos_violation",
+    }, headers=hdrs)
+    assert r.status_code == 200
+    assert r.json()["status"] == "revoked"