1 files changed, 18 insertions, 4 deletions
diff --git a/packages/meshbay-hub/tests/test_token_hardening.py b/packages/meshbay-hub/tests/test_token_hardening.py
index 0835180..f381f69 100644
--- a/packages/meshbay-hub/tests/test_token_hardening.py
+++ b/packages/meshbay-hub/tests/test_token_hardening.py
@@ -17,6 +17,7 @@ import time
 
 import pytest
 
+from meshbay_common.tokens import HUB_API_AUD
 from meshbay_hub import auth
 
 
@@ -33,7 +34,7 @@ async def test_token_without_exp_is_refused(client):
     import jwt
     assert _sk_pem_loaded()
     # A hub-signed token with a valid scope and sub but NO exp.
-    forged = jwt.encode({"sub": "u", "scope": "user"},
+    forged = jwt.encode({"sub": "u", "scope": "user", "aud": HUB_API_AUD},
                         auth.hub_private_key_pem(), algorithm="EdDSA")
     with pytest.raises(Exception):
         auth.decode_access_token(forged)
@@ -42,7 +43,8 @@ async def test_token_without_exp_is_refused(client):
 @pytest.mark.asyncio
 async def test_expired_token_is_refused(client):
     import jwt
-    forged = jwt.encode({"sub": "u", "scope": "user", "exp": int(time.time()) - 100},
+    forged = jwt.encode({"sub": "u", "scope": "user", "aud": HUB_API_AUD,
+                         "exp": int(time.time()) - 100},
                         auth.hub_private_key_pem(), algorithm="EdDSA")
     with pytest.raises(jwt.ExpiredSignatureError):
         auth.decode_access_token(forged)
@@ -51,7 +53,7 @@ async def test_expired_token_is_refused(client):
 @pytest.mark.asyncio
 async def test_token_without_scope_is_refused(client):
     import jwt
-    forged = jwt.encode({"sub": "u", "exp": int(time.time()) + 3600},
+    forged = jwt.encode({"sub": "u", "exp": int(time.time()) + 3600, "aud": HUB_API_AUD},
                         auth.hub_private_key_pem(), algorithm="EdDSA")
     with pytest.raises(Exception):
         auth.decode_access_token(forged)
@@ -60,13 +62,25 @@ async def test_token_without_scope_is_refused(client):
 @pytest.mark.asyncio
 async def test_unknown_scope_is_refused(client):
     import jwt
-    forged = jwt.encode({"sub": "u", "scope": "root", "exp": int(time.time()) + 3600},
+    forged = jwt.encode({"sub": "u", "scope": "root", "aud": HUB_API_AUD,
+                         "exp": int(time.time()) + 3600},
                         auth.hub_private_key_pem(), algorithm="EdDSA")
     with pytest.raises(jwt.InvalidTokenError):
         auth.decode_access_token(forged)
 
 
 @pytest.mark.asyncio
+async def test_an_mnp_token_is_refused_at_the_hub_api(client):
+    """The MNP token (aud=MNP_AUD) authorises a member to a node; it must not be
+    a session at the hub. A node operator holds one, and this is what stops them
+    replaying it against the hub API."""
+    from meshbay_hub.auth import issue_mnp_token
+    mnp = issue_mnp_token("some-user", groups=[])
+    with pytest.raises(Exception):
+        auth.decode_access_token(mnp)
+
+
+@pytest.mark.asyncio
 async def test_a_revocation_token_is_not_a_session(client):
     """The concrete cross-type case: revocation tokens are hub-signed, carry no
     exp/sub/scope, and are handed to admins and pushed to every node."""