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"}),