summaryrefslogtreecommitdiffstats
path: root/packages/meshbay-hub/tests/test_node_ws_auth.py
diff options
context:
space:
mode:
Diffstat (limited to 'packages/meshbay-hub/tests/test_node_ws_auth.py')
-rw-r--r--packages/meshbay-hub/tests/test_node_ws_auth.py38
1 files changed, 19 insertions, 19 deletions
diff --git a/packages/meshbay-hub/tests/test_node_ws_auth.py b/packages/meshbay-hub/tests/test_node_ws_auth.py
index 4ead0d7..f3ac3a2 100644
--- a/packages/meshbay-hub/tests/test_node_ws_auth.py
+++ b/packages/meshbay-hub/tests/test_node_ws_auth.py
@@ -72,7 +72,7 @@ async def test_ws_rejects_user_scoped_token(client):
"""C2: a browser token must never be able to register as a node."""
from meshbay_hub.api.revocation import _authorize_node_ws
- victim = await _make_user(client, "victim1")
+ victim = await _make_user(client, "victim1_test")
node_id = await _announce_node(client, victim)
resolved, detail = await _authorize_node_ws(victim["token"], node_id, None)
@@ -88,7 +88,7 @@ async def test_ws_rejects_foreign_node_id(client):
"""
from meshbay_hub.api.revocation import _authorize_node_ws
- victim = await _make_user(client, "victim2")
+ victim = await _make_user(client, "victim2_test")
attacker = await _make_user(client, "attacker2")
victim_node = await _announce_node(client, victim)
await _announce_node(client, attacker)
@@ -104,7 +104,7 @@ async def test_ws_rejects_unknown_node_id(client):
"""C2: an invented node_id must not register either."""
from meshbay_hub.api.revocation import _authorize_node_ws
- user = await _make_user(client, "user3")
+ user = await _make_user(client, "user3_test")
resolved, _ = await _authorize_node_ws(_node_token(user), "no-such-node", None)
assert resolved is None
@@ -114,7 +114,7 @@ async def test_ws_rejects_missing_node_id(client):
"""C2: identity may not fall back to the token subject."""
from meshbay_hub.api.revocation import _authorize_node_ws
- user = await _make_user(client, "user4")
+ user = await _make_user(client, "user4_test")
resolved, _ = await _authorize_node_ws(_node_token(user), "", None)
assert resolved is None
@@ -124,7 +124,7 @@ async def test_ws_accepts_own_node(client):
"""The legitimate path still works."""
from meshbay_hub.api.revocation import _authorize_node_ws
- user = await _make_user(client, "owner5")
+ user = await _make_user(client, "owner5_test")
node_id = await _announce_node(client, user)
resolved, groups = await _authorize_node_ws(_node_token(user), node_id, None)
@@ -140,7 +140,7 @@ async def test_ws_group_claims_cannot_widen_beyond_membership(client):
"""
from meshbay_hub.api.revocation import _authorize_node_ws
- user = await _make_user(client, "owner6")
+ user = await _make_user(client, "owner6_test")
node_id = await _announce_node(client, user)
r = await client.post(
@@ -168,7 +168,7 @@ async def test_signaling_rejects_non_member(client):
"""
from meshbay_hub.api import revocation as rev
- owner = await _make_user(client, "owner8")
+ owner = await _make_user(client, "owner8_test")
outsider = await _make_user(client, "outsider8")
node_id = await _announce_node(client, owner)
@@ -201,7 +201,7 @@ async def test_signaling_rejects_non_member(client):
@pytest.mark.asyncio
async def test_signaling_rejects_oversized_sdp(client):
"""H6: an SDP offer is ~2 KB; unbounded input is a memory amplifier."""
- user = await _make_user(client, "user9")
+ user = await _make_user(client, "user9_test")
resp = await client.post(
"/v1/nodes/whatever/webrtc/offer",
json={"sdp": "v=0" + ("x" * 200_000), "ice_candidates": []},
@@ -219,7 +219,7 @@ async def test_incoming_rejects_foreign_peer_ip(client):
"""
from meshbay_hub.api import revocation as rev
- owner = await _make_user(client, "owner10")
+ owner = await _make_user(client, "owner10_test")
node_id = await _announce_node(client, owner)
class _FakeWS:
@@ -243,7 +243,7 @@ async def test_ws_node_may_narrow_its_group_set(client):
"""A node hosting a subset of the operator's groups may say so."""
from meshbay_hub.api.revocation import _authorize_node_ws
- user = await _make_user(client, "owner7")
+ user = await _make_user(client, "owner7_test")
node_id = await _announce_node(client, user)
created = []
@@ -282,7 +282,7 @@ async def test_announce_requires_proof_of_possession(client):
the private key, so a user could announce a record carrying someone else's
node key.
"""
- user = await _make_user(client, "ann1")
+ user = await _make_user(client, "ann1_test")
r = await client.post(
"/v1/nodes/announce",
json={"pk_node": user["pk_ed"], "endpoint_hint": "test"},
@@ -294,7 +294,7 @@ async def test_announce_requires_proof_of_possession(client):
@pytest.mark.asyncio
async def test_announce_rejects_foreign_key(client):
"""M8: announcing someone else's public key must fail — no matching private key."""
- user = await _make_user(client, "ann2")
+ user = await _make_user(client, "ann2_test")
victim_sk = Ed25519PrivateKey.generate()
victim_pk = pk_to_b64(victim_sk.public_key())
@@ -312,7 +312,7 @@ async def test_announce_rejects_foreign_key(client):
async def test_announce_rejects_stale_timestamp(client):
"""M8: a captured announce must not be replayable later."""
import time as _t
- user = await _make_user(client, "ann3")
+ user = await _make_user(client, "ann3_test")
sk = Ed25519PrivateKey.generate()
payload = _announce_payload(
user["user_id"], sk, pk_to_b64(sk.public_key()), ts=int(_t.time()) - 3600)
@@ -327,7 +327,7 @@ async def test_announce_rejects_stale_timestamp(client):
@pytest.mark.asyncio
async def test_announce_with_valid_proof_succeeds_and_is_idempotent(client):
"""The legitimate path works, and re-announcing updates rather than piling up rows."""
- user = await _make_user(client, "ann4")
+ user = await _make_user(client, "ann4_test")
sk = Ed25519PrivateKey.generate()
pk_b64 = pk_to_b64(sk.public_key())
@@ -362,7 +362,7 @@ async def test_ws_absent_claim_registers_no_groups(client):
"""A node that declares nothing hosts nothing — it must not inherit the set."""
from meshbay_hub.api.revocation import _authorize_node_ws
- user = await _make_user(client, "empty1")
+ user = await _make_user(client, "empty1_test")
node_id = await _announce_node(client, user)
for name in ("has-one", "has-two"):
r = await client.post(
@@ -383,7 +383,7 @@ async def test_ws_explicit_empty_claim_registers_no_groups(client):
"""And the same when the node says so out loud, which it now does."""
from meshbay_hub.api.revocation import _authorize_node_ws
- user = await _make_user(client, "empty2")
+ user = await _make_user(client, "empty2_test")
node_id = await _announce_node(client, user)
r = await client.post(
"/v1/groups",
@@ -407,8 +407,8 @@ async def test_empty_node_cannot_shadow_another_members_group(client):
from meshbay_hub.api.revocation import (
_authorize_node_ws, _node_groups, get_online_nodes_for_group)
- host = await _make_user(client, "hoster")
- guest = await _make_user(client, "guest")
+ host = await _make_user(client, "hoster_test")
+ guest = await _make_user(client, "guest_test")
host_node = await _announce_node(client, host)
guest_node = await _announce_node(client, guest)
@@ -421,7 +421,7 @@ async def test_empty_node_cannot_shadow_another_members_group(client):
group_id = r.json()["group_id"]
r = await client.post(
- f"/v1/groups/{group_id}/members/{'guest'}",
+ f"/v1/groups/{group_id}/members/{'guest_test'}",
headers={"Authorization": f"Bearer {host['token']}"},
)
assert r.status_code == 201, r.text