Merge pull request #788 from christianlouis/sentinel/fix-b310-urllib-httpx-11046306234862582289

🛡️ Sentinel: [MEDIUM] Fix B310 Vulnerability - Use httpx instead of urllib.request

.jules/sentinel.md

@@ -2,3 +2,7 @@
 **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.
+## 2026-03-22 - B310: urllib.request.urlopen replaced with httpx
+**Vulnerability:** The `_test_webdav_connection` function used `urllib.request.urlopen`, which natively supports dangerous schemes like `file://` or `ftp://` and follows redirects by default, potentially allowing SSRF bypasses or Local File Inclusion.
+**Learning:** `urllib.request` should be avoided for user-supplied URLs. Even when URL schemes are manually validated, `urllib`'s default redirect following behavior can bypass SSRF protections (e.g. redirecting to `127.0.0.1`).
+**Prevention:** Use a modern, safer HTTP client like `httpx` with `follow_redirects=False` when testing user-provided URLs.

app/api/integrations.py

@@ -608,7 +608,7 @@ def _test_dropbox_connection(config: dict[str, Any] | None, credentials: dict[st
 
 def _test_webdav_connection(config: dict[str, Any] | None, credentials: dict[str, Any] | None) -> dict[str, Any]:
     """Test a WebDAV/Nextcloud connection by issuing an HTTP PROPFIND."""
-    import urllib.request
+    import httpx
 
     cfg = config or {}
     creds = credentials or {}
@@ -635,17 +635,14 @@ def _test_webdav_connection(config: dict[str, Any] | None, credentials: dict[str
             return {"success": False, "message": "URLs pointing to internal or private networks are not allowed"}
 
     try:
-        import base64
+        auth = (username, password) if username and password else None
+        headers = {"Depth": "0"}
 
-        req = urllib.request.Request(url, method="PROPFIND")  # noqa: S310
+        # Use httpx for secure connection testing, avoiding urllib vulnerabilities
-        if username and password:
+        resp = httpx.request("PROPFIND", url, auth=auth, headers=headers, timeout=10.0, follow_redirects=False)
-            token = base64.b64encode(f"{username}:{password}".encode()).decode()
+        if resp.status_code < 400:
-            req.add_header("Authorization", f"Basic {token}")
+            return {"success": True, "message": "WebDAV connection successful"}
-        req.add_header("Depth", "0")
+        return {"success": False, "message": f"WebDAV returned HTTP {resp.status_code}"}
-        with urllib.request.urlopen(req, timeout=10) as resp:  # noqa: S310
-            if resp.status < 400:
-                return {"success": True, "message": "WebDAV connection successful"}
-            return {"success": False, "message": f"WebDAV returned HTTP {resp.status}"}
     except Exception as exc:  # noqa: BLE001
         logger.warning("WebDAV connection error for %s: %s", hostname, exc)
         return {"success": False, "message": "WebDAV connection failed — check URL and credentials"}

tests/test_api_integrations.py

@@ -1,5 +1,7 @@
 """Tests for the per-user integrations API (app/api/integrations.py)."""
 
+import unittest.mock
+
 import pytest
 from fastapi.testclient import TestClient
 from sqlalchemy import create_engine
@@ -1061,6 +1063,69 @@ class TestConnectionTestEndpoint:
         assert data["success"] is False
         assert "scheme" in data["message"].lower()
 
+    @unittest.mock.patch("httpx.request")
+    def test_test_webdav_success(self, mock_request, int_client):
+        """WebDAV test succeeds with valid credentials and a valid status code."""
+        mock_response = unittest.mock.MagicMock()
+        mock_response.status_code = 207  # Typical WebDAV success for PROPFIND
+        mock_request.return_value = mock_response
+
+        payload = {
+            "integration_type": "WEBDAV",
+            "config": {"url": "https://example.com/webdav"},
+            "credentials": {"username": "user1", "password": "password123"},
+        }
+        resp = int_client.post("/api/integrations/test", json=payload)
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["success"] is True
+
+        mock_request.assert_called_once_with(
+            "PROPFIND",
+            "https://example.com/webdav",
+            auth=("user1", "password123"),
+            headers={"Depth": "0"},
+            timeout=10.0,
+            follow_redirects=False,
+        )
+
+    @unittest.mock.patch("httpx.request")
+    def test_test_webdav_failure_status(self, mock_request, int_client):
+        """WebDAV test fails if the server returns a 4xx or 5xx status code."""
+        mock_response = unittest.mock.MagicMock()
+        mock_response.status_code = 401
+        mock_request.return_value = mock_response
+
+        payload = {
+            "integration_type": "WEBDAV",
+            "config": {"url": "https://example.com/webdav"},
+            "credentials": {"username": "user1", "password": "wrong"},
+        }
+        resp = int_client.post("/api/integrations/test", json=payload)
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["success"] is False
+        assert "401" in data["message"]
+
+    @unittest.mock.patch("httpx.request")
+    def test_test_webdav_exception(self, mock_request, int_client):
+        """WebDAV test fails gracefully if an exception occurs during the request."""
+        mock_request.side_effect = Exception("Connection error")
+
+        payload = {
+            "integration_type": "WEBDAV",
+            "config": {"url": "https://example.com/webdav"},
+            "credentials": {},
+        }
+        resp = int_client.post("/api/integrations/test", json=payload)
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["success"] is False
+        assert "failed" in data["message"].lower()
+
 
 # ---------------------------------------------------------------------------
 # Quota endpoint tests