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 <noreply@anthropic.com>

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,

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

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:

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

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

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

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

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"

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