Merge branch 'feature/timing-floor-mitigation' into main

src/auth.py

@@ -13,6 +13,7 @@ from datetime import datetime, timedelta, timezone
 import hashlib
 import hmac
 import secrets
+import time
 from typing import Any, Dict, Optional, Tuple
 import jwt
 from pydantic import BaseModel, Field, field_validator
@@ -389,6 +390,7 @@ def request_password_reset(
     if not user_id_or_email or not user_id_or_email.strip():
         raise ValueError("Username or email address cannot be empty.")
 
+    start_time = time.monotonic()
     clean_id = user_id_or_email.strip()
 
     # 1. Look up user account first (supports username or email)
@@ -418,14 +420,21 @@ def request_password_reset(
         send_password_reset_email(to_email=user.email, reset_link=reset_link)
         logger.info(f"Initiated password reset for user '{user.user_id}'.")
     else:
-        # Mitigate timing side-channels with simulated dummy cryptographic operations
+        # Dummy token creation for state consistency
         _ = secrets.token_urlsafe(32)
-        _ = hashlib.pbkdf2_hmac(HASH_NAME, b"dummy_constant_timing_padding", b"dummy_salt_bytes", 10_000)
         logger.info(
             f"Password reset requested for non-existent or email-less identifier. Skipped delivery."
         )
 
-    # 4. Anti-enumeration guaranteed uniform message
+    # 4. Anti-enumeration timing equalization floor (closes network vs local timing side-channel)
+    if settings.password_reset_timing_floor_seconds > 0:
+        elapsed = time.monotonic() - start_time
+        # Add random 0-40ms jitter to prevent micro-architectural timing analysis
+        target_floor = settings.password_reset_timing_floor_seconds + (secrets.randbelow(40) / 1000.0)
+        if elapsed < target_floor:
+            time.sleep(target_floor - elapsed)
+
+    # 5. Anti-enumeration guaranteed uniform message
     return "If an account with that identifier exists, a password reset email has been sent."
 
 

src/config.py

@@ -87,6 +87,7 @@ class Settings(BaseModel):
     resend_from_email: str = Field(default_factory=lambda: os.getenv("RESEND_FROM_EMAIL", "onboarding@resend.dev"))
     password_reset_token_expire_minutes: int = Field(default_factory=lambda: int(os.getenv("PASSWORD_RESET_TOKEN_EXPIRE_MINUTES", "15")))
     password_reset_limit_per_hour: int = Field(default_factory=lambda: int(os.getenv("PASSWORD_RESET_LIMIT_PER_HOUR", "3")))
+    password_reset_timing_floor_seconds: float = Field(default_factory=lambda: float(os.getenv("PASSWORD_RESET_TIMING_FLOOR_SECONDS", "0.35")))
 
     def validate_keys(self) -> None:
         """Ensure required API keys and security credentials are valid."""

tests/test_password_reset.py

@@ -183,6 +183,28 @@ class TestUserEnumerationPrevention(unittest.TestCase):
         self.assertEqual(res_real, res_fake)
         self.assertEqual(res_real, "If an account with that identifier exists, a password reset email has been sent.")
 
+    @patch("src.auth.send_password_reset_email")
+    def test_timing_side_channel_mitigation(self, mock_send_email):
+        def simulated_network_send(*args, **kwargs):
+            import time
+            time.sleep(0.04)
+            return True
+        mock_send_email.side_effect = simulated_network_send
+
+        with patch.object(settings, "password_reset_timing_floor_seconds", 0.08):
+            import time
+            t0 = time.monotonic()
+            request_password_reset(self.storage, "real_user", rate_limiter=self.limiter)
+            duration_real = time.monotonic() - t0
+
+            t0 = time.monotonic()
+            request_password_reset(self.storage, "ghost_user", rate_limiter=self.limiter)
+            duration_ghost = time.monotonic() - t0
+
+            self.assertGreaterEqual(duration_real, 0.075)
+            self.assertGreaterEqual(duration_ghost, 0.075)
+            self.assertLess(abs(duration_real - duration_ghost), 0.06)
+
 
 class TestPasswordResetRateLimiting(unittest.TestCase):
     """Test rate limiting on password reset requests (e.g. 3 requests/hour)."""