🛡️ Sentinel: [HIGH] Fix SSRF in WebDAV connection test

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>

.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.

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:
+        from app.utils.network import is_private_ip
-            addr = ipaddress.ip_address(hostname)
+        if is_private_ip(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"}
-                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"}
 
     try:
         import base64

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).

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

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

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 = [