diff --git a/backend/secuscan/notification_service.py b/backend/secuscan/notification_service.py index 54145f9f..72904972 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 8c310496..cbc1ab1e 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,