diff options
Diffstat (limited to 'packages/meshbay-hub/tests/test_password_reset.py')
| -rw-r--r-- | packages/meshbay-hub/tests/test_password_reset.py | 204 |
1 files changed, 204 insertions, 0 deletions
diff --git a/packages/meshbay-hub/tests/test_password_reset.py b/packages/meshbay-hub/tests/test_password_reset.py new file mode 100644 index 0000000..80977c5 --- /dev/null +++ b/packages/meshbay-hub/tests/test_password_reset.py @@ -0,0 +1,204 @@ +""" +Passphrase reset by e-mail code — Flow B of docs/auth-confirm.md §4.2. + +The hub's part re-opens sign-in only: it swaps the auth_key verifier, kills +every session, and drops every registered device key so a stored one cannot +sign back in past the reset. Restoring group access is the client's job with +the recovery key and is not exercised here. +""" + +import base64 +import time +from datetime import datetime, timedelta, timezone + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.crypto import pk_to_b64 +from meshbay_hub.db.models import EmailVerification, IPLog, User +from sqlalchemy import select + + +def _email(username: str) -> str: + return f"{username}@example.com" + + +async def _register(client, username, auth_key="k" * 44): + r = await client.post("/v1/users/register", json={ + "username": username, "email": _email(username), + "auth_key": auth_key}) + assert r.status_code in (200, 201), r.text + + +async def _request_reset(client, username, email=None): + return await client.post("/v1/users/password/reset-request", json={ + "username": username, "email": email or _email(username)}) + + +async def _reset_code(db_session, username) -> str: + uid = (await db_session.execute( + select(User.id).where(User.username == username))).scalar_one() + row = (await db_session.execute( + select(EmailVerification).where( + EmailVerification.user_id == uid, + EmailVerification.purpose == "password_reset", + EmailVerification.verified_at.is_(None), + ).order_by(EmailVerification.created_at.desc()))).scalars().first() + return row.code if row else None + + +@pytest.mark.asyncio +async def test_reset_lets_the_user_sign_in_with_a_new_passphrase(client, db_session): + await _register(client, "alice", "old" + "a" * 41) + r = await _request_reset(client, "alice") + assert r.status_code == 200 and r.json()["status"] == "sent_if_exists" + + code = await _reset_code(db_session, "alice") + assert code + + new = "new" + "b" * 41 + r = await client.post("/v1/users/password/reset", json={ + "username": "alice", "code": code, "new_auth_key": new}) + assert r.status_code == 200, r.text + + assert (await client.post("/v1/users/login", json={ + "username": "alice", "auth_key": "old" + "a" * 41})).status_code == 401 + assert (await client.post("/v1/users/login", json={ + "username": "alice", "auth_key": new})).status_code == 200 + + +@pytest.mark.asyncio +async def test_reset_request_never_reveals_whether_an_account_exists( + client, db_session): + r = await _request_reset(client, "ghost") + assert r.status_code == 200 + assert r.json()["status"] == "sent_if_exists" + rows = (await db_session.execute(select(EmailVerification))).scalars().all() + assert rows == [] + + +@pytest.mark.asyncio +async def test_reset_request_needs_the_username_and_email_to_match(client, db_session): + await _register(client, "hank") + + # Right username, wrong e-mail — answered exactly like an unknown account, + # and no code is created. + r = await _request_reset(client, "hank", email="someone.else@example.com") + assert r.status_code == 200 + assert r.json()["status"] == "sent_if_exists" + assert (await db_session.execute( + select(EmailVerification))).scalars().all() == [] + + # The real pair does create one. + await _request_reset(client, "hank") + assert (await db_session.execute( + select(EmailVerification))).scalars().first() is not None + + +@pytest.mark.asyncio +async def test_reset_request_rejects_a_malformed_email(client): + await _register(client, "iris") + r = await client.post("/v1/users/password/reset-request", json={ + "username": "iris", "email": "not-an-email"}) + assert r.status_code == 422 + + +@pytest.mark.asyncio +async def test_a_wrong_code_is_rejected_and_counts_against_the_limit( + client, db_session): + await _register(client, "bob") + await _request_reset(client, "bob") + + for _ in range(10): + r = await client.post("/v1/users/password/reset", json={ + "username": "bob", "code": "000000", "new_auth_key": "x" * 44}) + assert r.status_code == 400 + r = await client.post("/v1/users/password/reset", json={ + "username": "bob", "code": "000000", "new_auth_key": "x" * 44}) + assert r.status_code == 429 + + +@pytest.mark.asyncio +async def test_an_expired_code_is_refused(client, db_session): + await _register(client, "carol") + await _request_reset(client, "carol") + + uid = (await db_session.execute( + select(User.id).where(User.username == "carol"))).scalar_one() + row = (await db_session.execute(select(EmailVerification).where( + EmailVerification.user_id == uid))).scalars().one() + row.expires_at = datetime.now(timezone.utc) - timedelta(minutes=1) + await db_session.commit() + + r = await client.post("/v1/users/password/reset", json={ + "username": "carol", "code": row.code, "new_auth_key": "y" * 44}) + assert r.status_code == 410 + + +@pytest.mark.asyncio +async def test_a_reset_code_works_once(client, db_session): + await _register(client, "dave") + await _request_reset(client, "dave") + code = await _reset_code(db_session, "dave") + + first = await client.post("/v1/users/password/reset", json={ + "username": "dave", "code": code, "new_auth_key": "z" * 44}) + assert first.status_code == 200 + second = await client.post("/v1/users/password/reset", json={ + "username": "dave", "code": code, "new_auth_key": "z" * 44}) + assert second.status_code == 404 + + +@pytest.mark.asyncio +async def test_reset_revokes_sessions_and_wipes_devices(client, db_session): + await _register(client, "erin", "erin" + "a" * 40) + login = await client.post("/v1/users/login", json={ + "username": "erin", "auth_key": "erin" + "a" * 40}) + refresh_token = login.json()["refresh_token"] + token = login.json()["access_token"] + + sk = Ed25519PrivateKey.generate() + dev = await client.post( + "/v1/users/devices", + json={"pk_auth_ed25519": pk_to_b64(sk.public_key()), "label": "laptop"}, + headers={"Authorization": f"Bearer {token}"}) + assert dev.status_code == 201, dev.text + + await _request_reset(client, "erin") + code = await _reset_code(db_session, "erin") + r = await client.post("/v1/users/password/reset", json={ + "username": "erin", "code": code, "new_auth_key": "erin-new" + "b" * 36}) + assert r.status_code == 200 + + # Old refresh token is dead. + assert (await client.post("/v1/users/token/refresh", json={ + "refresh_token": refresh_token})).status_code == 401 + + # Every device key is gone; the stored one can no longer sign in. + uid = (await db_session.execute( + select(User.id).where(User.username == "erin"))).scalar_one() + from meshbay_hub.db.models import UserDevice + devices = (await db_session.execute( + select(UserDevice).where(UserDevice.user_id == uid))).scalars().all() + assert devices == [] + + ts = int(time.time()) + msg = f"meshbay:user_auth:erin:{ts}".encode() + da = await client.post("/v1/users/auth", json={ + "username": "erin", "timestamp": ts, + "signature": base64.b64encode(sk.sign(msg)).decode()}) + assert da.status_code == 401 + + +@pytest.mark.asyncio +async def test_the_request_and_the_reset_are_logged(client, db_session): + await _register(client, "frank") + await _request_reset(client, "frank") + code = await _reset_code(db_session, "frank") + await client.post("/v1/users/password/reset", json={ + "username": "frank", "code": code, "new_auth_key": "f" * 44}) + + uid = (await db_session.execute( + select(User.id).where(User.username == "frank"))).scalar_one() + events = {e.event for e in (await db_session.execute( + select(IPLog).where(IPLog.user_id == uid))).scalars().all()} + assert {"password_reset_request", "password_reset"} <= events |