feat(security): apply uniform verify delay for all outcomes

Use min_delay/max_delay for every verify_key response and deprecate rrd to timing signal leakage while keeping backward compatibility.

Author: Athroniaeth

src/fastapi_api_key/services/base.py

@@ -1,10 +1,11 @@
 import asyncio
 import os
+import warnings
 from dataclasses import dataclass
 from datetime import datetime
 from random import SystemRandom
 from abc import ABC, abstractmethod
-from typing import Optional, Tuple, List
+from typing import List, Optional, Tuple
 
 from fastapi_api_key.domain.entities import ApiKey
 from fastapi_api_key.domain.errors import KeyNotProvided, KeyNotFound, InvalidKey, ConfigurationError
@@ -45,7 +46,9 @@ class AbstractApiKeyService(ABC):
         hasher: Hasher for hashing secrets. Defaults to Argon2ApiKeyHasher.
         separator: Separator in API key format. Defaults to "-".
         global_prefix: Prefix for API keys. Defaults to "ak".
-        rrd: Random response delay for timing attack mitigation. Defaults to 1/3.
+        rrd: Deprecated random response delay. Ignored if provided.
+        min_delay: Minimum delay (seconds) applied to all verify responses.
+        max_delay: Maximum delay (seconds) applied to all verify responses.
 
     Notes:
         The global key_id is pure cosmetic, it is not used for anything else.
@@ -59,16 +62,33 @@ def __init__(
         hasher: ApiKeyHasher,
         separator: str = DEFAULT_SEPARATOR,
         global_prefix: str = DEFAULT_GLOBAL_PREFIX,
-        rrd: float = 1 / 3,
+        rrd: Optional[float] = None,
+        min_delay: float = 0.1,
+        max_delay: float = 0.3,
     ) -> None:
         # Warning developer that separator is automatically added to the global key_id
         if separator in global_prefix:
             raise ValueError("Separator must not be in the global key_id")
 
+        if rrd is not None:
+            warnings.warn(
+                "rrd is deprecated and ignored. Use min_delay/max_delay instead.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+
+        if min_delay < 0 or max_delay < 0:
+            raise ValueError("min_delay and max_delay must be non-negative")
+
+        if max_delay < min_delay:
+            raise ValueError("max_delay must be greater than or equal to min_delay")
+
         self._repo = repo
         self._hasher = hasher
 
         self.rrd = rrd
+        self.min_delay = min_delay
+        self.max_delay = max_delay
         self.separator = separator
         self.global_prefix = global_prefix
         self._system_random = SystemRandom()
@@ -216,14 +236,21 @@ async def verify_key(self, api_key: str, required_scopes: Optional[List[str]] =
             If the entity is inactive or expired, an exception is raised.
             If the check between the provided plain key and the stored hash fails,
             an InvalidKey exception is raised. Else, the entity is returned.
+            A randomized delay is always applied to reduce timing signals.
         """
         try:
-            return await self._verify_key(api_key, required_scopes)
-        except Exception as e:
-            # Add a small jitter to make timing-based probing harder to profile.
-            wait = self._system_random.uniform(self.rrd, self.rrd * 2)
-            await asyncio.sleep(wait)
-            raise e
+            result = await self._verify_key(api_key, required_scopes)
+        except Exception as exc:
+            await self._apply_delay()
+            raise exc
+
+        await self._apply_delay()
+        return result
+
+    async def _apply_delay(self) -> None:
+        """Apply a randomized delay to reduce timing signals."""
+        wait = self._system_random.uniform(self.min_delay, self.max_delay)
+        await asyncio.sleep(wait)
 
     @abstractmethod
     async def _verify_key(self, api_key: str, required_scopes: Optional[List[str]] = None) -> ApiKey:
@@ -249,6 +276,7 @@ async def _verify_key(self, api_key: str, required_scopes: Optional[List[str]] =
             If the entity is inactive or expired, an exception is raised.
             If the check between the provided plain key and the stored hash fails,
             an InvalidKey exception is raised. Else, the entity is returned.
+            A randomized delay is always applied to reduce timing signals.
         """
         ...
 
@@ -280,14 +308,18 @@ def __init__(
         hasher: ApiKeyHasher,
         separator: str = DEFAULT_SEPARATOR,
         global_prefix: str = DEFAULT_GLOBAL_PREFIX,
-        rrd: float = 1 / 3,
+        rrd: Optional[float] = None,
+        min_delay: float = 0.1,
+        max_delay: float = 0.3,
     ) -> None:
         super().__init__(
             repo=repo,
             hasher=hasher,
             separator=separator,
             global_prefix=global_prefix,
             rrd=rrd,
+            min_delay=min_delay,
+            max_delay=max_delay,
         )
 
     async def load_dotenv(self, envvar_prefix: str = "API_KEY_"):

src/fastapi_api_key/services/cached.py

@@ -6,7 +6,7 @@
     ) from e
 
 import hashlib
-from typing import Optional, List
+from typing import List, Optional
 
 import aiocache
 from aiocache import BaseCache
@@ -59,14 +59,18 @@ def __init__(
         cache_ttl: int = 300,
         separator: str = DEFAULT_SEPARATOR,
         global_prefix: str = "ak",
-        rrd: float = 1 / 3,
+        rrd: Optional[float] = None,
+        min_delay: float = 0.1,
+        max_delay: float = 0.3,
     ):
         super().__init__(
             repo=repo,
             hasher=hasher,
             separator=separator,
             global_prefix=global_prefix,
             rrd=rrd,
+            min_delay=min_delay,
+            max_delay=max_delay,
         )
         self.cache_prefix = cache_prefix
         self.cache_ttl = cache_ttl
@@ -112,7 +116,7 @@ async def _verify_key(self, api_key: Optional[str] = None, required_scopes: Opti
 
         # Compute cache key from the full API key (secure: requires complete key)
         cache_key = _compute_cache_key(parsed.raw)
-        cached_entity: ApiKey = await self.cache.get(cache_key)
+        cached_entity: Optional[ApiKey] = await self.cache.get(cache_key)
 
         if cached_entity:
             # Cache hit: the full API key is correct (hash matched)

tests/unit/test_api.py

@@ -36,7 +36,8 @@ def service(repo: InMemoryApiKeyRepository) -> ApiKeyService:
     return ApiKeyService(
         repo=repo,
         hasher=MockApiKeyHasher(pepper="test-pepper"),
-        rrd=0,  # No random delay for tests
+        min_delay=0,
+        max_delay=0,
     )
 
 

tests/unit/test_cached_service.py

@@ -38,7 +38,8 @@ def service(mock_cache: AsyncMock) -> CachedApiKeyService:
         hasher=MockApiKeyHasher(pepper="test-pepper"),
         separator="-",
         global_prefix="ak",
-        rrd=0,
+        min_delay=0,
+        max_delay=0,
     )
 
 
@@ -268,7 +269,8 @@ async def test_cache_set_uses_default_ttl(
             repo=InMemoryApiKeyRepository(),
             cache=mock_cache,
             hasher=MockApiKeyHasher(pepper="test"),
-            rrd=0,
+            min_delay=0,
+            max_delay=0,
         )
         entity, api_key = await service.create(name="test")
         mock_cache.get.return_value = None  # Cache miss
@@ -290,7 +292,8 @@ async def test_cache_set_uses_custom_ttl(
             cache=mock_cache,
             cache_ttl=60,
             hasher=MockApiKeyHasher(pepper="test"),
-            rrd=0,
+            min_delay=0,
+            max_delay=0,
         )
         entity, api_key = await service.create(name="test")
         mock_cache.get.return_value = None
@@ -305,42 +308,7 @@ def test_default_ttl_is_300(self):
         service = CachedApiKeyService(
             repo=InMemoryApiKeyRepository(),
             hasher=MockApiKeyHasher(pepper="test"),
-            rrd=0,
+            min_delay=0,
+            max_delay=0,
         )
         assert service.cache_ttl == 300
-
-
-class TestDefaultCache:
-    """Tests for default cache behavior."""
-
-    @pytest.mark.asyncio
-    async def test_default_cache_is_simple_memory(self):
-        """Service uses SimpleMemoryCache by default."""
-        import aiocache
-
-        service = CachedApiKeyService(
-            repo=InMemoryApiKeyRepository(),
-            hasher=MockApiKeyHasher(pepper="test"),
-            rrd=0,
-        )
-
-        assert isinstance(service.cache, aiocache.SimpleMemoryCache)
-
-    @pytest.mark.asyncio
-    async def test_service_works_without_explicit_cache(self):
-        """Service works end-to-end with default cache."""
-        service = CachedApiKeyService(
-            repo=InMemoryApiKeyRepository(),
-            hasher=MockApiKeyHasher(pepper="test"),
-            rrd=0,
-        )
-
-        entity, api_key = await service.create(name="test")
-
-        # First verify (cache miss)
-        result = await service.verify_key(api_key)
-        assert result.id_ == entity.id_
-
-        # Second verify (cache hit)
-        result = await service.verify_key(api_key)
-        assert result.id_ == entity.id_

tests/unit/test_cli.py

@@ -36,7 +36,8 @@ def service(repo: InMemoryApiKeyRepository) -> ApiKeyService:
     return ApiKeyService(
         repo=repo,
         hasher=MockApiKeyHasher(pepper="test-pepper"),
-        rrd=0,  # No random delay for tests
+        min_delay=0,
+        max_delay=0,
     )
 
 

tests/unit/test_service.py

@@ -35,7 +35,8 @@ def service() -> ApiKeyService:
         hasher=MockApiKeyHasher(pepper="test-pepper"),
         separator=".",
         global_prefix="ak",
-        rrd=0,  # No delay for tests
+        min_delay=0,
+        max_delay=0,
     )
 
 
@@ -310,15 +311,16 @@ async def test_verify_no_required_scopes(self, service: ApiKeyService):
 
 
 class TestServiceTimingAttackMitigation:
-    """Tests for RRD (random response delay) timing attack mitigation."""
+    """Tests for response delay timing attack mitigation."""
 
     @pytest.mark.asyncio
-    async def test_rrd_adds_delay_on_error(self):
+    async def test_delay_applies_on_error(self):
         """verify_key() adds random delay on verification failure."""
         service = ApiKeyService(
             repo=InMemoryApiKeyRepository(),
             hasher=MockApiKeyHasher(pepper="test"),
-            rrd=0.1,  # 100ms base delay
+            min_delay=0.1,
+            max_delay=0.3,
         )
 
         with patch("asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
@@ -328,7 +330,35 @@ async def test_rrd_adds_delay_on_error(self):
             mock_sleep.assert_awaited_once()
             assert mock_sleep.await_args, "Expected sleep to be called"
             delay = mock_sleep.await_args.args[0]
-            assert 0.1 <= delay <= 0.2  # Between rrd and rrd*2
+            assert 0.1 <= delay <= 0.3
+
+    @pytest.mark.asyncio
+    async def test_delay_applies_on_success(self):
+        """verify_key() adds random delay on successful verification."""
+        service = ApiKeyService(
+            repo=InMemoryApiKeyRepository(),
+            hasher=MockApiKeyHasher(pepper="test"),
+            min_delay=0.1,
+            max_delay=0.3,
+        )
+        _, full_key = await service.create(name="ok")
+
+        with patch("asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
+            await service.verify_key(full_key)
+
+            mock_sleep.assert_awaited_once()
+            assert mock_sleep.await_args, "Expected sleep to be called"
+            delay = mock_sleep.await_args.args[0]
+            assert 0.1 <= delay <= 0.3
+
+    def test_rrd_warns_and_is_ignored(self):
+        """rrd emits a deprecation warning when provided."""
+        with pytest.warns(DeprecationWarning, match="rrd is deprecated"):
+            ApiKeyService(
+                repo=InMemoryApiKeyRepository(),
+                hasher=MockApiKeyHasher(pepper="test"),
+                rrd=0.1,
+            )
 
 
 class TestServiceConstructor:
@@ -351,10 +381,33 @@ def test_custom_prefix_and_separator(self):
             hasher=MockApiKeyHasher(pepper="test"),
             separator=":",
             global_prefix="KEY",
+            min_delay=0,
+            max_delay=0,
         )
+
         assert service.separator == ":"
         assert service.global_prefix == "KEY"
 
+    def test_negative_delay_raises(self):
+        """Constructor rejects negative delay values."""
+        with pytest.raises(ValueError, match="non-negative"):
+            ApiKeyService(
+                repo=InMemoryApiKeyRepository(),
+                hasher=MockApiKeyHasher(pepper="test"),
+                min_delay=-0.1,
+                max_delay=0.1,
+            )
+
+    def test_max_delay_less_than_min_raises(self):
+        """Constructor rejects max_delay below min_delay."""
+        with pytest.raises(ValueError, match="greater than or equal"):
+            ApiKeyService(
+                repo=InMemoryApiKeyRepository(),
+                hasher=MockApiKeyHasher(pepper="test"),
+                min_delay=0.2,
+                max_delay=0.1,
+            )
+
 
 class TestServiceListFindCount:
     """Tests for list(), find(), count() methods."""
@@ -364,7 +417,8 @@ def service(self) -> ApiKeyService:
         return ApiKeyService(
             repo=InMemoryApiKeyRepository(),
             hasher=MockApiKeyHasher(pepper="test"),
-            rrd=0,
+            min_delay=0,
+            max_delay=0,
         )
 
     @pytest.mark.asyncio
@@ -419,7 +473,8 @@ async def test_load_dotenv_creates_keys(self, monkeypatch: pytest.MonkeyPatch):
             hasher=MockApiKeyHasher(pepper="test"),
             separator="-",
             global_prefix="ak",
-            rrd=0,
+            min_delay=0,
+            max_delay=0,
         )
 
         # Set environment variables
@@ -442,7 +497,8 @@ async def test_load_dotenv_no_keys_raises(self, monkeypatch: pytest.MonkeyPatch)
         service = ApiKeyService(
             repo=InMemoryApiKeyRepository(),
             hasher=MockApiKeyHasher(pepper="test"),
-            rrd=0,
+            min_delay=0,
+            max_delay=0,
         )
 
         # Clear any existing API_KEY_ vars
@@ -461,7 +517,8 @@ async def test_load_dotenv_custom_prefix(self, monkeypatch: pytest.MonkeyPatch):
             hasher=MockApiKeyHasher(pepper="test"),
             separator="-",
             global_prefix="ak",
-            rrd=0,
+            min_delay=0,
+            max_delay=0,
         )
 
         monkeypatch.setenv("MYAPP_KEY_TEST", "ak-abc123def456ghij-secretsecretsecretsecretsecretsecretsecretsecret12")
@@ -483,7 +540,8 @@ def service(self) -> ApiKeyService:
             hasher=MockApiKeyHasher(pepper="test"),
             separator=".",
             global_prefix="ak",
-            rrd=0,
+            min_delay=0,
+            max_delay=0,
         )
 
     @pytest.mark.asyncio