diff --git a/app/utils/user_notification.py b/app/utils/user_notification.py index 791b2219..a42ea9e7 100644 --- a/app/utils/user_notification.py +++ b/app/utils/user_notification.py @@ -12,11 +12,13 @@ import smtplib from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from typing import Any +from urllib.parse import urlparse import httpx from app.database import SessionLocal from app.models import InAppNotification, UserNotificationPreference, UserNotificationTarget +from app.utils.network import is_private_ip logger = logging.getLogger(__name__) @@ -28,6 +30,11 @@ USER_EVENT_LABELS: dict[str, str] = { EVENT_DOCUMENT_PROCESSED: "Document Processed", EVENT_DOCUMENT_FAILED: "Document Processing Failed", } +METADATA_ENDPOINTS = { + "169.254.169.254", + "169.254.169.253", + "metadata.google.internal", +} def create_in_app_notification( @@ -128,6 +135,20 @@ def _send_webhook_notification(target_config: dict[str, Any], event_type: str, t logger.warning("Webhook notification target missing url") return False + parsed_url = urlparse(url) + if parsed_url.scheme not in {"http", "https"}: + logger.warning("Webhook notification to %s blocked: invalid scheme %s", url, parsed_url.scheme) + return False + + hostname = parsed_url.hostname + if not hostname: + logger.warning("Webhook notification to %s blocked: missing hostname", url) + return False + + if hostname in METADATA_ENDPOINTS or is_private_ip(hostname): + logger.warning("Webhook notification to %s blocked: private or metadata endpoint", url) + return False + payload = { "event": event_type, "title": title, diff --git a/app/utils/webhook.py b/app/utils/webhook.py index 0d181e71..9273e75d 100644 --- a/app/utils/webhook.py +++ b/app/utils/webhook.py @@ -18,11 +18,13 @@ import json import logging import time from typing import Any +from urllib.parse import urlparse import requests from app.database import SessionLocal from app.models import WebhookConfig +from app.utils.network import is_private_ip logger = logging.getLogger(__name__) @@ -40,6 +42,11 @@ VALID_EVENTS: frozenset[str] = frozenset( #: Timeout (seconds) for outgoing webhook HTTP requests. WEBHOOK_TIMEOUT = 10 +METADATA_ENDPOINTS = { + "169.254.169.254", + "169.254.169.253", + "metadata.google.internal", +} def compute_signature(payload_bytes: bytes, secret: str) -> str: @@ -67,6 +74,20 @@ def deliver_webhook(url: str, payload: dict[str, Any], secret: str | None = None Returns: ``True`` when the remote server responds with a 2xx status. """ + parsed_url = urlparse(url) + if parsed_url.scheme not in {"http", "https"}: + logger.warning("Webhook to %s blocked: invalid scheme %s", url, parsed_url.scheme) + return False + + hostname = parsed_url.hostname + if not hostname: + logger.warning("Webhook to %s blocked: missing hostname", url) + return False + + if hostname in METADATA_ENDPOINTS or is_private_ip(hostname): + logger.warning("Webhook to %s blocked: private or metadata endpoint", url) + return False + body = json.dumps(payload, default=str, sort_keys=True) body_bytes = body.encode("utf-8") diff --git a/tests/test_user_notification_service.py b/tests/test_user_notification_service.py index 79742757..c9c3cbfd 100644 --- a/tests/test_user_notification_service.py +++ b/tests/test_user_notification_service.py @@ -235,7 +235,10 @@ class TestSendWebhookNotification: mock_response.status_code = 200 mock_response.raise_for_status = MagicMock() - with patch("app.utils.user_notification.httpx.post", return_value=mock_response) as mock_post: + with ( + patch("app.utils.user_notification.is_private_ip", return_value=False), + patch("app.utils.user_notification.httpx.post", return_value=mock_response) as mock_post, + ): result = _send_webhook_notification( {"url": "https://hook.example.com/test", "secret": "mysecret"}, "document.processed", @@ -256,7 +259,10 @@ class TestSendWebhookNotification: mock_response.status_code = 200 mock_response.raise_for_status = MagicMock() - with patch("app.utils.user_notification.httpx.post", return_value=mock_response) as mock_post: + with ( + patch("app.utils.user_notification.is_private_ip", return_value=False), + patch("app.utils.user_notification.httpx.post", return_value=mock_response) as mock_post, + ): result = _send_webhook_notification( {"url": "https://hook.example.com/test"}, "document.failed", @@ -272,9 +278,12 @@ class TestSendWebhookNotification: """_send_webhook_notification returns False when httpx raises.""" from app.utils.user_notification import _send_webhook_notification - with patch( - "app.utils.user_notification.httpx.post", - side_effect=Exception("connection error"), + with ( + patch("app.utils.user_notification.is_private_ip", return_value=False), + patch( + "app.utils.user_notification.httpx.post", + side_effect=Exception("connection error"), + ), ): result = _send_webhook_notification( {"url": "https://hook.example.com/test"}, @@ -300,7 +309,10 @@ class TestSendWebhookNotification: ) ) - with patch("app.utils.user_notification.httpx.post", return_value=mock_response): + with ( + patch("app.utils.user_notification.is_private_ip", return_value=False), + patch("app.utils.user_notification.httpx.post", return_value=mock_response), + ): result = _send_webhook_notification( {"url": "https://hook.example.com/test"}, "document.processed", @@ -310,6 +322,69 @@ class TestSendWebhookNotification: assert result is False + def test_blocks_private_webhook_target(self): + """Webhook delivery is skipped for private network targets.""" + from app.utils.user_notification import _send_webhook_notification + + with ( + patch("app.utils.user_notification.is_private_ip", return_value=True), + patch("app.utils.user_notification.httpx.post") as mock_post, + ): + result = _send_webhook_notification( + {"url": "https://10.0.0.5/test"}, + "document.processed", + "T", + "M", + ) + + assert result is False + mock_post.assert_not_called() + + def test_blocks_metadata_webhook_target(self): + """Webhook delivery is skipped for cloud metadata endpoints.""" + from app.utils.user_notification import _send_webhook_notification + + with patch("app.utils.user_notification.httpx.post") as mock_post: + result = _send_webhook_notification( + {"url": "http://169.254.169.254/latest/meta-data"}, + "document.processed", + "T", + "M", + ) + + assert result is False + mock_post.assert_not_called() + + def test_blocks_invalid_webhook_scheme(self): + """Webhook delivery is skipped for unsupported URL schemes.""" + from app.utils.user_notification import _send_webhook_notification + + with patch("app.utils.user_notification.httpx.post") as mock_post: + result = _send_webhook_notification( + {"url": "file:///etc/passwd"}, + "document.processed", + "T", + "M", + ) + + assert result is False + mock_post.assert_not_called() + + def test_blocks_webhook_without_hostname(self): + """Webhook delivery is skipped when the URL has no hostname.""" + from app.utils.user_notification import _send_webhook_notification + + with patch("app.utils.user_notification.httpx.post") as mock_post: + result = _send_webhook_notification( + {"url": "https:///missing-host"}, + "document.processed", + "T", + "M", + ) + + assert result is False + mock_post.assert_not_called() + # --------------------------------------------------------------------------- # dispatch_user_notification – preference loop diff --git a/tests/test_webhooks.py b/tests/test_webhooks.py index ea6925a4..fc2cf9b1 100644 --- a/tests/test_webhooks.py +++ b/tests/test_webhooks.py @@ -81,6 +81,7 @@ class TestDeliverWebhook: def test_success_returns_true(self, mocker): """A 200 response returns True.""" + mocker.patch("app.utils.webhook.is_private_ip", return_value=False) mock_post = mocker.patch("app.utils.webhook.requests.post") mock_post.return_value = MagicMock(ok=True, status_code=200) @@ -90,6 +91,7 @@ class TestDeliverWebhook: def test_non_2xx_returns_false(self, mocker): """A non-2xx response returns False.""" + mocker.patch("app.utils.webhook.is_private_ip", return_value=False) mock_post = mocker.patch("app.utils.webhook.requests.post") mock_post.return_value = MagicMock(ok=False, status_code=500) @@ -100,6 +102,7 @@ class TestDeliverWebhook: """A network error returns False.""" import requests + mocker.patch("app.utils.webhook.is_private_ip", return_value=False) mocker.patch("app.utils.webhook.requests.post", side_effect=requests.ConnectionError("fail")) result = deliver_webhook("https://example.com/hook", {"event": "test"}) @@ -107,6 +110,7 @@ class TestDeliverWebhook: def test_signature_header_included_when_secret(self, mocker): """X-Webhook-Signature header is present when a secret is supplied.""" + mocker.patch("app.utils.webhook.is_private_ip", return_value=False) mock_post = mocker.patch("app.utils.webhook.requests.post") mock_post.return_value = MagicMock(ok=True, status_code=200) @@ -118,6 +122,7 @@ class TestDeliverWebhook: def test_no_signature_header_without_secret(self, mocker): """X-Webhook-Signature header is absent when no secret is supplied.""" + mocker.patch("app.utils.webhook.is_private_ip", return_value=False) mock_post = mocker.patch("app.utils.webhook.requests.post") mock_post.return_value = MagicMock(ok=True, status_code=200) @@ -126,6 +131,43 @@ class TestDeliverWebhook: headers = call_kwargs.kwargs.get("headers") or call_kwargs[1].get("headers") assert "X-Webhook-Signature" not in headers + def test_private_target_is_blocked(self, mocker): + """Private network webhook targets are not called.""" + mocker.patch("app.utils.webhook.is_private_ip", return_value=True) + mock_post = mocker.patch("app.utils.webhook.requests.post") + + result = deliver_webhook("https://10.0.0.5/hook", {"event": "test"}) + + assert result is False + mock_post.assert_not_called() + + def test_metadata_target_is_blocked(self, mocker): + """Cloud metadata webhook targets are not called.""" + mock_post = mocker.patch("app.utils.webhook.requests.post") + + result = deliver_webhook("http://169.254.169.254/latest/meta-data", {"event": "test"}) + + assert result is False + mock_post.assert_not_called() + + def test_invalid_scheme_is_blocked(self, mocker): + """Unsupported webhook URL schemes are not called.""" + mock_post = mocker.patch("app.utils.webhook.requests.post") + + result = deliver_webhook("file:///etc/passwd", {"event": "test"}) + + assert result is False + mock_post.assert_not_called() + + def test_missing_hostname_is_blocked(self, mocker): + """Webhook URLs without a hostname are not called.""" + mock_post = mocker.patch("app.utils.webhook.requests.post") + + result = deliver_webhook("https:///missing-host", {"event": "test"}) + + assert result is False + mock_post.assert_not_called() + # --------------------------------------------------------------------------- # Unit tests – get_active_webhooks_for_event (DB)