From 8c6963ffaa12796eba95d6e71b947efb396dcf0b Mon Sep 17 00:00:00 2001 From: Naman Singh Date: Wed, 8 Jul 2026 12:39:41 +0530 Subject: [PATCH] fix: report webhook redirects as failures instead of silent success send_webhook() with follow_redirects=False was validating redirect target IPs against the SSRF blocklist but then returning (True, None) without actually delivering the payload. Security notifications would silently fail to arrive at redirected webhook endpoints. Now properly returns (False, ...) for all redirect outcomes: - Redirect to blocked IP (unchanged) - Redirect to allowed IP (new: reports non-delivery) - Empty Location header (new: reports non-delivery) - Relative redirect with no hostname (new: reports non-delivery) - Unresolvable redirect target (unchanged) Fixes #1739 --- backend/secuscan/notification_service.py | 69 +++++++++------ .../backend/unit/test_notification_service.py | 87 +++++++++++++++++++ 2 files changed, 129 insertions(+), 27 deletions(-) diff --git a/backend/secuscan/notification_service.py b/backend/secuscan/notification_service.py index 54145f9f7..729049728 100644 --- a/backend/secuscan/notification_service.py +++ b/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: diff --git a/testing/backend/unit/test_notification_service.py b/testing/backend/unit/test_notification_service.py index 8c310496b..cbc1ab1ef 100644 --- a/testing/backend/unit/test_notification_service.py +++ b/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,