UN-3815 [FIX] Validate webhook URLs in one place, at both sinks

backend/notification_v2/internal_views.py

@@ -30,6 +30,7 @@
     WebhookTestSerializer,
 )
 from notification_v2.models import Notification
+from unstract.core.network.ssrf import is_safe_webhook_url
 
 logger = logging.getLogger(__name__)
 
@@ -337,6 +338,15 @@ def post(self, request):
             validated_data = serializer.validated_data
             headers = self._build_headers(validated_data)
 
+            # Same guard as the delivery sinks. This endpoint is behind
+            # INTERNAL_SERVICE_API_KEY and not tenant-reachable, but it takes
+            # an arbitrary URL and so gets the same treatment.
+            if not is_safe_webhook_url(validated_data["url"]):
+                return Response(
+                    {"error": "URL must resolve to a public address."},
+                    status=status.HTTP_400_BAD_REQUEST,
+                )
+
             import requests
 
             try:
@@ -345,16 +355,18 @@ def post(self, request):
                     json=validated_data["payload"],
                     headers=headers,
                     timeout=validated_data["timeout"],
+                    allow_redirects=False,
                 )
 
+                # Status only. The response body and headers are not the
+                # caller's to read, and request_headers carried back the
+                # Authorization value built from authorization_key.
                 test_result = {
-                    "success": response.status_code < 400,
+                    # 2xx only: redirects are not followed, so a 301/302 means
+                    # the payload never reached the final destination.
+                    "success": 200 <= response.status_code < 300,
                     "status_code": response.status_code,
-                    "response_headers": dict(response.headers),
-                    "response_body": response.text[:1000],
                     "url": validated_data["url"],
-                    "request_headers": headers,
-                    "request_payload": validated_data["payload"],
                 }
 
                 logger.info(

backend/notification_v2/serializers.py

@@ -1,6 +1,8 @@
 from rest_framework import serializers
 from utils.input_sanitizer import validate_name_field
 
+from unstract.core.network.ssrf import is_safe_webhook_url
+
 from .enums import AuthorizationType, NotificationType, PlatformType
 from .models import Notification
 
@@ -34,8 +36,33 @@ def validate(self, data):
         # General validation for the relationship between api and pipeline
         self._validate_api_or_pipeline(data)
         self._validate_authorization(data)
+        self._validate_url(data)
         return data
 
+    def _validate_url(self, data):
+        """Reject internal webhook targets when a URL is supplied.
+
+        URLField only checks the shape, so an internal address would be stored
+        and only refused later at the sink, silently and out of the user's
+        sight. Fail here instead; the sink guard stays as the real control.
+
+        Only checks a URL the caller actually sent. Re-resolving the stored one
+        would make an unrelated PATCH fail whenever DNS is briefly unavailable
+        or a legacy record predates this check.
+
+        resolve=False keeps DNS off the request thread — getaddrinfo honours no
+        timeout, so a slow resolver here would stall the worker serving the
+        request. Literal internal addresses are still refused; a hostname
+        pointing inward is caught at the sink, which is the real control.
+        """
+        if "url" not in data:
+            return
+        url = data["url"]
+        if url and not is_safe_webhook_url(url, resolve=False):
+            raise serializers.ValidationError(
+                {"url": "URL must resolve to a public address."}
+            )
+
     def _validate_api_or_pipeline(self, data):
         """Ensure either 'api' or 'pipeline' is provided, but not both."""
         api = data.get("api", getattr(self.instance, "api", None))

backend/notification_v2/tests/test_webhook_ssrf.py

@@ -0,0 +1,124 @@
+"""Webhook URL egress controls on the backend side.
+
+The sink guard in ``unstract.core`` is the real control; these cover the two
+backend surfaces that also accept a URL — the notification serializer, which
+should refuse an internal target at creation rather than at delivery time, and
+the internal webhook-test endpoint, which used to return the response body.
+"""
+
+from unittest.mock import Mock, patch
+
+import pytest
+from django.test import SimpleTestCase
+from notification_v2.internal_views import WebhookTestAPIView
+from notification_v2.serializers import NotificationSerializer
+from rest_framework import status
+from rest_framework.exceptions import ValidationError
+from rest_framework.parsers import JSONParser
+from rest_framework.request import Request
+from rest_framework.test import APIRequestFactory
+
+INTERNAL_URLS = [
+    "http://169.254.169.254/latest/meta-data/",
+    "http://127.0.0.1:8000/admin/",
+    r"https://127.0.0.1:6666\@1.1.1.1",
+]
+
+# Stub DNS so nothing here depends on the network. The serializer path does not
+# resolve at all; the endpoint path does, and would otherwise make a real
+# lookup for example.com and fail in an isolated runner.
+_FAKE_DNS = {"example.com": "93.184.216.34"}
+
+
+@pytest.fixture(autouse=True)
+def stub_dns(monkeypatch):
+    def fake_getaddrinfo(host, *_args, **_kwargs):
+        if host not in _FAKE_DNS:
+            raise OSError(f"unresolvable in test: {host}")
+        return [(None, None, None, "", (_FAKE_DNS[host], 0))]
+
+    monkeypatch.setattr(
+        "unstract.core.network.ssrf.socket.getaddrinfo", fake_getaddrinfo
+    )
+
+
+def _notification_data(url):
+    """Minimum that reaches the URL check in ``NotificationSerializer.validate``."""
+    return {"pipeline": Mock(), "authorization_type": "NONE", "url": url}
+
+
+class NotificationSerializerUrlTest(SimpleTestCase):
+    """URLField only checks the shape, so an internal target would persist."""
+
+    def test_internal_urls_are_rejected(self):
+        for url in INTERNAL_URLS:
+            with self.subTest(url=url):
+                with self.assertRaises(ValidationError) as caught:
+                    NotificationSerializer().validate(_notification_data(url))
+                assert "url" in caught.exception.detail
+
+    def test_public_url_is_accepted(self):
+        data = _notification_data("https://example.com/hook")
+        assert NotificationSerializer().validate(data) == data
+
+    def test_patch_that_omits_url_is_not_revalidated(self):
+        """A PATCH touching other fields must not re-resolve the stored URL.
+
+        Otherwise a brief DNS failure, or a record predating this check, makes
+        an unrelated edit fail on a field the caller never sent.
+        """
+        # api=None so the api/pipeline check doesn't trip on Mock's truthy
+        # auto-attribute before the URL check is reached.
+        instance = Mock(api=None, url="http://127.0.0.1:8000/legacy")
+        serializer = NotificationSerializer(instance=instance)
+
+        data = {"pipeline": Mock(), "authorization_type": "NONE", "max_retries": 2}
+        assert serializer.validate(data) == data
+
+
+class WebhookTestEndpointTest(SimpleTestCase):
+    """This endpoint had no URL check, and returned the response body."""
+
+    def _post(self, url):
+        request = Request(
+            APIRequestFactory().post(
+                "/internal/webhook/test/", {"url": url, "payload": {}}, format="json"
+            ),
+            parsers=[JSONParser()],
+        )
+        return WebhookTestAPIView().post(request)
+
+    def test_internal_url_is_refused_before_any_request(self):
+        for url in INTERNAL_URLS:
+            with self.subTest(url=url):
+                with patch("requests.post") as post:
+                    response = self._post(url)
+                assert response.status_code == status.HTTP_400_BAD_REQUEST
+                post.assert_not_called()
+
+    def test_response_body_and_headers_are_not_echoed(self):
+        with patch("requests.post") as post:
+            post.return_value.status_code = 200
+            post.return_value.headers = {"X-Internal-Secret": "leaked"}
+            post.return_value.text = "internal response body"
+            response = self._post("https://example.com/hook")
+
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["status_code"] == 200
+        assert post.call_args.kwargs["allow_redirects"] is False
+
+        # Nothing about the upstream response comes back, and neither do the
+        # request headers — those carry the Authorization value we built.
+        for leaked in ("response_body", "response_headers", "request_headers"):
+            assert leaked not in response.data, f"{leaked} is echoed to the caller"
+
+    def test_redirect_is_not_reported_as_success(self):
+        """Redirects are not followed, so a 3xx means the payload never landed."""
+        with patch("requests.post") as post:
+            post.return_value.status_code = 302
+            post.return_value.headers = {}
+            post.return_value.text = ""
+            response = self._post("https://example.com/hook")
+
+        assert response.data["status_code"] == 302
+        assert response.data["success"] is False

unstract/core/src/unstract/core/network/__init__.py

@@ -1,5 +1,6 @@
 from unstract.core.network.enums import HTTPMethod
 from unstract.core.network.http_client import HttpClient
 from unstract.core.network.retry import get_retry_session
+from unstract.core.network.ssrf import is_safe_webhook_url
 
-__all__ = ["HTTPMethod", "get_retry_session", "HttpClient"]
+__all__ = ["HTTPMethod", "HttpClient", "get_retry_session", "is_safe_webhook_url"]

unstract/core/src/unstract/core/network/ssrf.py

@@ -0,0 +1,160 @@
+"""Shared egress guard for user-supplied webhook URLs.
+
+Both webhook sinks — prompt postprocessing and pipeline notifications — take a
+URL from a tenant and hand it to ``requests``. This module is the single place
+that decides whether such a URL may be dialled, so a new sink does not have to
+carry its own copy of the rules.
+
+Three things are checked, in order:
+
+1. **Parser agreement.** This module reads the URL with ``urllib.parse`` while
+   the transport underneath ``requests`` resolves it with ``urllib3``. The two
+   do not always agree on the host, and where they disagree the URL is refused,
+   because the host approved here is not the host the socket connects to.
+   Comparing the two parsers is an invariant rather than a list of characters to
+   reject, so it holds as either parser changes.
+2. **Scheme and userinfo.** Anything outside the caller's allowlist is refused,
+   as is a URL carrying credentials.
+3. **Resolved address.** Every address the host resolves to must be publicly
+   routable. Loopback, private, link-local (which covers the cloud metadata
+   endpoints), reserved and multicast ranges are all refused.
+
+Note the ceiling: resolve-then-connect cannot cover a name that is re-resolved
+to an internal address between this check and the socket. The control for that
+is an egress policy on the worker pods, not application code.
+"""
+
+import ipaddress
+import logging
+import socket
+from urllib.parse import urlparse
+
+from urllib3.exceptions import LocationParseError
+from urllib3.util import parse_url
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_ALLOWED_SCHEMES = ("http", "https")
+
+
+def _normalize_host(host: str | None) -> str:
+    """Reduce a host to a form the two parsers can be compared on.
+
+    urllib3 keeps the brackets on an IPv6 literal and punycodes a unicode host;
+    ``urlparse`` does neither. Comparing the raw strings would refuse both of
+    those legitimate URLs.
+    """
+    if not host:
+        return ""
+    host = host.strip().strip("[]").rstrip(".").lower()
+    try:
+        return host.encode("idna").decode("ascii")
+    except UnicodeError:
+        # Not IDNA-encodable (empty label, over-long label). Compare as-is;
+        # the parsers still have to agree for the URL to be accepted.
+        return host
+
+
+def _resolve(host: str) -> set[str]:
+    """Return every IP the host resolves to, or an empty set on failure."""
+    try:
+        ipaddress.ip_address(host)
+        return {host}
+    except ValueError:
+        pass
+    try:
+        return {
+            sockaddr[0]
+            for *_, sockaddr in socket.getaddrinfo(host, None, type=socket.SOCK_STREAM)
+        }
+    except (OSError, UnicodeError):
+        # UnicodeError: getaddrinfo IDNA-encodes internally and raises, not
+        # returns, on an over-long or empty label. Unresolvable either way.
+        return set()
+
+
+def _is_public(addr: str) -> bool:
+    """Whether an address is globally routable.
+
+    ``is_global`` is an allowlist maintained against the IANA special-purpose
+    registries, so it stays correct as ranges are added. Enumerating the
+    negative flags instead misses ranges that belong to none of them — RFC 6598
+    shared address space (100.64.0.0/10) is private in practice but is not
+    ``is_private``, ``is_reserved`` or any of the rest.
+    """
+    try:
+        return ipaddress.ip_address(addr).is_global
+    except ValueError:
+        return False
+
+
+def is_safe_webhook_url(
+    url: str | None,
+    allowed_schemes: tuple[str, ...] = DEFAULT_ALLOWED_SCHEMES,
+    resolve: bool = True,
+) -> bool:
+    """Whether ``url`` may be dialled from inside the network.
+
+    Args:
+        url: The tenant-supplied URL.
+        allowed_schemes: Schemes to accept. Callers that already require TLS
+            should pass ``("https",)`` rather than widening to the default.
+        resolve: Whether to resolve the host and check every address. Leave it
+            on at the sinks — that is the real control. Turn it off on request
+            handling threads: ``socket.getaddrinfo`` honours no timeout, so a
+            slow or hostile resolver would stall the worker serving the
+            request. With it off, the syntactic checks still run and a literal
+            internal IP is still refused; a hostname that resolves internally
+            is caught at the sink instead.
+
+    Returns:
+        True only if the URL is well-formed, unambiguous to both parsers, and
+        (when ``resolve`` is set) maps entirely to public addresses.
+    """
+    if not url:
+        return False
+
+    try:
+        parsed = urlparse(url)
+    except ValueError:
+        return False
+
+    if parsed.scheme not in allowed_schemes:
+        return False
+
+    # Credentials in the URL are the vehicle for the parser confusion above and
+    # have no legitimate use on a webhook target.
+    if parsed.username or parsed.password or "@" in (parsed.netloc or ""):
+        return False
+
+    try:
+        transport_host = parse_url(url).host
+    except LocationParseError:
+        # The transport cannot parse it, so nothing here can predict where it
+        # would connect.
+        return False
+
+    host = _normalize_host(parsed.hostname)
+    if host != _normalize_host(transport_host):
+        logger.warning("Refusing webhook URL: validator and transport disagree on host")
+        return False
+
+    if not host:
+        return False
+
+    if not resolve:
+        # No DNS on this path. A literal address is still checked, since that
+        # needs no lookup and is how most internal targets are written.
+        try:
+            ipaddress.ip_address(host)
+        except ValueError:
+            return True
+        return _is_public(host)
+
+    # Resolve the normalized host: that is the canonical form the transport
+    # ends up dialling, so the addresses checked here are the ones used.
+    addrs = _resolve(host)
+    if not addrs:
+        return False
+
+    return all(_is_public(addr) for addr in addrs)