respond to pr comments. adds a blacklist class and exports string equivalency in a util function

Author: AlexSCorey

src/aap_eda/api/blacklist.py

@@ -0,0 +1,98 @@
+#  Copyright 2026 Red Hat, Inc.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+import logging
+
+from django.conf import settings
+from django.core.cache import cache
+from rest_framework.exceptions import AuthenticationFailed
+
+logger = logging.getLogger(__name__)
+
+
+class BlacklistManager:
+    """Rate-limit and blacklist IPs that fail event stream authentication.
+
+    Tracks auth failures and invalid UUID probes per client IP using
+    Django's cache framework. All entries are evicted automatically
+    by cache TTL — no manual cleanup required.
+    """
+
+    AUTH_FAILURE_PREFIX = "es_auth_fail"
+    INVALID_UUID_PREFIX = "es_invalid_uuid"
+    BLACKLIST_PREFIX = "es_blacklist"
+
+    def check_blacklist(self, client_ip: str) -> None:
+        """Raise AuthenticationFailed if the IP is globally blacklisted."""
+        key = f"{self.BLACKLIST_PREFIX}:global:{client_ip}"
+        if cache.get(key):
+            raise AuthenticationFailed("Too many failed attempts")
+
+    def record_auth_failure(self, client_ip: str) -> None:
+        """Record an authentication failure for the given IP.
+
+        After EVENT_STREAM_AUTH_FAILURE_THRESHOLD failures within
+        EVENT_STREAM_AUTH_FAILURE_WINDOW seconds, the IP is globally
+        blacklisted for EVENT_STREAM_BLACKLIST_DURATION seconds.
+        """
+        counter_key = f"{self.AUTH_FAILURE_PREFIX}:{client_ip}"
+        failures = cache.get(counter_key, 0) + 1
+        cache.set(
+            counter_key,
+            failures,
+            settings.EVENT_STREAM_AUTH_FAILURE_WINDOW,
+        )
+
+        if failures >= settings.EVENT_STREAM_AUTH_FAILURE_THRESHOLD:
+            blacklist_key = f"{self.BLACKLIST_PREFIX}:global:{client_ip}"
+            cache.set(
+                blacklist_key,
+                True,
+                settings.EVENT_STREAM_BLACKLIST_DURATION,
+            )
+            logger.warning(
+                "Globally blacklisted IP %s after %d auth failures",
+                client_ip,
+                failures,
+            )
+            cache.delete(counter_key)
+
+    def record_invalid_uuid(self, client_ip: str) -> None:
+        """Record an invalid event stream UUID attempt for the given IP.
+
+        After EVENT_STREAM_INVALID_UUID_THRESHOLD attempts within
+        EVENT_STREAM_INVALID_UUID_WINDOW seconds, the IP is globally
+        blacklisted for EVENT_STREAM_BLACKLIST_DURATION seconds.
+        """
+        counter_key = f"{self.INVALID_UUID_PREFIX}:{client_ip}"
+        failures = cache.get(counter_key, 0) + 1
+        cache.set(
+            counter_key,
+            failures,
+            settings.EVENT_STREAM_INVALID_UUID_WINDOW,
+        )
+
+        if failures >= settings.EVENT_STREAM_INVALID_UUID_THRESHOLD:
+            blacklist_key = f"{self.BLACKLIST_PREFIX}:global:{client_ip}"
+            cache.set(
+                blacklist_key,
+                True,
+                settings.EVENT_STREAM_BLACKLIST_DURATION,
+            )
+            logger.warning(
+                "Globally blacklisted IP %s after %d invalid UUID attempts",
+                client_ip,
+                failures,
+            )
+            cache.delete(counter_key)

src/aap_eda/api/event_stream_authentication.py

@@ -33,6 +33,7 @@
 
 from aap_eda.core.enums import SignatureEncodingType
 from aap_eda.core.utils.credentials import validate_x509_subject_match
+from aap_eda.core.utils.crypto import timing_safe_compare
 
 logger = logging.getLogger(__name__)
 DEFAULT_TIMEOUT = 30
@@ -81,9 +82,7 @@ def authenticate(self, body: bytes):
             logger.warning(message)
             raise AuthenticationFailed(message)
 
-        if not hmac.compare_digest(
-            expected_signature.encode(), self.signature.encode()
-        ):
+        if not timing_safe_compare(expected_signature, self.signature):
             message = "Signature mismatch, check your payload and secret"
             logger.warning(message)
             raise AuthenticationFailed(message)
@@ -98,9 +97,7 @@ class TokenAuthentication(EventStreamAuthentication):
 
     def authenticate(self, _body=None):
         """Handle Token authentication."""
-        if not hmac.compare_digest(
-            self.token.encode(), _token_sans_bearer(self.value).encode()
-        ):
+        if not timing_safe_compare(self.token, _token_sans_bearer(self.value)):
             message = "Token mismatch, check your token"
             logger.warning(message)
             raise AuthenticationFailed(message)
@@ -156,9 +153,9 @@ def authenticate(self, _body=None):
         if self.authorization.startswith("Basic"):
             auth_str = self.authorization.split("Basic ")[1]
 
-        user_pass = f"{self.username}:{self.password}"
+        user_pass = f"{self.username}: {self.password}"
         b64_value = base64.b64encode(user_pass.encode()).decode()
-        if not hmac.compare_digest(auth_str.encode(), b64_value.encode()):
+        if not timing_safe_compare(auth_str, b64_value):
             message = "Credential mismatch"
             logger.warning(message)
             raise AuthenticationFailed(message)

src/aap_eda/api/views/external_event_stream.py

@@ -23,7 +23,6 @@
     validate_x_trusted_proxy_header,
 )
 from django.conf import settings
-from django.core.cache import cache
 from django.core.exceptions import ValidationError
 from django.db import transaction
 from django.db.models import F
@@ -35,6 +34,7 @@
 from rest_framework.permissions import AllowAny
 from rest_framework.response import Response
 
+from aap_eda.api.blacklist import BlacklistManager
 from aap_eda.api.event_stream_authentication import (
     BasicAuthentication,
     EcdsaAuthentication,
@@ -50,11 +50,10 @@
 from aap_eda.core.utils.credentials import get_resolved_secrets
 from aap_eda.services.pg_notify import PGNotify
 
-FAILURE_THRESHOLD = 5
-FAILURE_WINDOW = 60  # seconds
 logger = logging.getLogger(__name__)
 UNSAFE_HEADER_KEYS = {"X-Trusted-Proxy", "X-Forwarded-For", "X-Real-IP"}
 REDACTED_STRING = "********"
+blacklist_manager = BlacklistManager()
 
 
 class ExternalEventStreamViewSet(viewsets.GenericViewSet):
@@ -286,41 +285,32 @@ def _handle_auth(self, request, inputs):
             raise
 
     def _get_client_ip(self, request):
+        """Return the client IP from the request.
+
+        Uses X-Forwarded-For when trusted proxy validation is enabled,
+        otherwise falls back to REMOTE_ADDR.
+        """
         if settings.EVENT_STREAM_REQUIRE_TRUSTED_PROXY:
             x_forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR")
             if x_forwarded_for:
                 return x_forwarded_for.split(",")[0].strip()
         return request.META.get("REMOTE_ADDR")
 
-    def _check_rate_limit(self, request, event_stream_uuid):
-        key = (
-            f"es_auth_fail: {event_stream_uuid}: "
-            f"{self._get_client_ip(request)}"
-        )
-        failures = cache.get(key, 0)
-        if failures >= FAILURE_THRESHOLD:
-            raise AuthenticationFailed("Too many failed attempts")
-
-    def _record_failure(self, request, event_stream_uuid):
-        key = (
-            f"es_auth_fail: {event_stream_uuid}: "
-            f"{self._get_client_ip(request)}"
-        )
-        failures = cache.get(key, 0)
-        cache.set(key, failures + 1, FAILURE_WINDOW)
-
     @extend_schema(exclude=True)
     @action(detail=True, methods=["POST"], rbac_action=None)
     def post(self, request, *_args, **kwargs):
         """Handle posts from external vendors."""
+        client_ip = self._get_client_ip(request)
+        blacklist_manager.check_blacklist(client_ip)
+
         try:
             self.event_stream = EventStream.objects.get(uuid=kwargs["pk"])
         except (EventStream.DoesNotExist, ValidationError) as exc:
+            blacklist_manager.record_invalid_uuid(client_ip)
             raise ParseError("bad uuid specified") from exc
 
         # Validate X-Trusted-Proxy header from Gateway/Envoy
         self._validate_trusted_proxy_header(request)
-        self._check_rate_limit(request, kwargs["pk"])
 
         try:
             inputs = get_resolved_secrets(self.event_stream.eda_credential)
@@ -341,10 +331,11 @@ def post(self, request, *_args, **kwargs):
                     headers=yaml.dump(event_headers),
                 )
             raise ParseError(message)
+
         try:
             self._handle_auth(request, inputs)
         except AuthenticationFailed:
-            self._record_failure(request, kwargs["pk"])
+            blacklist_manager.record_auth_failure(client_ip)
             raise
 
         body = self._parse_body(

src/aap_eda/core/utils/crypto/__init__.py

@@ -11,3 +11,18 @@
 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
+
+import hmac
+
+
+def timing_safe_compare(a: str, b: str) -> bool:
+    """Compare two strings in constant time using hmac.compare_digest.
+
+    Python's hmac.compare_digest raises TypeError when either str
+    operand contains non-ASCII characters — not just on a str-vs-bytes
+    mismatch. Since Django provides HTTP header values as str with no
+    ASCII guarantee, callers comparing credentials from headers must
+    encode to bytes first. This function handles that conversion so
+    callers don't need to remember the footgun.
+    """
+    return hmac.compare_digest(a.encode(), b.encode())

src/aap_eda/settings/defaults.py

@@ -238,6 +238,17 @@
 # Set to False for local development without proxy:
 #   export EDA_EVENT_STREAM_REQUIRE_TRUSTED_PROXY=False
 EVENT_STREAM_REQUIRE_TRUSTED_PROXY: bool = True
+
+# IP blacklisting for event stream abuse prevention
+# Auth failure: threshold and window before globally blacklisting an IP
+EVENT_STREAM_AUTH_FAILURE_THRESHOLD: int = 5
+EVENT_STREAM_AUTH_FAILURE_WINDOW: int = 60  # seconds
+# Invalid UUID probing: threshold and window before blacklisting an IP globally
+EVENT_STREAM_INVALID_UUID_THRESHOLD: int = 3
+EVENT_STREAM_INVALID_UUID_WINDOW: int = 60  # seconds
+# Duration an IP remains blacklisted after exceeding a threshold
+EVENT_STREAM_BLACKLIST_DURATION: int = 3600  # seconds (1 hour)
+
 MAX_PG_NOTIFY_MESSAGE_SIZE: int = 6144
 
 # Database credentials for the event streams user