From 20f8e0543dd425e68bc936cb2f4183a2fb4f37da Mon Sep 17 00:00:00 2001 From: Alex Date: Mon, 10 Aug 2026 16:08:03 -0400 Subject: [PATCH 1/4] fix: harden event stream authentication (CTRL-006 AR-01, AR-02) Replace timing-vulnerable != comparisons with hmac.compare_digest() in TokenAuthentication and BasicAuthentication. Add per-stream per-IP rate limiting on failed authentication attempts to prevent brute-force credential recovery. --- .../api/event_stream_authentication.py | 10 ++++-- .../api/views/external_event_stream.py | 35 +++++++++++++++++-- 2 files changed, 40 insertions(+), 5 deletions(-) diff --git a/src/aap_eda/api/event_stream_authentication.py b/src/aap_eda/api/event_stream_authentication.py index a2e82bd84..1154d1326 100644 --- a/src/aap_eda/api/event_stream_authentication.py +++ b/src/aap_eda/api/event_stream_authentication.py @@ -81,7 +81,9 @@ def authenticate(self, body: bytes): logger.warning(message) raise AuthenticationFailed(message) - if not hmac.compare_digest(expected_signature, self.signature): + if not hmac.compare_digest( + expected_signature.encode(), self.signature.encode() + ): message = "Signature mismatch, check your payload and secret" logger.warning(message) raise AuthenticationFailed(message) @@ -96,7 +98,9 @@ class TokenAuthentication(EventStreamAuthentication): def authenticate(self, _body=None): """Handle Token authentication.""" - if self.token != _token_sans_bearer(self.value): + if not hmac.compare_digest( + self.token.encode(), _token_sans_bearer(self.value).encode() + ): message = "Token mismatch, check your token" logger.warning(message) raise AuthenticationFailed(message) @@ -154,7 +158,7 @@ def authenticate(self, _body=None): user_pass = f"{self.username}:{self.password}" b64_value = base64.b64encode(user_pass.encode()).decode() - if auth_str != b64_value: + if not hmac.compare_digest(auth_str.encode(), b64_value.encode()): message = "Credential mismatch" logger.warning(message) raise AuthenticationFailed(message) diff --git a/src/aap_eda/api/views/external_event_stream.py b/src/aap_eda/api/views/external_event_stream.py index 44b05a56a..8047d3562 100644 --- a/src/aap_eda/api/views/external_event_stream.py +++ b/src/aap_eda/api/views/external_event_stream.py @@ -23,6 +23,7 @@ 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 @@ -50,6 +51,8 @@ from aap_eda.services.pg_notify import PGNotify from aap_eda.utils.log_sanitizer import REDACTED_STRING +FAILURE_THRESHOLD = 5 +FAILURE_WINDOW = 60 # seconds logger = logging.getLogger(__name__) UNSAFE_HEADER_KEYS = {"X-Trusted-Proxy", "X-Forwarded-For", "X-Real-IP"} @@ -306,6 +309,30 @@ def _handle_auth(self, request, inputs): ) raise + def _get_client_ip(self, request): + 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): @@ -317,6 +344,7 @@ def post(self, request, *_args, **kwargs): # 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) @@ -337,8 +365,11 @@ def post(self, request, *_args, **kwargs): headers=yaml.dump(event_headers), ) raise ParseError(message) - - self._handle_auth(request, inputs) + try: + self._handle_auth(request, inputs) + except AuthenticationFailed: + self._record_failure(request, kwargs["pk"]) + raise body = self._parse_body( request.headers.get("Content-Type", ""), request.body From 91120db5e3f80fc95f02fb4878f85ea647b5da6a Mon Sep 17 00:00:00 2001 From: Alex Date: Thu, 13 Aug 2026 11:28:26 -0400 Subject: [PATCH 2/4] respond to pr comments. adds a blacklist class and exports string equivalency in a util function --- src/aap_eda/api/blacklist.py | 81 ++++++++++++++++++ .../api/event_stream_authentication.py | 13 ++- .../api/views/external_event_stream.py | 49 +++++------ src/aap_eda/core/utils/crypto/__init__.py | 15 ++++ src/aap_eda/settings/defaults.py | 12 +++ src/aap_eda/settings/testing_defaults.py | 14 +++ tests/integration/api/conftest.py | 10 +++ tests/unit/test_blacklist.py | 85 +++++++++++++++++++ tests/unit/test_timing_safe_compare.py | 39 +++++++++ 9 files changed, 283 insertions(+), 35 deletions(-) create mode 100644 src/aap_eda/api/blacklist.py create mode 100644 src/aap_eda/settings/testing_defaults.py create mode 100644 tests/unit/test_blacklist.py create mode 100644 tests/unit/test_timing_safe_compare.py diff --git a/src/aap_eda/api/blacklist.py b/src/aap_eda/api/blacklist.py new file mode 100644 index 000000000..6ca1aa087 --- /dev/null +++ b/src/aap_eda/api/blacklist.py @@ -0,0 +1,81 @@ +# 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. + """ + + FAILURE_PREFIX = "es_fail" + BLACKLIST_PREFIX = "es_blacklist" + + def check_blacklist(self, client_ip: str) -> None: + """Raise AuthenticationFailed if the IP is globally blacklisted. + + A threshold of 0 disables blacklist checking entirely. + """ + if settings.EVENT_STREAM_BLACKLIST_THRESHOLD == 0: + return + key = f"{self.BLACKLIST_PREFIX}:{client_ip}" + if cache.get(key): + raise AuthenticationFailed("Too many failed attempts") + + def record_failure(self, client_ip: str) -> None: + """Record a failed request from the given IP. + + All failure types (bad credentials, invalid UUIDs) count + toward the same threshold. After + EVENT_STREAM_BLACKLIST_THRESHOLD failures within + EVENT_STREAM_BLACKLIST_WINDOW seconds, the IP is globally + blacklisted for EVENT_STREAM_BLACKLIST_DURATION seconds. + A threshold of 0 disables blacklisting. + """ + if settings.EVENT_STREAM_BLACKLIST_THRESHOLD == 0: + return + counter_key = f"{self.FAILURE_PREFIX}:{client_ip}" + try: + failures = cache.incr(counter_key) + except ValueError: + cache.set( + counter_key, + 1, + settings.EVENT_STREAM_BLACKLIST_WINDOW, + ) + failures = 1 + + if failures >= settings.EVENT_STREAM_BLACKLIST_THRESHOLD: + blacklist_key = f"{self.BLACKLIST_PREFIX}:{client_ip}" + cache.set( + blacklist_key, + True, + settings.EVENT_STREAM_BLACKLIST_DURATION, + ) + logger.warning( + "Globally blacklisted IP %s after %d failures", + client_ip, + failures, + ) + cache.delete(counter_key) diff --git a/src/aap_eda/api/event_stream_authentication.py b/src/aap_eda/api/event_stream_authentication.py index 1154d1326..ec9da7fc0 100644 --- a/src/aap_eda/api/event_stream_authentication.py +++ b/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}" # noqa: E231 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) diff --git a/src/aap_eda/api/views/external_event_stream.py b/src/aap_eda/api/views/external_event_stream.py index 8047d3562..087ad2941 100644 --- a/src/aap_eda/api/views/external_event_stream.py +++ b/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, @@ -51,10 +51,9 @@ from aap_eda.services.pg_notify import PGNotify from aap_eda.utils.log_sanitizer import REDACTED_STRING -FAILURE_THRESHOLD = 5 -FAILURE_WINDOW = 60 # seconds logger = logging.getLogger(__name__) UNSAFE_HEADER_KEYS = {"X-Trusted-Proxy", "X-Forwarded-For", "X-Real-IP"} +blacklist_manager = BlacklistManager() class ExternalEventStreamViewSet(viewsets.GenericViewSet): @@ -310,42 +309,37 @@ def _handle_auth(self, request, inputs): raise def _get_client_ip(self, request): + """Return the client IP from the request. + + Uses the rightmost X-Forwarded-For IP (appended by the + trusted proxy) when 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) + return x_forwarded_for.split(",")[-1].strip() + remote_addr = request.META.get("REMOTE_ADDR") + if not remote_addr: + raise AuthenticationFailed("Unable to determine client IP") + return remote_addr @extend_schema(exclude=True) @action(detail=True, methods=["POST"], rbac_action=None) def post(self, request, *_args, **kwargs): """Handle posts from external vendors.""" + # Validate X-Trusted-Proxy header from Gateway/Envoy + self._validate_trusted_proxy_header(request) + + 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_failure(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) except CredentialPluginError as err: @@ -365,10 +359,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_failure(client_ip) raise body = self._parse_body( diff --git a/src/aap_eda/core/utils/crypto/__init__.py b/src/aap_eda/core/utils/crypto/__init__.py index 56d762bcd..bd8b8febb 100644 --- a/src/aap_eda/core/utils/crypto/__init__.py +++ b/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()) diff --git a/src/aap_eda/settings/defaults.py b/src/aap_eda/settings/defaults.py index 754921f8e..848223a89 100644 --- a/src/aap_eda/settings/defaults.py +++ b/src/aap_eda/settings/defaults.py @@ -238,6 +238,18 @@ # 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 +# Set threshold to 0 to disable blacklisting entirely +# Note: blacklisting requires a shared cache backend (Redis, +# Memcached) to work across multiple workers. The default +# LocMemCache is per-process and will not share state. +# For test deployments: export EDA_EVENT_STREAM_BLACKLIST_THRESHOLD=0 +# or set EDA_MODE=testing (loads testing_defaults.py) +EVENT_STREAM_BLACKLIST_THRESHOLD: int = 5 +EVENT_STREAM_BLACKLIST_WINDOW: int = 60 # seconds +EVENT_STREAM_BLACKLIST_DURATION: int = 3600 # seconds (1 hour) + MAX_PG_NOTIFY_MESSAGE_SIZE: int = 6144 # Database credentials for the event streams user diff --git a/src/aap_eda/settings/testing_defaults.py b/src/aap_eda/settings/testing_defaults.py new file mode 100644 index 000000000..c31d6230a --- /dev/null +++ b/src/aap_eda/settings/testing_defaults.py @@ -0,0 +1,14 @@ +# 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. +EVENT_STREAM_BLACKLIST_THRESHOLD = 0 diff --git a/tests/integration/api/conftest.py b/tests/integration/api/conftest.py index 0aae1b83c..6c73448d2 100644 --- a/tests/integration/api/conftest.py +++ b/tests/integration/api/conftest.py @@ -1,9 +1,19 @@ """Shared test configuration for API integration tests.""" import pytest +from django.core.cache import cache from django.test.utils import override_settings +@pytest.fixture(autouse=True) +def disable_blacklisting(settings): + """Disable IP blacklisting and clear cache for tests.""" + cache.clear() + settings.EVENT_STREAM_BLACKLIST_THRESHOLD = 0 + yield + cache.clear() + + @pytest.fixture(autouse=True) def disable_trusted_proxy_validation_for_tests(request): """Disable X-Trusted-Proxy header validation for all tests except diff --git a/tests/unit/test_blacklist.py b/tests/unit/test_blacklist.py new file mode 100644 index 000000000..ecffb7ec8 --- /dev/null +++ b/tests/unit/test_blacklist.py @@ -0,0 +1,85 @@ +# 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 pytest +from django.core.cache import cache +from rest_framework.exceptions import AuthenticationFailed + +from aap_eda.api.blacklist import BlacklistManager + + +@pytest.fixture(autouse=True) +def clear_cache(): + cache.clear() + yield + cache.clear() + + +@pytest.fixture +def manager(): + return BlacklistManager() + + +@pytest.fixture +def blacklist_settings(settings): + settings.EVENT_STREAM_BLACKLIST_THRESHOLD = 5 + settings.EVENT_STREAM_BLACKLIST_WINDOW = 60 + settings.EVENT_STREAM_BLACKLIST_DURATION = 3600 + return settings + + +class TestBlacklisting: + def test_single_failure_not_blacklisted(self, manager, blacklist_settings): + manager.record_failure("10.0.0.1") + manager.check_blacklist("10.0.0.1") + + def test_threshold_triggers_blacklist(self, manager, blacklist_settings): + blacklist_settings.EVENT_STREAM_BLACKLIST_THRESHOLD = 3 + + for _ in range(3): + manager.record_failure("10.0.0.1") + + with pytest.raises(AuthenticationFailed): + manager.check_blacklist("10.0.0.1") + + def test_below_threshold_not_blacklisted( + self, manager, blacklist_settings + ): + for _ in range(4): + manager.record_failure("10.0.0.1") + + manager.check_blacklist("10.0.0.1") + + def test_per_ip_isolation(self, manager, blacklist_settings): + blacklist_settings.EVENT_STREAM_BLACKLIST_THRESHOLD = 3 + + for _ in range(3): + manager.record_failure("10.0.0.1") + + manager.check_blacklist("10.0.0.2") + + def test_clean_ip_passes(self, manager): + manager.check_blacklist("10.0.0.1") + + +class TestDisabledBlacklisting: + def test_zero_threshold_disables_blacklisting( + self, manager, blacklist_settings + ): + blacklist_settings.EVENT_STREAM_BLACKLIST_THRESHOLD = 0 + + for _ in range(10): + manager.record_failure("10.0.0.1") + + manager.check_blacklist("10.0.0.1") diff --git a/tests/unit/test_timing_safe_compare.py b/tests/unit/test_timing_safe_compare.py new file mode 100644 index 000000000..0779ebc40 --- /dev/null +++ b/tests/unit/test_timing_safe_compare.py @@ -0,0 +1,39 @@ +# 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. + +from aap_eda.core.utils.crypto import timing_safe_compare + + +def test_equal_strings(): + assert timing_safe_compare("abc", "abc") is True + + +def test_unequal_strings(): + assert timing_safe_compare("abc", "xyz") is False + + +def test_empty_strings(): + assert timing_safe_compare("", "") is True + + +def test_empty_vs_nonempty(): + assert timing_safe_compare("", "a") is False + + +def test_non_ascii_equal(): + assert timing_safe_compare("café", "café") is True + + +def test_non_ascii_unequal(): + assert timing_safe_compare("café", "naïve") is False From 955cda6cf63504d92fd7231de1baee8cabd42f9f Mon Sep 17 00:00:00 2001 From: Alex Date: Mon, 17 Aug 2026 15:48:01 -0400 Subject: [PATCH 3/4] feat: add per-org event stream settings with DB-driven IP management Introduce EventStreamSetting model (one per organization) to replace static Dynaconf settings for IP allowlists, blocklists, and auto- blacklist configuration. Adds a CRUD API endpoint with a clear-blocked action, Django signal-based cache invalidation, and per-org IP policy enforcement in the external event stream view. Co-Authored-By: Claude Opus 4.6 (1M context) --- src/aap_eda/api/blacklist.py | 96 ++++++--- src/aap_eda/api/filters/__init__.py | 3 + .../api/filters/event_stream_setting.py | 23 +++ src/aap_eda/api/serializers/__init__.py | 7 + .../api/serializers/event_stream_setting.py | 113 +++++++++++ src/aap_eda/api/urls.py | 1 + src/aap_eda/api/views/__init__.py | 3 + src/aap_eda/api/views/event_stream_setting.py | 157 ++++++++++++++ .../api/views/external_event_stream.py | 5 +- src/aap_eda/core/apps.py | 1 + src/aap_eda/core/enums.py | 1 + .../migrations/0074_eventstreamsetting.py | 122 +++++++++++ src/aap_eda/core/models/__init__.py | 3 + .../core/models/event_stream_setting.py | 66 ++++++ .../services/event_stream_settings_cache.py | 98 +++++++++ .../api/test_event_stream_setting.py | 192 ++++++++++++++++++ tests/unit/test_blacklist_org_aware.py | 158 ++++++++++++++ tests/unit/test_event_stream_setting_model.py | 98 +++++++++ 18 files changed, 1120 insertions(+), 27 deletions(-) create mode 100644 src/aap_eda/api/filters/event_stream_setting.py create mode 100644 src/aap_eda/api/serializers/event_stream_setting.py create mode 100644 src/aap_eda/api/views/event_stream_setting.py create mode 100644 src/aap_eda/core/migrations/0074_eventstreamsetting.py create mode 100644 src/aap_eda/core/models/event_stream_setting.py create mode 100644 src/aap_eda/services/event_stream_settings_cache.py create mode 100644 tests/integration/api/test_event_stream_setting.py create mode 100644 tests/unit/test_blacklist_org_aware.py create mode 100644 tests/unit/test_event_stream_setting_model.py diff --git a/src/aap_eda/api/blacklist.py b/src/aap_eda/api/blacklist.py index 6ca1aa087..93f9b5403 100644 --- a/src/aap_eda/api/blacklist.py +++ b/src/aap_eda/api/blacklist.py @@ -24,16 +24,21 @@ 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. + Supports both global (pre-org-resolution) and per-org modes. + + Global mode uses Dynaconf settings and flat cache keys. Per-org + mode reads thresholds from the EventStreamSetting DB model + (cached) and uses org-namespaced cache keys for isolation. """ FAILURE_PREFIX = "es_fail" BLACKLIST_PREFIX = "es_blacklist" def check_blacklist(self, client_ip: str) -> None: - """Raise AuthenticationFailed if the IP is globally blacklisted. + """Global blacklist check (pre-org-resolution). + + Used before the EventStream UUID is resolved, when the + organization is unknown. Checks global Dynaconf settings. A threshold of 0 disables blacklist checking entirely. """ @@ -43,39 +48,78 @@ def check_blacklist(self, client_ip: str) -> None: if cache.get(key): raise AuthenticationFailed("Too many failed attempts") - def record_failure(self, client_ip: str) -> None: + def check_ip_policy(self, client_ip: str, org_id: int) -> None: + """Full per-org IP policy check. + + Order: + 1. Admin-managed blocked_ips (DB, cached) + 2. Auto-blacklist from cache + 3. Allowlist enforcement (if non-empty) + """ + from aap_eda.services.event_stream_settings_cache import ( + get_org_settings, + ) + + org_settings = get_org_settings(org_id) + + if client_ip in org_settings["blocked_ips"]: + raise AuthenticationFailed("IP address is blocked") + + threshold = org_settings["blacklist_threshold"] + if threshold > 0: + key = f"{self.BLACKLIST_PREFIX}:{org_id}:{client_ip}" + if cache.get(key): + raise AuthenticationFailed("Too many failed attempts") + + if ( + org_settings["allowed_ips"] + and client_ip not in org_settings["allowed_ips"] + ): + raise AuthenticationFailed("IP address not in allowlist") + + def record_failure( + self, client_ip: str, org_id: int | None = None + ) -> None: """Record a failed request from the given IP. - All failure types (bad credentials, invalid UUIDs) count - toward the same threshold. After - EVENT_STREAM_BLACKLIST_THRESHOLD failures within - EVENT_STREAM_BLACKLIST_WINDOW seconds, the IP is globally - blacklisted for EVENT_STREAM_BLACKLIST_DURATION seconds. - A threshold of 0 disables blacklisting. + When org_id is provided, uses per-org settings and + org-namespaced cache keys. Otherwise falls back to global + Dynaconf settings (for pre-org-resolution failures like + bad UUID lookups). """ - if settings.EVENT_STREAM_BLACKLIST_THRESHOLD == 0: + if org_id is not None: + from aap_eda.services.event_stream_settings_cache import ( + get_org_settings, + ) + + org_settings = get_org_settings(org_id) + threshold = org_settings["blacklist_threshold"] + window = org_settings["blacklist_window"] + duration = org_settings["lockout_duration"] + key_suffix = f"{org_id}:{client_ip}" + else: + threshold = settings.EVENT_STREAM_BLACKLIST_THRESHOLD + window = settings.EVENT_STREAM_BLACKLIST_WINDOW + duration = settings.EVENT_STREAM_BLACKLIST_DURATION + key_suffix = client_ip + + if threshold == 0: return - counter_key = f"{self.FAILURE_PREFIX}:{client_ip}" + + counter_key = f"{self.FAILURE_PREFIX}:{key_suffix}" try: failures = cache.incr(counter_key) except ValueError: - cache.set( - counter_key, - 1, - settings.EVENT_STREAM_BLACKLIST_WINDOW, - ) + cache.set(counter_key, 1, window) failures = 1 - if failures >= settings.EVENT_STREAM_BLACKLIST_THRESHOLD: - blacklist_key = f"{self.BLACKLIST_PREFIX}:{client_ip}" - cache.set( - blacklist_key, - True, - settings.EVENT_STREAM_BLACKLIST_DURATION, - ) + if failures >= threshold: + blacklist_key = f"{self.BLACKLIST_PREFIX}:{key_suffix}" + cache.set(blacklist_key, True, duration) logger.warning( - "Globally blacklisted IP %s after %d failures", + "Blacklisted IP %s (org=%s) after %d failures", client_ip, + org_id, failures, ) cache.delete(counter_key) diff --git a/src/aap_eda/api/filters/__init__.py b/src/aap_eda/api/filters/__init__.py index 8e18e1e11..23220d348 100644 --- a/src/aap_eda/api/filters/__init__.py +++ b/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", ) diff --git a/src/aap_eda/api/filters/event_stream_setting.py b/src/aap_eda/api/filters/event_stream_setting.py new file mode 100644 index 000000000..7870b3a0f --- /dev/null +++ b/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"] diff --git a/src/aap_eda/api/serializers/__init__.py b/src/aap_eda/api/serializers/__init__.py index 9c072c486..acb475a63 100644 --- a/src/aap_eda/api/serializers/__init__.py +++ b/src/aap_eda/api/serializers/__init__.py @@ -52,6 +52,10 @@ EdaCredentialUpdateSerializer, ) from .event_stream import EventStreamInSerializer, EventStreamOutSerializer +from .event_stream_setting import ( + EventStreamSettingCreateSerializer, + EventStreamSettingOutSerializer, +) from .organization import ( OrganizationCreateSerializer, OrganizationRefSerializer, @@ -155,4 +159,7 @@ # event streams "EventStreamInSerializer", "EventStreamOutSerializer", + # event stream settings + "EventStreamSettingCreateSerializer", + "EventStreamSettingOutSerializer", ) diff --git a/src/aap_eda/api/serializers/event_stream_setting.py b/src/aap_eda/api/serializers/event_stream_setting.py new file mode 100644 index 000000000..fda14b3db --- /dev/null +++ b/src/aap_eda/api/serializers/event_stream_setting.py @@ -0,0 +1,113 @@ +# 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} IP addresses allowed." + ) + for ip in value: + try: + ipaddress.ip_address(ip.strip()) + except ValueError: + raise serializers.ValidationError( + f"'{ip}' is not a valid IPv4 or IPv6 address." + ) + return value + + +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", + "blacklist_threshold", + "blacklist_window", + "lockout_duration", + ] + + def validate_allowed_ips(self, value): + return _validate_ip_list(value) + + def validate_blocked_ips(self, value): + return _validate_ip_list(value) + + +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", + "blacklist_threshold", + "blacklist_window", + "lockout_duration", + "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 diff --git a/src/aap_eda/api/urls.py b/src/aap_eda/api/urls.py index 83c673d29..e84c745ae 100644 --- a/src/aap_eda/api/urls.py +++ b/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, diff --git a/src/aap_eda/api/views/__init__.py b/src/aap_eda/api/views/__init__.py index 3b61fdf77..5b0540b5c 100644 --- a/src/aap_eda/api/views/__init__.py +++ b/src/aap_eda/api/views/__init__.py @@ -20,6 +20,7 @@ from .decision_environment import DecisionEnvironmentViewSet from .eda_credential import EdaCredentialViewSet from .event_stream import EventStreamViewSet +from .event_stream_setting import EventStreamSettingViewSet from .external_event_stream import ExternalEventStreamViewSet from .organization import OrganizationViewSet from .project import ProjectViewSet @@ -61,6 +62,8 @@ "ConfigView", # Event stream "EventStreamViewSet", + # Event stream settings + "EventStreamSettingViewSet", # External event stream "ExternalEventStreamViewSet", ) diff --git a/src/aap_eda/api/views/event_stream_setting.py b/src/aap_eda/api/views/event_stream_setting.py new file mode 100644 index 000000000..b86f0e923 --- /dev/null +++ b/src/aap_eda/api/views/event_stream_setting.py @@ -0,0 +1,157 @@ +# 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. +"""EventStreamSetting API — per-org IP security settings.""" + +import logging + +from django_filters import rest_framework as defaultfilters +from drf_spectacular.utils import OpenApiResponse, extend_schema +from rest_framework import mixins, status, viewsets +from rest_framework.decorators import action +from rest_framework.response import Response + +from aap_eda.api import filters, serializers +from aap_eda.api.views.mixins import ( + CreateModelMixin, + PartialUpdateOnlyModelMixin, + ResponseSerializerMixin, +) +from aap_eda.core import models +from aap_eda.core.enums import ResourceType + +logger = logging.getLogger(__name__) + + +class EventStreamSettingViewSet( + ResponseSerializerMixin, + CreateModelMixin, + PartialUpdateOnlyModelMixin, + mixins.RetrieveModelMixin, + mixins.ListModelMixin, + viewsets.GenericViewSet, +): + queryset = models.EventStreamSetting.objects.order_by("-created_at") + filter_backends = (defaultfilters.DjangoFilterBackend,) + filterset_class = filters.EventStreamSettingFilter + rbac_resource_type = ResourceType.EVENT_STREAM_SETTING + + def get_serializer_class(self): + if self.action in ("create", "partial_update"): + return serializers.EventStreamSettingCreateSerializer + return serializers.EventStreamSettingOutSerializer + + def get_response_serializer_class(self): + return serializers.EventStreamSettingOutSerializer + + def filter_queryset(self, queryset): + return super().filter_queryset( + queryset.model.access_qs(self.request.user, queryset=queryset) + ) + + @extend_schema( + description="Get event stream settings by id", + responses={ + status.HTTP_200_OK: OpenApiResponse( + serializers.EventStreamSettingOutSerializer, + description="Return the event stream settings.", + ), + }, + ) + def retrieve(self, request, *args, **kwargs): + return super().retrieve(request, *args, **kwargs) + + @extend_schema( + description="List event stream settings", + responses={ + status.HTTP_200_OK: OpenApiResponse( + serializers.EventStreamSettingOutSerializer(many=True), + description="Return a list of event stream settings.", + ), + }, + ) + def list(self, request, *args, **kwargs): + return super().list(request, *args, **kwargs) + + @extend_schema( + request=serializers.EventStreamSettingCreateSerializer, + responses={ + status.HTTP_201_CREATED: OpenApiResponse( + serializers.EventStreamSettingOutSerializer, + description="Return the new event stream settings.", + ), + }, + ) + def create(self, request, *args, **kwargs): + return super().create(request, *args, **kwargs) + + @extend_schema( + request=serializers.EventStreamSettingCreateSerializer, + responses={ + status.HTTP_200_OK: OpenApiResponse( + serializers.EventStreamSettingOutSerializer, + description="Return the updated event stream settings.", + ), + }, + ) + def partial_update(self, request, *args, **kwargs): + return super().partial_update(request, *args, **kwargs) + + @extend_schema( + description=( + "Clear blocked IPs for this organization. " + "Clears both admin-managed and auto-blacklisted IPs." + ), + request=None, + responses={ + status.HTTP_200_OK: OpenApiResponse( + serializers.EventStreamSettingOutSerializer, + description="Return the updated event stream settings.", + ), + }, + ) + @action(detail=True, methods=["post"], url_path="clear-blocked") + def clear_blocked(self, request, pk=None): + """Clear blocked IPs — DB-stored and cache-auto-blacklisted.""" + setting = self.get_object() + org_id = setting.organization_id + + setting.blocked_ips = [] + setting.save(update_fields=["blocked_ips", "modified_at"]) + + _clear_org_blacklist_cache(org_id) + + logger.info( + "Cleared blocked IPs for org %s (user: %s)", + org_id, + request.user, + ) + return Response( + serializers.EventStreamSettingOutSerializer(setting).data, + status=status.HTTP_200_OK, + ) + + +def _clear_org_blacklist_cache(org_id: int) -> None: + """Best-effort clear of auto-blacklist cache keys for an org. + + Django's cache API does not support wildcard deletion. We + invalidate the org settings cache so the next check re-reads + from DB (where blocked_ips is now empty). Auto-blacklist + entries in cache will expire naturally via TTL. + """ + from aap_eda.services.event_stream_settings_cache import ( + invalidate_org_settings, + ) + + invalidate_org_settings(org_id) diff --git a/src/aap_eda/api/views/external_event_stream.py b/src/aap_eda/api/views/external_event_stream.py index 087ad2941..cdca3439b 100644 --- a/src/aap_eda/api/views/external_event_stream.py +++ b/src/aap_eda/api/views/external_event_stream.py @@ -340,6 +340,9 @@ def post(self, request, *_args, **kwargs): blacklist_manager.record_failure(client_ip) raise ParseError("bad uuid specified") from exc + org_id = self.event_stream.organization_id + blacklist_manager.check_ip_policy(client_ip, org_id) + try: inputs = get_resolved_secrets(self.event_stream.eda_credential) except CredentialPluginError as err: @@ -363,7 +366,7 @@ def post(self, request, *_args, **kwargs): try: self._handle_auth(request, inputs) except AuthenticationFailed: - blacklist_manager.record_failure(client_ip) + blacklist_manager.record_failure(client_ip, org_id=org_id) raise body = self._parse_body( diff --git a/src/aap_eda/core/apps.py b/src/aap_eda/core/apps.py index ba63a5c22..4627f198f 100644 --- a/src/aap_eda/core/apps.py +++ b/src/aap_eda/core/apps.py @@ -28,6 +28,7 @@ class CoreConfig(AppConfig): def ready(self): # make sure we apply DAB decorations in case they are not yet imported from aap_eda.api.views import dab_decorate # noqa: F401 + from aap_eda.services import event_stream_settings_cache # noqa: F401 # Enable default dispatcher config. Workers may override this dispatcher_setup(settings.DISPATCHERD_DEFAULT_SETTINGS) diff --git a/src/aap_eda/core/enums.py b/src/aap_eda/core/enums.py index aa2e3a007..40ecc8a50 100644 --- a/src/aap_eda/core/enums.py +++ b/src/aap_eda/core/enums.py @@ -55,6 +55,7 @@ class ResourceType(DjangoStrEnum): ORGANIZATION = "organization" TEAM = "team" EVENT_STREAM = "event_stream" + EVENT_STREAM_SETTING = "event_stream_setting" CREDENTIAL_INPUT_SOURCE = "credential_input_source" diff --git a/src/aap_eda/core/migrations/0074_eventstreamsetting.py b/src/aap_eda/core/migrations/0074_eventstreamsetting.py new file mode 100644 index 000000000..93139863f --- /dev/null +++ b/src/aap_eda/core/migrations/0074_eventstreamsetting.py @@ -0,0 +1,122 @@ +# Generated by Django 5.2 + +import django.db.models.deletion +from django.conf import settings +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("core", "0073_activation_k8s_pod_tolerations"), + migrations.swappable_dependency(settings.AUTH_USER_MODEL), + ] + + operations = [ + migrations.CreateModel( + name="EventStreamSetting", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ( + "created_by", + models.ForeignKey( + default=None, + editable=False, + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="%(class)s_created+", + to=settings.AUTH_USER_MODEL, + ), + ), + ( + "modified_by", + models.ForeignKey( + default=None, + editable=False, + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="%(class)s_modified+", + to=settings.AUTH_USER_MODEL, + ), + ), + ( + "organization", + models.OneToOneField( + on_delete=django.db.models.deletion.CASCADE, + related_name="event_stream_setting", + to="core.organization", + ), + ), + ( + "allowed_ips", + models.JSONField( + blank=True, + default=list, + help_text=( + "IP allowlist. When non-empty, only these IPs " + "may post events to any event stream in this " + "organization." + ), + ), + ), + ( + "blocked_ips", + models.JSONField( + blank=True, + default=list, + help_text=( + "Admin-managed list of permanently blocked IPs." + ), + ), + ), + ( + "blacklist_threshold", + models.PositiveIntegerField( + default=5, + help_text=( + "Number of auth failures before an IP is " + "auto-blacklisted. Set to 0 to disable " + "auto-blacklisting." + ), + ), + ), + ( + "blacklist_window", + models.PositiveIntegerField( + default=60, + help_text=( + "Seconds within which failures are counted." + ), + ), + ), + ( + "lockout_duration", + models.PositiveIntegerField( + default=3600, + help_text=( + "Seconds an auto-blacklisted IP stays blocked." + ), + ), + ), + ( + "created_at", + models.DateTimeField(auto_now_add=True), + ), + ( + "modified_at", + models.DateTimeField(auto_now=True), + ), + ], + options={ + "db_table": "core_event_stream_setting", + "ordering": ("-created_at",), + }, + ), + ] diff --git a/src/aap_eda/core/models/__init__.py b/src/aap_eda/core/models/__init__.py index fba9fd3d6..8d34416a9 100644 --- a/src/aap_eda/core/models/__init__.py +++ b/src/aap_eda/core/models/__init__.py @@ -20,6 +20,7 @@ from .decision_environment import DecisionEnvironment from .eda_credential import EdaCredential from .event_stream import EventStream +from .event_stream_setting import EventStreamSetting from .job import ( ActivationInstanceJobInstance, Job, @@ -65,6 +66,7 @@ "Organization", "Team", "EventStream", + "EventStreamSetting", "Setting", ] @@ -77,6 +79,7 @@ Organization, Team, EventStream, + EventStreamSetting, parent_field_name="organization", ) permission_registry.register( diff --git a/src/aap_eda/core/models/event_stream_setting.py b/src/aap_eda/core/models/event_stream_setting.py new file mode 100644 index 000000000..6e2c0f846 --- /dev/null +++ b/src/aap_eda/core/models/event_stream_setting.py @@ -0,0 +1,66 @@ +# 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. + +from django.db import models + +from .base import PrimordialModel + + +class EventStreamSetting(PrimordialModel): + """Per-organization IP security settings for event streams. + + Controls which IPs may post events (allowlist), which are + permanently blocked (blocklist), and auto-blacklist behavior + (threshold, window, lockout duration). + """ + + class Meta: + db_table = "core_event_stream_setting" + ordering = ("-created_at",) + + organization = models.OneToOneField( + "Organization", + on_delete=models.CASCADE, + related_name="event_stream_setting", + ) + allowed_ips = models.JSONField( + default=list, + blank=True, + help_text=( + "IP allowlist. When non-empty, only these IPs may post " + "events to any event stream in this organization." + ), + ) + blocked_ips = models.JSONField( + default=list, + blank=True, + help_text="Admin-managed list of permanently blocked IPs.", + ) + blacklist_threshold = models.PositiveIntegerField( + default=5, + help_text=( + "Number of auth failures before an IP is auto-blacklisted. " + "Set to 0 to disable auto-blacklisting." + ), + ) + blacklist_window = models.PositiveIntegerField( + default=60, + help_text="Seconds within which failures are counted.", + ) + lockout_duration = models.PositiveIntegerField( + default=3600, + help_text="Seconds an auto-blacklisted IP stays blocked.", + ) + created_at = models.DateTimeField(auto_now_add=True) + modified_at = models.DateTimeField(auto_now=True) diff --git a/src/aap_eda/services/event_stream_settings_cache.py b/src/aap_eda/services/event_stream_settings_cache.py new file mode 100644 index 000000000..2c8f2d2a0 --- /dev/null +++ b/src/aap_eda/services/event_stream_settings_cache.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. + +"""Per-organization event stream settings cache with signal-based invalidation. + +Settings are cached for SETTINGS_CACHE_TTL seconds. A post_save signal +on EventStreamSetting invalidates the cache so changes propagate +immediately (within the same cache backend). +""" + +import logging +from typing import Any + +from django.core.cache import cache +from django.db.models.signals import post_save +from django.dispatch import receiver + +logger = logging.getLogger(__name__) + +SETTINGS_CACHE_PREFIX = "es_org_settings" +SETTINGS_CACHE_TTL = 300 + + +def _cache_key(org_id: int) -> str: + return f"{SETTINGS_CACHE_PREFIX}:{org_id}" + + +def get_org_settings(org_id: int) -> dict: + """Get per-org event stream settings, from cache or DB. + + Falls back to global Dynaconf defaults if no DB row exists + for the given organization. + """ + key = _cache_key(org_id) + cached = cache.get(key) + if cached is not None: + return cached + + from aap_eda.core.models import EventStreamSetting + + try: + setting = EventStreamSetting.objects.get(organization_id=org_id) + data = { + "allowed_ips": set(setting.allowed_ips), + "blocked_ips": set(setting.blocked_ips), + "blacklist_threshold": setting.blacklist_threshold, + "blacklist_window": setting.blacklist_window, + "lockout_duration": setting.lockout_duration, + } + except EventStreamSetting.DoesNotExist: + from django.conf import settings as django_settings + + data = { + "allowed_ips": set(), + "blocked_ips": set(), + "blacklist_threshold": getattr( + django_settings, "EVENT_STREAM_BLACKLIST_THRESHOLD", 5 + ), + "blacklist_window": getattr( + django_settings, "EVENT_STREAM_BLACKLIST_WINDOW", 60 + ), + "lockout_duration": getattr( + django_settings, "EVENT_STREAM_BLACKLIST_DURATION", 3600 + ), + } + + cache.set(key, data, SETTINGS_CACHE_TTL) + return data + + +def invalidate_org_settings(org_id: int) -> None: + """Delete the cached settings for an organization.""" + cache.delete(_cache_key(org_id)) + + +@receiver(post_save, sender="core.EventStreamSetting") +def on_event_stream_setting_saved( + sender: Any, + instance: Any, + **kwargs: Any, +) -> None: + """Invalidate cache when settings are saved.""" + invalidate_org_settings(instance.organization_id) + logger.info( + "Invalidated event stream settings cache for org %s", + instance.organization_id, + ) diff --git a/tests/integration/api/test_event_stream_setting.py b/tests/integration/api/test_event_stream_setting.py new file mode 100644 index 000000000..a8a0efca5 --- /dev/null +++ b/tests/integration/api/test_event_stream_setting.py @@ -0,0 +1,192 @@ +# 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 pytest +from rest_framework import status +from rest_framework.test import APIClient + +from aap_eda.core import models +from tests.integration.constants import api_url_v1 + +SETTINGS_URL = f"{api_url_v1}/event-stream-settings" + + +@pytest.fixture +def default_event_stream_setting(default_organization): + return models.EventStreamSetting.objects.create( + organization=default_organization, + allowed_ips=["10.0.0.1", "10.0.0.2"], + blocked_ips=["192.168.1.100"], + blacklist_threshold=3, + blacklist_window=30, + lockout_duration=1800, + ) + + +@pytest.mark.django_db +class TestEventStreamSettingCreate: + def test_create(self, admin_client: APIClient, default_organization): + response = admin_client.post( + f"{SETTINGS_URL}/", + data={ + "organization_id": default_organization.id, + "allowed_ips": ["10.0.0.1"], + "blocked_ips": [], + "blacklist_threshold": 5, + "blacklist_window": 60, + "lockout_duration": 3600, + }, + ) + assert response.status_code == status.HTTP_201_CREATED + assert response.data["allowed_ips"] == ["10.0.0.1"] + assert response.data["blacklist_threshold"] == 5 + + def test_create_duplicate_org_rejected( + self, + admin_client: APIClient, + default_event_stream_setting, + default_organization, + ): + response = admin_client.post( + f"{SETTINGS_URL}/", + data={ + "organization_id": default_organization.id, + "allowed_ips": [], + }, + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + + def test_create_with_defaults( + self, admin_client: APIClient, default_organization + ): + response = admin_client.post( + f"{SETTINGS_URL}/", + data={"organization_id": default_organization.id}, + ) + assert response.status_code == status.HTTP_201_CREATED + assert response.data["allowed_ips"] == [] + assert response.data["blocked_ips"] == [] + assert response.data["blacklist_threshold"] == 5 + assert response.data["blacklist_window"] == 60 + assert response.data["lockout_duration"] == 3600 + + +@pytest.mark.django_db +class TestEventStreamSettingRead: + def test_retrieve( + self, + admin_client: APIClient, + default_event_stream_setting, + ): + pk = default_event_stream_setting.id + response = admin_client.get(f"{SETTINGS_URL}/{pk}/") + assert response.status_code == status.HTTP_200_OK + assert response.data["allowed_ips"] == ["10.0.0.1", "10.0.0.2"] + assert response.data["organization"] is not None + + def test_list( + self, + admin_client: APIClient, + default_event_stream_setting, + ): + response = admin_client.get(f"{SETTINGS_URL}/") + assert response.status_code == status.HTTP_200_OK + assert len(response.data["results"]) == 1 + + +@pytest.mark.django_db +class TestEventStreamSettingUpdate: + def test_partial_update_allowed_ips( + self, + admin_client: APIClient, + default_event_stream_setting, + ): + pk = default_event_stream_setting.id + response = admin_client.patch( + f"{SETTINGS_URL}/{pk}/", + data={"allowed_ips": ["172.16.0.1"]}, + ) + assert response.status_code == status.HTTP_200_OK + assert response.data["allowed_ips"] == ["172.16.0.1"] + + def test_partial_update_lockout_duration( + self, + admin_client: APIClient, + default_event_stream_setting, + ): + pk = default_event_stream_setting.id + response = admin_client.patch( + f"{SETTINGS_URL}/{pk}/", + data={"lockout_duration": 7200}, + ) + assert response.status_code == status.HTTP_200_OK + assert response.data["lockout_duration"] == 7200 + + +@pytest.mark.django_db +class TestEventStreamSettingValidation: + def test_invalid_ip_rejected( + self, admin_client: APIClient, default_organization + ): + response = admin_client.post( + f"{SETTINGS_URL}/", + data={ + "organization_id": default_organization.id, + "allowed_ips": ["not-an-ip"], + }, + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + + def test_max_255_ips_enforced( + self, admin_client: APIClient, default_organization + ): + ips = [f"10.0.{i // 256}.{i % 256}" for i in range(256)] + response = admin_client.post( + f"{SETTINGS_URL}/", + data={ + "organization_id": default_organization.id, + "allowed_ips": ips, + }, + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + + def test_ipv6_accepted( + self, admin_client: APIClient, default_organization + ): + response = admin_client.post( + f"{SETTINGS_URL}/", + data={ + "organization_id": default_organization.id, + "allowed_ips": ["::1", "2001:db8::1"], + }, + ) + assert response.status_code == status.HTTP_201_CREATED + + +@pytest.mark.django_db +class TestClearBlocked: + def test_clear_blocked( + self, + admin_client: APIClient, + default_event_stream_setting, + ): + pk = default_event_stream_setting.id + assert default_event_stream_setting.blocked_ips == ["192.168.1.100"] + response = admin_client.post( + f"{SETTINGS_URL}/{pk}/clear-blocked/", + ) + assert response.status_code == status.HTTP_200_OK + assert response.data["blocked_ips"] == [] + default_event_stream_setting.refresh_from_db() + assert default_event_stream_setting.blocked_ips == [] diff --git a/tests/unit/test_blacklist_org_aware.py b/tests/unit/test_blacklist_org_aware.py new file mode 100644 index 000000000..3ffc0af3e --- /dev/null +++ b/tests/unit/test_blacklist_org_aware.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 pytest +from django.core.cache import cache +from rest_framework.exceptions import AuthenticationFailed + +from aap_eda.api.blacklist import BlacklistManager +from aap_eda.core import models + + +@pytest.fixture(autouse=True) +def clear_cache(): + cache.clear() + yield + cache.clear() + + +@pytest.fixture +def manager(): + return BlacklistManager() + + +@pytest.fixture +def blacklist_settings(settings): + settings.EVENT_STREAM_BLACKLIST_THRESHOLD = 5 + settings.EVENT_STREAM_BLACKLIST_WINDOW = 60 + settings.EVENT_STREAM_BLACKLIST_DURATION = 3600 + return settings + + +@pytest.mark.django_db +class TestCheckIpPolicy: + def test_blocked_ip_rejected( + self, manager, default_organization, blacklist_settings + ): + models.EventStreamSetting.objects.create( + organization=default_organization, + blocked_ips=["10.0.0.1"], + ) + with pytest.raises(AuthenticationFailed, match="blocked"): + manager.check_ip_policy("10.0.0.1", default_organization.id) + + def test_unblocked_ip_passes( + self, manager, default_organization, blacklist_settings + ): + models.EventStreamSetting.objects.create( + organization=default_organization, + blocked_ips=["10.0.0.1"], + ) + manager.check_ip_policy("10.0.0.2", default_organization.id) + + def test_allowed_ips_enforced( + self, manager, default_organization, blacklist_settings + ): + models.EventStreamSetting.objects.create( + organization=default_organization, + allowed_ips=["10.0.0.1", "10.0.0.2"], + ) + with pytest.raises(AuthenticationFailed, match="allowlist"): + manager.check_ip_policy("10.0.0.99", default_organization.id) + + def test_allowed_ips_passes_listed_ip( + self, manager, default_organization, blacklist_settings + ): + models.EventStreamSetting.objects.create( + organization=default_organization, + allowed_ips=["10.0.0.1"], + ) + manager.check_ip_policy("10.0.0.1", default_organization.id) + + def test_empty_allowed_ips_allows_all( + self, manager, default_organization, blacklist_settings + ): + models.EventStreamSetting.objects.create( + organization=default_organization, + allowed_ips=[], + ) + manager.check_ip_policy("10.0.0.99", default_organization.id) + + def test_auto_blacklisted_ip_rejected( + self, manager, default_organization, blacklist_settings + ): + models.EventStreamSetting.objects.create( + organization=default_organization, + blacklist_threshold=2, + ) + org_id = default_organization.id + manager.record_failure("10.0.0.1", org_id=org_id) + manager.record_failure("10.0.0.1", org_id=org_id) + with pytest.raises(AuthenticationFailed, match="Too many"): + manager.check_ip_policy("10.0.0.1", org_id) + + def test_no_settings_row_uses_global_defaults( + self, manager, default_organization, blacklist_settings + ): + manager.check_ip_policy("10.0.0.1", default_organization.id) + + +@pytest.mark.django_db +class TestOrgAwareRecordFailure: + def test_per_org_threshold( + self, manager, default_organization, blacklist_settings + ): + models.EventStreamSetting.objects.create( + organization=default_organization, + blacklist_threshold=2, + blacklist_window=60, + lockout_duration=3600, + ) + org_id = default_organization.id + manager.record_failure("10.0.0.1", org_id=org_id) + manager.record_failure("10.0.0.1", org_id=org_id) + key = f"es_blacklist:{org_id}:10.0.0.1" + assert cache.get(key) is True + + def test_org_isolation(self, manager, blacklist_settings): + org_a = models.Organization.objects.create(name="Org A") + org_b = models.Organization.objects.create(name="Org B") + models.EventStreamSetting.objects.create( + organization=org_a, blacklist_threshold=2 + ) + models.EventStreamSetting.objects.create( + organization=org_b, blacklist_threshold=2 + ) + manager.record_failure("10.0.0.1", org_id=org_a.id) + manager.record_failure("10.0.0.1", org_id=org_a.id) + manager.check_ip_policy("10.0.0.1", org_b.id) + + def test_zero_threshold_disables( + self, manager, default_organization, blacklist_settings + ): + models.EventStreamSetting.objects.create( + organization=default_organization, + blacklist_threshold=0, + ) + org_id = default_organization.id + for _ in range(10): + manager.record_failure("10.0.0.1", org_id=org_id) + manager.check_ip_policy("10.0.0.1", org_id) + + def test_global_fallback_without_org_id(self, manager, blacklist_settings): + blacklist_settings.EVENT_STREAM_BLACKLIST_THRESHOLD = 2 + manager.record_failure("10.0.0.1") + manager.record_failure("10.0.0.1") + with pytest.raises(AuthenticationFailed): + manager.check_blacklist("10.0.0.1") diff --git a/tests/unit/test_event_stream_setting_model.py b/tests/unit/test_event_stream_setting_model.py new file mode 100644 index 000000000..4034838d0 --- /dev/null +++ b/tests/unit/test_event_stream_setting_model.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 pytest +from django.core.cache import cache + +from aap_eda.core import models +from aap_eda.services.event_stream_settings_cache import ( + get_org_settings, + invalidate_org_settings, +) + + +@pytest.fixture(autouse=True) +def clear_cache(): + cache.clear() + yield + cache.clear() + + +@pytest.mark.django_db +class TestGetOrgSettings: + def test_returns_db_values(self, default_organization): + models.EventStreamSetting.objects.create( + organization=default_organization, + allowed_ips=["10.0.0.1", "10.0.0.2"], + blocked_ips=["192.168.1.100"], + blacklist_threshold=3, + blacklist_window=30, + lockout_duration=1800, + ) + result = get_org_settings(default_organization.id) + assert result["allowed_ips"] == {"10.0.0.1", "10.0.0.2"} + assert result["blocked_ips"] == {"192.168.1.100"} + assert result["blacklist_threshold"] == 3 + assert result["blacklist_window"] == 30 + assert result["lockout_duration"] == 1800 + + def test_caches_result(self, default_organization): + models.EventStreamSetting.objects.create( + organization=default_organization, + allowed_ips=["10.0.0.1"], + ) + result1 = get_org_settings(default_organization.id) + models.EventStreamSetting.objects.filter( + organization=default_organization + ).update(allowed_ips=["10.0.0.99"]) + result2 = get_org_settings(default_organization.id) + assert result1["allowed_ips"] == result2["allowed_ips"] + + def test_fallback_to_global_defaults(self, default_organization, settings): + settings.EVENT_STREAM_BLACKLIST_THRESHOLD = 7 + settings.EVENT_STREAM_BLACKLIST_WINDOW = 120 + settings.EVENT_STREAM_BLACKLIST_DURATION = 7200 + result = get_org_settings(default_organization.id) + assert result["allowed_ips"] == set() + assert result["blocked_ips"] == set() + assert result["blacklist_threshold"] == 7 + assert result["blacklist_window"] == 120 + assert result["lockout_duration"] == 7200 + + def test_invalidate_clears_cache(self, default_organization): + models.EventStreamSetting.objects.create( + organization=default_organization, + allowed_ips=["10.0.0.1"], + ) + get_org_settings(default_organization.id) + invalidate_org_settings(default_organization.id) + models.EventStreamSetting.objects.filter( + organization=default_organization + ).update(allowed_ips=["10.0.0.99"]) + result = get_org_settings(default_organization.id) + assert result["allowed_ips"] == {"10.0.0.99"} + + +@pytest.mark.django_db +class TestSignalInvalidation: + def test_post_save_invalidates_cache(self, default_organization): + setting = models.EventStreamSetting.objects.create( + organization=default_organization, + allowed_ips=["10.0.0.1"], + ) + get_org_settings(default_organization.id) + setting.allowed_ips = ["10.0.0.99"] + setting.save() + result = get_org_settings(default_organization.id) + assert result["allowed_ips"] == {"10.0.0.99"} From 725cb15e9b7e4eebd4639e140289a52c3288d348 Mon Sep 17 00:00:00 2001 From: Alex Date: Mon, 17 Aug 2026 16:50:53 -0400 Subject: [PATCH 4/4] refactor: replace cache-based blacklist with allowlist-first IP policy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Remove threshold-based auto-blacklisting in favor of per-org IP allowlists. Blocked IPs are now tracked in the DB for admin visibility — admins can promote them to the allowlist or remove them. Adding an IP to the allowlist auto-removes it from blocked. --- src/aap_eda/api/blacklist.py | 153 +++++------ .../api/event_stream_authentication.py | 2 +- src/aap_eda/api/serializers/__init__.py | 2 + .../api/serializers/event_stream_setting.py | 83 ++++-- src/aap_eda/api/views/event_stream_setting.py | 87 ++++--- .../api/views/external_event_stream.py | 15 +- .../migrations/0074_eventstreamsetting.py | 46 +--- .../core/models/event_stream_setting.py | 30 +-- .../services/event_stream_settings_cache.py | 42 +-- src/aap_eda/settings/defaults.py | 11 - src/aap_eda/settings/testing_defaults.py | 1 - tests/integration/api/conftest.py | 5 +- .../api/test_event_stream_setting.py | 239 ++++++++++++++++-- tests/integration/api/test_root.py | 2 + tests/integration/conftest.py | 9 + tests/integration/dab_rbac/conftest.py | 4 +- .../integration/dab_rbac/test_organization.py | 5 + tests/unit/test_blacklist.py | 121 +++++---- tests/unit/test_blacklist_org_aware.py | 138 +++------- tests/unit/test_event_stream_setting_model.py | 22 +- 20 files changed, 584 insertions(+), 433 deletions(-) diff --git a/src/aap_eda/api/blacklist.py b/src/aap_eda/api/blacklist.py index 93f9b5403..859ed8dc7 100644 --- a/src/aap_eda/api/blacklist.py +++ b/src/aap_eda/api/blacklist.py @@ -12,114 +12,117 @@ # See the License for the specific language governing permissions and # limitations under the License. +import ipaddress import logging -from django.conf import settings -from django.core.cache import cache from rest_framework.exceptions import AuthenticationFailed logger = logging.getLogger(__name__) +MAX_BLOCKED_IPS = 1000 -class BlacklistManager: - """Rate-limit and blacklist IPs that fail event stream authentication. - Supports both global (pre-org-resolution) and per-org modes. +def normalize_ip(ip_str: str) -> str: + """Normalize an IP address string. - Global mode uses Dynaconf settings and flat cache keys. Per-org - mode reads thresholds from the EventStreamSetting DB model - (cached) and uses org-namespaced cache keys for isolation. + 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) - FAILURE_PREFIX = "es_fail" - BLACKLIST_PREFIX = "es_blacklist" - def check_blacklist(self, client_ip: str) -> None: - """Global blacklist check (pre-org-resolution). +def ip_in_allowlist(client_ip: str, allowed_ips: set) -> bool: + """Check if an IP matches any entry in the allowlist. - Used before the EventStream UUID is resolved, when the - organization is unknown. Checks global Dynaconf settings. + 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 - A threshold of 0 disables blacklist checking entirely. - """ - if settings.EVENT_STREAM_BLACKLIST_THRESHOLD == 0: - return - key = f"{self.BLACKLIST_PREFIX}:{client_ip}" - if cache.get(key): - raise AuthenticationFailed("Too many failed attempts") + +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: - """Full per-org IP policy check. + """Check if the IP is allowed for this organization. - Order: - 1. Admin-managed blocked_ips (DB, cached) - 2. Auto-blacklist from cache - 3. Allowlist enforcement (if non-empty) + 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 - if client_ip in org_settings["blocked_ips"]: - raise AuthenticationFailed("IP address is blocked") - - threshold = org_settings["blacklist_threshold"] - if threshold > 0: - key = f"{self.BLACKLIST_PREFIX}:{org_id}:{client_ip}" - if cache.get(key): - raise AuthenticationFailed("Too many failed attempts") + normalized = normalize_ip(client_ip) - if ( - org_settings["allowed_ips"] - and client_ip not in org_settings["allowed_ips"] + 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_failure( - self, client_ip: str, org_id: int | None = None - ) -> None: - """Record a failed request from the given IP. + def record_blocked_ip(self, client_ip: str, org_id: int) -> None: + """Add an IP to the org's blocked_ips list for visibility. - When org_id is provided, uses per-org settings and - org-namespaced cache keys. Otherwise falls back to global - Dynaconf settings (for pre-org-resolution failures like - bad UUID lookups). + Only adds the IP if it is not already tracked. Caps the + list at MAX_BLOCKED_IPS to prevent unbounded growth. """ - if org_id is not None: - from aap_eda.services.event_stream_settings_cache import ( - get_org_settings, - ) + from aap_eda.core.models import EventStreamSetting - org_settings = get_org_settings(org_id) - threshold = org_settings["blacklist_threshold"] - window = org_settings["blacklist_window"] - duration = org_settings["lockout_duration"] - key_suffix = f"{org_id}:{client_ip}" - else: - threshold = settings.EVENT_STREAM_BLACKLIST_THRESHOLD - window = settings.EVENT_STREAM_BLACKLIST_WINDOW - duration = settings.EVENT_STREAM_BLACKLIST_DURATION - key_suffix = client_ip - - if threshold == 0: - return + normalized = normalize_ip(client_ip) - counter_key = f"{self.FAILURE_PREFIX}:{key_suffix}" try: - failures = cache.incr(counter_key) - except ValueError: - cache.set(counter_key, 1, window) - failures = 1 - - if failures >= threshold: - blacklist_key = f"{self.BLACKLIST_PREFIX}:{key_suffix}" - cache.set(blacklist_key, True, duration) + 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( - "Blacklisted IP %s (org=%s) after %d failures", - client_ip, + "Blocked IPs cap (%d) reached for org %s", + MAX_BLOCKED_IPS, org_id, - failures, ) - cache.delete(counter_key) + 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, + ) diff --git a/src/aap_eda/api/event_stream_authentication.py b/src/aap_eda/api/event_stream_authentication.py index ec9da7fc0..a8d8a84de 100644 --- a/src/aap_eda/api/event_stream_authentication.py +++ b/src/aap_eda/api/event_stream_authentication.py @@ -153,7 +153,7 @@ def authenticate(self, _body=None): if self.authorization.startswith("Basic"): auth_str = self.authorization.split("Basic ")[1] - user_pass = f"{self.username}:{self.password}" # noqa: E231 + user_pass = "%s:%s" % (self.username, self.password) b64_value = base64.b64encode(user_pass.encode()).decode() if not timing_safe_compare(auth_str, b64_value): message = "Credential mismatch" diff --git a/src/aap_eda/api/serializers/__init__.py b/src/aap_eda/api/serializers/__init__.py index acb475a63..bf144f359 100644 --- a/src/aap_eda/api/serializers/__init__.py +++ b/src/aap_eda/api/serializers/__init__.py @@ -55,6 +55,7 @@ from .event_stream_setting import ( EventStreamSettingCreateSerializer, EventStreamSettingOutSerializer, + RemoveBlockedIpsSerializer, ) from .organization import ( OrganizationCreateSerializer, @@ -162,4 +163,5 @@ # event stream settings "EventStreamSettingCreateSerializer", "EventStreamSettingOutSerializer", + "RemoveBlockedIpsSerializer", ) diff --git a/src/aap_eda/api/serializers/event_stream_setting.py b/src/aap_eda/api/serializers/event_stream_setting.py index fda14b3db..b2a42ac19 100644 --- a/src/aap_eda/api/serializers/event_stream_setting.py +++ b/src/aap_eda/api/serializers/event_stream_setting.py @@ -27,19 +27,35 @@ def _validate_ip_list(value): if len(value) > MAX_IPS_PER_LIST: raise serializers.ValidationError( - f"Maximum {MAX_IPS_PER_LIST} IP addresses allowed." + f"Maximum {MAX_IPS_PER_LIST} entries allowed." ) - for ip in value: - try: - ipaddress.ip_address(ip.strip()) - except ValueError: - raise serializers.ValidationError( - f"'{ip}' is not a valid IPv4 or IPv6 address." - ) - return value - - -class EventStreamSettingCreateSerializer(serializers.ModelSerializer): + 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, @@ -63,19 +79,40 @@ class Meta: "organization_id", "allowed_ips", "blocked_ips", - "blacklist_threshold", - "blacklist_window", - "lockout_duration", ] + 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): +class EventStreamSettingOutSerializer( + serializers.ModelSerializer, +): organization = serializers.SerializerMethodField() created_by = BasicUserFieldSerializer() modified_by = BasicUserFieldSerializer() @@ -91,9 +128,6 @@ class Meta: "organization", "allowed_ips", "blocked_ips", - "blacklist_threshold", - "blacklist_window", - "lockout_duration", "created_by", "modified_by", *read_only_fields, @@ -111,3 +145,14 @@ def to_representation(self, 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) diff --git a/src/aap_eda/api/views/event_stream_setting.py b/src/aap_eda/api/views/event_stream_setting.py index b86f0e923..bd10d6328 100644 --- a/src/aap_eda/api/views/event_stream_setting.py +++ b/src/aap_eda/api/views/event_stream_setting.py @@ -29,6 +29,9 @@ ) from aap_eda.core import models from aap_eda.core.enums import ResourceType +from aap_eda.services.event_stream_settings_cache import ( + invalidate_org_settings, +) logger = logging.getLogger(__name__) @@ -64,7 +67,7 @@ def filter_queryset(self, queryset): responses={ status.HTTP_200_OK: OpenApiResponse( serializers.EventStreamSettingOutSerializer, - description="Return the event stream settings.", + description=("Return the event stream settings."), ), }, ) @@ -76,7 +79,7 @@ def retrieve(self, request, *args, **kwargs): responses={ status.HTTP_200_OK: OpenApiResponse( serializers.EventStreamSettingOutSerializer(many=True), - description="Return a list of event stream settings.", + description=("Return a list of event stream settings."), ), }, ) @@ -88,7 +91,7 @@ def list(self, request, *args, **kwargs): responses={ status.HTTP_201_CREATED: OpenApiResponse( serializers.EventStreamSettingOutSerializer, - description="Return the new event stream settings.", + description=("Return the new event stream settings."), ), }, ) @@ -100,7 +103,7 @@ def create(self, request, *args, **kwargs): responses={ status.HTTP_200_OK: OpenApiResponse( serializers.EventStreamSettingOutSerializer, - description="Return the updated event stream settings.", + description=("Return the updated event stream settings."), ), }, ) @@ -108,50 +111,70 @@ def partial_update(self, request, *args, **kwargs): return super().partial_update(request, *args, **kwargs) @extend_schema( - description=( - "Clear blocked IPs for this organization. " - "Clears both admin-managed and auto-blacklisted IPs." - ), - request=None, + description=("Remove specific IPs from the blocked list."), + request=serializers.RemoveBlockedIpsSerializer, responses={ status.HTTP_200_OK: OpenApiResponse( serializers.EventStreamSettingOutSerializer, - description="Return the updated event stream settings.", + description=("Return the updated event stream settings."), ), }, ) - @action(detail=True, methods=["post"], url_path="clear-blocked") - def clear_blocked(self, request, pk=None): - """Clear blocked IPs — DB-stored and cache-auto-blacklisted.""" + @action( + detail=True, + methods=["post"], + url_path="remove-blocked", + ) + def remove_blocked(self, request, pk=None): + """Remove specific IPs from the blocked list.""" setting = self.get_object() - org_id = setting.organization_id + sz = serializers.RemoveBlockedIpsSerializer(data=request.data) + sz.is_valid(raise_exception=True) + ips_to_remove = set(sz.validated_data["ips"]) - setting.blocked_ips = [] + setting.blocked_ips = [ + ip for ip in setting.blocked_ips if ip not in ips_to_remove + ] setting.save(update_fields=["blocked_ips", "modified_at"]) - - _clear_org_blacklist_cache(org_id) + invalidate_org_settings(setting.organization_id) logger.info( - "Cleared blocked IPs for org %s (user: %s)", - org_id, - request.user, + "Removed %d IPs from blocked list for org %s", + len(ips_to_remove), + setting.organization_id, ) return Response( serializers.EventStreamSettingOutSerializer(setting).data, status=status.HTTP_200_OK, ) - -def _clear_org_blacklist_cache(org_id: int) -> None: - """Best-effort clear of auto-blacklist cache keys for an org. - - Django's cache API does not support wildcard deletion. We - invalidate the org settings cache so the next check re-reads - from DB (where blocked_ips is now empty). Auto-blacklist - entries in cache will expire naturally via TTL. - """ - from aap_eda.services.event_stream_settings_cache import ( - invalidate_org_settings, + @extend_schema( + description="Clear all IPs from the blocked list.", + request=None, + responses={ + status.HTTP_200_OK: OpenApiResponse( + serializers.EventStreamSettingOutSerializer, + description=("Return the updated event stream settings."), + ), + }, ) + @action( + detail=True, + methods=["post"], + url_path="clear-blocked", + ) + def clear_blocked(self, request, pk=None): + """Clear all blocked IPs.""" + setting = self.get_object() + setting.blocked_ips = [] + setting.save(update_fields=["blocked_ips", "modified_at"]) + invalidate_org_settings(setting.organization_id) - invalidate_org_settings(org_id) + logger.info( + "Cleared all blocked IPs for org %s", + setting.organization_id, + ) + return Response( + serializers.EventStreamSettingOutSerializer(setting).data, + status=status.HTTP_200_OK, + ) diff --git a/src/aap_eda/api/views/external_event_stream.py b/src/aap_eda/api/views/external_event_stream.py index cdca3439b..58ab5ce8e 100644 --- a/src/aap_eda/api/views/external_event_stream.py +++ b/src/aap_eda/api/views/external_event_stream.py @@ -309,20 +309,23 @@ def _handle_auth(self, request, inputs): raise def _get_client_ip(self, request): - """Return the client IP from the request. + """Return the normalized client IP from the request. Uses the rightmost X-Forwarded-For IP (appended by the trusted proxy) when proxy validation is enabled, otherwise - falls back to REMOTE_ADDR. + falls back to REMOTE_ADDR. Normalizes IPv4-mapped IPv6 + addresses to IPv4 form for consistent allowlist matching. """ + from aap_eda.api.blacklist import normalize_ip + 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(",")[-1].strip() + return normalize_ip(x_forwarded_for.split(",")[-1]) remote_addr = request.META.get("REMOTE_ADDR") if not remote_addr: raise AuthenticationFailed("Unable to determine client IP") - return remote_addr + return normalize_ip(remote_addr) @extend_schema(exclude=True) @action(detail=True, methods=["POST"], rbac_action=None) @@ -332,12 +335,10 @@ def post(self, request, *_args, **kwargs): self._validate_trusted_proxy_header(request) 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_failure(client_ip) raise ParseError("bad uuid specified") from exc org_id = self.event_stream.organization_id @@ -366,7 +367,7 @@ def post(self, request, *_args, **kwargs): try: self._handle_auth(request, inputs) except AuthenticationFailed: - blacklist_manager.record_failure(client_ip, org_id=org_id) + blacklist_manager.record_blocked_ip(client_ip, org_id) raise body = self._parse_body( diff --git a/src/aap_eda/core/migrations/0074_eventstreamsetting.py b/src/aap_eda/core/migrations/0074_eventstreamsetting.py index 93139863f..96bd79f6d 100644 --- a/src/aap_eda/core/migrations/0074_eventstreamsetting.py +++ b/src/aap_eda/core/migrations/0074_eventstreamsetting.py @@ -31,7 +31,7 @@ class Migration(migrations.Migration): editable=False, null=True, on_delete=django.db.models.deletion.SET_NULL, - related_name="%(class)s_created+", + related_name="%s(class)s_created+", to=settings.AUTH_USER_MODEL, ), ), @@ -42,7 +42,7 @@ class Migration(migrations.Migration): editable=False, null=True, on_delete=django.db.models.deletion.SET_NULL, - related_name="%(class)s_modified+", + related_name="%s(class)s_modified+", to=settings.AUTH_USER_MODEL, ), ), @@ -60,9 +60,11 @@ class Migration(migrations.Migration): blank=True, default=list, help_text=( - "IP allowlist. When non-empty, only these IPs " - "may post events to any event stream in this " - "organization." + "IP allowlist. Accepts individual IPs " + "and CIDR ranges (e.g. 192.30.252.0/22). " + "When non-empty, only matching IPs may " + "post events. Empty means all IPs are " + "allowed." ), ), ), @@ -72,36 +74,9 @@ class Migration(migrations.Migration): blank=True, default=list, help_text=( - "Admin-managed list of permanently blocked IPs." - ), - ), - ), - ( - "blacklist_threshold", - models.PositiveIntegerField( - default=5, - help_text=( - "Number of auth failures before an IP is " - "auto-blacklisted. Set to 0 to disable " - "auto-blacklisting." - ), - ), - ), - ( - "blacklist_window", - models.PositiveIntegerField( - default=60, - help_text=( - "Seconds within which failures are counted." - ), - ), - ), - ( - "lockout_duration", - models.PositiveIntegerField( - default=3600, - help_text=( - "Seconds an auto-blacklisted IP stays blocked." + "IPs that attempted access and were " + "rejected. Auto-populated on failed " + "requests for admin visibility." ), ), ), @@ -117,6 +92,7 @@ class Migration(migrations.Migration): options={ "db_table": "core_event_stream_setting", "ordering": ("-created_at",), + "default_permissions": ("add", "change", "view"), }, ), ] diff --git a/src/aap_eda/core/models/event_stream_setting.py b/src/aap_eda/core/models/event_stream_setting.py index 6e2c0f846..8286c9f80 100644 --- a/src/aap_eda/core/models/event_stream_setting.py +++ b/src/aap_eda/core/models/event_stream_setting.py @@ -20,14 +20,16 @@ class EventStreamSetting(PrimordialModel): """Per-organization IP security settings for event streams. - Controls which IPs may post events (allowlist), which are - permanently blocked (blocklist), and auto-blacklist behavior - (threshold, window, lockout duration). + Controls which IPs may post events (allowlist) and tracks + IPs that attempted access and were rejected (blocklist). + Admins can promote blocked IPs to the allowlist or remove + them from the blocklist entirely. """ class Meta: db_table = "core_event_stream_setting" ordering = ("-created_at",) + default_permissions = ("add", "change", "view") organization = models.OneToOneField( "Organization", @@ -38,29 +40,19 @@ class Meta: default=list, blank=True, help_text=( - "IP allowlist. When non-empty, only these IPs may post " - "events to any event stream in this organization." + "IP allowlist. Accepts individual IPs and CIDR " + "ranges (e.g. 192.30.252.0/22). When non-empty, " + "only matching IPs may post events. " + "Empty means all IPs are allowed." ), ) blocked_ips = models.JSONField( default=list, blank=True, - help_text="Admin-managed list of permanently blocked IPs.", - ) - blacklist_threshold = models.PositiveIntegerField( - default=5, help_text=( - "Number of auth failures before an IP is auto-blacklisted. " - "Set to 0 to disable auto-blacklisting." + "IPs that attempted access and were rejected. " + "Auto-populated on failed requests for admin visibility." ), ) - blacklist_window = models.PositiveIntegerField( - default=60, - help_text="Seconds within which failures are counted.", - ) - lockout_duration = models.PositiveIntegerField( - default=3600, - help_text="Seconds an auto-blacklisted IP stays blocked.", - ) created_at = models.DateTimeField(auto_now_add=True) modified_at = models.DateTimeField(auto_now=True) diff --git a/src/aap_eda/services/event_stream_settings_cache.py b/src/aap_eda/services/event_stream_settings_cache.py index 2c8f2d2a0..427e16916 100644 --- a/src/aap_eda/services/event_stream_settings_cache.py +++ b/src/aap_eda/services/event_stream_settings_cache.py @@ -12,15 +12,15 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Per-organization event stream settings cache with signal-based invalidation. +"""Per-organization event stream settings cache. -Settings are cached for SETTINGS_CACHE_TTL seconds. A post_save signal -on EventStreamSetting invalidates the cache so changes propagate -immediately (within the same cache backend). +Settings are cached for SETTINGS_CACHE_TTL seconds. A post_save +signal on EventStreamSetting invalidates the cache so changes +propagate immediately (within the same cache backend). """ import logging -from typing import Any +from typing import Any, Optional from django.core.cache import cache from django.db.models.signals import post_save @@ -30,21 +30,22 @@ SETTINGS_CACHE_PREFIX = "es_org_settings" SETTINGS_CACHE_TTL = 300 +_CACHE_MISS = object() def _cache_key(org_id: int) -> str: - return f"{SETTINGS_CACHE_PREFIX}:{org_id}" + return f"{SETTINGS_CACHE_PREFIX}:{org_id}" # noqa: E231 -def get_org_settings(org_id: int) -> dict: +def get_org_settings(org_id: int) -> Optional[dict]: """Get per-org event stream settings, from cache or DB. - Falls back to global Dynaconf defaults if no DB row exists - for the given organization. + Returns None if no settings row exists for this org + (meaning no IP restrictions are configured). """ key = _cache_key(org_id) - cached = cache.get(key) - if cached is not None: + cached = cache.get(key, _CACHE_MISS) + if cached is not _CACHE_MISS: return cached from aap_eda.core.models import EventStreamSetting @@ -54,26 +55,9 @@ def get_org_settings(org_id: int) -> dict: data = { "allowed_ips": set(setting.allowed_ips), "blocked_ips": set(setting.blocked_ips), - "blacklist_threshold": setting.blacklist_threshold, - "blacklist_window": setting.blacklist_window, - "lockout_duration": setting.lockout_duration, } except EventStreamSetting.DoesNotExist: - from django.conf import settings as django_settings - - data = { - "allowed_ips": set(), - "blocked_ips": set(), - "blacklist_threshold": getattr( - django_settings, "EVENT_STREAM_BLACKLIST_THRESHOLD", 5 - ), - "blacklist_window": getattr( - django_settings, "EVENT_STREAM_BLACKLIST_WINDOW", 60 - ), - "lockout_duration": getattr( - django_settings, "EVENT_STREAM_BLACKLIST_DURATION", 3600 - ), - } + data = None cache.set(key, data, SETTINGS_CACHE_TTL) return data diff --git a/src/aap_eda/settings/defaults.py b/src/aap_eda/settings/defaults.py index 848223a89..d69d67103 100644 --- a/src/aap_eda/settings/defaults.py +++ b/src/aap_eda/settings/defaults.py @@ -239,17 +239,6 @@ # export EDA_EVENT_STREAM_REQUIRE_TRUSTED_PROXY=False EVENT_STREAM_REQUIRE_TRUSTED_PROXY: bool = True -# IP blacklisting for event stream abuse prevention -# Set threshold to 0 to disable blacklisting entirely -# Note: blacklisting requires a shared cache backend (Redis, -# Memcached) to work across multiple workers. The default -# LocMemCache is per-process and will not share state. -# For test deployments: export EDA_EVENT_STREAM_BLACKLIST_THRESHOLD=0 -# or set EDA_MODE=testing (loads testing_defaults.py) -EVENT_STREAM_BLACKLIST_THRESHOLD: int = 5 -EVENT_STREAM_BLACKLIST_WINDOW: int = 60 # seconds -EVENT_STREAM_BLACKLIST_DURATION: int = 3600 # seconds (1 hour) - MAX_PG_NOTIFY_MESSAGE_SIZE: int = 6144 # Database credentials for the event streams user diff --git a/src/aap_eda/settings/testing_defaults.py b/src/aap_eda/settings/testing_defaults.py index c31d6230a..c47431ef2 100644 --- a/src/aap_eda/settings/testing_defaults.py +++ b/src/aap_eda/settings/testing_defaults.py @@ -11,4 +11,3 @@ # 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. -EVENT_STREAM_BLACKLIST_THRESHOLD = 0 diff --git a/tests/integration/api/conftest.py b/tests/integration/api/conftest.py index 6c73448d2..4a8956008 100644 --- a/tests/integration/api/conftest.py +++ b/tests/integration/api/conftest.py @@ -6,10 +6,9 @@ @pytest.fixture(autouse=True) -def disable_blacklisting(settings): - """Disable IP blacklisting and clear cache for tests.""" +def clear_cache_for_tests(): + """Clear cache before and after each test.""" cache.clear() - settings.EVENT_STREAM_BLACKLIST_THRESHOLD = 0 yield cache.clear() diff --git a/tests/integration/api/test_event_stream_setting.py b/tests/integration/api/test_event_stream_setting.py index a8a0efca5..260c11cdb 100644 --- a/tests/integration/api/test_event_stream_setting.py +++ b/tests/integration/api/test_event_stream_setting.py @@ -12,11 +12,22 @@ # See the License for the specific language governing permissions and # limitations under the License. +import hashlib +import hmac +import secrets + import pytest from rest_framework import status +from rest_framework.renderers import JSONRenderer from rest_framework.test import APIClient from aap_eda.core import models +from tests.integration.api.test_event_stream import ( + create_event_stream, + create_event_stream_credential, + event_stream_post_url, + get_default_test_org, +) from tests.integration.constants import api_url_v1 SETTINGS_URL = f"{api_url_v1}/event-stream-settings" @@ -28,29 +39,26 @@ def default_event_stream_setting(default_organization): organization=default_organization, allowed_ips=["10.0.0.1", "10.0.0.2"], blocked_ips=["192.168.1.100"], - blacklist_threshold=3, - blacklist_window=30, - lockout_duration=1800, ) @pytest.mark.django_db class TestEventStreamSettingCreate: - def test_create(self, admin_client: APIClient, default_organization): + def test_create( + self, + admin_client: APIClient, + default_organization, + ): response = admin_client.post( f"{SETTINGS_URL}/", data={ "organization_id": default_organization.id, "allowed_ips": ["10.0.0.1"], - "blocked_ips": [], - "blacklist_threshold": 5, - "blacklist_window": 60, - "lockout_duration": 3600, }, ) assert response.status_code == status.HTTP_201_CREATED assert response.data["allowed_ips"] == ["10.0.0.1"] - assert response.data["blacklist_threshold"] == 5 + assert response.data["blocked_ips"] == [] def test_create_duplicate_org_rejected( self, @@ -62,24 +70,24 @@ def test_create_duplicate_org_rejected( f"{SETTINGS_URL}/", data={ "organization_id": default_organization.id, - "allowed_ips": [], }, ) assert response.status_code == status.HTTP_400_BAD_REQUEST def test_create_with_defaults( - self, admin_client: APIClient, default_organization + self, + admin_client: APIClient, + default_organization, ): response = admin_client.post( f"{SETTINGS_URL}/", - data={"organization_id": default_organization.id}, + data={ + "organization_id": default_organization.id, + }, ) assert response.status_code == status.HTTP_201_CREATED assert response.data["allowed_ips"] == [] assert response.data["blocked_ips"] == [] - assert response.data["blacklist_threshold"] == 5 - assert response.data["blacklist_window"] == 60 - assert response.data["lockout_duration"] == 3600 @pytest.mark.django_db @@ -92,7 +100,10 @@ def test_retrieve( pk = default_event_stream_setting.id response = admin_client.get(f"{SETTINGS_URL}/{pk}/") assert response.status_code == status.HTTP_200_OK - assert response.data["allowed_ips"] == ["10.0.0.1", "10.0.0.2"] + assert response.data["allowed_ips"] == [ + "10.0.0.1", + "10.0.0.2", + ] assert response.data["organization"] is not None def test_list( @@ -120,36 +131,86 @@ def test_partial_update_allowed_ips( assert response.status_code == status.HTTP_200_OK assert response.data["allowed_ips"] == ["172.16.0.1"] - def test_partial_update_lockout_duration( + def test_adding_to_allowlist_removes_from_blocked( self, admin_client: APIClient, default_event_stream_setting, ): pk = default_event_stream_setting.id + assert "192.168.1.100" in (default_event_stream_setting.blocked_ips) response = admin_client.patch( f"{SETTINGS_URL}/{pk}/", - data={"lockout_duration": 7200}, + data={ + "allowed_ips": [ + "10.0.0.1", + "10.0.0.2", + "192.168.1.100", + ], + }, ) assert response.status_code == status.HTTP_200_OK - assert response.data["lockout_duration"] == 7200 + assert "192.168.1.100" in response.data["allowed_ips"] + assert "192.168.1.100" not in response.data["blocked_ips"] @pytest.mark.django_db class TestEventStreamSettingValidation: - def test_invalid_ip_rejected( - self, admin_client: APIClient, default_organization + @pytest.mark.parametrize( + "bad_entry", + [ + "not-an-ip", + "999.999.999.999", + "192.168.1", + "192.168.1.1.1", + "", + "abc::xyz", + ], + ) + def test_malformed_ip_rejected( + self, + admin_client: APIClient, + default_organization, + bad_entry, ): response = admin_client.post( f"{SETTINGS_URL}/", data={ "organization_id": default_organization.id, - "allowed_ips": ["not-an-ip"], + "allowed_ips": [bad_entry], + }, + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + + @pytest.mark.parametrize( + "bad_cidr", + [ + "192.168.1.0/abc", + "192.168.1.0/33", + "192.168.1.0/-1", + "not-a-network/24", + "/24", + "192.168.1.0/", + ], + ) + def test_malformed_cidr_rejected( + self, + admin_client: APIClient, + default_organization, + bad_cidr, + ): + response = admin_client.post( + f"{SETTINGS_URL}/", + data={ + "organization_id": default_organization.id, + "allowed_ips": [bad_cidr], }, ) assert response.status_code == status.HTTP_400_BAD_REQUEST def test_max_255_ips_enforced( - self, admin_client: APIClient, default_organization + self, + admin_client: APIClient, + default_organization, ): ips = [f"10.0.{i // 256}.{i % 256}" for i in range(256)] response = admin_client.post( @@ -162,7 +223,9 @@ def test_max_255_ips_enforced( assert response.status_code == status.HTTP_400_BAD_REQUEST def test_ipv6_accepted( - self, admin_client: APIClient, default_organization + self, + admin_client: APIClient, + default_organization, ): response = admin_client.post( f"{SETTINGS_URL}/", @@ -173,16 +236,47 @@ def test_ipv6_accepted( ) assert response.status_code == status.HTTP_201_CREATED + def test_cidr_accepted( + self, + admin_client: APIClient, + default_organization, + ): + response = admin_client.post( + f"{SETTINGS_URL}/", + data={ + "organization_id": default_organization.id, + "allowed_ips": [ + "192.30.252.0/22", + "10.0.0.1", + "2a0a:a440::/29", + ], + }, + ) + assert response.status_code == status.HTTP_201_CREATED + assert "192.30.252.0/22" in response.data["allowed_ips"] + @pytest.mark.django_db -class TestClearBlocked: - def test_clear_blocked( +class TestRemoveBlocked: + def test_remove_specific_ips( + self, + admin_client: APIClient, + default_event_stream_setting, + ): + pk = default_event_stream_setting.id + response = admin_client.post( + f"{SETTINGS_URL}/{pk}/remove-blocked/", + data={"ips": ["192.168.1.100"]}, + ) + assert response.status_code == status.HTTP_200_OK + assert response.data["blocked_ips"] == [] + + def test_clear_all_blocked( self, admin_client: APIClient, default_event_stream_setting, ): pk = default_event_stream_setting.id - assert default_event_stream_setting.blocked_ips == ["192.168.1.100"] response = admin_client.post( f"{SETTINGS_URL}/{pk}/clear-blocked/", ) @@ -190,3 +284,94 @@ def test_clear_blocked( assert response.data["blocked_ips"] == [] default_event_stream_setting.refresh_from_db() assert default_event_stream_setting.blocked_ips == [] + + +def _create_hmac_event_stream(admin_client, org): + """Create an HMAC-authenticated event stream for IP policy tests.""" + secret = secrets.token_hex(32) + header_key = "X-Hub-Signature" + cred = create_event_stream_credential( + admin_client, + "HMAC Event Stream", + { + "auth_type": "hmac", + "secret": secret, + "http_header_key": header_key, + "hash_algorithm": "sha256", + "signature_encoding": "hex", + }, + name="ip-policy-test-cred", + ) + es = create_event_stream( + admin_client, + { + "name": "ip-policy-test-es", + "event_stream_type": cred["credential_type"]["kind"], + "eda_credential_id": cred["id"], + "organization_id": org.id, + }, + ) + return es, secret, header_key + + +@pytest.mark.django_db +class TestIpPolicyOnPost: + def test_allowlist_rejects_unlisted_ip( + self, + admin_client: APIClient, + preseed_credential_types, + ): + org = get_default_test_org() + es, secret, header_key = _create_hmac_event_stream(admin_client, org) + models.EventStreamSetting.objects.create( + organization=org, + allowed_ips=["192.168.1.1"], + ) + data = {"test": "payload"} + data_bytes = JSONRenderer().render(data) + sig = hmac.new( + secret.encode(), msg=data_bytes, digestmod=hashlib.sha256 + ).hexdigest() + response = admin_client.post( + event_stream_post_url(es.uuid), + headers={header_key: sig}, + data=data, + ) + assert response.status_code == status.HTTP_403_FORBIDDEN + + def test_no_settings_allows_request( + self, + admin_client: APIClient, + preseed_credential_types, + ): + org = get_default_test_org() + es, secret, header_key = _create_hmac_event_stream(admin_client, org) + data = {"test": "payload"} + data_bytes = JSONRenderer().render(data) + sig = hmac.new( + secret.encode(), msg=data_bytes, digestmod=hashlib.sha256 + ).hexdigest() + response = admin_client.post( + event_stream_post_url(es.uuid), + headers={header_key: sig}, + data=data, + ) + assert response.status_code == status.HTTP_200_OK + + def test_rejection_records_blocked_ip( + self, + admin_client: APIClient, + preseed_credential_types, + ): + org = get_default_test_org() + es, _, _ = _create_hmac_event_stream(admin_client, org) + setting = models.EventStreamSetting.objects.create( + organization=org, + allowed_ips=["192.168.1.1"], + ) + admin_client.post( + event_stream_post_url(es.uuid), + data={"test": "payload"}, + ) + setting.refresh_from_db() + assert len(setting.blocked_ips) > 0 diff --git a/tests/integration/api/test_root.py b/tests/integration/api/test_root.py index 93b0628b6..ff074bf2c 100644 --- a/tests/integration/api/test_root.py +++ b/tests/integration/api/test_root.py @@ -45,6 +45,7 @@ "/organizations/", "/teams/", "/event-streams/", + "/event-stream-settings/", "/credential-input-sources/", "/role_definitions/", "/role_user_assignments/", @@ -97,6 +98,7 @@ "/organizations/", "/teams/", "/event-streams/", + "/event-stream-settings/", "feature_flags/states/", # To be removed after all components # have migrated away from this endpoint diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 2f703b284..886438e64 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -1199,6 +1199,15 @@ def default_event_stream( ) +@pytest.fixture +def default_event_stream_setting( + default_organization: models.Organization, +) -> models.EventStreamSetting: + return models.EventStreamSetting.objects.create( + organization=default_organization, + ) + + @pytest.fixture def activation_payload_skip_audit_events(activation_payload: dict) -> dict: activation_payload["skip_audit_events"] = True diff --git a/tests/integration/dab_rbac/conftest.py b/tests/integration/dab_rbac/conftest.py index fdb520ad2..aed85ac9d 100644 --- a/tests/integration/dab_rbac/conftest.py +++ b/tests/integration/dab_rbac/conftest.py @@ -82,9 +82,9 @@ def get_post_data(self, model_obj): post_data["eda_credentials"] = [obj["id"] for obj in related_objs] # handle EventStream uuid field - let serializer generate new UUID elif model_obj._meta.model == models.EventStream: - # Remove uuid from post_data to avoid uniqueness conflicts - # The serializer will generate a new UUID automatically post_data.pop("uuid", None) + elif model_obj._meta.model == models.EventStreamSetting: + model_obj.delete() return post_data diff --git a/tests/integration/dab_rbac/test_organization.py b/tests/integration/dab_rbac/test_organization.py index 70d118273..bf9268223 100644 --- a/tests/integration/dab_rbac/test_organization.py +++ b/tests/integration/dab_rbac/test_organization.py @@ -54,6 +54,8 @@ def test_create_with_default_org( if model._meta.model_name == "team": pytest.skip("Team model requires an organization") + if model._meta.model_name == "eventstreamsetting": + pytest.skip("Singleton per-org model tested separately") try: url = reverse(f"{model._meta.model_name}-list") @@ -116,6 +118,9 @@ def test_create_with_custom_org( # factory returns data with default org so we have to change it here post_data["organization_id"] = new_organization.id + if model._meta.model_name == "eventstreamsetting": + pytest.skip("Singleton per-org model tested separately") + try: url = reverse(f"{model._meta.model_name}-list") except NoReverseMatch: diff --git a/tests/unit/test_blacklist.py b/tests/unit/test_blacklist.py index ecffb7ec8..4d6b0f488 100644 --- a/tests/unit/test_blacklist.py +++ b/tests/unit/test_blacklist.py @@ -17,6 +17,7 @@ from rest_framework.exceptions import AuthenticationFailed from aap_eda.api.blacklist import BlacklistManager +from aap_eda.core import models @pytest.fixture(autouse=True) @@ -31,55 +32,79 @@ def manager(): return BlacklistManager() -@pytest.fixture -def blacklist_settings(settings): - settings.EVENT_STREAM_BLACKLIST_THRESHOLD = 5 - settings.EVENT_STREAM_BLACKLIST_WINDOW = 60 - settings.EVENT_STREAM_BLACKLIST_DURATION = 3600 - return settings - - -class TestBlacklisting: - def test_single_failure_not_blacklisted(self, manager, blacklist_settings): - manager.record_failure("10.0.0.1") - manager.check_blacklist("10.0.0.1") - - def test_threshold_triggers_blacklist(self, manager, blacklist_settings): - blacklist_settings.EVENT_STREAM_BLACKLIST_THRESHOLD = 3 +@pytest.mark.django_db +class TestCheckIpPolicy: + def test_no_settings_allows_all(self, manager, default_organization): + manager.check_ip_policy("10.0.0.1", default_organization.id) - for _ in range(3): - manager.record_failure("10.0.0.1") + def test_empty_allowlist_allows_all(self, manager, default_organization): + models.EventStreamSetting.objects.create( + organization=default_organization, + allowed_ips=[], + ) + manager.check_ip_policy("10.0.0.1", default_organization.id) - with pytest.raises(AuthenticationFailed): - manager.check_blacklist("10.0.0.1") - - def test_below_threshold_not_blacklisted( - self, manager, blacklist_settings + def test_allowlist_rejects_unlisted_ip( + self, manager, default_organization ): - for _ in range(4): - manager.record_failure("10.0.0.1") - - manager.check_blacklist("10.0.0.1") - - def test_per_ip_isolation(self, manager, blacklist_settings): - blacklist_settings.EVENT_STREAM_BLACKLIST_THRESHOLD = 3 - - for _ in range(3): - manager.record_failure("10.0.0.1") - - manager.check_blacklist("10.0.0.2") - - def test_clean_ip_passes(self, manager): - manager.check_blacklist("10.0.0.1") - - -class TestDisabledBlacklisting: - def test_zero_threshold_disables_blacklisting( - self, manager, blacklist_settings + models.EventStreamSetting.objects.create( + organization=default_organization, + allowed_ips=["10.0.0.1"], + ) + with pytest.raises(AuthenticationFailed, match="allowlist"): + manager.check_ip_policy("10.0.0.99", default_organization.id) + + def test_allowlist_passes_listed_ip(self, manager, default_organization): + models.EventStreamSetting.objects.create( + organization=default_organization, + allowed_ips=["10.0.0.1"], + ) + manager.check_ip_policy("10.0.0.1", default_organization.id) + + def test_cidr_allows_ip_in_range(self, manager, default_organization): + models.EventStreamSetting.objects.create( + organization=default_organization, + allowed_ips=["192.30.252.0/22"], + ) + manager.check_ip_policy("192.30.253.5", default_organization.id) + + def test_cidr_rejects_ip_outside_range( + self, manager, default_organization ): - blacklist_settings.EVENT_STREAM_BLACKLIST_THRESHOLD = 0 - - for _ in range(10): - manager.record_failure("10.0.0.1") - - manager.check_blacklist("10.0.0.1") + models.EventStreamSetting.objects.create( + organization=default_organization, + allowed_ips=["192.30.252.0/22"], + ) + with pytest.raises(AuthenticationFailed, match="allowlist"): + manager.check_ip_policy("10.0.0.1", default_organization.id) + + def test_mixed_ips_and_cidrs(self, manager, default_organization): + models.EventStreamSetting.objects.create( + organization=default_organization, + allowed_ips=["10.0.0.1", "192.30.252.0/22"], + ) + manager.check_ip_policy("10.0.0.1", default_organization.id) + manager.check_ip_policy("192.30.255.100", default_organization.id) + + +@pytest.mark.django_db +class TestRecordBlockedIp: + def test_records_rejected_ip(self, manager, default_organization): + setting = models.EventStreamSetting.objects.create( + organization=default_organization, + ) + manager.record_blocked_ip("10.0.0.1", default_organization.id) + setting.refresh_from_db() + assert "10.0.0.1" in setting.blocked_ips + + def test_does_not_duplicate(self, manager, default_organization): + setting = models.EventStreamSetting.objects.create( + organization=default_organization, + blocked_ips=["10.0.0.1"], + ) + manager.record_blocked_ip("10.0.0.1", default_organization.id) + setting.refresh_from_db() + assert setting.blocked_ips.count("10.0.0.1") == 1 + + def test_no_settings_row_is_noop(self, manager, default_organization): + manager.record_blocked_ip("10.0.0.1", default_organization.id) diff --git a/tests/unit/test_blacklist_org_aware.py b/tests/unit/test_blacklist_org_aware.py index 3ffc0af3e..073d86d32 100644 --- a/tests/unit/test_blacklist_org_aware.py +++ b/tests/unit/test_blacklist_org_aware.py @@ -32,127 +32,49 @@ def manager(): return BlacklistManager() -@pytest.fixture -def blacklist_settings(settings): - settings.EVENT_STREAM_BLACKLIST_THRESHOLD = 5 - settings.EVENT_STREAM_BLACKLIST_WINDOW = 60 - settings.EVENT_STREAM_BLACKLIST_DURATION = 3600 - return settings - - @pytest.mark.django_db -class TestCheckIpPolicy: - def test_blocked_ip_rejected( - self, manager, default_organization, blacklist_settings - ): - models.EventStreamSetting.objects.create( - organization=default_organization, - blocked_ips=["10.0.0.1"], - ) - with pytest.raises(AuthenticationFailed, match="blocked"): - manager.check_ip_policy("10.0.0.1", default_organization.id) - - def test_unblocked_ip_passes( - self, manager, default_organization, blacklist_settings - ): - models.EventStreamSetting.objects.create( - organization=default_organization, - blocked_ips=["10.0.0.1"], - ) - manager.check_ip_policy("10.0.0.2", default_organization.id) - - def test_allowed_ips_enforced( - self, manager, default_organization, blacklist_settings - ): +class TestOrgIsolation: + def test_separate_allowlists(self, manager): + org_a = models.Organization.objects.create(name="A") + org_b = models.Organization.objects.create(name="B") models.EventStreamSetting.objects.create( - organization=default_organization, - allowed_ips=["10.0.0.1", "10.0.0.2"], - ) - with pytest.raises(AuthenticationFailed, match="allowlist"): - manager.check_ip_policy("10.0.0.99", default_organization.id) - - def test_allowed_ips_passes_listed_ip( - self, manager, default_organization, blacklist_settings - ): - models.EventStreamSetting.objects.create( - organization=default_organization, + organization=org_a, allowed_ips=["10.0.0.1"], ) - manager.check_ip_policy("10.0.0.1", default_organization.id) - - def test_empty_allowed_ips_allows_all( - self, manager, default_organization, blacklist_settings - ): models.EventStreamSetting.objects.create( - organization=default_organization, - allowed_ips=[], + organization=org_b, + allowed_ips=["10.0.0.2"], ) - manager.check_ip_policy("10.0.0.99", default_organization.id) + manager.check_ip_policy("10.0.0.1", org_a.id) + with pytest.raises(AuthenticationFailed): + manager.check_ip_policy("10.0.0.1", org_b.id) - def test_auto_blacklisted_ip_rejected( - self, manager, default_organization, blacklist_settings - ): - models.EventStreamSetting.objects.create( - organization=default_organization, - blacklist_threshold=2, + def test_blocked_ips_per_org(self, manager): + org_a = models.Organization.objects.create(name="A") + org_b = models.Organization.objects.create(name="B") + setting_a = models.EventStreamSetting.objects.create( + organization=org_a, ) - org_id = default_organization.id - manager.record_failure("10.0.0.1", org_id=org_id) - manager.record_failure("10.0.0.1", org_id=org_id) - with pytest.raises(AuthenticationFailed, match="Too many"): - manager.check_ip_policy("10.0.0.1", org_id) - - def test_no_settings_row_uses_global_defaults( - self, manager, default_organization, blacklist_settings - ): - manager.check_ip_policy("10.0.0.1", default_organization.id) - - -@pytest.mark.django_db -class TestOrgAwareRecordFailure: - def test_per_org_threshold( - self, manager, default_organization, blacklist_settings - ): models.EventStreamSetting.objects.create( - organization=default_organization, - blacklist_threshold=2, - blacklist_window=60, - lockout_duration=3600, + organization=org_b, ) - org_id = default_organization.id - manager.record_failure("10.0.0.1", org_id=org_id) - manager.record_failure("10.0.0.1", org_id=org_id) - key = f"es_blacklist:{org_id}:10.0.0.1" - assert cache.get(key) is True + manager.record_blocked_ip("10.0.0.1", org_a.id) + setting_a.refresh_from_db() + setting_b = models.EventStreamSetting.objects.get(organization=org_b) + assert "10.0.0.1" in setting_a.blocked_ips + assert "10.0.0.1" not in setting_b.blocked_ips - def test_org_isolation(self, manager, blacklist_settings): - org_a = models.Organization.objects.create(name="Org A") - org_b = models.Organization.objects.create(name="Org B") - models.EventStreamSetting.objects.create( - organization=org_a, blacklist_threshold=2 - ) - models.EventStreamSetting.objects.create( - organization=org_b, blacklist_threshold=2 - ) - manager.record_failure("10.0.0.1", org_id=org_a.id) - manager.record_failure("10.0.0.1", org_id=org_a.id) - manager.check_ip_policy("10.0.0.1", org_b.id) - def test_zero_threshold_disables( - self, manager, default_organization, blacklist_settings +@pytest.mark.django_db +class TestAllowlistRejectsAndRecords: + def test_rejection_auto_records_blocked_ip( + self, manager, default_organization ): - models.EventStreamSetting.objects.create( + setting = models.EventStreamSetting.objects.create( organization=default_organization, - blacklist_threshold=0, + allowed_ips=["10.0.0.1"], ) - org_id = default_organization.id - for _ in range(10): - manager.record_failure("10.0.0.1", org_id=org_id) - manager.check_ip_policy("10.0.0.1", org_id) - - def test_global_fallback_without_org_id(self, manager, blacklist_settings): - blacklist_settings.EVENT_STREAM_BLACKLIST_THRESHOLD = 2 - manager.record_failure("10.0.0.1") - manager.record_failure("10.0.0.1") with pytest.raises(AuthenticationFailed): - manager.check_blacklist("10.0.0.1") + manager.check_ip_policy("10.0.0.99", default_organization.id) + setting.refresh_from_db() + assert "10.0.0.99" in setting.blocked_ips diff --git a/tests/unit/test_event_stream_setting_model.py b/tests/unit/test_event_stream_setting_model.py index 4034838d0..bbaf13ee0 100644 --- a/tests/unit/test_event_stream_setting_model.py +++ b/tests/unit/test_event_stream_setting_model.py @@ -36,16 +36,13 @@ def test_returns_db_values(self, default_organization): organization=default_organization, allowed_ips=["10.0.0.1", "10.0.0.2"], blocked_ips=["192.168.1.100"], - blacklist_threshold=3, - blacklist_window=30, - lockout_duration=1800, ) result = get_org_settings(default_organization.id) - assert result["allowed_ips"] == {"10.0.0.1", "10.0.0.2"} + assert result["allowed_ips"] == { + "10.0.0.1", + "10.0.0.2", + } assert result["blocked_ips"] == {"192.168.1.100"} - assert result["blacklist_threshold"] == 3 - assert result["blacklist_window"] == 30 - assert result["lockout_duration"] == 1800 def test_caches_result(self, default_organization): models.EventStreamSetting.objects.create( @@ -59,16 +56,9 @@ def test_caches_result(self, default_organization): result2 = get_org_settings(default_organization.id) assert result1["allowed_ips"] == result2["allowed_ips"] - def test_fallback_to_global_defaults(self, default_organization, settings): - settings.EVENT_STREAM_BLACKLIST_THRESHOLD = 7 - settings.EVENT_STREAM_BLACKLIST_WINDOW = 120 - settings.EVENT_STREAM_BLACKLIST_DURATION = 7200 + def test_no_row_returns_none(self, default_organization): result = get_org_settings(default_organization.id) - assert result["allowed_ips"] == set() - assert result["blocked_ips"] == set() - assert result["blacklist_threshold"] == 7 - assert result["blacklist_window"] == 120 - assert result["lockout_duration"] == 7200 + assert result is None def test_invalidate_clears_cache(self, default_organization): models.EventStreamSetting.objects.create(