Improvements

Author: RobertoPrevato

CHANGELOG.md

@@ -12,20 +12,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   authentication attempts consistently.
 - Add an `InvalidCredentialsError` exception. `AuthenticationHandler` implementations
   can raise `InvalidCredentialsError` when invalid credentials are provided, to
-  enable automatic logging and, if desired, brute-force protection.
-- Integrate `RateLimiter` into `AuthenticationStrategy` with automatic tracking of
-  failed authentication attempts and support for blocking excessive requests.
-- Add `RateLimiter` class that cam block authentication attempts after a configurable
+  enable automatic logging and, if enabled, brute-force protection.
+- Add `RateLimiter` class that can block authentication attempts after a configurable
   threshold is exceeded. By default stores failed attempts in-memory. This feature is
   disabled unless a `key_getter` function is provided (a function to obtain a context
   key from the user-defined authentication context, like a client IP).
+- Integrate `RateLimiter` into `AuthenticationStrategy` with automatic tracking of
+  failed authentication attempts and support for blocking excessive requests.
 - Add Python `3.14` and remove `3.9` from the build matrix.
-- Improve exceptions raised for invalid `JWTs` to include the source exception
-  (`exc.__cause__`).
 - Drop support for Python `3.9` (it reached EOL in October 2025).
 - Add an optional dependency on `essentials`, to use its `Secret` class to handle
   secrets for JWT validation with symmetric encryption. This is useful to support
   rotating secrets by updating env variables.
+- Improve exceptions raised for invalid `JWTs` to include the source exception
+  (`exc.__cause__`).
 
 ## [1.0.3] - 2025-10-04 :trident:
 

guardpost/authentication.py

@@ -8,7 +8,6 @@
 from rodi import ContainerProtocol
 
 from guardpost.abc import BaseStrategy
-from guardpost.errors import RateLimitExceededError
 from guardpost.protection import InvalidCredentialsError, RateLimiter
 
 
@@ -115,10 +114,23 @@ def __init__(
         rate_limiter: Optional[RateLimiter] = None,
         logger: Optional[Logger] = None,
     ):
+        """
+        Initializes an AuthenticationStrategy instance.
+
+        Args:
+            *handlers: One or more authentication handler instances or types to be used
+                for authentication.
+            container: Optional dependency injection container for resolving handler
+                instances.
+            rate_limiter: Optional RateLimiter to apply rate limiting to authentication
+                attempts.
+            logger: Optional logger instance for logging authentication events. If not
+                provided, defaults to `logging.getLogger("guardpost")`
+        """
         super().__init__(container)
         self.handlers = list(handlers)
-        self._rate_limiter = rate_limiter or RateLimiter()
         self._logger = logger or logging.getLogger("guardpost")
+        self._rate_limiter = rate_limiter
 
     def add(self, handler: AuthenticationHandlerConfType) -> "AuthenticationStrategy":
         self.handlers.append(handler)
@@ -159,15 +171,14 @@ async def authenticate(
         self, context: Any, authentication_schemes: Optional[Sequence[str]] = None
     ) -> Optional[Identity]:
         """
-        Tries to obtain the user for a context, applying authentication rules.
+        Tries to obtain the user for a context, applying authentication rules and
+        optional rate limiting.
         """
         if not context:
             raise ValueError("Missing context to evaluate authentication")
 
-        valid_context = await self._rate_limiter.allow_authentication_attempt(context)
-
-        if not valid_context:
-            raise RateLimitExceededError()
+        if self._rate_limiter:
+            await self._rate_limiter.validate_authentication_attempt(context)
 
         identity = None
         for handler in self._get_handlers_by_schemes(authentication_schemes, context):
@@ -182,9 +193,10 @@ async def authenticate(
                     invalid_credentials_error.client_ip,
                     handler.scheme,
                 )
-                await self._rate_limiter.store_authentication_failure(
-                    invalid_credentials_error
-                )
+                if self._rate_limiter:
+                    await self._rate_limiter.store_authentication_failure(
+                        invalid_credentials_error
+                    )
 
             if identity:
                 try:

guardpost/jwts/__init__.py

@@ -2,9 +2,9 @@
 from typing import Any, Dict, List, Optional, Protocol, Sequence, Union
 
 import jwt
-from jwt.exceptions import InvalidIssuerError, InvalidTokenError
-
 from essentials.secrets import Secret
+from jwt.exceptions import InvalidTokenError
+
 from guardpost.errors import AuthException
 
 from ..jwks import JWK, JWKS, KeysProvider

guardpost/protection.py

@@ -2,11 +2,13 @@
 This module provides classes to protect against brute-force attacks.
 """
 
+import logging
 from abc import ABC, abstractmethod
 from datetime import UTC, datetime
+from logging import Logger
 from typing import Any, Callable, Optional, Sequence
 
-from guardpost.errors import InvalidCredentialsError
+from guardpost.errors import InvalidCredentialsError, RateLimitExceededError
 
 
 class FailedAuthenticationAttempts:
@@ -140,6 +142,7 @@ def __init__(
         block_time: int = 60,
         store: Optional[AuthenticationAttemptsStore] = None,
         trusted_keys: Optional[Sequence[str]] = None,
+        logger: Optional[Logger] = None,
     ) -> None:
         """
         Initialize a RateLimiter instance for brute-force protection.
@@ -171,25 +174,50 @@ def __init__(
             max_entry_age=self._block_time + 5
         )
         self._key_getter = key_getter
+        self._logger = logger or logging.getLogger("guardpost")
 
     def get_context_key(self, context: Any) -> str:
         """
-        Extracts the rate limiting key from the authentication context.
-
-        Raises an AuthException if no key_getter is provided. Custom extractors
-        should be used carefully as they may introduce security vulnerabilities.
+        Obtains the rate limiting key from the authentication context.
+        If a key_getter is not configured, rate limiting is disabled and
+        returns an empty string.
         """
         if not self._key_getter:
             return ""
         return self._key_getter(context)
 
-    async def allow_authentication_attempt(self, context: Any) -> bool:
+    async def validate_authentication_attempt(self, context: Any) -> None:
+        """
+        Checks if an authentication attempt should be allowed for the given context.
+        It always returns true if a `key_getter` function is not configured (function to
+        obtain a key from the user-defined authentication context).
+
+        Args:
+            context: The authentication context (e.g., request object).
+
+        Raises:
+            **RateLimitExceededError**: If the rate limit has been exceeded and the
+            attempt is blocked.
+        """
+        key = self.get_context_key(context)
+
+        valid_context = await self.allow_authentication_attempt(key)
+
+        if not valid_context:
+            self._logger.warning(
+                "Blocking authentication attempt for '%s' because of rate limiting.",
+                key,
+            )
+            raise RateLimitExceededError()
+
+    async def allow_authentication_attempt(self, key: Any) -> bool:
         """
         Determines if an authentication attempt should be allowed based on rate limiting
         rules. Returns True if the attempt should proceed, False if it should be
-        blocked.
+        blocked. Key can be a str or a context.
         """
-        key = self.get_context_key(context)
+        if not isinstance(key, str):
+            key = self.get_context_key(key)
 
         if not key:
             # BF protection disabled for backward compatibility.
@@ -212,7 +240,9 @@ async def allow_authentication_attempt(self, context: Any) -> bool:
 
         return True
 
-    async def store_authentication_failure(self, error: InvalidCredentialsError):
+    async def store_authentication_failure(
+        self, error: InvalidCredentialsError
+    ) -> FailedAuthenticationAttempts:
         """
         Tracks information about a failed authentication attempt.
         """
@@ -225,6 +255,7 @@ async def store_authentication_failure(self, error: InvalidCredentialsError):
             else:
                 failed_attempt.increase_counter()
         await self._store.set_failed_attempts(failed_attempt)
+        return failed_attempt
 
 
 class SelfCleaningInMemoryAuthenticationAttemptsStore(

tests/test_protection.py

@@ -1,4 +1,3 @@
-import warnings
 from datetime import UTC, datetime, timedelta
 from unittest.mock import AsyncMock