1 files changed, 78 insertions, 1 deletions
diff --git a/packages/meshbay-hub/tests/test_availability_between_members.py b/packages/meshbay-hub/tests/test_availability_between_members.py
index bfbfcc0..d1a4dcb 100644
--- a/packages/meshbay-hub/tests/test_availability_between_members.py
+++ b/packages/meshbay-hub/tests/test_availability_between_members.py
@@ -338,6 +338,9 @@ async def test_a_relay_must_prove_it_holds_the_approved_key(client, monkeypatch)
     """
     from meshbay_hub.api import relay as relay_mod
 
+    # The registry ships closed (`relay.RELAYS_ENABLED`); the proof it demands
+    # is still what will be wanted the day it opens.
+    monkeypatch.setattr(relay_mod, "RELAYS_ENABLED", True)
     sk = Ed25519PrivateKey.generate()
     pk = pk_to_b64(sk.public_key())
     relay_mod._relays["r1"] = {"pk": pk, "active": False}
@@ -370,10 +373,84 @@ async def test_a_relay_must_prove_it_holds_the_approved_key(client, monkeypatch)
 
 
 @pytest.mark.asyncio
-async def test_a_captured_relay_registration_is_not_replayable(client):
+async def test_every_relay_route_is_closed_as_the_hub_ships(client):
+    """Nothing in the tree uses the registry, and two of its routes take no account.
+
+    A dependency on the router, so a route added later is closed too. The flag
+    is read as shipped, not set here — a test that closes the gate itself
+    would keep passing the day somebody opens it.
+    """
+    admin = await _make_user(client, "relayadmin")
+    from meshbay_hub.api.deps import set_admin_usernames
+    set_admin_usernames(["relayadmin"])
+    auth = {"Authorization": f"Bearer {admin['token']}"}
+
+    for method, path in (("get", "/v1/relays"),
+                         ("post", "/v1/relays/register"),
+                         ("post", "/v1/relays/approve")):
+        kwargs = {"headers": auth} if method == "get" else {"json": {}, "headers": auth}
+        r = await getattr(client, method)(path, **kwargs)
+        assert r.status_code == 503, (path, r.status_code, r.text)
+
+
+@pytest.mark.asyncio
+async def test_a_stranger_who_locks_your_name_does_not_sign_you_out(client, db_session):
+    """AV26. The sign-in lockout is keyed by username, and usernames are public.
+
+    So anyone can spend your attempts, and the design has to make that cost as
+    little as possible: a lockout refuses *passphrase* sign-in and nothing else.
+    The session you already have keeps working and keeps renewing, and a reset
+    code sent to your own address ends the lockout at once.
+    """
+    victim_key = base64.b64encode(b"k" * 32).decode()
+    r = await client.post("/v1/users/register", json={
+        "username": "victim26", "email": "victim26@example.test", "auth_key": victim_key})
+    assert r.status_code == 201, r.text
+    session = (await client.post("/v1/users/login", json={
+        "username": "victim26", "auth_key": victim_key})).json()
+
+    # The stranger needs no account at all — only the name.
+    for _ in range(4):
+        r = await client.post("/v1/users/login", json={
+            "username": "victim26", "auth_key": "guess" + "x" * 39})
+        assert r.status_code == 401
+    r = await client.post("/v1/users/login", json={
+        "username": "victim26", "auth_key": victim_key})
+    assert r.status_code == 429, r.text
+
+    # Still signed in, and still able to stay signed in.
+    auth = {"Authorization": f"Bearer {session['access_token']}"}
+    assert (await client.get("/v1/users/me", headers=auth)).status_code == 200
+    r = await client.post("/v1/users/token/refresh",
+                          json={"refresh_token": session["refresh_token"]})
+    assert r.status_code == 200, r.text
+
+    # The way out that needs nobody's help: a code to the address on file.
+    from meshbay_hub.db.models import EmailVerification, User
+    from sqlalchemy import select
+    r = await client.post("/v1/users/password/reset-request", json={
+        "username": "victim26", "email": "victim26@example.test"})
+    assert r.status_code == 200, r.text
+    uid = (await db_session.execute(
+        select(User.id).where(User.username == "victim26"))).scalar_one()
+    code = (await db_session.execute(select(EmailVerification.code).where(
+        EmailVerification.user_id == uid,
+        EmailVerification.purpose == "password_reset"))).scalar_one()
+    new_key = base64.b64encode(b"n" * 32).decode()
+    r = await client.post("/v1/users/password/reset", json={
+        "username": "victim26", "code": code, "new_auth_key": new_key})
+    assert r.status_code == 200, r.text
+    r = await client.post("/v1/users/login", json={
+        "username": "victim26", "auth_key": new_key})
+    assert r.status_code == 200, r.text
+
+
+@pytest.mark.asyncio
+async def test_a_captured_relay_registration_is_not_replayable(client, monkeypatch):
     """Same reason /v1/nodes/announce bounds its timestamp."""
     from meshbay_hub.api import relay as relay_mod
 
+    monkeypatch.setattr(relay_mod, "RELAYS_ENABLED", True)
     sk = Ed25519PrivateKey.generate()
     pk = pk_to_b64(sk.public_key())
     relay_mod._relays["r2"] = {"pk": pk, "active": False}