From a17f5c167b24b8639baa7c5fc5841d6a7e3e6bcb Mon Sep 17 00:00:00 2001 From: Dirk Julich <202084408+djulich@users.noreply.github.com> Date: Wed, 21 Oct 2026 15:02:29 +0200 Subject: [PATCH 1/3] Add SSRF egress filtering and allowlist to notification backends (#16679) * AAP-94827 Add SSRF egress filtering and allowlist to notification backends Add url_validation.py to block notification URLs that resolve to private, loopback, link-local, or reserved addresses. Admins can allowlist specific IPs, CIDR ranges, or hostnames via the new NOTIFICATION_IP_ALLOW_LIST system setting. The allowlist ships together with the filter to avoid breaking webhook integrations targeting internal (RFC1918) endpoints. * AAP-94827 Mock validate_url in functional redirect test The functional test uses a fake hostname ('some.fake.url') that fails DNS resolution when validate_url tries to resolve it in CI. * AAP-94163 Add non-overridable deny list for link-local addresses Link-local addresses (169.254.0.0/16, fe80::/10) are now always blocked regardless of allowlist settings, protecting cloud metadata endpoints even if DNS is subverted for an allowlisted hostname. Requested by Product Security as a condition for approving hostname support in the NOTIFICATION_IP_ALLOW_LIST setting. * AAP-94163 Block IPv4-mapped link-local addresses in deny list Unwrap IPv4-mapped IPv6 addresses (e.g. ::ffff:169.254.169.254) before checking the link-local deny list, preventing a bypass where the mapped address doesn't match either the IPv4 or IPv6 deny network directly. --------- Co-authored-by: Claude Opus 4.6 --- awx/main/conf.py | 46 +++ awx/main/notifications/grafana_backend.py | 13 +- awx/main/notifications/mattermost_backend.py | 12 +- awx/main/notifications/rocketchat_backend.py | 12 +- awx/main/notifications/url_validation.py | 90 ++++++ awx/main/notifications/webhook_backend.py | 16 ++ .../tests/functional/test_notifications.py | 3 +- .../unit/notifications/test_url_validation.py | 263 ++++++++++++++++++ .../tests/unit/notifications/test_webhook.py | 6 + 9 files changed, 457 insertions(+), 4 deletions(-) create mode 100644 awx/main/notifications/url_validation.py create mode 100644 awx/main/tests/unit/notifications/test_url_validation.py diff --git a/awx/main/conf.py b/awx/main/conf.py index 1babd1b70c..f0d4d2ee56 100644 --- a/awx/main/conf.py +++ b/awx/main/conf.py @@ -1,5 +1,7 @@ # Python +import ipaddress import logging +import re # Django from django.core.checks import Error @@ -103,6 +105,27 @@ register( category_slug='system', ) +register( + 'NOTIFICATION_IP_ALLOW_LIST', + field_class=fields.StringListField, + default=[], + label=_('Notification Connections Allowed List'), + help_text=_( + "List of IP addresses, CIDR ranges, and hostnames allowed as notification destinations. " + "By default, notifications to private, loopback, link-local, and reserved " + "addresses are blocked to prevent server-side request forgery (SSRF). " + "Add entries here to permit specific internal endpoints. " + "IP/CIDR entries (e.g. 10.50.0.0/24) allow matching resolved addresses. " + "Hostname entries (e.g. elastic.internal.example.com) bypass address checking for that host. " + "Use a dot prefix (e.g. .internal.example.com) to allow all subdomains. " + "Note: hostname entries trust DNS resolution for the named host; " + "use IP/CIDR entries where possible for stronger guarantees." + ), + category=_('System'), + category_slug='system', + required=False, +) + register( 'CSRF_TRUSTED_ORIGINS', default=[], @@ -1116,6 +1139,29 @@ def csrf_trusted_origins_validate(serializer, attrs): register_validate('system', csrf_trusted_origins_validate) +_HOSTNAME_RE = re.compile(r'^\.?(?:[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)*[a-zA-Z]{2,}$') + + +def notification_ip_allow_list_validate(serializer, attrs): + if 'NOTIFICATION_IP_ALLOW_LIST' not in attrs: + return attrs + errors = [] + for entry in attrs['NOTIFICATION_IP_ALLOW_LIST']: + try: + ipaddress.ip_network(entry, strict=False) + continue + except ValueError: + pass + if not _HOSTNAME_RE.match(entry): + errors.append(f"{entry!r} is not a valid IP address, CIDR range, or hostname.") + if errors: + raise serializers.ValidationError(_('\n'.join(errors))) + return attrs + + +register_validate('system', notification_ip_allow_list_validate) + + register( 'OPA_HOST', field_class=fields.CharField, diff --git a/awx/main/notifications/grafana_backend.py b/awx/main/notifications/grafana_backend.py index a48a1f389b..2765e96168 100644 --- a/awx/main/notifications/grafana_backend.py +++ b/awx/main/notifications/grafana_backend.py @@ -12,6 +12,7 @@ from django.utils.translation import gettext_lazy as _ from awx.main.notifications.base import AWXBaseEmailBackend from awx.main.notifications.custom_notification_base import CustomNotificationBase +from awx.main.notifications.url_validation import SSRFBlockedError, validate_url DEFAULT_MSG = CustomNotificationBase.DEFAULT_MSG @@ -96,8 +97,18 @@ class GrafanaBackend(AWXBaseEmailBackend, CustomNotificationBase): grafana_data['text'] = m.subject grafana_headers['Authorization'] = "Bearer {}".format(self.grafana_key) grafana_headers['Content-Type'] = "application/json" + + url = "{}/api/annotations".format(m.recipients()[0]) + try: + validate_url(url) + except SSRFBlockedError as e: + logger.error(str(e)) + if not self.fail_silently: + raise Exception(str(e)) + continue + r = requests.post( - "{}/api/annotations".format(m.recipients()[0]), + url, json=grafana_data, headers=grafana_headers, verify=(not self.grafana_no_verify_ssl), diff --git a/awx/main/notifications/mattermost_backend.py b/awx/main/notifications/mattermost_backend.py index 70476eaace..b3eb70bd76 100644 --- a/awx/main/notifications/mattermost_backend.py +++ b/awx/main/notifications/mattermost_backend.py @@ -9,6 +9,7 @@ from django.utils.translation import gettext_lazy as _ from awx.main.notifications.base import AWXBaseEmailBackend from awx.main.notifications.custom_notification_base import CustomNotificationBase +from awx.main.notifications.url_validation import SSRFBlockedError, validate_url logger = logging.getLogger('awx.main.notifications.mattermost_backend') @@ -41,7 +42,16 @@ class MattermostBackend(AWXBaseEmailBackend, CustomNotificationBase): payload['text'] = m.subject - r = requests.post("{}".format(m.recipients()[0]), json=payload, verify=(not self.mattermost_no_verify_ssl), allow_redirects=False) + url = "{}".format(m.recipients()[0]) + try: + validate_url(url) + except SSRFBlockedError as e: + logger.error(str(e)) + if not self.fail_silently: + raise Exception(str(e)) + continue + + r = requests.post(url, json=payload, verify=(not self.mattermost_no_verify_ssl), allow_redirects=False) if r.status_code < 200 or r.status_code >= 300: logger.error(smart_str(_("Error sending notification mattermost: {}").format(r.status_code))) if not self.fail_silently: diff --git a/awx/main/notifications/rocketchat_backend.py b/awx/main/notifications/rocketchat_backend.py index a3eac782c2..95a59c3576 100644 --- a/awx/main/notifications/rocketchat_backend.py +++ b/awx/main/notifications/rocketchat_backend.py @@ -11,6 +11,7 @@ from django.utils.translation import gettext_lazy as _ from awx.main.notifications.base import AWXBaseEmailBackend from awx.main.utils import get_awx_http_client_headers from awx.main.notifications.custom_notification_base import CustomNotificationBase +from awx.main.notifications.url_validation import SSRFBlockedError, validate_url logger = logging.getLogger('awx.main.notifications.rocketchat_backend') @@ -38,8 +39,17 @@ class RocketChatBackend(AWXBaseEmailBackend, CustomNotificationBase): if optvalue is not None: payload[optval] = optvalue.strip() + url = "{}".format(m.recipients()[0]) + try: + validate_url(url) + except SSRFBlockedError as e: + logger.error(str(e)) + if not self.fail_silently: + raise Exception(str(e)) + continue + r = requests.post( - "{}".format(m.recipients()[0]), + url, data=json.dumps(payload), headers=get_awx_http_client_headers(), verify=(not self.rocketchat_no_verify_ssl), diff --git a/awx/main/notifications/url_validation.py b/awx/main/notifications/url_validation.py new file mode 100644 index 0000000000..3ee4960ad1 --- /dev/null +++ b/awx/main/notifications/url_validation.py @@ -0,0 +1,90 @@ +import ipaddress +import logging +import socket +from urllib.parse import urlparse + +from django.conf import settings + +logger = logging.getLogger('awx.main.notifications.url_validation') + +# Link-local addresses are always blocked regardless of allowlist settings. +# Protects cloud metadata endpoints (169.254.169.254) even if DNS is subverted +# for an allowlisted hostname. +_DENY_NETWORKS = ( + ipaddress.ip_network('169.254.0.0/16'), + ipaddress.ip_network('fe80::/10'), +) + + +class SSRFBlockedError(Exception): + pass + + +def _is_always_denied(ip): + check_ip = getattr(ip, 'ipv4_mapped', None) or ip + for net in _DENY_NETWORKS: + if check_ip in net: + return True + return False + + +def _is_ip_allowed(ip, allow_list): + for entry in allow_list: + try: + if ip in ipaddress.ip_network(entry, strict=False): + return True + except ValueError: + pass + return False + + +def _is_hostname_allowed(hostname, allow_list): + hostname = hostname.lower() + for entry in allow_list: + try: + ipaddress.ip_network(entry, strict=False) + continue + except ValueError: + pass + entry = entry.lower() + if hostname == entry: + return True + # ".example.com" matches "foo.example.com" but not "example.com" + if entry.startswith('.') and hostname.endswith(entry): + return True + return False + + +def validate_url(url): + """Resolve the URL's hostname and reject private, loopback, and link-local addresses + unless the resolved IP or hostname is in NOTIFICATION_IP_ALLOW_LIST. + Link-local addresses (169.254.0.0/16, fe80::/10) are always blocked.""" + parsed = urlparse(url) + hostname = parsed.hostname + if not hostname: + raise SSRFBlockedError(f"Notification URL has no hostname: {url}") + + allow_list = getattr(settings, 'NOTIFICATION_IP_ALLOW_LIST', []) + hostname_allowed = _is_hostname_allowed(hostname, allow_list) + + try: + addrinfo = socket.getaddrinfo(hostname, parsed.port or 443, proto=socket.IPPROTO_TCP) + except socket.gaierror as e: + if hostname_allowed: + return url + raise SSRFBlockedError(f"Cannot resolve notification URL hostname {hostname!r}: {e}") + + for family, _, _, _, sockaddr in addrinfo: + ip = ipaddress.ip_address(sockaddr[0]) + if _is_always_denied(ip): + raise SSRFBlockedError(f"Notification URL {hostname!r} resolves to link-local address {ip} (always blocked)") + if not hostname_allowed: + if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved: + if not _is_ip_allowed(ip, allow_list): + raise SSRFBlockedError(f"Notification URL {hostname!r} resolves to blocked address {ip}") + + return url + + +def get_hostname(url): + return urlparse(url).hostname diff --git a/awx/main/notifications/webhook_backend.py b/awx/main/notifications/webhook_backend.py index 92f85b93fa..eab740d3c9 100644 --- a/awx/main/notifications/webhook_backend.py +++ b/awx/main/notifications/webhook_backend.py @@ -11,6 +11,7 @@ import requests from awx.main.notifications.base import AWXBaseEmailBackend from awx.main.utils import get_awx_http_client_headers from awx.main.notifications.custom_notification_base import CustomNotificationBase +from awx.main.notifications.url_validation import SSRFBlockedError, validate_url logger = logging.getLogger('awx.main.notifications.webhook_backend') @@ -72,6 +73,15 @@ class WebhookBackend(AWXBaseEmailBackend, CustomNotificationBase): err = None + try: + validate_url(url) + except SSRFBlockedError as e: + err = str(e) + logger.error(err) + if not self.fail_silently: + raise Exception(err) + continue + def _origin(value): parsed = urlparse(value) scheme = parsed.scheme.lower() @@ -112,6 +122,12 @@ class WebhookBackend(AWXBaseEmailBackend, CustomNotificationBase): err = f"Webhook notification received redirect to a blank URL from {url_log_safe}. Response headers={resp.headers}" break + try: + validate_url(url) + except SSRFBlockedError as e: + err = str(e) + break + if _origin(url) != original_origin: logger.warning("Redirect changed origin; stripping credentials") auth = None diff --git a/awx/main/tests/functional/test_notifications.py b/awx/main/tests/functional/test_notifications.py index 0ee99e03ee..898ca94f5a 100644 --- a/awx/main/tests/functional/test_notifications.py +++ b/awx/main/tests/functional/test_notifications.py @@ -208,8 +208,9 @@ def mock_post(*args, **kwargs): @pytest.mark.django_db +@mock.patch('awx.main.notifications.webhook_backend.validate_url', side_effect=lambda url: url) @mock.patch('requests.post', side_effect=mock_post) -def test_webhook_notification_pointed_to_a_redirect_launch_endpoint(post, admin, organization): +def test_webhook_notification_pointed_to_a_redirect_launch_endpoint(post, mock_validate, admin, organization): n1 = NotificationTemplate.objects.create( name="test-webhook", description="test webhook", diff --git a/awx/main/tests/unit/notifications/test_url_validation.py b/awx/main/tests/unit/notifications/test_url_validation.py new file mode 100644 index 0000000000..017a552078 --- /dev/null +++ b/awx/main/tests/unit/notifications/test_url_validation.py @@ -0,0 +1,263 @@ +import socket +from unittest import mock + +import pytest + +from awx.main.notifications.url_validation import SSRFBlockedError, get_hostname, validate_url + + +@pytest.mark.parametrize( + "ip,should_block", + [ + ("127.0.0.1", True), + ("10.0.0.1", True), + ("172.16.0.1", True), + ("192.168.1.1", True), + ("169.254.169.254", True), + ("::1", True), + ("8.8.8.8", False), + ("140.82.121.3", False), + ], +) +def test_validate_url_blocked_addresses(ip, should_block): + addrinfo = [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, '', (ip, 443))] + with mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=addrinfo): + if should_block: + with pytest.raises(SSRFBlockedError): + validate_url("https://example.com/hook") + else: + assert validate_url("https://example.com/hook") == "https://example.com/hook" + + +def test_validate_url_no_hostname(): + with pytest.raises(SSRFBlockedError, match="no hostname"): + validate_url("not-a-url") + + +def test_validate_url_unresolvable(): + with mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', side_effect=socket.gaierror("Name or service not known")): + with pytest.raises(SSRFBlockedError, match="Cannot resolve"): + validate_url("https://does.not.exist.example.com/hook") + + +def test_get_hostname(): + assert get_hostname("https://example.com/path") == "example.com" + assert get_hostname("https://other.host:8443/api") == "other.host" + + +class TestAllowList: + """Tests for the NOTIFICATION_IP_ALLOW_LIST setting.""" + + def _addrinfo(self, ip): + return [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, '', (ip, 443))] + + def test_cidr_allows_private_ip(self): + with ( + mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo('10.50.12.40')), + mock.patch('awx.main.notifications.url_validation.settings') as mock_settings, + ): + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['10.50.0.0/16'] + assert validate_url("https://elastic.internal/hook") == "https://elastic.internal/hook" + + def test_non_allowlisted_private_ip_still_blocked(self): + with ( + mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo('10.50.12.40')), + mock.patch('awx.main.notifications.url_validation.settings') as mock_settings, + ): + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['172.16.0.0/12'] + with pytest.raises(SSRFBlockedError, match="blocked address"): + validate_url("https://elastic.internal/hook") + + def test_individual_ip_allowed(self): + with ( + mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo('192.168.1.1')), + mock.patch('awx.main.notifications.url_validation.settings') as mock_settings, + ): + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['192.168.1.1'] + assert validate_url("https://grafana.local/hook") == "https://grafana.local/hook" + + def test_public_ip_unaffected_by_empty_allow_list(self): + with ( + mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo('8.8.8.8')), + mock.patch('awx.main.notifications.url_validation.settings') as mock_settings, + ): + mock_settings.NOTIFICATION_IP_ALLOW_LIST = [] + assert validate_url("https://example.com/hook") == "https://example.com/hook" + + def test_loopback_allowed_when_explicitly_listed(self): + with ( + mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo('127.0.0.1')), + mock.patch('awx.main.notifications.url_validation.settings') as mock_settings, + ): + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['127.0.0.0/8'] + assert validate_url("https://localhost/hook") == "https://localhost/hook" + + def test_metadata_ip_blocked_even_when_explicitly_listed(self): + with ( + mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo('169.254.169.254')), + mock.patch('awx.main.notifications.url_validation.settings') as mock_settings, + ): + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['169.254.169.254'] + with pytest.raises(SSRFBlockedError, match="link-local"): + validate_url("https://metadata.internal/hook") + + def test_multiple_cidrs(self): + with ( + mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo('172.16.20.15')), + mock.patch('awx.main.notifications.url_validation.settings') as mock_settings, + ): + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['10.0.0.0/8', '172.16.0.0/12'] + assert validate_url("https://alerts.internal/hook") == "https://alerts.internal/hook" + + def test_default_empty_allow_list_blocks_private(self): + with ( + mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo('10.0.0.1')), + mock.patch('awx.main.notifications.url_validation.settings') as mock_settings, + ): + mock_settings.NOTIFICATION_IP_ALLOW_LIST = [] + with pytest.raises(SSRFBlockedError, match="blocked address"): + validate_url("https://example.com/hook") + + def test_invalid_allow_list_entry_ignored(self): + with ( + mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo('10.0.0.1')), + mock.patch('awx.main.notifications.url_validation.settings') as mock_settings, + ): + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['not-a-cidr'] + with pytest.raises(SSRFBlockedError, match="blocked address"): + validate_url("https://example.com/hook") + + +class TestHostnameAllowList: + """Tests for hostname entries in NOTIFICATION_IP_ALLOW_LIST.""" + + def _addrinfo(self, ip): + return [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, '', (ip, 443))] + + def test_exact_hostname_bypasses_ssrf_check(self): + with mock.patch('awx.main.notifications.url_validation.settings') as mock_settings: + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['elastic.internal.example.com'] + with mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo('10.50.12.40')): + assert validate_url("https://elastic.internal.example.com/hook") == "https://elastic.internal.example.com/hook" + + def test_hostname_match_is_case_insensitive(self): + with mock.patch('awx.main.notifications.url_validation.settings') as mock_settings: + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['Elastic.Internal.Example.Com'] + with mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo('10.50.12.40')): + assert validate_url("https://elastic.internal.example.com/hook") == "https://elastic.internal.example.com/hook" + + def test_dot_prefix_matches_subdomains(self): + with mock.patch('awx.main.notifications.url_validation.settings') as mock_settings: + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['.internal.example.com'] + with mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo('10.50.12.40')): + assert validate_url("https://elastic.internal.example.com/hook") == "https://elastic.internal.example.com/hook" + + def test_dot_prefix_does_not_match_base_domain(self): + with ( + mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo('10.50.12.40')), + mock.patch('awx.main.notifications.url_validation.settings') as mock_settings, + ): + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['.internal.example.com'] + with pytest.raises(SSRFBlockedError, match="blocked address"): + validate_url("https://internal.example.com/hook") + + def test_non_matching_hostname_still_blocked(self): + with ( + mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo('10.50.12.40')), + mock.patch('awx.main.notifications.url_validation.settings') as mock_settings, + ): + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['elastic.internal.example.com'] + with pytest.raises(SSRFBlockedError, match="blocked address"): + validate_url("https://other.internal.example.com/hook") + + def test_hostname_allowed_when_dns_fails(self): + """When hostname is allowlisted but DNS resolution fails, the request is still allowed.""" + with mock.patch('awx.main.notifications.url_validation.settings') as mock_settings: + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['elastic.internal.example.com'] + with mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', side_effect=socket.gaierror("Name or service not known")): + assert validate_url("https://elastic.internal.example.com/hook") == "https://elastic.internal.example.com/hook" + + def test_mixed_cidr_and_hostname_entries(self): + with mock.patch('awx.main.notifications.url_validation.settings') as mock_settings: + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['10.0.0.0/8', 'grafana.corp.internal'] + with mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo('172.16.1.5')): + assert validate_url("https://grafana.corp.internal/hook") == "https://grafana.corp.internal/hook" + + def test_cidr_still_works_with_hostname_entries_present(self): + with ( + mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo('10.50.12.40')), + mock.patch('awx.main.notifications.url_validation.settings') as mock_settings, + ): + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['10.50.0.0/16', 'grafana.corp.internal'] + assert validate_url("https://unknown.host/hook") == "https://unknown.host/hook" + + +class TestLinkLocalDenyList: + """Link-local addresses are always blocked, even if allowlisted.""" + + def _addrinfo(self, ip): + return [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, '', (ip, 443))] + + def _addrinfo6(self, ip): + return [(socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP, '', (ip, 443, 0, 0))] + + def test_link_local_blocked_with_cidr_allowlist(self): + with ( + mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo('169.254.169.254')), + mock.patch('awx.main.notifications.url_validation.settings') as mock_settings, + ): + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['169.254.0.0/16'] + with pytest.raises(SSRFBlockedError, match="link-local"): + validate_url("https://metadata.example.com/hook") + + def test_link_local_blocked_with_hostname_allowlist(self): + with ( + mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo('169.254.169.254')), + mock.patch('awx.main.notifications.url_validation.settings') as mock_settings, + ): + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['metadata.example.com'] + with pytest.raises(SSRFBlockedError, match="link-local"): + validate_url("https://metadata.example.com/hook") + + def test_ipv6_link_local_blocked_with_allowlist(self): + with ( + mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo6('fe80::1')), + mock.patch('awx.main.notifications.url_validation.settings') as mock_settings, + ): + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['fe80::/10'] + with pytest.raises(SSRFBlockedError, match="link-local"): + validate_url("https://metadata.example.com/hook") + + def test_hostname_allowed_for_non_link_local_private_ip(self): + with ( + mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo('10.50.1.20')), + mock.patch('awx.main.notifications.url_validation.settings') as mock_settings, + ): + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['elastic.internal.example.com'] + assert validate_url("https://elastic.internal.example.com/hook") == "https://elastic.internal.example.com/hook" + + def test_ipv4_mapped_link_local_blocked_with_hostname_allowlist(self): + with ( + mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo6('::ffff:169.254.169.254')), + mock.patch('awx.main.notifications.url_validation.settings') as mock_settings, + ): + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['metadata.example.com'] + with pytest.raises(SSRFBlockedError, match="link-local"): + validate_url("https://metadata.example.com/hook") + + def test_ipv4_mapped_link_local_blocked_with_cidr_allowlist(self): + with ( + mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo6('::ffff:169.254.169.254')), + mock.patch('awx.main.notifications.url_validation.settings') as mock_settings, + ): + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['169.254.0.0/16'] + with pytest.raises(SSRFBlockedError, match="link-local"): + validate_url("https://metadata.example.com/hook") + + def test_loopback_still_allowable(self): + with ( + mock.patch('awx.main.notifications.url_validation.socket.getaddrinfo', return_value=self._addrinfo('127.0.0.1')), + mock.patch('awx.main.notifications.url_validation.settings') as mock_settings, + ): + mock_settings.NOTIFICATION_IP_ALLOW_LIST = ['127.0.0.0/8'] + assert validate_url("https://localhost/hook") == "https://localhost/hook" diff --git a/awx/main/tests/unit/notifications/test_webhook.py b/awx/main/tests/unit/notifications/test_webhook.py index 0a6e1a1e45..4cc3303d29 100644 --- a/awx/main/tests/unit/notifications/test_webhook.py +++ b/awx/main/tests/unit/notifications/test_webhook.py @@ -385,6 +385,8 @@ def test_send_messages_redirect_strips_credentials_on_host_change(): with ( mock.patch('awx.main.notifications.webhook_backend.requests') as requests_mock, mock.patch('awx.main.notifications.webhook_backend.get_awx_http_client_headers') as version_mock, + # no-op: redirect tests use fake hostnames that would fail DNS resolution in validate_url + mock.patch('awx.main.notifications.webhook_backend.validate_url', side_effect=lambda url: url), ): requests_mock.post.side_effect = [ mock.Mock(status_code=301, headers={"Location": "http://other-host.com/hook"}), @@ -410,6 +412,8 @@ def test_send_messages_redirect_strips_credentials_on_scheme_downgrade(): with ( mock.patch('awx.main.notifications.webhook_backend.requests') as requests_mock, mock.patch('awx.main.notifications.webhook_backend.get_awx_http_client_headers') as version_mock, + # no-op: redirect tests use fake hostnames that would fail DNS resolution in validate_url + mock.patch('awx.main.notifications.webhook_backend.validate_url', side_effect=lambda url: url), ): requests_mock.post.side_effect = [ mock.Mock(status_code=301, headers={"Location": "http://example.com/hook"}), @@ -430,6 +434,8 @@ def test_send_messages_redirect_keeps_auth_on_same_origin(): with ( mock.patch('awx.main.notifications.webhook_backend.requests') as requests_mock, mock.patch('awx.main.notifications.webhook_backend.get_awx_http_client_headers') as version_mock, + # no-op: redirect tests use fake hostnames that would fail DNS resolution in validate_url + mock.patch('awx.main.notifications.webhook_backend.validate_url', side_effect=lambda url: url), ): requests_mock.post.side_effect = [ mock.Mock(status_code=301, headers={"Location": "http://example.com/new-path"}), From 85c334e3a4476a77f4680aa1c009442080f3d7b9 Mon Sep 17 00:00:00 2001 From: Jake Jackson <24478650+thedoubl3j@users.noreply.github.com> Date: Thu, 22 Oct 2026 09:34:30 -0400 Subject: [PATCH 2/3] AAP-96178: Fix unified_jobs pagination count to respect RBAC filtering (#16751) * AAP-96178: Fix unified_jobs pagination count to respect RBAC filtering Non-admin users saw the unfiltered table-wide count instead of the count matching their RBAC-visible results. Use the same primary-key subquery approach already proven on the /api/v2/jobs/ endpoint. * AAP-96178: Add docstrings and mixed-role count test for UnifiedJobPaginator Clarify the PK subquery approach with expanded docstring and inline comments per review feedback. Add a functional test verifying that a user with a mix of JT and inventory roles sees a count matching exactly their permitted results. * AAP-96178: Add deduplication test for multiple RBAC access routes Verify that a job reachable through three overlapping permission paths (direct JT role, org audit role, and team JT role) appears exactly once in both results and count. Assisted-By: Claude Opus 4.6 --- awx/api/pagination.py | 14 ++- .../dab_rbac/test_unified_job_access.py | 87 +++++++++++++++++-- awx/main/tests/unit/api/test_pagination.py | 32 ++++--- 3 files changed, 113 insertions(+), 20 deletions(-) diff --git a/awx/api/pagination.py b/awx/api/pagination.py index 517fa40dfc..0f44dd9dc2 100644 --- a/awx/api/pagination.py +++ b/awx/api/pagination.py @@ -16,7 +16,7 @@ from django.utils.translation import gettext_lazy as _ from ansible_base.rest_filters.rest_framework.field_lookup_backend import ( FieldLookupBackend, ) -from awx.main.models import ActivityStream, UnifiedJob +from awx.main.models import ActivityStream class DisabledPaginator(DjangoPaginator): @@ -38,11 +38,19 @@ class ActivityStreamPaginator(DjangoPaginator): class UnifiedJobPaginator(DjangoPaginator): - """Use unfiltered table count for unified job pagination.""" + """Count only the unified jobs the requesting user can see. + + Uses a PK subquery so the database can index-scan the RBAC-filtered + primary keys instead of counting the full UNION queryset directly. + Same approach as JobPaginator. + """ @cached_property def count(self): - return UnifiedJob.objects.count() + # Collect PKs from the RBAC-filtered queryset (stripping ORDER BY for efficiency), + # then COUNT against the base table using an index-only scan on the PK. + visible_pks = self.object_list.order_by().values("pk") + return self.object_list.model.objects.filter(pk__in=visible_pks).count() class JobPaginator(DjangoPaginator): diff --git a/awx/main/tests/functional/dab_rbac/test_unified_job_access.py b/awx/main/tests/functional/dab_rbac/test_unified_job_access.py index 69d604e28f..9e5bbccd35 100644 --- a/awx/main/tests/functional/dab_rbac/test_unified_job_access.py +++ b/awx/main/tests/functional/dab_rbac/test_unified_job_access.py @@ -205,10 +205,10 @@ def test_unified_job_list_filtered_count_matches_results(admin, organization, in @pytest.mark.django_db -def test_unified_job_list_pagination_uses_unfiltered_count(rando, setup_managed_roles, get): - """The pagination count should reflect total unified job rows, not - the RBAC-filtered subset. The RBAC-filtered COUNT is catastrophically - slow on large tables with pk__in UNION subqueries.""" +def test_unified_job_list_pagination_count_matches_rbac_filtered_results(rando, setup_managed_roles, get): + """The pagination count should reflect the RBAC-filtered result set, + not the unfiltered table total. A user who cannot see any jobs + should see count == 0.""" org = Organization.objects.create(name='uj-count-org') inventory = org.inventories.create(name='uj-count-inv') project = Project.objects.create(name='uj-count-project', organization=org) @@ -221,7 +221,84 @@ def test_unified_job_list_pagination_uses_unfiltered_count(rando, setup_managed_ response = get(reverse('api:unified_job_list'), rando) assert response.status_code == 200 assert len(response.data['results']) == 0 - assert response.data['count'] == total_jobs + assert response.data['count'] == 0 + + +@pytest.mark.django_db +def test_unified_job_list_count_matches_for_mixed_roles(admin, user, setup_managed_roles, get): + """A user with a mix of roles should see a count that matches exactly + the number of results they are permitted to view, not the global total.""" + org = Organization.objects.create(name='uj-mixed-count-org') + inventory = org.inventories.create(name='uj-mixed-count-inv') + project = Project.objects.create(name='uj-mixed-count-project', organization=org) + + # Two job templates — user will only get access to jt_visible + jt_visible = JobTemplate.objects.create(name='uj-mixed-visible-jt', project=project, inventory=inventory, organization=org) + jt_hidden = JobTemplate.objects.create(name='uj-mixed-hidden-jt', project=project, inventory=inventory, organization=org) + + visible_job_1 = jt_visible.create_unified_job() + visible_job_2 = jt_visible.create_unified_job() + jt_hidden.create_unified_job() + jt_hidden.create_unified_job() + + # Inventory source — user will get access via inventory view + inv_src = InventorySource.objects.create(name='uj-mixed-count-invsrc', inventory=inventory, source='ec2') + visible_inv_update = InventoryUpdate.objects.create(inventory_source=inv_src, source=inv_src.source) + + # Ad hoc command — also visible via inventory view + visible_adhoc = AdHocCommand.objects.create(name='uj-mixed-count-adhoc', inventory=inventory) + + mixed_user = user('uj-mixed-role-user') + RoleDefinition.objects.get(name='JobTemplate Execute').give_permission(mixed_user, jt_visible) + RoleDefinition.objects.get(name='Inventory Admin').give_permission(mixed_user, inventory) + + expected_visible_ids = {visible_job_1.pk, visible_job_2.pk, visible_inv_update.pk, visible_adhoc.pk} + + # Admin sees everything + admin_response = get(reverse('api:unified_job_list'), admin) + assert admin_response.status_code == 200 + assert admin_response.data['count'] == len(admin_response.data['results']) + + # Mixed-role user sees only their permitted jobs, and count matches + user_response = get(reverse('api:unified_job_list'), mixed_user) + assert user_response.status_code == 200 + user_result_ids = {r['id'] for r in user_response.data['results']} + assert user_result_ids == expected_visible_ids + assert user_response.data['count'] == len(expected_visible_ids) + assert user_response.data['count'] < admin_response.data['count'] + + +@pytest.mark.django_db +def test_unified_job_count_deduplicates_multiple_access_routes(user, setup_managed_roles, get): + """A job reachable through multiple permission paths (direct JT role, + org auditor role, and team role) must appear exactly once in the + results and count.""" + org = Organization.objects.create(name="triple-route-org") + inventory = org.inventories.create(name="triple-route-inventory") + project = Project.objects.create(name="triple-route-project", organization=org) + jt = JobTemplate.objects.create( + name="triple-route-template", + project=project, + inventory=inventory, + organization=org, + ) + job = jt.create_unified_job() + + viewer = user("triple-route-viewer") + + RoleDefinition.objects.get(name="JobTemplate Execute").give_permission(viewer, jt) + RoleDefinition.objects.get(name="Organization Audit").give_permission(viewer, org) + + team = Team.objects.create(name="triple-route-team", organization=org) + team.member_role.members.add(viewer) + RoleDefinition.objects.get(name="JobTemplate Execute").give_permission(team, jt) + + response = get(reverse("api:unified_job_list"), viewer) + + assert response.status_code == 200 + ids = [result["id"] for result in response.data["results"]] + assert ids == [job.pk] + assert response.data["count"] == 1 @pytest.mark.django_db diff --git a/awx/main/tests/unit/api/test_pagination.py b/awx/main/tests/unit/api/test_pagination.py index ebd8192d38..2ddcc10448 100644 --- a/awx/main/tests/unit/api/test_pagination.py +++ b/awx/main/tests/unit/api/test_pagination.py @@ -72,20 +72,28 @@ class TestActivityStreamPagination: class TestUnifiedJobPaginator: - def test_count_uses_unfiltered_table_count(self): - with patch('awx.api.pagination.UnifiedJob') as mock_uj: - mock_uj.objects.count.return_value = 42000 - paginator = UnifiedJobPaginator(object_list=[], per_page=25) - assert paginator.count == 42000 - mock_uj.objects.count.assert_called_once() + def test_count_uses_primary_key_subquery(self): + object_list = MagicMock() + object_list.order_by.return_value.values.return_value = "visible_pks" + object_list.model.objects.filter.return_value.count.return_value = 713 + + paginator = UnifiedJobPaginator(object_list=object_list, per_page=25) + + assert paginator.count == 713 + object_list.order_by.assert_called_once_with() + object_list.order_by.return_value.values.assert_called_once_with("pk") + object_list.model.objects.filter.assert_called_once_with(pk__in="visible_pks") def test_count_is_cached(self): - with patch('awx.api.pagination.UnifiedJob') as mock_uj: - mock_uj.objects.count.return_value = 500 - paginator = UnifiedJobPaginator(object_list=[], per_page=25) - _ = paginator.count - _ = paginator.count - mock_uj.objects.count.assert_called_once() + object_list = MagicMock() + object_list.order_by.return_value.values.return_value = "visible_pks" + object_list.model.objects.filter.return_value.count.return_value = 500 + paginator = UnifiedJobPaginator(object_list=object_list, per_page=25) + + _ = paginator.count + _ = paginator.count + + object_list.model.objects.filter.assert_called_once_with(pk__in="visible_pks") class TestJobPaginator: From c9b547d6d950618650106377e40d3a29fed5bb90 Mon Sep 17 00:00:00 2001 From: Peter Braun Date: Thu, 22 Oct 2026 15:46:11 +0200 Subject: [PATCH 3/3] fix: do not call .filter() on list before converting to queryset (#16752) * fix: do not call .filter() on list before converting to queryset * fix: optimize slicing of queryset * fix: preserve existing filters and ordering on the queryset --- awx/main/models/inventory.py | 20 +++----- .../tests/functional/tasks/test_fact_cache.py | 51 +++++++++++++++++++ 2 files changed, 59 insertions(+), 12 deletions(-) diff --git a/awx/main/models/inventory.py b/awx/main/models/inventory.py index 4157d96921..a66724fe35 100644 --- a/awx/main/models/inventory.py +++ b/awx/main/models/inventory.py @@ -282,21 +282,17 @@ class Inventory(CommonModelNameNotUnique, RelatedJobsMixin, OpaQueryPathMixin): def get_sliced_hosts(self, host_queryset, slice_number, slice_count): """ - Returns a slice of Hosts given a slice number and total slice count, or - the original queryset if slicing is not requested. - - NOTE: If slicing is performed, this will return a List[Host] with the - resulting slice. If slicing is not performed it will return the - original queryset (not evaluating it or forcing it to a list). This - puts the burden on the caller to check the resulting type. This is - non-ideal because it's easy to get wrong, but I think the only way - around it is to force the queryset which has memory implications for - large inventories. + Returns a queryset of Hosts for a given slice number and total slice + count, or the original queryset if slicing is not requested. """ - if slice_count > 1 and slice_number > 0: offset = slice_number - 1 - host_queryset = host_queryset[offset::slice_count] + # Fetch only PKs to avoid loading the full host records into memory + # before slicing the query set. Unfortunately it seems that there is + # no good way to slice a queryset without converting it to a list first + all_pks = list(host_queryset.values_list('pk', flat=True)) + sliced_pks = all_pks[offset::slice_count] + host_queryset = host_queryset.filter(pk__in=sliced_pks) return host_queryset def get_script_data(self, hostvars=False, towervars=False, show_all=False, slice_number=1, slice_count=1): diff --git a/awx/main/tests/functional/tasks/test_fact_cache.py b/awx/main/tests/functional/tasks/test_fact_cache.py index 616bf67d65..fb6566389e 100644 --- a/awx/main/tests/functional/tasks/test_fact_cache.py +++ b/awx/main/tests/functional/tasks/test_fact_cache.py @@ -221,3 +221,54 @@ class TestConstructedInventoryFactCache: assert source_host.ansible_facts == {'role': 'web', 'deployed': True} assert not constructed_host.ansible_facts, f'Facts were stored on the constructed host: {constructed_host.ansible_facts!r}' + + +@pytest.mark.django_db +class TestSlicedJobFactCache: + """finish_fact_cache must work when get_hosts_for_fact_cache returns a sliced result.""" + + def test_sliced_hosts_persist_facts(self, organization, artifacts_dir): + """Sliced jobs must persist facts via get_sliced_hosts queryset. + + Regression test for AAP-97902: get_sliced_hosts() used to return a list + when slice_count > 1, causing finish_fact_cache() to crash with + AttributeError when calling .filter(). get_sliced_hosts() now converts + the step-sliced list back to a QuerySet. + """ + inv = Inventory.objects.create(organization=organization, name='slice-inv') + hosts = [inv.hosts.create(name=f'host{i}') for i in range(4)] + + for h in hosts: + h.ansible_facts = {'old': True} + h.ansible_facts_modified = now() + h.save(update_fields=['ansible_facts', 'ansible_facts_modified']) + + host_qs = inv.hosts.all() + sliced_qs = inv.get_sliced_hosts(host_qs, slice_number=1, slice_count=2) + + # get_sliced_hosts must return a QuerySet, not a list + assert hasattr(sliced_qs, 'filter'), 'get_sliced_hosts must return a QuerySet' + + sliced_hosts = list(sliced_qs) + start_fact_cache(sliced_qs, artifacts_dir=artifacts_dir, timeout=0, inventory_id=inv.id) + + # Simulate ansible writing new facts for each host in the slice + for h in sliced_hosts: + fact_file = os.path.join(artifacts_dir, 'fact_cache', h.name) + future = time.time() + 60 + with open(fact_file, 'w') as f: + json.dump({'updated': True, 'host': h.name}, f) + os.utime(fact_file, (future, future)) + + # This would crash with AttributeError before the fix + finish_fact_cache(sliced_qs, artifacts_dir=artifacts_dir, inventory_id=inv.id) + + for h in hosts: + h.refresh_from_db() + + sliced_names = {h.name for h in sliced_hosts} + for h in hosts: + if h.name in sliced_names: + assert h.ansible_facts.get('updated') is True, f'{h.name} should have updated facts' + else: + assert h.ansible_facts == {'old': True}, f'{h.name} should be unchanged'