diff --git a/backend/notification_v2/internal_views.py b/backend/notification_v2/internal_views.py
index 596945d01f..fb15ba773b 100644
--- a/backend/notification_v2/internal_views.py
+++ b/backend/notification_v2/internal_views.py
@@ -5,6 +5,7 @@
 import logging
 from typing import Any
 
+import requests
 from rest_framework import status, viewsets
 from rest_framework.decorators import action
 from rest_framework.response import Response
@@ -21,6 +22,7 @@
     WebhookTestSerializer,
 )
 from notification_v2.models import Notification
+from unstract.core.network.ssrf import is_safe_webhook_url
 
 logger = logging.getLogger(__name__)
 
@@ -115,7 +117,14 @@ def post(self, request):
             validated_data = serializer.validated_data
             headers = self._build_headers(validated_data)
 
-            import requests
+            # 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,
+                )
 
             try:
                 response = requests.post(
@@ -123,16 +132,19 @@ 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 `headers` (built above) carries the
+                # Authorization value built from authorization_key, so it is
+                # not echoed back either.
                 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(
@@ -142,12 +154,14 @@ def post(self, request):
                 return Response(test_result)
 
             except requests.exceptions.RequestException as e:
+                # Same rule as the success branch above: `headers` carries the
+                # Authorization value built from authorization_key, so it is not
+                # echoed back. A target that times out or refuses the connection
+                # is the most common way to get here.
                 test_result = {
                     "success": False,
                     "error": str(e),
                     "url": validated_data["url"],
-                    "request_headers": headers,
-                    "request_payload": validated_data["payload"],
                 }
 
                 return Response(test_result, status=status.HTTP_400_BAD_REQUEST)
diff --git a/backend/notification_v2/serializers.py b/backend/notification_v2/serializers.py
index 3cd7085751..c5444c1d91 100644
--- a/backend/notification_v2/serializers.py
+++ b/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,38 @@ 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 a URL that can never be dialled, at save time.
+
+        is_safe_webhook_url refuses a disallowed scheme, credentials in the URL,
+        a host the two parsers disagree on, and an internal address literal —
+        not only the last of those. The sink is still the real control.
+
+        resolve=False keeps DNS off the request thread — getaddrinfo takes no
+        timeout. A hostname pointing inward is accepted here and refused at the
+        sink, which resolves. Only a URL the caller actually sent is checked;
+        re-checking the stored one would 400 an unrelated PATCH on a legacy row.
+        """
+        notification_type = data.get(
+            "notification_type", getattr(self.instance, "notification_type", None)
+        )
+        url = data.get("url", getattr(self.instance, "url", None))
+
+        if not url:
+            if notification_type == NotificationType.WEBHOOK.value:
+                raise serializers.ValidationError(
+                    {"url": "A webhook notification requires a URL."}
+                )
+            return
+
+        if "url" in data and not is_safe_webhook_url(url, resolve=False):
+            raise serializers.ValidationError(
+                {"url": "URL must not be an internal or ambiguous 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))
diff --git a/backend/notification_v2/tests/test_webhook_ssrf.py b/backend/notification_v2/tests/test_webhook_ssrf.py
new file mode 100644
index 0000000000..a66f0d111c
--- /dev/null
+++ b/backend/notification_v2/tests/test_webhook_ssrf.py
@@ -0,0 +1,228 @@
+"""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
+import requests
+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_webhook_create_without_a_url_is_rejected(self):
+        """``url`` is null=True on the model, so DRF makes it optional.
+
+        Without this check a webhook notification persists with no destination
+        and returns 201; at dispatch the user is told the URL "is not an
+        allowed public destination" for a URL that was never set.
+        """
+        for data in (
+            # omitted entirely
+            {
+                "pipeline": Mock(),
+                "authorization_type": "NONE",
+                "notification_type": "WEBHOOK",
+            },
+            # explicitly null
+            {
+                "pipeline": Mock(),
+                "authorization_type": "NONE",
+                "notification_type": "WEBHOOK",
+                "url": None,
+            },
+        ):
+            with self.subTest(data=sorted(data)):
+                with self.assertRaises(ValidationError) as caught:
+                    NotificationSerializer().validate(data)
+                assert "url" in caught.exception.detail
+
+    def test_webhook_patch_that_omits_url_keeps_the_stored_one(self):
+        """The create check must not break the documented PATCH case."""
+        instance = Mock(api=None, notification_type="WEBHOOK", url="https://a.example")
+        serializer = NotificationSerializer(instance=instance, partial=True)
+
+        data = {"pipeline": Mock(), "authorization_type": "NONE", "max_retries": 2}
+        assert serializer.validate(data) == data
+
+    def test_patch_switching_a_url_less_record_to_webhook_is_rejected(self):
+        """``self.partial`` alone is the wrong gate for the required-URL check.
+
+        Turning an existing URL-less notification into a WEBHOOK creates a
+        destination-less webhook just as surely as a create does, so the type
+        change has to be checked as well as ``partial``.
+        """
+        instance = Mock(api=None, notification_type="EMAIL", url=None)
+        serializer = NotificationSerializer(instance=instance, partial=True)
+
+        data = {
+            "pipeline": Mock(),
+            "authorization_type": "NONE",
+            "notification_type": "WEBHOOK",
+        }
+        with self.assertRaises(ValidationError) as caught:
+            serializer.validate(data)
+        assert "url" in caught.exception.detail
+
+    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_transport_failure_does_not_echo_the_authorization_header(self):
+        """The error branch is the common path, and it built the credential.
+
+        A public host that simply does not answer never reaches the guard, so
+        this is reachable for any well-formed URL. The success-branch test
+        above cannot catch it: it only stubs a 200.
+        """
+        request = Request(
+            APIRequestFactory().post(
+                "/internal/webhook/test/",
+                {
+                    "url": "https://example.com/hook",
+                    "payload": {},
+                    "authorization_type": "BEARER",
+                    "authorization_key": "super-secret-token",
+                },
+                format="json",
+            ),
+            parsers=[JSONParser()],
+        )
+        with patch("requests.post") as post:
+            post.side_effect = requests.exceptions.ConnectTimeout("timed out")
+            response = WebhookTestAPIView().post(request)
+
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert response.data["success"] is False
+        for leaked in ("request_headers", "request_payload"):
+            assert leaked not in response.data, f"{leaked} is echoed to the caller"
+        assert "super-secret-token" not in str(response.data)
+
+    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
+
+
+class UrlLessWebhookRowTest(SimpleTestCase):
+    """A stored webhook with no URL is invalid however it got that way."""
+
+    def test_patch_on_a_url_less_webhook_row_is_rejected(self):
+        """Even when the PATCH is about something else entirely.
+
+        Gating on `partial` let a legacy WEBHOOK row with url=None survive an
+        unrelated edit and stay undeliverable. Gating on the stored URL does
+        not, and leaves rows that already have one alone.
+        """
+        instance = Mock(api=None, notification_type="WEBHOOK", url=None)
+        serializer = NotificationSerializer(instance=instance, partial=True)
+
+        data = {"pipeline": Mock(), "authorization_type": "NONE", "max_retries": 2}
+        with self.assertRaises(ValidationError) as caught:
+            serializer.validate(data)
+        assert "url" in caught.exception.detail
diff --git a/unstract/core/src/unstract/core/network/__init__.py b/unstract/core/src/unstract/core/network/__init__.py
index bc3e7960a2..91fb4376d5 100644
--- a/unstract/core/src/unstract/core/network/__init__.py
+++ b/unstract/core/src/unstract/core/network/__init__.py
@@ -1,5 +1,21 @@
 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 (
+    UNRESOLVABLE,
+    is_retryable_refusal,
+    is_safe_webhook_url,
+    safe_host,
+    webhook_url_refusal,
+)
 
-__all__ = ["HTTPMethod", "get_retry_session", "HttpClient"]
+__all__ = [
+    "UNRESOLVABLE",
+    "HTTPMethod",
+    "HttpClient",
+    "get_retry_session",
+    "is_retryable_refusal",
+    "is_safe_webhook_url",
+    "safe_host",
+    "webhook_url_refusal",
+]
diff --git a/unstract/core/src/unstract/core/network/ssrf.py b/unstract/core/src/unstract/core/network/ssrf.py
new file mode 100644
index 0000000000..1a63d9d640
--- /dev/null
+++ b/unstract/core/src/unstract/core/network/ssrf.py
@@ -0,0 +1,263 @@
+"""Shared egress guard for user-supplied webhook URLs.
+
+Both webhook sinks — prompt postprocessing and pipeline notifications — hand a
+tenant-supplied URL to ``requests``. This module is the single place that
+decides whether one may be dialled, so a new sink does not carry its own copy
+of the rules.
+
+``is_safe_webhook_url`` answers the yes/no question and logs the reason;
+``webhook_url_refusal`` returns the reason, so a sink can tell a resolver
+outage (retryable) from a URL that will never be allowed.
+
+Note the ceiling: resolve-then-connect cannot cover a name that is re-resolved
+to an internal address between the 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
+
+import idna
+from urllib3.exceptions import LocationParseError
+from urllib3.util import parse_url
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_ALLOWED_SCHEMES = ("http", "https")
+
+# Refusal reasons. UNRESOLVABLE is the only transient one — a resolver outage
+# clears on its own, so a sink may retry it. Every other reason is a property
+# of the URL itself and cannot change between attempts.
+UNRESOLVABLE = "unresolvable"
+REFUSED_EMPTY_URL = "empty-url"
+REFUSED_UNPARSEABLE = "unparseable-url"
+REFUSED_SCHEME = "scheme-not-allowed"
+REFUSED_CREDENTIALS = "credentials-in-url"
+REFUSED_PARSER_DISAGREEMENT = "parser-disagreement"
+REFUSED_EMPTY_HOST = "empty-host"
+REFUSED_INTERNAL_LITERAL = "internal-literal"
+REFUSED_NON_PUBLIC = "non-public-address"
+
+# RFC 6761 reserves these to loopback, so no lookup is needed to know where
+# they point.
+_LOOPBACK_NAMES = ("localhost",)
+
+# Ranges IANA marks as not globally reachable that ``is_global`` still admits,
+# because the stdlib's copy of the registries predates them or omits them.
+# Multicast is handled separately in ``_is_public`` via ``is_multicast``.
+# These are registry entries, not deployment addresses: they name ranges this
+# guard must refuse, so they are hardcoded by definition. NOSONAR
+_NOT_GLOBALLY_REACHABLE = (
+    ipaddress.ip_network("192.88.99.0/24"),  # NOSONAR - 6to4 relay anycast (RFC 7526)
+    ipaddress.ip_network("5f00::/16"),  # NOSONAR - SRv6 SIDs (RFC 9602)
+)
+
+
+def _normalize_host(host: str | None) -> str:
+    """Reduce a host to the form the transport will dial.
+
+    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.
+
+    The encoding mirrors ``urllib3.util.url._idna_encode`` exactly — ASCII hosts
+    are only lowercased, non-ASCII hosts are encoded per label with the ``idna``
+    package. The stdlib ``"idna"`` codec is *not* interchangeable here: it is
+    IDNA-2003 with nameprep, so it maps ``faß.de`` to ``fass.de`` where the
+    transport produces ``xn--fa-hia.de``, and the parser-agreement check below
+    would refuse every such host.
+    """
+    if not host:
+        return ""
+    host = host.strip().strip("[]").rstrip(".").lower()
+    if host.isascii():
+        return host
+    try:
+        return ".".join(
+            idna.encode(label, strict=True, std3_rules=True).decode("ascii")
+            for label in host.split(".")
+        )
+    except (idna.IDNAError, UnicodeError):
+        # Not IDNA-encodable (empty label, over-long label, disallowed
+        # codepoint). Compare as-is; the parsers still have to agree for the
+        # URL to be accepted.
+        return host
+
+
+def _as_ip(host: str) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None:
+    """Parse a host that is an address literal, or None if it is a name.
+
+    ``ipaddress.ip_address`` only accepts dotted-quad IPv4, so on its own it
+    reads ``2130706433``, ``0177.0.0.1`` and ``127.1`` as hostnames — all three
+    are ``127.0.0.1`` to the resolver. ``inet_aton`` accepts the same legacy
+    forms the C resolver does, which is what the transport ends up using.
+    """
+    try:
+        return ipaddress.ip_address(host)
+    except ValueError:
+        pass
+    try:
+        return ipaddress.IPv4Address(socket.inet_aton(host))
+    except (OSError, ipaddress.AddressValueError):
+        return None
+
+
+def _resolve(host: str) -> set[str]:
+    """Return every IP the host resolves to, or an empty set on failure."""
+    literal = _as_ip(host)
+    if literal is not None:
+        return {str(literal)}
+    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 the base check because it is an allowlist maintained
+    against the IANA registries; enumerating the negative flags instead misses
+    ranges that belong to none of them, such as RFC 6598 shared address space.
+    It is not sufficient alone — it reports multicast and the
+    ``_NOT_GLOBALLY_REACHABLE`` ranges as global, so those are refused here.
+    """
+    try:
+        ip = ipaddress.ip_address(addr)
+    except ValueError:
+        return False
+    if not ip.is_global or ip.is_multicast:
+        return False
+    # ``in`` is version-safe: _BaseNetwork.__contains__ returns False on a
+    # version mismatch rather than raising, so no explicit guard is needed.
+    return not any(ip in net for net in _NOT_GLOBALLY_REACHABLE)
+
+
+def is_retryable_refusal(reason: str | None) -> bool:
+    """Whether retrying a refusal could ever produce a different outcome.
+
+    Only :data:`UNRESOLVABLE` can: a resolver outage clears on its own. Every
+    other reason is a property of the URL itself. Kept next to the reasons so a
+    new one has to be classified here rather than at each sink.
+    """
+    return reason == UNRESOLVABLE
+
+
+def webhook_url_refusal(
+    url: str | None,
+    allowed_schemes: tuple[str, ...] = DEFAULT_ALLOWED_SCHEMES,
+    resolve: bool = True,
+) -> str | None:
+    """Why ``url`` may not be dialled, or None if it may.
+
+    Same checks as :func:`is_safe_webhook_url`; use this one where the sink
+    needs to act on *why* it was refused. Compare the result against
+    :data:`UNRESOLVABLE` to separate a resolver outage, which may clear, from a
+    refusal that never will.
+    """
+    if not url:
+        return REFUSED_EMPTY_URL
+
+    try:
+        parsed = urlparse(url)
+    except ValueError:
+        return REFUSED_UNPARSEABLE
+
+    if parsed.scheme not in allowed_schemes:
+        return REFUSED_SCHEME
+
+    # Credentials in the URL are the vehicle for the parser confusion below and
+    # have no legitimate use on a webhook target.
+    if parsed.username or parsed.password or "@" in (parsed.netloc or ""):
+        return REFUSED_CREDENTIALS
+
+    try:
+        transport_host = parse_url(url).host
+    except LocationParseError:
+        # The transport cannot parse it, so nothing here can predict where it
+        # would connect.
+        return REFUSED_UNPARSEABLE
+
+    # ``urlparse`` decided the host above; urllib3 is what the transport under
+    # ``requests`` actually dials. Where the two disagree, the host approved
+    # here is not the host the socket connects to. Comparing them is an
+    # invariant rather than a list of characters to reject, so it holds as
+    # either parser changes.
+    host = _normalize_host(parsed.hostname)
+    if host != _normalize_host(transport_host):
+        return REFUSED_PARSER_DISAGREEMENT
+
+    if not host:
+        return REFUSED_EMPTY_HOST
+
+    literal = _as_ip(host)
+
+    if not resolve:
+        # No DNS on this path. An address literal is still checked, since that
+        # needs no lookup and is how most internal targets are written — in any
+        # of the encodings ``_as_ip`` understands, not just dotted-quad. A
+        # hostname is accepted here and caught at the sink.
+        if literal is not None and not _is_public(str(literal)):
+            return REFUSED_INTERNAL_LITERAL
+        if host in _LOOPBACK_NAMES or host.endswith(".localhost"):
+            return REFUSED_INTERNAL_LITERAL
+        return None
+
+    # 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 UNRESOLVABLE
+
+    if not all(_is_public(addr) for addr in addrs):
+        return REFUSED_NON_PUBLIC
+
+    return None
+
+
+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.
+
+    Leave ``resolve`` 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 an address literal is still refused,
+    but a hostname that resolves inward is accepted here and caught at the sink.
+    """
+    reason = webhook_url_refusal(url, allowed_schemes=allowed_schemes, resolve=resolve)
+    if reason is None:
+        return True
+
+    # The host, not the URL: a webhook URL routinely carries a token in its
+    # query string, and this line goes to shared logs. Without the reason,
+    # support cannot tell a scheme rejection from a resolver outage — five
+    # different remediations behind one message.
+    logger.warning(
+        "Refusing webhook URL: %s (host=%s)",
+        reason,
+        safe_host(url),
+    )
+    return False
+
+
+def safe_host(url: str | None) -> str:
+    """The host of ``url`` for logging, never the path or query string.
+
+    Public because every sink that logs a refusal needs it, and a second copy
+    is a second chance to drop the ``ValueError`` guard on a malformed literal.
+    """
+    try:
+        return urlparse(url or "").hostname or "<none>"
+    except ValueError:
+        return "<unparseable>"
diff --git a/unstract/core/src/unstract/core/notification_utils.py b/unstract/core/src/unstract/core/notification_utils.py
index e62f6cd347..342d91e384 100644
--- a/unstract/core/src/unstract/core/notification_utils.py
+++ b/unstract/core/src/unstract/core/notification_utils.py
@@ -11,6 +11,11 @@
 
 import requests
 
+from unstract.core.network.ssrf import (
+    is_retryable_refusal,
+    safe_host,
+    webhook_url_refusal,
+)
 from unstract.core.notification_enums import AuthorizationType
 
 logger = logging.getLogger(__name__)
@@ -136,6 +141,34 @@ def send_webhook_request(
     Raises:
         requests.exceptions.RequestException: If request fails after all retries
     """
+    # Guard at the sink, not at the callers, so no future caller can skip it.
+    refusal = webhook_url_refusal(url)
+    if refusal is not None:
+        # A resolver outage clears on its own, so that one stays retryable.
+        # Every other reason is a property of the URL and cannot change between
+        # attempts — retrying it wastes an attempt (and, when the refusal came
+        # from resolution, a DNS lookup on a tenant-supplied hostname) without
+        # ever changing the outcome, and delays dead-lettering.
+        retryable = is_retryable_refusal(refusal)
+        logger.error(
+            "Refusing webhook: %s (host=%s, retryable=%s)",
+            refusal,
+            safe_host(url),
+            retryable,
+        )
+        return {
+            "success": False,
+            "error": (
+                "Webhook URL could not be resolved"
+                if retryable
+                else "Webhook URL is not an allowed public destination"
+            ),
+            "refusal_reason": refusal,
+            "retryable": retryable,
+            "attempts": current_retry + 1,
+            "url": url,
+        }
+
     # Serialize payload to handle UUIDs and datetimes
     serialized_payload = serialize_notification_data(payload)
 
@@ -143,7 +176,13 @@ def send_webhook_request(
         logger.debug(f"Sending webhook to {url} (attempt {current_retry + 1})")
 
         response = requests.post(
-            url=url, json=serialized_payload, headers=headers or {}, timeout=timeout
+            url=url,
+            json=serialized_payload,
+            headers=headers or {},
+            timeout=timeout,
+            # A 302 to an internal host would otherwise be followed, and
+            # 302/303 rewrites POST to GET.
+            allow_redirects=False,
         )
 
         # Check response status
diff --git a/unstract/core/tests/test_ssrf_guard.py b/unstract/core/tests/test_ssrf_guard.py
new file mode 100644
index 0000000000..79f0015154
--- /dev/null
+++ b/unstract/core/tests/test_ssrf_guard.py
@@ -0,0 +1,392 @@
+"""Tests for the shared webhook egress guard.
+
+The parser-differential cases are URLs where ``urlparse`` and ``urllib3``
+disagree on the host, in both directions. The IPv6/IDN cases are the false
+positives a naive string compare of the two hosts produces — those are
+legitimate targets and must still be allowed.
+
+DNS is stubbed so the suite does not depend on the network; the resolver is
+exercised separately through the public-address cases.
+"""
+
+import socket
+from unittest.mock import patch
+
+import pytest
+
+from unstract.core.network.ssrf import (
+    REFUSED_INTERNAL_LITERAL,
+    REFUSED_NON_PUBLIC,
+    REFUSED_SCHEME,
+    REFUSED_UNPARSEABLE,
+    UNRESOLVABLE,
+    _normalize_host,
+    is_retryable_refusal,
+    is_safe_webhook_url,
+    safe_host,
+    webhook_url_refusal,
+)
+from unstract.core.notification_utils import send_webhook_request
+
+_REAL_GETADDRINFO = socket.getaddrinfo
+
+# Hosts the stub resolver answers for. Anything else fails to resolve.
+_FAKE_DNS = {
+    "example.com": {"93.184.216.34"},
+    "webhook.site": {"46.4.105.116"},
+    "internal.corp": {"10.0.0.5"},
+    "rebind.test": {"93.184.216.34", "127.0.0.1"},
+    "xn--e1afmkfd.xn--p1ai": {"93.184.216.34"},
+    # UTS-46 forms of faß.de and σόλος.gr. The stdlib "idna" codec maps these
+    # to fass.de and xn--wxaikc6b.gr instead, which is the bug the
+    # normalization test below pins.
+    "xn--fa-hia.de": {"93.184.216.34"},
+    "xn--wxaijb9b.gr": {"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, "", (addr, 0)) for addr in _FAKE_DNS[host]]
+
+    monkeypatch.setattr("unstract.core.network.ssrf.socket.getaddrinfo", fake_getaddrinfo)
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        # urlparse reads 1.1.1.1 here; urllib3 connects to 127.0.0.1.
+        r"https://127.0.0.1:6666\@1.1.1.1",
+        # The differential runs both ways.
+        r"https://1.1.1.1:80\@127.0.0.1/",
+    ],
+)
+def test_parser_differential_is_refused(url):
+    assert is_safe_webhook_url(url) is False
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "https://127.0.0.1/hook",
+        "https://localhost/hook",
+        "https://169.254.169.254/latest/meta-data/",  # cloud metadata
+        "https://[::1]/hook",
+        "https://10.0.0.5/hook",
+        "https://192.168.1.1/hook",
+        "https://172.16.0.1/hook",
+        "https://0.0.0.0/hook",
+        "https://internal.corp/hook",
+        # Ranges that belong to no single "is_private"-style flag but are not
+        # globally routable. Enumerating negative flags misses these.
+        "https://100.64.0.1/hook",  # RFC 6598 shared address space (CGNAT)
+        "https://198.18.0.1/hook",  # RFC 2544 benchmarking
+        "https://192.0.0.1/hook",  # IETF protocol assignments
+    ],
+)
+def test_internal_targets_are_refused(url):
+    assert is_safe_webhook_url(url) is False
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        # IANA marks all of these as not globally reachable, but
+        # ``ipaddress.is_global`` reports them as global on CPython 3.12, so
+        # the guard has to refuse them itself. If a future CPython folds one of
+        # these in, this test keeps passing.
+        "https://224.0.0.1/hook",  # IPv4 multicast, all-hosts
+        "https://239.255.255.250/hook",  # IPv4 multicast, SSDP
+        "https://[ff02::1]/hook",  # IPv6 multicast, all-nodes
+        "https://192.88.99.1/hook",  # 6to4 relay anycast
+        "https://[5f00::1]/hook",  # SRv6 SIDs
+    ],
+)
+def test_ranges_the_stdlib_calls_global_are_still_refused(url):
+    assert is_safe_webhook_url(url) is False
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "ftp://example.com/hook",
+        "file:///etc/passwd",
+        "gopher://example.com/",
+        "https://user:pass@example.com/hook",  # credentials in URL
+        "not-a-url",
+        "",
+        None,
+    ],
+)
+def test_malformed_and_disallowed_schemes_are_refused(url):
+    assert is_safe_webhook_url(url) is False
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "https://example.com/hook",
+        "https://webhook.site/abc-123",
+        "https://example.com./hook",  # trailing dot
+        "https://EXAMPLE.com/hook",  # uppercase host
+        "https://xn--e1afmkfd.xn--p1ai/hook",  # punycode IDN
+        "https://пример.рф/hook",  # raw unicode IDN, same host
+    ],
+)
+def test_public_targets_are_allowed(url):
+    assert is_safe_webhook_url(url) is True
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "https://" + "a" * 64 + ".com/hook",  # label over the 63-char limit
+        "https://ex..ample.com/hook",  # empty label
+    ],
+)
+def test_unresolvable_hosts_return_false_rather_than_raising(url, monkeypatch):
+    """getaddrinfo raises UnicodeError on these instead of failing to resolve.
+
+    Callers treat this as a boolean check, so an escaping exception becomes a
+    500 in the notification serializer and an error in the delivery task.
+
+    Runs against the real resolver: the failure is inside getaddrinfo, so the
+    stub above would make this pass for the wrong reason. No lookup is issued
+    — both hosts are rejected before any query goes out.
+    """
+    monkeypatch.setattr(
+        "unstract.core.network.ssrf.socket.getaddrinfo", _REAL_GETADDRINFO
+    )
+    assert is_safe_webhook_url(url) is False
+
+
+class TestWithoutResolution:
+    """resolve=False keeps DNS off request-handling threads.
+
+    getaddrinfo honours no timeout, so a slow resolver would stall the worker
+    serving the request. The syntactic checks still run.
+    """
+
+    @pytest.mark.parametrize(
+        "url",
+        [
+            "https://127.0.0.1/hook",
+            "https://169.254.169.254/latest/meta-data/",
+            "https://10.0.0.5/hook",
+            "https://100.64.0.1/hook",
+            "https://[::1]/hook",
+            r"https://127.0.0.1:6666\@1.1.1.1",  # parsers disagree
+            "https://user:pass@example.com/hook",  # credentials
+            "ftp://example.com/hook",  # scheme
+        ],
+    )
+    def test_literal_and_syntactic_cases_still_refused(self, url):
+        assert is_safe_webhook_url(url, resolve=False) is False
+
+    def test_no_lookup_is_issued(self, monkeypatch):
+        def explode(*_a, **_k):
+            raise AssertionError("DNS was resolved on a resolve=False call")
+
+        monkeypatch.setattr("unstract.core.network.ssrf.socket.getaddrinfo", explode)
+        assert is_safe_webhook_url("https://anything.internal/hook", resolve=False)
+
+    def test_hostname_pointing_inward_is_left_to_the_sink(self):
+        """Accepted here by design — the sink still resolves and refuses it."""
+        assert is_safe_webhook_url("https://internal.corp/hook", resolve=False) is True
+        assert is_safe_webhook_url("https://internal.corp/hook") is False
+
+
+def test_any_internal_address_in_a_multi_answer_rrset_refuses():
+    """A host that also answers with a loopback address is not safe."""
+    assert is_safe_webhook_url("https://rebind.test/hook") is False
+
+
+def test_http_is_allowed_by_default_but_not_for_tls_only_callers():
+    assert is_safe_webhook_url("http://example.com/hook") is True
+    assert (
+        is_safe_webhook_url("http://example.com/hook", allowed_schemes=("https",))
+        is False
+    )
+
+
+@pytest.mark.parametrize(
+    "raw,expected",
+    [
+        ("[::1]", "::1"),
+        ("EXAMPLE.com.", "example.com"),
+        ("пример.рф", "xn--e1afmkfd.xn--p1ai"),
+        (None, ""),
+        # UTS-46, matching urllib3. The stdlib "idna" codec is IDNA-2003 and
+        # would give "fass.de" and "xn--wxaikc6b.gr" — a host the transport
+        # never dials, so the parser-agreement check would refuse both.
+        ("faß.de", "xn--fa-hia.de"),
+        ("σόλος.gr", "xn--wxaijb9b.gr"),
+    ],
+)
+def test_normalize_host(raw, expected):
+    assert _normalize_host(raw) == expected
+
+
+@pytest.mark.parametrize("host", ["faß.de", "σόλος.gr", "пример.рф"])
+def test_normalization_agrees_with_the_transport(host):
+    """The two parsers must reduce a host the same way, or every IDN is refused.
+
+    Pinned against urllib3 itself rather than a hardcoded expectation, so this
+    fails if either encoder moves.
+    """
+    from urllib3.util import parse_url
+
+    assert _normalize_host(host) == _normalize_host(parse_url(f"https://{host}/").host)
+    assert is_safe_webhook_url(f"https://{host}/hook") is True
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "https://2130706433/hook",  # decimal
+        "https://0177.0.0.1/hook",  # octal dotted
+        "https://127.1/hook",  # short form
+        "https://localhost/hook",  # RFC 6761, no lookup needed
+        "https://api.localhost/hook",  # RFC 6761 reserves the whole subtree
+        "https://DB.LocalHost/hook",  # case-insensitive after normalization
+    ],
+)
+def test_legacy_loopback_encodings_are_refused_without_dns(url):
+    """All of these are 127.0.0.1 to the resolver.
+
+    ``ipaddress.ip_address`` parses none of the numeric forms, so on the
+    no-resolve path they would otherwise be accepted as if they were
+    hostnames — exactly the encodings used to slip a literal past a check.
+    """
+    assert is_safe_webhook_url(url, resolve=False) is False
+    assert is_safe_webhook_url(url) is False
+
+
+class TestRefusalReason:
+    """Each sink needs the reason, not just the boolean.
+
+    Without it a resolver outage and a genuinely internal target produce the
+    same log line and the same error, and the delivery task cannot tell which
+    of the two is worth retrying.
+    """
+
+    def test_public_url_has_no_reason(self):
+        assert webhook_url_refusal("https://example.com/hook") is None
+
+    def test_resolver_failure_is_reported_as_transient(self):
+        assert webhook_url_refusal("https://nowhere.invalid/hook") == UNRESOLVABLE
+
+    def test_internal_target_is_not_transient(self):
+        assert webhook_url_refusal("https://internal.corp/hook") == REFUSED_NON_PUBLIC
+        assert (
+            webhook_url_refusal("https://127.0.0.1/hook", resolve=False)
+            == REFUSED_INTERNAL_LITERAL
+        )
+
+    def test_reason_distinguishes_the_syntactic_checks(self):
+        assert (
+            webhook_url_refusal("http://example.com/hook", allowed_schemes=("https",))
+            == REFUSED_SCHEME
+        )
+
+
+class TestNotificationSink:
+    """The guard sits inside ``send_webhook_request`` so no caller can skip it.
+
+    Redirects are off on this path as well: a 302 to an internal host would
+    otherwise be followed, and 302/303 rewrites POST to GET.
+    """
+
+    @pytest.mark.parametrize(
+        "url",
+        [
+            "https://169.254.169.254/latest/meta-data/",
+            "https://127.0.0.1:8000/admin/",
+            r"https://127.0.0.1:6666\@1.1.1.1",
+        ],
+    )
+    def test_blocked_url_never_reaches_the_network(self, url):
+        with patch("unstract.core.notification_utils.requests.post") as post:
+            result = send_webhook_request(url=url, payload={"payload": 1})
+        post.assert_not_called()
+        assert result["success"] is False
+
+    def test_redirects_are_not_followed(self):
+        with patch("unstract.core.notification_utils.requests.post") as post:
+            post.return_value.status_code = 200
+            post.return_value.text = "ok"
+            send_webhook_request(url="https://example.com/hook", payload={})
+
+        assert post.call_args.kwargs["allow_redirects"] is False
+
+    def test_public_url_is_still_delivered(self):
+        with patch("unstract.core.notification_utils.requests.post") as post:
+            post.return_value.status_code = 200
+            post.return_value.text = "ok"
+            result = send_webhook_request(url="https://example.com/hook", payload={})
+
+        post.assert_called_once()
+        assert result["success"] is True
+
+    def test_a_refused_url_is_marked_not_retryable(self):
+        """The retry loop cannot change the answer, so it must not run.
+
+        Every attempt would re-issue getaddrinfo for a tenant-supplied
+        hostname and delay dead-lettering by up to max_retries × retry_delay.
+        """
+        with patch("unstract.core.notification_utils.requests.post"):
+            result = send_webhook_request(url="https://internal.corp/hook", payload={})
+
+        assert result["success"] is False
+        assert result["retryable"] is False
+
+    def test_a_resolver_outage_stays_retryable(self):
+        """A blip is not a security refusal, and must not be reported as one."""
+        with patch("unstract.core.notification_utils.requests.post"):
+            result = send_webhook_request(url="https://nowhere.invalid/hook", payload={})
+
+        assert result["success"] is False
+        assert result["retryable"] is True
+        assert result["refusal_reason"] == UNRESOLVABLE
+
+    def test_a_url_that_breaks_urlparse_is_refused_not_raised(self):
+        """The refusal log re-parses the URL; it must not raise doing so.
+
+        ``urlparse("http://[::1")`` raises ValueError. ``webhook_url_refusal``
+        catches that and returns REFUSED_UNPARSEABLE, so a second unguarded
+        parse in the sink turned a deterministic refusal into an exception that
+        the caller wraps as DeliveryError and retries like a transient failure.
+        """
+        with patch("unstract.core.notification_utils.requests.post") as post:
+            result = send_webhook_request(url="http://[::1", payload={})
+
+        post.assert_not_called()
+        assert result["success"] is False
+        assert result["retryable"] is False
+        assert result["refusal_reason"] == REFUSED_UNPARSEABLE
+
+
+class TestRetryabilityIsClassifiedOnce:
+    """Every sink asks the guard whether a refusal can clear, rather than
+    re-deriving it — a new reason has to be classified in one place."""
+
+    def test_only_a_resolver_outage_is_retryable(self):
+        assert is_retryable_refusal(UNRESOLVABLE) is True
+        for reason in (
+            REFUSED_INTERNAL_LITERAL,
+            REFUSED_NON_PUBLIC,
+            REFUSED_SCHEME,
+            REFUSED_UNPARSEABLE,
+        ):
+            assert is_retryable_refusal(reason) is False
+
+    def test_no_refusal_is_not_retryable(self):
+        assert is_retryable_refusal(None) is False
+
+    def test_safe_host_never_raises_and_never_leaks_the_query_string(self):
+        assert safe_host("https://example.com/hook?token=secret") == "example.com"
+        assert safe_host("http://[::1") == "<unparseable>"
+        assert safe_host(None) == "<none>"
diff --git a/workers/executor/executors/answer_prompt.py b/workers/executor/executors/answer_prompt.py
index 4184b78b71..460d77fa2a 100644
--- a/workers/executor/executors/answer_prompt.py
+++ b/workers/executor/executors/answer_prompt.py
@@ -10,12 +10,9 @@
 are integrated at the caller level (LegacyExecutor).
 """
 
-import ipaddress
 import logging
 import os
-import socket
 from typing import Any
-from urllib.parse import urlparse
 
 from executor.executors.constants import PromptServiceConstants as PSKeys
 from executor.executors.exceptions import LegacyExecutorError, RateLimitError
@@ -23,59 +20,6 @@
 logger = logging.getLogger(__name__)
 
 
-def _resolve_host_addresses(host: str) -> set[str]:
-    """Resolve a hostname or IP string to a set of IP address strings."""
-    try:
-        ipaddress.ip_address(host)
-        return {host}
-    except ValueError:
-        pass
-    try:
-        return {
-            sockaddr[0]
-            for _family, _type, _proto, _canonname, sockaddr in socket.getaddrinfo(
-                host, None, type=socket.SOCK_STREAM
-            )
-        }
-    except Exception:
-        return set()
-
-
-def _is_safe_public_url(url: str) -> bool:
-    """Validate webhook URL for SSRF protection.
-
-    Only allows HTTPS and blocks private/loopback/internal addresses.
-    """
-    try:
-        p = urlparse(url)
-        if p.scheme not in ("https",):
-            return False
-        host = p.hostname or ""
-        if host in ("localhost",):
-            return False
-
-        addrs = _resolve_host_addresses(host)
-        if not addrs:
-            return False
-
-        for addr in addrs:
-            try:
-                ip = ipaddress.ip_address(addr)
-            except ValueError:
-                return False
-            if (
-                ip.is_private
-                or ip.is_loopback
-                or ip.is_link_local
-                or ip.is_reserved
-                or ip.is_multicast
-            ):
-                return False
-        return True
-    except Exception:
-        return False
-
-
 class AnswerPromptService:
     @staticmethod
     def extract_variable(
@@ -395,9 +339,9 @@ def _run_webhook_postprocess(
         if not webhook_url:
             logger.warning("Postprocessing webhook enabled but URL missing; skipping.")
             return parsed_data, None
-        if not _is_safe_public_url(webhook_url):
-            logger.warning("Postprocessing webhook URL is not allowed; skipping.")
-            return parsed_data, None
+        # No URL check here: _make_webhook_request applies the identical guard
+        # at the sink. Duplicating it only bought a second blocking getaddrinfo
+        # per prompt per document.
         try:
             return postprocess_data(
                 parsed_data,
diff --git a/workers/executor/executors/postprocessor.py b/workers/executor/executors/postprocessor.py
index bf14a56698..d179ca9895 100644
--- a/workers/executor/executors/postprocessor.py
+++ b/workers/executor/executors/postprocessor.py
@@ -9,6 +9,8 @@
 
 import requests
 
+from unstract.core.network.ssrf import is_safe_webhook_url
+
 logger = logging.getLogger(__name__)
 
 
@@ -55,8 +57,14 @@ def _process_successful_response(
 
 def _make_webhook_request(
     webhook_url: str, payload: dict, timeout: float
-) -> tuple[dict[str, Any], list | None] | None:
+) -> dict[str, Any] | None:
     """Make webhook request and return processed response or None on failure."""
+    # Guard at the sink so it cannot be skipped by a caller. This path has
+    # always required TLS, so keep it to https.
+    if not is_safe_webhook_url(webhook_url, allowed_schemes=("https",)):
+        logger.warning("Postprocessing webhook URL is not allowed; skipping.")
+        return None
+
     try:
         response = requests.post(
             webhook_url,
diff --git a/workers/notification/tasks.py b/workers/notification/tasks.py
index f1c780ae0c..be87de5f01 100644
--- a/workers/notification/tasks.py
+++ b/workers/notification/tasks.py
@@ -30,6 +30,17 @@
 
 logger = WorkerLogger.get_logger(__name__)
 
+
+class TerminalWebhookRefusal(Exception):
+    """A refusal no retry can clear, already dead-lettered by its raiser.
+
+    Distinct from the errors below it so ``send_webhook_notification``'s broad
+    ``except Exception`` cannot route it back into ``self.retry`` — a plain
+    ``Exception`` here was retried ``max_retries`` times for a URL that will
+    never be dialable, and dead-lettered the buffer rows a second time.
+    """
+
+
 # Initialize worker configuration
 config = WorkerConfig.from_env("NOTIFICATION")
 
@@ -311,10 +322,28 @@ def send_webhook_notification(
             _mark_buffer_outcome(buffer_row_ids, organization_id, dispatched=True)
             return None  # Success - matches original behavior
         else:
-            # Failed delivery - raise exception for retry handling
             error_message = result.get("message", "Unknown webhook delivery error")
+
+            # A refusal the sink marked non-retryable is a property of the URL
+            # itself, so no attempt can change it. Dead-letter now instead of
+            # re-resolving a tenant-supplied hostname up to max_retries times.
+            if result.get("details", {}).get("retryable") is False:
+                logger.error(
+                    f"Webhook to {url} refused and not retryable: {error_message}"
+                )
+                _mark_buffer_outcome(buffer_row_ids, organization_id, dispatched=False)
+                if raise_on_final_failure:
+                    raise TerminalWebhookRefusal(error_message)
+                return None
+
+            # Failed delivery - raise exception for retry handling
             raise Exception(error_message)
 
+    except TerminalWebhookRefusal:
+        # Buffer rows are already dead-lettered above and no attempt can change
+        # the outcome. Surface FAILURE to the caller without re-entering retry.
+        raise
+
     except (ValidationError, DeliveryError) as e:
         # Handle provider-specific errors
         if max_retries is not None:
diff --git a/workers/tests/test_notification_terminal_refusal.py b/workers/tests/test_notification_terminal_refusal.py
new file mode 100644
index 0000000000..897cad8a10
--- /dev/null
+++ b/workers/tests/test_notification_terminal_refusal.py
@@ -0,0 +1,103 @@
+"""A webhook refusal no retry can clear must not be retried.
+
+``send_webhook_notification`` dead-letters the buffer rows the moment the sink
+reports ``retryable: False``, then honours ``raise_on_final_failure``. That
+raise happens inside the task's own ``try``, so a plain ``Exception`` was
+caught by the broad handler below it and fed straight back into
+``self.retry(...)`` — the URL got re-resolved up to ``max_retries`` times and
+``_mark_buffer_outcome(dispatched=False)`` ran a second time on exhaustion.
+
+``TerminalWebhookRefusal`` exists to cross that handler untouched. These tests
+pin the behaviour rather than the mechanism: one POST, one dead-letter mark,
+and no ``Retry``.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+from celery.exceptions import Retry
+from notification.tasks import TerminalWebhookRefusal, send_webhook_notification
+
+_URL = "https://127.0.0.1/hook"
+_BUFFER_IDS = ["b1", "b2"]
+_ORG = 7
+
+
+class _RefusingProvider:
+    """Stands in for the sink refusing a URL that can never be dialled."""
+
+    def __init__(self, *, retryable: bool) -> None:
+        self.posts = 0
+        self._retryable = retryable
+
+    def send(self, notification_data: dict) -> dict:
+        self.posts += 1
+        return {
+            "success": False,
+            "message": "Webhook URL is not an allowed public destination",
+            "details": {"retryable": self._retryable},
+        }
+
+
+def _run(*, retryable: bool, max_retries: int, raise_on_final_failure: bool):
+    provider = _RefusingProvider(retryable=retryable)
+    marks: list[bool] = []
+    with (
+        patch(
+            "notification.tasks._get_webhook_provider_for_url", return_value=provider
+        ),
+        patch(
+            "notification.tasks._mark_buffer_outcome",
+            side_effect=lambda ids, org, *, dispatched: marks.append(dispatched),
+        ),
+    ):
+        raised: BaseException | None = None
+        try:
+            send_webhook_notification.apply(
+                args=[_URL, {"text": "hi"}, {"Content-Type": "application/json"}, 30],
+                kwargs={
+                    "max_retries": max_retries,
+                    "retry_delay": 10,
+                    "platform": None,
+                    "raise_on_final_failure": raise_on_final_failure,
+                    "buffer_row_ids": _BUFFER_IDS,
+                    "organization_id": _ORG,
+                },
+                throw=True,
+            )
+        except BaseException as exc:  # noqa: BLE001 - the type is the assertion
+            raised = exc
+    return provider, marks, raised
+
+
+def test_terminal_refusal_is_not_retried_when_it_must_raise():
+    # raise_on_final_failure=True with retries left is the case that regressed:
+    # the raise has to reach the caller as a FAILURE, not loop back into retry.
+    provider, marks, raised = _run(
+        retryable=False, max_retries=3, raise_on_final_failure=True
+    )
+    assert isinstance(raised, TerminalWebhookRefusal)
+    assert not isinstance(raised, Retry)
+    assert provider.posts == 1  # the URL is resolved once, not max_retries times
+    assert marks == [False]  # dead-lettered exactly once
+
+
+def test_terminal_refusal_without_raise_returns_quietly():
+    provider, marks, raised = _run(
+        retryable=False, max_retries=3, raise_on_final_failure=False
+    )
+    assert raised is None
+    assert provider.posts == 1
+    assert marks == [False]
+
+
+@pytest.mark.parametrize("raise_on_final", [True, False])
+def test_a_retryable_failure_still_retries(raise_on_final):
+    """The fast path must not swallow a resolver outage, which can clear."""
+    provider, marks, raised = _run(
+        retryable=True, max_retries=3, raise_on_final_failure=raise_on_final
+    )
+    assert isinstance(raised, Retry)
+    assert marks == []  # not dead-lettered while attempts remain
diff --git a/workers/tests/test_variable_replacement_postprocessor.py b/workers/tests/test_variable_replacement_postprocessor.py
index cbd2215fb8..c759a6b77b 100644
--- a/workers/tests/test_variable_replacement_postprocessor.py
+++ b/workers/tests/test_variable_replacement_postprocessor.py
@@ -276,6 +276,19 @@ class TestPostprocessor:
     PARSED = {"field": "original"}
     HIGHLIGHT = [{"page": 1, "spans": []}]
 
+    @pytest.fixture(autouse=True)
+    def _allow_webhook_url(self):
+        """Let the fictional test URL past the egress guard.
+
+        These tests exercise postprocessing behaviour, not URL safety, and
+        ``hook.example.com`` does not resolve. The guard itself is covered by
+        ``test_webhook_ssrf_sink`` and ``unstract/core``'s ``test_ssrf_guard``.
+        """
+        with patch(
+            "executor.executors.postprocessor.is_safe_webhook_url", return_value=True
+        ):
+            yield
+
     # --- disabled / no-op paths ---
 
     def test_disabled_returns_original(self):
diff --git a/workers/tests/test_webhook_ssrf_sink.py b/workers/tests/test_webhook_ssrf_sink.py
new file mode 100644
index 0000000000..b15a753032
--- /dev/null
+++ b/workers/tests/test_webhook_ssrf_sink.py
@@ -0,0 +1,132 @@
+"""The postprocessing webhook guard must live in the sink, not in its caller.
+
+The URL check used to run one frame up in ``answer_prompt``, which left any
+new caller of ``_make_webhook_request`` to remember it. It now sits in the
+sink; these tests call the sink directly.
+
+Both directions are covered on purpose. Refusing every blocked URL is easy to
+get right and easy to over-do: a guard that refuses *everything* silently
+disables postprocessing for every tenant, and — because a refusal returns the
+unprocessed data on a run still reported successful — nothing else in the
+suite would notice. The allow-path cases are what pin that.
+
+The notification sink's equivalent tests live in
+``unstract/core/tests/test_ssrf_guard.py``, next to that sink.
+"""
+
+from unittest.mock import patch
+
+import pytest
+from executor.executors.answer_prompt import AnswerPromptService
+from executor.executors.postprocessor import _make_webhook_request
+
+BLOCKED_URLS = [
+    "https://169.254.169.254/latest/meta-data/",  # cloud metadata
+    "https://127.0.0.1:8000/admin/",
+    r"https://127.0.0.1:6666\@1.1.1.1",  # parsers disagree on the host
+    "http://example.com/hook",  # this path has always required TLS
+]
+
+# Stub DNS so the allow-path cases do not depend on the network.
+_FAKE_DNS = {"hook.example.com": "93.184.216.34"}
+
+
+@pytest.fixture
+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
+    )
+
+
+@pytest.mark.parametrize("url", BLOCKED_URLS)
+def test_postprocessor_sink_refuses_without_calling_out(url):
+    with patch("executor.executors.postprocessor.requests.post") as post:
+        assert _make_webhook_request(url, {"payload": 1}, timeout=5) is None
+    post.assert_not_called()
+
+
+def test_postprocessor_sink_still_calls_a_public_https_url(stub_dns):
+    """Without this, a guard that refuses everything keeps the suite green."""
+    with patch("executor.executors.postprocessor.requests.post") as post:
+        post.return_value.status_code = 200
+        post.return_value.json.return_value = {"structured_output": {"field": "new"}}
+        result = _make_webhook_request(
+            "https://hook.example.com/hook", {"payload": 1}, timeout=5
+        )
+
+    post.assert_called_once()
+    assert post.call_args.kwargs["allow_redirects"] is False
+    assert result == {"structured_output": {"field": "new"}}
+
+
+class TestRunWebhookPostprocess:
+    """``_run_webhook_postprocess`` is the caller that used to hold the guard.
+
+    It no longer checks the URL itself — the sink does — so what matters here
+    is that it still reaches the sink, and that a refusal leaves the caller's
+    data untouched rather than raising into the executor.
+    """
+
+    PARSED = {"field": "original"}
+
+    def test_missing_url_skips_without_touching_the_network(self):
+        with patch("executor.executors.postprocessor.requests.post") as post:
+            result, highlights = AnswerPromptService._run_webhook_postprocess(
+                parsed_data=self.PARSED, webhook_url=None, highlight_data=None
+            )
+        post.assert_not_called()
+        assert result == self.PARSED
+        assert highlights is None
+
+    def test_refused_url_returns_the_original_data(self, stub_dns):
+        with patch("executor.executors.postprocessor.requests.post") as post:
+            result, _ = AnswerPromptService._run_webhook_postprocess(
+                parsed_data=self.PARSED,
+                webhook_url="https://127.0.0.1/hook",
+                highlight_data=None,
+            )
+        post.assert_not_called()
+        assert result == self.PARSED
+
+    def test_allowed_url_is_delivered_and_its_output_used(self, stub_dns):
+        """The guard is applied once, in the sink, and does not block a real URL."""
+        with patch("executor.executors.postprocessor.requests.post") as post:
+            post.return_value.status_code = 200
+            post.return_value.json.return_value = {
+                "structured_output": {"field": "processed"}
+            }
+            result, _ = AnswerPromptService._run_webhook_postprocess(
+                parsed_data=self.PARSED,
+                webhook_url="https://hook.example.com/hook",
+                highlight_data=None,
+            )
+
+        post.assert_called_once()
+        assert result == {"field": "processed"}
+
+    def test_the_guard_runs_only_once_per_call(self, monkeypatch):
+        """The caller-side check was removed; resolving twice was its only cost."""
+        lookups = []
+
+        def counting_getaddrinfo(host, *_args, **_kwargs):
+            lookups.append(host)
+            return [(None, None, None, "", (_FAKE_DNS[host], 0))]
+
+        monkeypatch.setattr(
+            "unstract.core.network.ssrf.socket.getaddrinfo", counting_getaddrinfo
+        )
+        with patch("executor.executors.postprocessor.requests.post") as post:
+            post.return_value.status_code = 200
+            post.return_value.json.return_value = {"structured_output": {}}
+            AnswerPromptService._run_webhook_postprocess(
+                parsed_data=self.PARSED,
+                webhook_url="https://hook.example.com/hook",
+                highlight_data=None,
+            )
+
+        assert lookups == ["hook.example.com"]