feat: enhance refresh token handling with grace period and rotation detection

backend/tests/test_auth_api.py

@@ -1,11 +1,11 @@
 import pytest
 from werkzeug.security import generate_password_hash, check_password_hash
 from flask import Flask
-from api.auth_api import auth_api
+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
+from datetime import datetime, timedelta, timezone
 from tests.conftest import TEST_SECRET_KEY, TEST_REFRESH_TOKEN_EXPIRY_DAYS
 
 @pytest.fixture
@@ -188,4 +188,245 @@ def test_migration_script_hashes_plain_text_passwords():
 
     # Check user2 password unchanged
     user2_dict = users_db.get(Query().email == 'test2@example.com')
-    assert user2_dict['password'] == already_hashed
+    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