fix: report webhook redirects as failures instead of silent success

backend/secuscan/notification_service.py

@@ -423,33 +423,48 @@ async def send_webhook(
 
         if response.status_code in (301, 302, 303, 307, 308):
             redirect_url = response.headers.get("location", "")
-            if redirect_url:
-                from urllib.parse import urlparse
-
-                parsed_redirect = urlparse(redirect_url)
-                if parsed_redirect.hostname:
-                    try:
-                        redirect_ips = socket.getaddrinfo(
-                            parsed_redirect.hostname, parsed_redirect.port or 443
-                        )
-                        for _family, _stype, _proto, _cname, sockaddr in redirect_ips:
-                            rip = ipaddress.ip_address(sockaddr[0])
-                            for blocked_cidr in settings.notification_blocked_ip_ranges:
-                                try:
-                                    if rip in ipaddress.ip_network(
-                                        blocked_cidr, strict=False
-                                    ):
-                                        return (
-                                            False,
-                                            f"Redirect to blocked IP range: {blocked_cidr}",
-                                        )
-                                except ValueError:
-                                    continue
-                    except OSError:
-                        return (
-                            False,
-                            f"Could not resolve redirect target: {redirect_url}",
-                        )
+            if not redirect_url:
+                return (
+                    False,
+                    f"Webhook returned HTTP {response.status_code} with no Location header, payload not delivered",
+                )
+
+            from urllib.parse import urlparse
+
+            parsed_redirect = urlparse(redirect_url)
+            if not parsed_redirect.hostname:
+                return (
+                    False,
+                    f"Webhook returned HTTP {response.status_code} redirect with no hostname: {redirect_url}, payload not delivered",
+                )
+
+            try:
+                redirect_ips = socket.getaddrinfo(
+                    parsed_redirect.hostname, parsed_redirect.port or 443
+                )
+                for _family, _stype, _proto, _cname, sockaddr in redirect_ips:
+                    rip = ipaddress.ip_address(sockaddr[0])
+                    for blocked_cidr in settings.notification_blocked_ip_ranges:
+                        try:
+                            if rip in ipaddress.ip_network(
+                                blocked_cidr, strict=False
+                            ):
+                                return (
+                                    False,
+                                    f"Redirect to blocked IP range: {blocked_cidr}",
+                                )
+                        except ValueError:
+                            continue
+            except OSError:
+                return (
+                    False,
+                    f"Could not resolve redirect target: {redirect_url}",
+                )
+
+            return (
+                False,
+                f"Webhook returned HTTP {response.status_code} redirect to {redirect_url}, payload not delivered",
+            )
 
         return True, None
     except httpx.HTTPError as exc:

testing/backend/unit/test_notification_service.py

@@ -575,6 +575,93 @@ def fake_getaddrinfo(hostname, port=None, *args, **kwargs):
     assert "blocked" in err.lower()
 
 
+@pytest.mark.asyncio
+async def test_send_webhook_redirect_to_allowed_ip_reports_failure():
+    """Redirect to a non-blocked IP reports failure (payload was not delivered)."""
+    from backend.secuscan.notification_service import send_webhook
+
+    mock_response = AsyncMock()
+    mock_response.status_code = 302
+    mock_response.headers = {"location": "https://new-hooks.example.com/v2/alert"}
+    mock_post = AsyncMock(return_value=mock_response)
+
+    def fake_getaddrinfo(hostname, port=None, *args, **kwargs):
+        if "hooks.example.com" in hostname:
+            return [(socket.AF_INET, None, None, None, ("93.184.216.34", 443))]
+        return [(socket.AF_INET, None, None, None, ("93.184.216.35", 443))]
+
+    with (
+        patch("httpx.AsyncClient", return_value=_mock_async_client(mock_post)),
+        patch(
+            "backend.secuscan.notification_service.socket.getaddrinfo",
+            side_effect=fake_getaddrinfo,
+        ),
+    ):
+        ok, err = await send_webhook(
+            "https://hooks.example.com/alert", {"event": "test"}
+        )
+
+    assert ok is False
+    assert "redirect" in err.lower()
+    assert "not delivered" in err.lower()
+
+
+@pytest.mark.asyncio
+async def test_send_webhook_redirect_empty_location():
+    """Redirect with no Location header reports failure."""
+    from backend.secuscan.notification_service import send_webhook
+
+    mock_response = AsyncMock()
+    mock_response.status_code = 302
+    mock_response.headers = {}
+    mock_post = AsyncMock(return_value=mock_response)
+
+    def fake_getaddrinfo(*args, **kwargs):
+        return [(socket.AF_INET, None, None, None, ("93.184.216.34", 443))]
+
+    with (
+        patch("httpx.AsyncClient", return_value=_mock_async_client(mock_post)),
+        patch(
+            "backend.secuscan.notification_service.socket.getaddrinfo",
+            side_effect=fake_getaddrinfo,
+        ),
+    ):
+        ok, err = await send_webhook(
+            "https://hooks.example.com/alert", {"event": "test"}
+        )
+
+    assert ok is False
+    assert "no Location" in err.lower() or "no location" in err.lower()
+
+
+@pytest.mark.asyncio
+async def test_send_webhook_redirect_relative_location():
+    """Redirect to a relative URL (no hostname) reports failure."""
+    from backend.secuscan.notification_service import send_webhook
+
+    mock_response = AsyncMock()
+    mock_response.status_code = 301
+    mock_response.headers = {"location": "/v2/alert"}
+    mock_post = AsyncMock(return_value=mock_response)
+
+    def fake_getaddrinfo(*args, **kwargs):
+        return [(socket.AF_INET, None, None, None, ("93.184.216.34", 443))]
+
+    with (
+        patch("httpx.AsyncClient", return_value=_mock_async_client(mock_post)),
+        patch(
+            "backend.secuscan.notification_service.socket.getaddrinfo",
+            side_effect=fake_getaddrinfo,
+        ),
+    ):
+        ok, err = await send_webhook(
+            "https://hooks.example.com/alert", {"event": "test"}
+        )
+
+    assert ok is False
+    assert "no hostname" in err.lower() or "redirect" in err.lower()
+
+
 @pytest.mark.asyncio
 async def test_send_webhook_https_delivery_pins_ip_and_preserves_tls_hostname(
     monkeypatch,