From 8079db78937237b6eef182168d66f0fd94b78999 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Sun, 15 Mar 2026 04:10:11 +0000 Subject: [PATCH] =?UTF-8?q?=F0=9F=9B=A1=EF=B8=8F=20Sentinel:=20[HIGH]=20Fi?= =?UTF-8?q?x=20SSRF=20in=20WebDAV=20connection=20test?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- .jules/sentinel.md | 4 ++++ app/api/integrations.py | 11 +++-------- app/api/url_upload.py | 32 +------------------------------- app/utils/network.py | 33 +++++++++++++++++++++++++++++++++ tests/test_coverage_polish.py | 4 ++-- tests/test_url_upload.py | 14 +++++++------- 6 files changed, 50 insertions(+), 48 deletions(-) create mode 100644 .jules/sentinel.md create mode 100644 app/utils/network.py diff --git a/.jules/sentinel.md b/.jules/sentinel.md new file mode 100644 index 00000000..7833fd12 --- /dev/null +++ b/.jules/sentinel.md @@ -0,0 +1,4 @@ +## 2024-05-24 - SSRF in WebDAV connection test +**Vulnerability:** The `_test_webdav_connection` function had a custom SSRF check that failed to resolve DNS names, allowing attackers to bypass the check by providing a domain that resolves to an internal IP (e.g., `127.0.0.1`). +**Learning:** DNS resolution is required for robust SSRF protection when validating URLs provided by users. +**Prevention:** Use a centralized `is_private_ip` function (now in `app/utils/network.py`) that resolves the hostname to its IPs and checks if any are private. diff --git a/app/api/integrations.py b/app/api/integrations.py index 25eddab1..ee5740cf 100644 --- a/app/api/integrations.py +++ b/app/api/integrations.py @@ -574,14 +574,9 @@ def _test_webdav_connection(config: dict[str, Any] | None, credentials: dict[str # Block requests to private/internal IPs to prevent SSRF hostname = parsed.hostname or "" if hostname: - try: - addr = ipaddress.ip_address(hostname) - if addr.is_private or addr.is_loopback or addr.is_link_local: - return {"success": False, "message": "URLs pointing to internal or private networks are not allowed"} - except ValueError: - # Hostname is not an IP literal — allow DNS names through - if hostname in ("localhost", "localhost.localdomain"): - return {"success": False, "message": "URLs pointing to localhost are not allowed"} + from app.utils.network import is_private_ip + if is_private_ip(hostname): + return {"success": False, "message": "URLs pointing to internal or private networks are not allowed"} try: import base64 diff --git a/app/api/url_upload.py b/app/api/url_upload.py index f8e34d7b..eaec0518 100644 --- a/app/api/url_upload.py +++ b/app/api/url_upload.py @@ -19,6 +19,7 @@ from app.config import settings from app.tasks.process_document import process_document from app.utils.allowed_types import ALLOWED_MIME_TYPES from app.utils.filename_utils import sanitize_filename +from app.utils.network import is_private_ip # Set up logging logger = logging.getLogger(__name__) @@ -42,37 +43,6 @@ class URLUploadRequest(BaseModel): return v -def is_private_ip(hostname: str) -> bool: - """ - Check if a hostname resolves to a private/internal IP address. - Protects against SSRF attacks by blocking access to internal networks. - """ - try: - # Try to parse as IP address directly - ip = ipaddress.ip_address(hostname) - return ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved - except ValueError: - # Not a direct IP, try to resolve hostname - try: - import socket - - # Get all IP addresses for this hostname - addr_info = socket.getaddrinfo(hostname, None) - for info in addr_info: - ip_str = info[4][0] - ip = ipaddress.ip_address(ip_str) - # Block if ANY resolved IP is private/internal - if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved: - return True - return False - except (socket.gaierror, socket.error): - # Cannot resolve - allow for testing/development - # In production, DNS should work properly - # Log this for debugging - logger.warning(f"Could not resolve hostname: {hostname}") - return False # Changed from True to False to allow external domains in tests - - def validate_url_safety(url: str) -> None: """ Validate that URL is safe to fetch (SSRF protection). diff --git a/app/utils/network.py b/app/utils/network.py new file mode 100644 index 00000000..6e988d32 --- /dev/null +++ b/app/utils/network.py @@ -0,0 +1,33 @@ +import ipaddress +import logging +import socket + +logger = logging.getLogger(__name__) + +def is_private_ip(hostname: str) -> bool: + """ + Check if a hostname resolves to a private/internal IP address. + Protects against SSRF attacks by blocking access to internal networks. + """ + try: + # Try to parse as IP address directly + ip = ipaddress.ip_address(hostname) + return ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved + except ValueError: + # Not a direct IP, try to resolve hostname + try: + # Get all IP addresses for this hostname + addr_info = socket.getaddrinfo(hostname, None) + for info in addr_info: + ip_str = info[4][0] + ip = ipaddress.ip_address(ip_str) + # Block if ANY resolved IP is private/internal + if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved: + return True + return False + except (socket.gaierror, socket.error): + # Cannot resolve - allow for testing/development + # In production, DNS should work properly + # Log this for debugging + logger.warning(f"Could not resolve hostname: {hostname}") + return False # Changed from True to False to allow external domains in tests diff --git a/tests/test_coverage_polish.py b/tests/test_coverage_polish.py index 6be04d7b..69857004 100644 --- a/tests/test_coverage_polish.py +++ b/tests/test_coverage_polish.py @@ -523,7 +523,7 @@ class TestURLUploadAdditionalCoverage: """Cover DNS resolution failure branch (lines 67-72).""" import socket as _socket - from app.api.url_upload import is_private_ip + from app.utils.network import is_private_ip with patch("socket.getaddrinfo", side_effect=_socket.gaierror("nope")): result = is_private_ip("nonexistent.invalid.hostname.test") @@ -531,7 +531,7 @@ class TestURLUploadAdditionalCoverage: def test_is_private_ip_hostname_resolves_to_private(self): """Cover branch where hostname resolves to a private IP (line 64-65).""" - from app.api.url_upload import is_private_ip + from app.utils.network import is_private_ip with patch("socket.getaddrinfo") as mock_gai: # Simulate resolving to a private IP diff --git a/tests/test_url_upload.py b/tests/test_url_upload.py index 5bb3c96e..3204d8a9 100644 --- a/tests/test_url_upload.py +++ b/tests/test_url_upload.py @@ -50,14 +50,14 @@ class TestURLUploadValidation: def test_is_private_ip_localhost(self): """Test that localhost is detected as private""" - from app.api.url_upload import is_private_ip + from app.utils.network import is_private_ip assert is_private_ip("127.0.0.1") is True assert is_private_ip("localhost") is True def test_is_private_ip_private_ranges(self): """Test that private IP ranges are detected""" - from app.api.url_upload import is_private_ip + from app.utils.network import is_private_ip # Private IP ranges assert is_private_ip("10.0.0.1") is True @@ -67,7 +67,7 @@ class TestURLUploadValidation: def test_is_private_ip_public_allowed(self): """Test that public IPs are allowed""" - from app.api.url_upload import is_private_ip + from app.utils.network import is_private_ip # Public IPs should not be blocked assert is_private_ip("8.8.8.8") is False @@ -548,14 +548,14 @@ class TestURLUploadEndpoint: def test_is_private_ip_ipv6_loopback(self): """Test that IPv6 loopback is detected as private""" - from app.api.url_upload import is_private_ip + from app.utils.network import is_private_ip # IPv6 loopback (::1) assert is_private_ip("::1") is True def test_is_private_ip_link_local(self): """Test that link-local addresses are detected as private""" - from app.api.url_upload import is_private_ip + from app.utils.network import is_private_ip # Link-local address assert is_private_ip("169.254.1.1") is True @@ -609,7 +609,7 @@ class TestURLUploadCoverageGaps: @patch("socket.getaddrinfo") def test_is_private_ip_hostname_resolves_to_public_ip(self, mock_getaddrinfo): """Test that a hostname resolving to a public IP returns False (lines 65->61, 67)""" - from app.api.url_upload import is_private_ip + from app.utils.network import is_private_ip # Mock DNS resolution to return a single public IP (8.8.8.8 is Google DNS) mock_getaddrinfo.return_value = [ @@ -624,7 +624,7 @@ class TestURLUploadCoverageGaps: @patch("socket.getaddrinfo") def test_is_private_ip_hostname_resolves_multiple_ips_all_public(self, mock_getaddrinfo): """Test hostname with multiple public IPs returns False (covers 65->61 loop branch)""" - from app.api.url_upload import is_private_ip + from app.utils.network import is_private_ip # Return two public IPs - neither is private, so loop runs twice (65->61) then returns False (67) mock_getaddrinfo.return_value = [