fix: harden event stream auth with per-org IP management (AAP-76184)

src/aap_eda/api/blacklist.py

@@ -0,0 +1,128 @@
+#  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 ipaddress
+import logging
+
+from rest_framework.exceptions import AuthenticationFailed
+
+logger = logging.getLogger(__name__)
+
+MAX_BLOCKED_IPS = 1000
+
+
+def normalize_ip(ip_str: str) -> str:
+    """Normalize an IP address string.
+
+    Converts IPv4-mapped IPv6 addresses (e.g. ::ffff:10.0.0.1)
+    to their IPv4 form so allowlist lookups match regardless of
+    how the proxy reports the client IP.
+    """
+    try:
+        addr = ipaddress.ip_address(ip_str.strip())
+    except ValueError:
+        return ip_str.strip()
+    if isinstance(addr, ipaddress.IPv6Address) and addr.ipv4_mapped:
+        return str(addr.ipv4_mapped)
+    return str(addr)
+
+
+def ip_in_allowlist(client_ip: str, allowed_ips: set) -> bool:
+    """Check if an IP matches any entry in the allowlist.
+
+    Supports both individual IPs and CIDR ranges.
+    """
+    if client_ip in allowed_ips:
+        return True
+    try:
+        addr = ipaddress.ip_address(client_ip)
+    except ValueError:
+        return False
+    for entry in allowed_ips:
+        if "/" in entry:
+            try:
+                if addr in ipaddress.ip_network(entry, strict=False):
+                    return True
+            except ValueError:
+                continue
+    return False
+
+
+class BlacklistManager:
+    """Allowlist-based IP policy for event streams.
+
+    When an organization has configured allowed_ips, only those
+    IPs may post events. IPs that are rejected are recorded in
+    blocked_ips for admin visibility — admins can then promote
+    a blocked IP to the allowlist or remove it entirely.
+
+    When no EventStreamSetting row exists for the org, or the
+    allowed_ips list is empty, all IPs are permitted.
+    """
+
+    def check_ip_policy(self, client_ip: str, org_id: int) -> None:
+        """Check if the IP is allowed for this organization.
+
+        Raises AuthenticationFailed if the org has a non-empty
+        allowlist and the IP is not in it.
+        """
+        from aap_eda.services.event_stream_settings_cache import (
+            get_org_settings,
+        )
+
+        org_settings = get_org_settings(org_id)
+        if org_settings is None:
+            return
+
+        normalized = normalize_ip(client_ip)
+
+        if org_settings["allowed_ips"] and not ip_in_allowlist(
+            normalized, org_settings["allowed_ips"]
+        ):
+            self.record_blocked_ip(normalized, org_id)
+            raise AuthenticationFailed("IP address not in allowlist")
+
+    def record_blocked_ip(self, client_ip: str, org_id: int) -> None:
+        """Add an IP to the org's blocked_ips list for visibility.
+
+        Only adds the IP if it is not already tracked. Caps the
+        list at MAX_BLOCKED_IPS to prevent unbounded growth.
+        """
+        from aap_eda.core.models import EventStreamSetting
+
+        normalized = normalize_ip(client_ip)
+
+        try:
+            setting = EventStreamSetting.objects.get(organization_id=org_id)
+        except EventStreamSetting.DoesNotExist:
+            return
+
+        if normalized in setting.blocked_ips:
+            return
+
+        if len(setting.blocked_ips) >= MAX_BLOCKED_IPS:
+            logger.warning(
+                "Blocked IPs cap (%d) reached for org %s",
+                MAX_BLOCKED_IPS,
+                org_id,
+            )
+            return
+
+        setting.blocked_ips = [*setting.blocked_ips, normalized]
+        setting.save(update_fields=["blocked_ips", "modified_at"])
+        logger.info(
+            "Recorded blocked IP %s for org %s",
+            normalized,
+            org_id,
+        )

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,7 +82,7 @@ def authenticate(self, body: bytes):
             logger.warning(message)
             raise AuthenticationFailed(message)
 
-        if not hmac.compare_digest(expected_signature, self.signature):
+        if not timing_safe_compare(expected_signature, self.signature):
             message = "Signature mismatch, check your payload and secret"
             logger.warning(message)
             raise AuthenticationFailed(message)
@@ -96,7 +97,7 @@ class TokenAuthentication(EventStreamAuthentication):
 
     def authenticate(self, _body=None):
         """Handle Token authentication."""
-        if self.token != _token_sans_bearer(self.value):
+        if not timing_safe_compare(self.token, _token_sans_bearer(self.value)):
             message = "Token mismatch, check your token"
             logger.warning(message)
             raise AuthenticationFailed(message)
@@ -152,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 = "%s:%s" % (self.username, self.password)
         b64_value = base64.b64encode(user_pass.encode()).decode()
-        if auth_str != b64_value:
+        if not timing_safe_compare(auth_str, b64_value):
             message = "Credential mismatch"
             logger.warning(message)
             raise AuthenticationFailed(message)

src/aap_eda/api/filters/__init__.py

@@ -22,6 +22,7 @@
 from .decision_environment import DecisionEnvironmentFilter
 from .eda_credential import EdaCredentialFilter
 from .event_stream import EventStreamFilter
+from .event_stream_setting import EventStreamSettingFilter
 from .organization import OrganizationFilter
 from .project import ProjectFilter
 from .rulebook import RulebookFilter
@@ -52,4 +53,6 @@
     "OrganizationTeamFilter",
     # EventStream
     "EventStreamFilter",
+    # EventStreamSetting
+    "EventStreamSettingFilter",
 )

src/aap_eda/api/filters/event_stream_setting.py

@@ -0,0 +1,23 @@
+#  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 django_filters
+
+from aap_eda.core import models
+
+
+class EventStreamSettingFilter(django_filters.FilterSet):
+    class Meta:
+        model = models.EventStreamSetting
+        fields = ["organization"]

src/aap_eda/api/serializers/__init__.py

@@ -52,6 +52,11 @@
     EdaCredentialUpdateSerializer,
 )
 from .event_stream import EventStreamInSerializer, EventStreamOutSerializer
+from .event_stream_setting import (
+    EventStreamSettingCreateSerializer,
+    EventStreamSettingOutSerializer,
+    RemoveBlockedIpsSerializer,
+)
 from .organization import (
     OrganizationCreateSerializer,
     OrganizationRefSerializer,
@@ -155,4 +160,8 @@
     # event streams
     "EventStreamInSerializer",
     "EventStreamOutSerializer",
+    # event stream settings
+    "EventStreamSettingCreateSerializer",
+    "EventStreamSettingOutSerializer",
+    "RemoveBlockedIpsSerializer",
 )

src/aap_eda/api/serializers/event_stream_setting.py

@@ -0,0 +1,158 @@
+#  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 ipaddress
+
+from rest_framework import serializers
+
+from aap_eda.api.serializers.fields.basic_user import BasicUserFieldSerializer
+from aap_eda.api.serializers.organization import OrganizationRefSerializer
+from aap_eda.api.serializers.user import BasicUserSerializer
+from aap_eda.core import models, validators
+
+MAX_IPS_PER_LIST = 255
+
+
+def _validate_ip_list(value):
+    if len(value) > MAX_IPS_PER_LIST:
+        raise serializers.ValidationError(
+            f"Maximum {MAX_IPS_PER_LIST} entries allowed."
+        )
+    from aap_eda.api.blacklist import normalize_ip
+
+    normalized = []
+    for entry in value:
+        entry = entry.strip()
+        if "/" in entry:
+            try:
+                net = ipaddress.ip_network(entry, strict=False)
+                normalized.append(str(net))
+            except ValueError:
+                raise serializers.ValidationError(
+                    f"'{entry}' is not a valid CIDR range."
+                )
+        else:
+            try:
+                ipaddress.ip_address(entry)
+            except ValueError:
+                raise serializers.ValidationError(
+                    f"'{entry}' is not a valid IP address."
+                )
+            normalized.append(normalize_ip(entry))
+    return normalized
+
+
+class EventStreamSettingCreateSerializer(
+    serializers.ModelSerializer,
+):
+    organization_id = serializers.IntegerField(
+        required=True,
+        allow_null=False,
+        validators=[validators.check_if_organization_exists],
+        error_messages={"null": "Organization is required."},
+    )
+    allowed_ips = serializers.ListField(
+        child=serializers.CharField(max_length=45),
+        required=False,
+        default=list,
+    )
+    blocked_ips = serializers.ListField(
+        child=serializers.CharField(max_length=45),
+        required=False,
+        default=list,
+    )
+
+    class Meta:
+        model = models.EventStreamSetting
+        fields = [
+            "organization_id",
+            "allowed_ips",
+            "blocked_ips",
+        ]
+
+    def validate_organization_id(self, value):
+        if (
+            not self.instance
+            and models.EventStreamSetting.objects.filter(
+                organization_id=value
+            ).exists()
+        ):
+            raise serializers.ValidationError(
+                "Settings already exist for this organization."
+            )
+        return value
+
+    def validate_allowed_ips(self, value):
+        return _validate_ip_list(value)
+
+    def validate_blocked_ips(self, value):
+        return _validate_ip_list(value)
+
+    def update(self, instance, validated_data):
+        new_allowed = validated_data.get("allowed_ips")
+        if new_allowed is not None:
+            added_ips = set(new_allowed) - set(instance.allowed_ips)
+            if added_ips and instance.blocked_ips:
+                instance.blocked_ips = [
+                    ip for ip in instance.blocked_ips if ip not in added_ips
+                ]
+        return super().update(instance, validated_data)
+
+
+class EventStreamSettingOutSerializer(
+    serializers.ModelSerializer,
+):
+    organization = serializers.SerializerMethodField()
+    created_by = BasicUserFieldSerializer()
+    modified_by = BasicUserFieldSerializer()
+
+    class Meta:
+        model = models.EventStreamSetting
+        read_only_fields = [
+            "id",
+            "created_at",
+            "modified_at",
+        ]
+        fields = [
+            "organization",
+            "allowed_ips",
+            "blocked_ips",
+            "created_by",
+            "modified_by",
+            *read_only_fields,
+        ]
+
+    def get_organization(self, obj):
+        return (
+            OrganizationRefSerializer(obj.organization).data
+            if obj.organization
+            else None
+        )
+
+    def to_representation(self, instance):
+        result = super().to_representation(instance)
+        result["created_by"] = BasicUserSerializer(instance.created_by).data
+        result["modified_by"] = BasicUserSerializer(instance.modified_by).data
+        return result
+
+
+class RemoveBlockedIpsSerializer(serializers.Serializer):
+    ips = serializers.ListField(
+        child=serializers.CharField(max_length=45),
+        required=True,
+        help_text="IPs to remove from the blocked list.",
+    )
+
+    def validate_ips(self, value):
+        return _validate_ip_list(value)

src/aap_eda/api/urls.py

@@ -56,6 +56,7 @@
 router.register("organizations", views.OrganizationViewSet)
 router.register("teams", views.TeamViewSet)
 router.register("event-streams", views.EventStreamViewSet)
+router.register("event-stream-settings", views.EventStreamSettingViewSet)
 router.register(
     "external_event_stream",
     views.ExternalEventStreamViewSet,