Merge branch 'devel' into cve-2026-102268-pyjwt

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

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/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):

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/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

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'

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/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:

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