1 files changed, 256 insertions, 0 deletions
diff --git a/packages/meshbay-hub/tests/test_login_lockout.py b/packages/meshbay-hub/tests/test_login_lockout.py
new file mode 100644
index 0000000..43300d0
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_login_lockout.py
@@ -0,0 +1,256 @@
+"""
+Per-account sign-in lockout (`login_throttle.py`).
+
+The per-IP limit bounds one address, and IPv6 gives every subscriber a /64 of
+them; an online guess targets an account, so the account is what is counted.
+The properties pinned here are the ones that make that safe to ship:
+
+  * the Nth wrong passphrase locks, and a locked account is refused **before**
+    the passphrase is checked — the right one is refused too
+  * an unknown username locks exactly like a real one, so the 429 says nothing
+    the 401 did not (M1)
+  * a right passphrase clears the count, and failures age out of the window
+  * `change_password` checks the same passphrase, so it counts on the same row
+  * the two numbers are the admin's to change, and zero turns it off
+
+What one account costs another through this — a stranger locking your name —
+is `test_availability_between_members.py`, because it takes two accounts.
+"""
+
+import asyncio
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from sqlalchemy import select, update
+
+from meshbay_hub.api.deps import set_admin_usernames
+from meshbay_hub.db.models import LoginThrottle
+from meshbay_hub.login_throttle import _key
+
+RIGHT = "r" * 44
+WRONG = "w" * 44
+
+
+async def _register(client, username, auth_key=RIGHT):
+    r = await client.post("/v1/users/register", json={
+        "username": username, "email": f"{username}@example.com",
+        "auth_key": auth_key})
+    assert r.status_code == 201, r.text
+
+
+async def _login(client, username, auth_key):
+    return await client.post("/v1/users/login", json={
+        "username": username, "auth_key": auth_key})
+
+
+async def _fail(client, username, times):
+    for _ in range(times):
+        r = await _login(client, username, WRONG)
+        assert r.status_code == 401, r.text
+
+
+@pytest.mark.asyncio
+async def test_the_fourth_failure_locks_and_the_right_passphrase_is_refused(client):
+    await _register(client, "alice")
+    await _fail(client, "alice", 4)
+
+    r = await _login(client, "alice", RIGHT)
+    assert r.status_code == 429, r.text
+    assert r.json()["detail"] == "account_locked"
+    # An hour, give or take the time the four failures took.
+    assert 3500 <= int(r.headers["retry-after"]) <= 3600
+
+
+@pytest.mark.asyncio
+async def test_an_unknown_name_locks_exactly_like_a_real_one(client):
+    """M1: the lockout must not become the enumeration oracle `login` avoids."""
+    await _register(client, "bob")
+    await _fail(client, "bob", 4)
+    await _fail(client, "nobody-by-this-name", 4)
+
+    real = await _login(client, "bob", WRONG)
+    ghost = await _login(client, "nobody-by-this-name", WRONG)
+    assert (real.status_code, real.json()) == (ghost.status_code, ghost.json())
+    assert real.status_code == 429
+
+
+@pytest.mark.asyncio
+async def test_the_right_passphrase_clears_the_count(client):
+    await _register(client, "carol")
+    await _fail(client, "carol", 3)
+    r = await _login(client, "carol", RIGHT)
+    assert r.status_code == 200, r.text
+
+    # Three more would have been seven in a row without the reset.
+    await _fail(client, "carol", 3)
+    assert (await _login(client, "carol", RIGHT)).status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_a_lockout_ends_when_its_window_does(client, db_session):
+    await _register(client, "dave")
+    await _fail(client, "dave", 4)
+    assert (await _login(client, "dave", RIGHT)).status_code == 429
+
+    await db_session.execute(
+        update(LoginThrottle).where(LoginThrottle.key == _key("dave"))
+        .values(last_failure_at=datetime.now(timezone.utc) - timedelta(minutes=61)))
+    await db_session.commit()
+
+    assert (await _login(client, "dave", RIGHT)).status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_old_failures_do_not_carry_into_a_new_window(client, db_session):
+    await _register(client, "erin")
+    await _fail(client, "erin", 3)
+    await db_session.execute(
+        update(LoginThrottle).where(LoginThrottle.key == _key("erin"))
+        .values(last_failure_at=datetime.now(timezone.utc) - timedelta(minutes=61)))
+    await db_session.commit()
+
+    # One stale window of three, then one fresh failure: a count of one, not four.
+    await _fail(client, "erin", 1)
+    assert (await _login(client, "erin", RIGHT)).status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_a_burst_of_concurrent_guesses_gets_no_more_than_the_limit(client):
+    """The attempt is taken before the check, in one statement.
+
+    Read-then-write would let every request in a burst read "no failures yet"
+    and be checked. On SQLite writes serialise anyway, so this pins the
+    behaviour rather than proving the statement under PostgreSQL's concurrency;
+    the statement is an `ON CONFLICT DO UPDATE … WHERE`, which both evaluate
+    against the row as locked.
+    """
+    await _register(client, "frank")
+    results = await asyncio.gather(*[_login(client, "frank", WRONG) for _ in range(10)])
+    codes = sorted(r.status_code for r in results)
+    assert codes.count(401) == 4, codes
+    assert codes.count(429) == 6, codes
+
+
+@pytest.mark.asyncio
+async def test_change_password_counts_on_the_same_row(client):
+    """It checks the same passphrase, so it is the same oracle."""
+    await _register(client, "grace")
+    token = (await _login(client, "grace", RIGHT)).json()["access_token"]
+    auth = {"Authorization": f"Bearer {token}"}
+
+    for _ in range(4):
+        r = await client.post("/v1/users/password", headers=auth, json={
+            "old_auth_key": WRONG, "new_auth_key": "n" * 44})
+        assert r.status_code == 403, r.text
+
+    r = await client.post("/v1/users/password", headers=auth, json={
+        "old_auth_key": RIGHT, "new_auth_key": "n" * 44})
+    assert r.status_code == 429, r.text
+    assert (await _login(client, "grace", RIGHT)).status_code == 429
+
+
+@pytest.mark.asyncio
+async def test_a_signed_in_session_is_told_its_own_lockout(client):
+    """A passphrase change re-wraps every node's bundle before the hub accepts
+    it, so the client must know not to start one the hub would then refuse."""
+    await _register(client, "olivia")
+    token = (await _login(client, "olivia", RIGHT)).json()["access_token"]
+    auth = {"Authorization": f"Bearer {token}"}
+    assert (await client.get("/v1/users/me", headers=auth)).json()["passphrase_locked_for"] == 0
+
+    await _fail(client, "olivia", 4)
+    left = (await client.get("/v1/users/me", headers=auth)).json()["passphrase_locked_for"]
+    assert 3500 <= left <= 3600
+
+
+@pytest.mark.asyncio
+async def test_an_attempt_that_checks_no_passphrase_is_not_counted(client, db_session):
+    """A legacy account asked to upgrade has been told nothing about its passphrase."""
+    from meshbay_hub.db.models import User
+
+    await _register(client, "heidi")
+    await db_session.execute(
+        update(User).where(User.username == "heidi").values(pw_version=2))
+    await db_session.commit()
+
+    for _ in range(6):
+        r = await _login(client, "heidi", RIGHT)
+        assert r.status_code == 401 and r.json()["detail"] == "auth_upgrade_required"
+    failures = await db_session.scalar(
+        select(LoginThrottle.failures).where(LoginThrottle.key == _key("heidi")))
+    assert not failures
+
+
+@pytest.mark.asyncio
+async def test_the_table_never_holds_what_was_typed(client, db_session):
+    """People type passphrases into the username field."""
+    await _login(client, "my-secret-passphrase-typed-in-the-wrong-box", WRONG)
+    keys = (await db_session.execute(select(LoginThrottle.key))).scalars().all()
+    assert keys and all("secret" not in k for k in keys)
+
+
+# ── The admin's two numbers ──────────────────────────────────────────────────
+
+async def _admin_headers(client, username="root"):
+    await _register(client, username)
+    set_admin_usernames([username])
+    token = (await _login(client, username, RIGHT)).json()["access_token"]
+    return {"Authorization": f"Bearer {token}"}
+
+
+@pytest.mark.asyncio
+async def test_the_admin_sets_the_limit_and_the_hub_applies_it(client):
+    admin = await _admin_headers(client)
+
+    r = await client.get("/v1/admin/settings", headers=admin)
+    assert r.json()["login"] == {"max_failures": 4, "lockout_minutes": 60}
+    assert r.json()["login_defaults"] == {"max_failures": 4, "lockout_minutes": 60}
+
+    r = await client.patch("/v1/admin/settings", headers=admin,
+                           json={"login": {"max_failures": 2, "lockout_minutes": 5}})
+    assert r.status_code == 200, r.text
+    assert r.json()["login"] == {"max_failures": 2, "lockout_minutes": 5}
+
+    await _register(client, "ivan")
+    await _fail(client, "ivan", 2)
+    r = await _login(client, "ivan", RIGHT)
+    assert r.status_code == 429
+    assert int(r.headers["retry-after"]) <= 300
+
+
+@pytest.mark.asyncio
+async def test_zero_failures_turns_the_lockout_off(client):
+    admin = await _admin_headers(client)
+    await client.patch("/v1/admin/settings", headers=admin,
+                       json={"login": {"max_failures": 0}})
+
+    await _register(client, "judy")
+    await _fail(client, "judy", 8)
+    assert (await _login(client, "judy", RIGHT)).status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_values_are_clamped_and_unknown_keys_refused(client):
+    admin = await _admin_headers(client)
+
+    r = await client.patch("/v1/admin/settings", headers=admin,
+                           json={"login": {"max_failures": -3, "lockout_minutes": 10**9}})
+    assert r.status_code == 200
+    low, _ = r.json()["login_bounds"]["max_failures"]
+    _, high = r.json()["login_bounds"]["lockout_minutes"]
+    assert r.json()["login"] == {"max_failures": low, "lockout_minutes": high}
+
+    r = await client.patch("/v1/admin/settings", headers=admin,
+                           json={"login": {"lockout_hours": 1}})
+    assert r.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_only_an_admin_changes_them(client):
+    await _admin_headers(client)          # an admin exists; this is someone else
+    await _register(client, "mallory")
+    token = (await _login(client, "mallory", RIGHT)).json()["access_token"]
+    r = await client.patch("/v1/admin/settings",
+                           headers={"Authorization": f"Bearer {token}"},
+                           json={"login": {"max_failures": 0}})
+    assert r.status_code == 403