microfish/backend/tests/test_password_reset.py

"""TDD gate: durable, single-use password reset tokens.

Password resets must use a one-time, expiring, hashed token (never plaintext),
be tied to the user's auth_version so a reset invalidates old sessions, and be
scoped. The invite-pending password can only be finalized by the owner.
"""

from datetime import timedelta

import pytest
from sqlalchemy import create_engine

from app.db import Base, create_session_factory
from app.services.password_reset import PasswordResetService


@pytest.fixture()
def session_factory():
    engine = create_engine("sqlite+pysqlite:///:memory:")
    Base.metadata.create_all(engine)
    factory = create_session_factory(engine)
    try:
        yield factory
    finally:
        engine.dispose()


def test_password_reset_create_and_consume(session_factory):
    session = session_factory()
    try:
        svc = PasswordResetService(session)
        token = svc.create_token(user_id="user-1")
        # Token is opaque and not the user id.
        assert token and token != "user-1"

        # Consuming with the correct token succeeds and increments auth_version.
        ok = svc.consume_token(token, user_id="user-1")
        assert ok is True

        # Token is single-use: a second consume fails.
        assert svc.consume_token(token, user_id="user-1") is False
    finally:
        session.close()


def test_password_reset_rejects_expired_token(session_factory):
    session = session_factory()
    try:
        svc = PasswordResetService(session, ttl=timedelta(seconds=5))
        token = svc.create_token(user_id="user-1")
        row = svc._latest_token(user_id="user-1")
        # Force expiry.
        row.expires_at = row.expires_at - timedelta(seconds=60)
        session.commit()
        assert svc.consume_token(token, user_id="user-1") is False
    finally:
        session.close()


def test_password_reset_rejects_wrong_user(session_factory):
    session = session_factory()
    try:
        svc = PasswordResetService(session)
        token = svc.create_token(user_id="user-1")
        assert svc.consume_token(token, user_id="user-2") is False
    finally:
        session.close()


def test_password_reset_never_stores_plaintext_token(session_factory):
    session = session_factory()
    try:
        svc = PasswordResetService(session)
        token = svc.create_token(user_id="user-1")
        row = svc._latest_token(user_id="user-1")
        blob = repr(row)  # hashed token in DB, plaintext token only returned to caller
        assert token not in blob
    finally:
        session.close()


def test_password_reset_migration_round_trip(tmp_path, monkeypatch):
    from pathlib import Path

    from alembic import command
    from alembic.config import Config as AlembicConfig
    from sqlalchemy import inspect

    from app.db import create_database_engine

    database_url = f"sqlite+pysqlite:///{tmp_path / 'pr-roundtrip.db'}"
    monkeypatch.setenv("DATABASE_URL", database_url)
    alembic_config = AlembicConfig(str(Path(__file__).resolve().parents[1] / "alembic.ini"))
    alembic_config.set_main_option("sqlalchemy.url", database_url)

    command.upgrade(alembic_config, "head")
    engine = create_database_engine(database_url)
    try:
        inspector = inspect(engine)
        assert "password_reset_tokens" in set(inspector.get_table_names())
        cols = {c["name"] for c in inspector.get_columns("password_reset_tokens")}
        assert {"user_id", "token_hash", "auth_version", "used", "expires_at"}.issubset(cols)
    finally:
        engine.dispose()

    command.downgrade(alembic_config, "0010_usage_events")
    engine = create_database_engine(database_url)
    try:
        assert "password_reset_tokens" not in set(inspect(engine).get_table_names())
    finally:
        engine.dispose()

    command.upgrade(alembic_config, "head")
    command.check(alembic_config)


def test_password_reset_endpoints_full_flow(tmp_path, monkeypatch, session_factory):
    """request → confirm flow works end to end and changes the password."""
    import secrets

    from flask import Flask, jsonify

    from app.api.auth import auth_bp
    from app.services.identity import IdentityRepository, PasswordService
    from app.services.password_reset import PasswordResetService
    from app.utils.api_errors import ApiError
    from app.utils.locale import t

    engine = create_engine("sqlite+pysqlite:///:memory:")
    Base.metadata.create_all(engine)
    factory = create_session_factory(engine)

    app = Flask(__name__)
    app.config.update(TESTING=True, SESSION_COOKIE_SECURE=False)
    app.config["SECRET_KEY"] = secrets.token_hex(32)
    app.extensions["crowdsight_session_factory"] = factory

    @app.errorhandler(ApiError)
    def handle_api_error(error):
        return jsonify(error.to_payload(t)), error.status_code

    app.register_blueprint(auth_bp, url_prefix="/api/auth")

    pr_email = "pr-flow@example.com"
    with factory() as session:
        repo = IdentityRepository(session)
        org = repo.create_organization(name="PR Org", slug="pr-org")
        user = repo.create_user(
            email=pr_email,
            password_hash=PasswordService.hash_password("old-password-12"),
        )
        repo.create_membership(user.id, org.id, "user")
        session.commit()

    client = app.test_client()
    # Request (enrollment-agnostic).
    r = client.post(
        "/api/auth/password-reset/request",
        json={"email": pr_email},
    )
    assert r.status_code == 200

    # Issue a token directly against the store and confirm it.
    with factory() as session:
        user = IdentityRepository(session).get_user_by_email(pr_email)
        token = PasswordResetService(session).create_token(user_id=user.id)
        session.commit()

    r2 = client.post(
        "/api/auth/password-reset/confirm",
        json={"email": pr_email, "token": token, "password": "brand-new-pass-99"},
    )
    assert r2.status_code == 200

    # Old password no longer works; new one does.
    r_old = client.post(
        "/api/auth/login",
        json={"email": pr_email, "password": "old-password-12"},
    )
    assert r_old.status_code == 401
    r_new = client.post(
        "/api/auth/login",
        json={"email": pr_email, "password": "brand-new-pass-99"},
    )
    assert r_new.status_code == 200
    engine.dispose()