diff options
Diffstat (limited to 'packages/meshbay-hub/tests/test_login_lockout.py')
| -rw-r--r-- | packages/meshbay-hub/tests/test_login_lockout.py | 84 |
1 files changed, 42 insertions, 42 deletions
diff --git a/packages/meshbay-hub/tests/test_login_lockout.py b/packages/meshbay-hub/tests/test_login_lockout.py index 43300d0..6f17d98 100644 --- a/packages/meshbay-hub/tests/test_login_lockout.py +++ b/packages/meshbay-hub/tests/test_login_lockout.py @@ -51,10 +51,10 @@ async def _fail(client, username, times): @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) + await _register(client, "alice_test") + await _fail(client, "alice_test", 4) - r = await _login(client, "alice", RIGHT) + r = await _login(client, "alice_test", 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. @@ -64,11 +64,11 @@ async def test_the_fourth_failure_locks_and_the_right_passphrase_is_refused(clie @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 _register(client, "bob_test") + await _fail(client, "bob_test", 4) await _fail(client, "nobody-by-this-name", 4) - real = await _login(client, "bob", WRONG) + real = await _login(client, "bob_test", 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 @@ -76,42 +76,42 @@ async def test_an_unknown_name_locks_exactly_like_a_real_one(client): @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) + await _register(client, "carol_test") + await _fail(client, "carol_test", 3) + r = await _login(client, "carol_test", 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 + await _fail(client, "carol_test", 3) + assert (await _login(client, "carol_test", 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 _register(client, "dave_test") + await _fail(client, "dave_test", 4) + assert (await _login(client, "dave_test", RIGHT)).status_code == 429 await db_session.execute( - update(LoginThrottle).where(LoginThrottle.key == _key("dave")) + update(LoginThrottle).where(LoginThrottle.key == _key("dave_test")) .values(last_failure_at=datetime.now(timezone.utc) - timedelta(minutes=61))) await db_session.commit() - assert (await _login(client, "dave", RIGHT)).status_code == 200 + assert (await _login(client, "dave_test", 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 _register(client, "erin_test") + await _fail(client, "erin_test", 3) await db_session.execute( - update(LoginThrottle).where(LoginThrottle.key == _key("erin")) + update(LoginThrottle).where(LoginThrottle.key == _key("erin_test")) .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 + await _fail(client, "erin_test", 1) + assert (await _login(client, "erin_test", RIGHT)).status_code == 200 @pytest.mark.asyncio @@ -124,8 +124,8 @@ async def test_a_burst_of_concurrent_guesses_gets_no_more_than_the_limit(client) 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)]) + await _register(client, "frank_test") + results = await asyncio.gather(*[_login(client, "frank_test", 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 @@ -134,8 +134,8 @@ async def test_a_burst_of_concurrent_guesses_gets_no_more_than_the_limit(client) @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"] + await _register(client, "grace_test") + token = (await _login(client, "grace_test", RIGHT)).json()["access_token"] auth = {"Authorization": f"Bearer {token}"} for _ in range(4): @@ -146,19 +146,19 @@ async def test_change_password_counts_on_the_same_row(client): 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 + assert (await _login(client, "grace_test", 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"] + await _register(client, "olivia_test") + token = (await _login(client, "olivia_test", 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) + await _fail(client, "olivia_test", 4) left = (await client.get("/v1/users/me", headers=auth)).json()["passphrase_locked_for"] assert 3500 <= left <= 3600 @@ -168,16 +168,16 @@ async def test_an_attempt_that_checks_no_passphrase_is_not_counted(client, db_se """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 _register(client, "heidi_test") await db_session.execute( - update(User).where(User.username == "heidi").values(pw_version=2)) + update(User).where(User.username == "heidi_test").values(pw_version=2)) await db_session.commit() for _ in range(6): - r = await _login(client, "heidi", RIGHT) + r = await _login(client, "heidi_test", 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"))) + select(LoginThrottle.failures).where(LoginThrottle.key == _key("heidi_test"))) assert not failures @@ -191,7 +191,7 @@ async def test_the_table_never_holds_what_was_typed(client, db_session): # ── The admin's two numbers ────────────────────────────────────────────────── -async def _admin_headers(client, username="root"): +async def _admin_headers(client, username="root_test"): await _register(client, username) set_admin_usernames([username]) token = (await _login(client, username, RIGHT)).json()["access_token"] @@ -211,9 +211,9 @@ async def test_the_admin_sets_the_limit_and_the_hub_applies_it(client): 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) + await _register(client, "ivan_test") + await _fail(client, "ivan_test", 2) + r = await _login(client, "ivan_test", RIGHT) assert r.status_code == 429 assert int(r.headers["retry-after"]) <= 300 @@ -224,9 +224,9 @@ async def test_zero_failures_turns_the_lockout_off(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 + await _register(client, "judy_test") + await _fail(client, "judy_test", 8) + assert (await _login(client, "judy_test", RIGHT)).status_code == 200 @pytest.mark.asyncio @@ -248,8 +248,8 @@ async def test_values_are_clamped_and_unknown_keys_refused(client): @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"] + await _register(client, "mallory_test") + token = (await _login(client, "mallory_test", RIGHT)).json()["access_token"] r = await client.patch("/v1/admin/settings", headers={"Authorization": f"Bearer {token}"}, json={"login": {"max_failures": 0}}) |