chore/backend/tests/test_auth_api.py

import pytest
from werkzeug.security import generate_password_hash, check_password_hash
from flask import Flask
from api.auth_api import auth_api, _hash_token
from db.db import users_db, refresh_tokens_db
from tinydb import Query
from models.user import User
from datetime import datetime, timedelta, timezone
from tests.conftest import TEST_SECRET_KEY, TEST_REFRESH_TOKEN_EXPIRY_DAYS

@pytest.fixture
def client():
    """Setup Flask test client with auth blueprint."""
    app = Flask(__name__)
    app.register_blueprint(auth_api, url_prefix='/auth')
    app.config['TESTING'] = True
    app.config['SECRET_KEY'] = TEST_SECRET_KEY
    app.config['REFRESH_TOKEN_EXPIRY_DAYS'] = TEST_REFRESH_TOKEN_EXPIRY_DAYS
    app.config['FRONTEND_URL'] = 'http://localhost:5173'
    with app.test_client() as client:
        yield client

def test_signup_hashes_password(client):
    """Test that signup hashes the password."""
    # Clean up any existing user
    users_db.remove(Query().email == 'test@example.com')
    
    data = {
        'first_name': 'Test',
        'last_name': 'User',
        'email': 'test@example.com',
        'password': 'password123'
    }
    response = client.post('/auth/signup', json=data)
    assert response.status_code == 201

    # Check that password is hashed in DB
    user_dict = users_db.get(Query().email == 'test@example.com')
    assert user_dict is not None
    assert user_dict['password'].startswith('scrypt:')

def test_login_with_correct_password(client):
    """Test login succeeds with correct password."""
    # Clean up and create a user with hashed password
    users_db.remove(Query().email == 'test@example.com')
    hashed_pw = generate_password_hash('password123')
    user = User(
        first_name='Test',
        last_name='User',
        email='test@example.com',
        password=hashed_pw,
        verified=True
    )
    users_db.insert(user.to_dict())

    data = {'email': 'test@example.com', 'password': 'password123'}
    response = client.post('/auth/login', json=data)
    assert response.status_code == 200
    cookies = response.headers.getlist('Set-Cookie')
    cookie_str = ' '.join(cookies)
    assert 'access_token=' in cookie_str
    assert 'refresh_token=' in cookie_str

def test_login_with_incorrect_password(client):
    """Test login fails with incorrect password."""
    # Clean up and create a user with hashed password
    users_db.remove(Query().email == 'test@example.com')
    hashed_pw = generate_password_hash('password123')
    user = User(
        first_name='Test',
        last_name='User',
        email='test@example.com',
        password=hashed_pw,
        verified=True
    )
    users_db.insert(user.to_dict())

    data = {'email': 'test@example.com', 'password': 'wrongpassword'}
    response = client.post('/auth/login', json=data)
    assert response.status_code == 401
    assert response.json['code'] == 'INVALID_CREDENTIALS'

def test_reset_password_hashes_new_password(client):
    """Test that reset-password hashes the new password."""
    # Clean up and create a user with reset token
    users_db.remove(Query().email == 'test@example.com')
    user = User(
        first_name='Test',
        last_name='User',
        email='test@example.com',
        password=generate_password_hash('oldpassword'),
        verified=True,
        reset_token='validtoken',
        reset_token_created=datetime.utcnow().isoformat()
    )
    users_db.insert(user.to_dict())

    data = {'token': 'validtoken', 'password': 'newpassword123'}
    response = client.post('/auth/reset-password', json=data)
    assert response.status_code == 200

    # Check that password is hashed in DB
    user_dict = users_db.get(Query().email == 'test@example.com')
    assert user_dict is not None
    assert user_dict['password'].startswith('scrypt:')
    assert check_password_hash(user_dict['password'], 'newpassword123')


def test_reset_password_invalidates_existing_jwt(client):
    users_db.remove(Query().email == 'test@example.com')
    user = User(
        first_name='Test',
        last_name='User',
        email='test@example.com',
        password=generate_password_hash('oldpassword123'),
        verified=True,
        reset_token='validtoken2',
        reset_token_created=datetime.utcnow().isoformat(),
    )
    users_db.insert(user.to_dict())

    login_response = client.post('/auth/login', json={'email': 'test@example.com', 'password': 'oldpassword123'})
    assert login_response.status_code == 200
    login_cookies = login_response.headers.getlist('Set-Cookie')
    login_cookie_str = ' '.join(login_cookies)
    assert 'access_token=' in login_cookie_str
    # Extract the old access token
    old_token = None
    for c in login_cookies:
        if c.startswith('access_token='):
            old_token = c.split('access_token=', 1)[1].split(';', 1)[0]
            break
    assert old_token

    reset_response = client.post('/auth/reset-password', json={'token': 'validtoken2', 'password': 'newpassword123'})
    assert reset_response.status_code == 200
    reset_cookies = reset_response.headers.getlist('Set-Cookie')
    reset_cookie_str = ' '.join(reset_cookies)
    assert 'access_token=' in reset_cookie_str

    # Verify all refresh tokens for this user are deleted
    user_dict = users_db.get(Query().email == 'test@example.com')
    user_tokens = refresh_tokens_db.search(Query().user_id == user_dict['id'])
    assert len(user_tokens) == 0

    # Set the old token as a cookie and test that it's now invalid
    client.set_cookie('access_token', old_token)
    me_response = client.get('/auth/me')
    assert me_response.status_code == 401
    assert me_response.json['code'] == 'INVALID_TOKEN'

def test_migration_script_hashes_plain_text_passwords():
    """Test the migration script hashes plain text passwords."""
    # Clean up
    users_db.remove(Query().email == 'test1@example.com')
    users_db.remove(Query().email == 'test2@example.com')
    
    # Create users with plain text passwords
    user1 = User(
        first_name='Test1',
        last_name='User',
        email='test1@example.com',
        password='plaintext1',
        verified=True
    )
    already_hashed = generate_password_hash('alreadyhashed')
    user2 = User(
        first_name='Test2',
        last_name='User',
        email='test2@example.com',
        password=already_hashed,  # Already hashed
        verified=True
    )
    users_db.insert(user1.to_dict())
    users_db.insert(user2.to_dict())

    # Run migration script
    import sys
    import os
    sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
    from scripts.hash_passwords import main
    main()

    # Check user1 password is now hashed
    user1_dict = users_db.get(Query().email == 'test1@example.com')
    assert user1_dict['password'].startswith('scrypt:')
    assert check_password_hash(user1_dict['password'], 'plaintext1')

    # Check user2 password unchanged
    user2_dict = users_db.get(Query().email == 'test2@example.com')
    assert user2_dict['password'] == already_hashed


def _extract_cookie_value(response, cookie_name):
    """Extract a raw cookie value from a response's Set-Cookie headers."""
    for cookie in response.headers.getlist('Set-Cookie'):
        if cookie.startswith(f'{cookie_name}='):
            return cookie.split(f'{cookie_name}=', 1)[1].split(';', 1)[0]
    return None


def _create_verified_user(email, password):
    """Insert a verified user with the given credentials."""
    users_db.remove(Query().email == email)
    user = User(
        first_name='Test',
        last_name='User',
        email=email,
        password=generate_password_hash(password),
        verified=True,
    )
    users_db.insert(user.to_dict())
    return user


def _set_refresh_cookie(client, raw_token):
    """
    Set the refresh token cookie on a test client so it is sent to /auth/refresh.
    Production uses path='/api/auth' because the frontend calls /api/auth/refresh,
    but the test fixture exposes the blueprint at /auth/refresh directly.
    """
    client.set_cookie('refresh_token', raw_token, path='/')


def test_refresh_rotates_token(client):
    """A successful refresh marks the old token used and issues a new one in the same family."""
    email = 'refresh-rotate@test.com'
    password = 'password123'
    _create_verified_user(email, password)

    login_response = client.post('/auth/login', json={'email': email, 'password': password})
    assert login_response.status_code == 200
    old_refresh = _extract_cookie_value(login_response, 'refresh_token')
    assert old_refresh

    _set_refresh_cookie(client, old_refresh)
    refresh_response = client.post('/auth/refresh')
    assert refresh_response.status_code == 200
    new_refresh = _extract_cookie_value(refresh_response, 'refresh_token')
    assert new_refresh
    assert new_refresh != old_refresh

    user_dict = users_db.get(Query().email == email)
    old_hash = _hash_token(old_refresh)
    new_hash = _hash_token(new_refresh)

    old_record = refresh_tokens_db.get(Query().token_hash == old_hash)
    assert old_record is not None
    assert old_record['is_used'] is True
    assert old_record['rotated_at'] is not None

    new_record = refresh_tokens_db.get(Query().token_hash == new_hash)
    assert new_record is not None
    assert new_record['is_used'] is False
    assert new_record['token_family'] == old_record['token_family']


def test_refresh_reuse_only_invalidates_family(client):
    """Replay of a used refresh token only kills its own family, not other devices."""
    email = 'refresh-family@test.com'
    password = 'password123'
    user = _create_verified_user(email, password)

    # Device A logs in
    client_a = client
    login_a = client_a.post('/auth/login', json={'email': email, 'password': password})
    assert login_a.status_code == 200
    refresh_a = _extract_cookie_value(login_a, 'refresh_token')

    # Device B logs in (separate client = separate cookie jar)
    app = client_a.application
    client_b = app.test_client()
    login_b = client_b.post('/auth/login', json={'email': email, 'password': password})
    assert login_b.status_code == 200
    refresh_b = _extract_cookie_value(login_b, 'refresh_token')

    assert refresh_a != refresh_b

    # Device A refreshes normally
    _set_refresh_cookie(client_a, refresh_a)
    refresh_a_response = client_a.post('/auth/refresh')
    assert refresh_a_response.status_code == 200

    # Capture families before any purge so we can assert afterwards.
    family_a = refresh_tokens_db.get(Query().token_hash == _hash_token(refresh_a))['token_family']
    family_b = refresh_tokens_db.get(Query().token_hash == _hash_token(refresh_b))['token_family']
    assert family_a != family_b

    # Backdate rotation so the replay is past the grace period and treated as theft.
    old_hash_a = _hash_token(refresh_a)
    backdated = (datetime.now(timezone.utc) - timedelta(seconds=60)).isoformat()
    refresh_tokens_db.update({'rotated_at': backdated}, Query().token_hash == old_hash_a)

    # Attacker replays device A's old token
    _set_refresh_cookie(client_a, refresh_a)
    reuse_response = client_a.post('/auth/refresh')
    assert reuse_response.status_code == 401
    assert reuse_response.json['code'] == 'REFRESH_TOKEN_REUSE'

    # Device B's refresh token should still be valid
    _set_refresh_cookie(client_b, refresh_b)
    refresh_b_response = client_b.post('/auth/refresh')
    assert refresh_b_response.status_code == 200

    # Only family A should be purged; family B should remain
    remaining = refresh_tokens_db.search(Query().user_id == user.id)
    remaining_families = {t['token_family'] for t in remaining}
    assert family_a not in remaining_families
    assert family_b in remaining_families


def test_refresh_reuse_within_grace_period_is_tolerated(client):
    """A replay within the grace period is treated as a race condition, not theft."""
    email = 'refresh-race@test.com'
    password = 'password123'
    _create_verified_user(email, password)

    login_response = client.post('/auth/login', json={'email': email, 'password': password})
    assert login_response.status_code == 200
    refresh_token = _extract_cookie_value(login_response, 'refresh_token')

    # First refresh marks the token as used
    _set_refresh_cookie(client, refresh_token)
    first_refresh = client.post('/auth/refresh')
    assert first_refresh.status_code == 200

    # Immediate replay (same legitimate client racing) should succeed
    _set_refresh_cookie(client, refresh_token)
    race_response = client.post('/auth/refresh')
    assert race_response.status_code == 200

    # The family should still be valid
    user_dict = users_db.get(Query().email == email)
    family = refresh_tokens_db.get(Query().token_hash == _hash_token(refresh_token))['token_family']
    family_tokens = refresh_tokens_db.search(
        (Query().user_id == user_dict['id']) & (Query().token_family == family)
    )
    assert len(family_tokens) >= 1
    assert any(t['is_used'] is False for t in family_tokens)


def test_refresh_reuse_after_grace_period_invalidates_family(client):
    """A replay after the grace period is treated as theft and kills only that family."""
    email = 'refresh-theft@test.com'
    password = 'password123'
    user = _create_verified_user(email, password)

    login_response = client.post('/auth/login', json={'email': email, 'password': password})
    assert login_response.status_code == 200
    refresh_token = _extract_cookie_value(login_response, 'refresh_token')

    # Refresh once, then backdate the rotation timestamp past the grace period
    _set_refresh_cookie(client, refresh_token)
    client.post('/auth/refresh')

    old_hash = _hash_token(refresh_token)
    old_record = refresh_tokens_db.get(Query().token_hash == old_hash)
    old_family = old_record['token_family']
    backdated = (datetime.now(timezone.utc) - timedelta(seconds=60)).isoformat()
    refresh_tokens_db.update({'rotated_at': backdated}, Query().token_hash == old_hash)

    # Replay now should be detected as theft
    _set_refresh_cookie(client, refresh_token)
    reuse_response = client.post('/auth/refresh')
    assert reuse_response.status_code == 401
    assert reuse_response.json['code'] == 'REFRESH_TOKEN_REUSE'

    remaining = refresh_tokens_db.search(Query().user_id == user.id)
    remaining_families = {t['token_family'] for t in remaining}
    assert old_family not in remaining_families


def test_refresh_reuse_without_rotated_at_invalidates_family(client):
    """Legacy used tokens without rotated_at are treated as theft, not race conditions."""
    email = 'refresh-legacy@test.com'
    password = 'password123'
    user = _create_verified_user(email, password)

    login_response = client.post('/auth/login', json={'email': email, 'password': password})
    assert login_response.status_code == 200
    refresh_token = _extract_cookie_value(login_response, 'refresh_token')

    # Rotate the token, then strip rotated_at to simulate pre-migration data
    _set_refresh_cookie(client, refresh_token)
    client.post('/auth/refresh')
    old_hash = _hash_token(refresh_token)
    refresh_tokens_db.update({'rotated_at': None}, Query().token_hash == old_hash)

    old_record = refresh_tokens_db.get(Query().token_hash == old_hash)
    old_family = old_record['token_family']

    # Replay should be treated as theft because rotated_at is missing
    _set_refresh_cookie(client, refresh_token)
    reuse_response = client.post('/auth/refresh')
    assert reuse_response.status_code == 401
    assert reuse_response.json['code'] == 'REFRESH_TOKEN_REUSE'

    remaining = refresh_tokens_db.search(Query().user_id == user.id)
    remaining_families = {t['token_family'] for t in remaining}
    assert old_family not in remaining_families


def test_refresh_reuse_with_zero_grace_period(client):
    """A grace period of 0 means any replay of a used token is treated as theft."""
    email = 'refresh-zero-grace@test.com'
    password = 'password123'
    user = _create_verified_user(email, password)

    # Configure the app with a 0-second grace period
    client.application.config['REFRESH_TOKEN_REUSE_GRACE_PERIOD_SECONDS'] = 0

    login_response = client.post('/auth/login', json={'email': email, 'password': password})
    assert login_response.status_code == 200
    refresh_token = _extract_cookie_value(login_response, 'refresh_token')

    # Rotate the token; rotated_at is within the normal default grace period
    _set_refresh_cookie(client, refresh_token)
    client.post('/auth/refresh')

    old_hash = _hash_token(refresh_token)
    old_record = refresh_tokens_db.get(Query().token_hash == old_hash)
    old_family = old_record['token_family']

    # Immediate replay should still be theft with a 0-second grace period
    _set_refresh_cookie(client, refresh_token)
    reuse_response = client.post('/auth/refresh')
    assert reuse_response.status_code == 401
    assert reuse_response.json['code'] == 'REFRESH_TOKEN_REUSE'

    remaining = refresh_tokens_db.search(Query().user_id == user.id)
    remaining_families = {t['token_family'] for t in remaining}
    assert old_family not in remaining_families