fix(integrations): sanitize error messages and add SSRF protection for WebDAV test

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

app/api/integrations.py

@@ -509,9 +509,11 @@ def _test_imap_connection(config: dict[str, Any] | None, credentials: dict[str,
         mail.logout()
         return {"success": True, "message": "IMAP connection successful"}
     except OSError as exc:
-        return {"success": False, "message": f"Connection error: {exc}"}
+        logger.warning("IMAP network error for %s@%s: %s", username, host, exc)
+        return {"success": False, "message": "IMAP connection failed — check host, port, and network connectivity"}
     except Exception as exc:  # noqa: BLE001
-        return {"success": False, "message": f"IMAP error: {exc}"}
+        logger.warning("IMAP error for %s@%s: %s", username, host, exc)
+        return {"success": False, "message": "IMAP authentication or connection failed"}
 
 
 def _test_s3_connection(config: dict[str, Any] | None, credentials: dict[str, Any] | None) -> dict[str, Any]:
@@ -541,9 +543,11 @@ def _test_s3_connection(config: dict[str, Any] | None, credentials: dict[str, An
         client.head_bucket(Bucket=bucket)
         return {"success": True, "message": f"S3 bucket '{bucket}' is accessible"}
     except (BotoCoreError, ClientError) as exc:
-        return {"success": False, "message": f"S3 error: {exc}"}
+        logger.warning("S3 connection error for bucket '%s': %s", bucket, exc)
+        return {"success": False, "message": "S3 connection failed — check bucket name, region, and credentials"}
     except Exception as exc:  # noqa: BLE001
-        return {"success": False, "message": f"Unexpected error: {exc}"}
+        logger.warning("S3 unexpected error for bucket '%s': %s", bucket, exc)
+        return {"success": False, "message": "S3 connection failed"}
 
 
 def _test_webdav_connection(config: dict[str, Any] | None, credentials: dict[str, Any] | None) -> dict[str, Any]:
@@ -560,12 +564,25 @@ def _test_webdav_connection(config: dict[str, Any] | None, credentials: dict[str
         return {"success": False, "message": "Missing required field: url"}
 
     # Only allow http/https to prevent file:// or other custom scheme attacks
+    import ipaddress
     from urllib.parse import urlparse
 
     parsed = urlparse(url)
     if parsed.scheme not in ("http", "https"):
         return {"success": False, "message": "URL must use http or https scheme"}
 
+    # 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"}
+
     try:
         import base64
 
@@ -579,7 +596,8 @@ def _test_webdav_connection(config: dict[str, Any] | None, credentials: dict[str
                 return {"success": True, "message": "WebDAV connection successful"}
             return {"success": False, "message": f"WebDAV returned HTTP {resp.status}"}
     except Exception as exc:  # noqa: BLE001
-        return {"success": False, "message": f"WebDAV error: {exc}"}
+        logger.warning("WebDAV connection error for %s: %s", hostname, exc)
+        return {"success": False, "message": "WebDAV connection failed — check URL and credentials"}
 
 
 _CONNECTION_TESTERS: dict[str, Any] = {

tests/test_api_integrations.py

@@ -945,6 +945,45 @@ class TestConnectionTestEndpoint:
         assert data["success"] is False
         assert "url" in data["message"].lower()
 
+    def test_test_webdav_blocks_private_ip(self, int_client):
+        """WebDAV test blocks requests to private/internal IPs (SSRF protection)."""
+        payload = {
+            "integration_type": "WEBDAV",
+            "config": {"url": "http://127.0.0.1/webdav"},
+            "credentials": {"username": "u", "password": "p"},
+        }
+        resp = int_client.post("/api/integrations/test", json=payload)
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["success"] is False
+        assert "internal" in data["message"].lower() or "private" in data["message"].lower()
+
+    def test_test_webdav_blocks_localhost(self, int_client):
+        """WebDAV test blocks requests to localhost."""
+        payload = {
+            "integration_type": "WEBDAV",
+            "config": {"url": "http://localhost/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 "localhost" in data["message"].lower()
+
+    def test_test_webdav_blocks_file_scheme(self, int_client):
+        """WebDAV test blocks file:// scheme."""
+        payload = {
+            "integration_type": "WEBDAV",
+            "config": {"url": "file:///etc/passwd"},
+            "credentials": {},
+        }
+        resp = int_client.post("/api/integrations/test", json=payload)
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["success"] is False
+        assert "scheme" in data["message"].lower()
+
 
 # ---------------------------------------------------------------------------
 # Quota endpoint tests