🛡️ Sentinel: [HIGH] Fix SSRF vulnerability in S3 connection test endpoint_url

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

.jules/sentinel.md

@@ -15,3 +15,7 @@
 **Vulnerability:** The `is_private_ip` function in `app/utils/network.py` failed open (returned `False`) when a hostname could not be resolved (`socket.gaierror`).
 **Learning:** This fail-open pattern was originally added to allow external domains in tests, but in production, it created a severe SSRF risk. An attacker could bypass SSRF protections by providing a URL that fails to resolve during the security check but resolves later (DNS rebinding), or by exploiting internal routing behaviors via unresolvable addresses.
 **Prevention:** Always fail securely in network authorization functions. If a domain cannot be resolved to verify its safety, the request must be blocked (`return True` / default-deny). Tests should mock DNS resolution correctly instead of compromising production security logic.
+## 2024-11-06 - SSRF in S3 Connection Test
+**Vulnerability:** The `_test_s3_connection` function in `app/api/integrations.py` accepted a user-provided `endpoint_url` without validation, passing it directly to `boto3.client`. An attacker could provide an internal URL (e.g., `http://169.254.169.254` or `http://localhost`) to execute Server-Side Request Forgery (SSRF) when `head_bucket` was called.
+**Learning:** Even SDK initializations (like AWS boto3) that accept URLs can be abused for SSRF. Input validation is required for any user-controlled URL before it is used to initialize an outbound network client.
+**Prevention:** Validate the scheme (must be HTTP/HTTPS) and use a centralized IP resolution function (e.g., `is_private_ip`) to ensure the hostname does not resolve to private or internal addresses before passing the `endpoint_url` to the SDK.

app/api/integrations.py

@@ -547,13 +547,28 @@ def _test_s3_connection(config: dict[str, Any] | None, credentials: dict[str, An
     if not bucket:
         return {"success": False, "message": "Missing required field: bucket"}
 
+    endpoint_url = cfg.get("endpoint_url")
+    if endpoint_url:
+        from urllib.parse import urlparse
+
+        parsed = urlparse(endpoint_url)
+        if parsed.scheme not in ("http", "https"):
+            return {"success": False, "message": "URL must use http or https scheme"}
+
+        hostname = parsed.hostname or ""
+        if hostname:
+            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:
         client = boto3.client(
             "s3",
             region_name=region,
             aws_access_key_id=creds.get("access_key_id", ""),
             aws_secret_access_key=creds.get("secret_access_key", ""),
-            endpoint_url=cfg.get("endpoint_url"),
+            endpoint_url=endpoint_url,
         )
         client.head_bucket(Bucket=bucket)
         return {"success": True, "message": f"S3 bucket '{bucket}' is accessible"}

tests/test_api_integrations.py

@@ -1011,6 +1011,45 @@ class TestConnectionTestEndpoint:
         assert data["success"] is False
         assert "bucket" in data["message"].lower()
 
+    def test_test_s3_blocks_private_ip(self, int_client):
+        """S3 test blocks requests to private/internal IPs via endpoint_url (SSRF protection)."""
+        payload = {
+            "integration_type": "S3",
+            "config": {"bucket": "my-bucket", "endpoint_url": "http://127.0.0.1/s3"},
+            "credentials": {"access_key_id": "AKIA", "secret_access_key": "secret"},
+        }
+        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_s3_blocks_localhost(self, int_client):
+        """S3 test blocks requests to localhost via endpoint_url."""
+        payload = {
+            "integration_type": "S3",
+            "config": {"bucket": "my-bucket", "endpoint_url": "http://localhost/s3"},
+            "credentials": {"access_key_id": "AKIA", "secret_access_key": "secret"},
+        }
+        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_s3_blocks_file_scheme(self, int_client):
+        """S3 test blocks file:// scheme via endpoint_url."""
+        payload = {
+            "integration_type": "S3",
+            "config": {"bucket": "my-bucket", "endpoint_url": "file:///etc/passwd"},
+            "credentials": {"access_key_id": "AKIA", "secret_access_key": "secret"},
+        }
+        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()
+
     def test_test_webdav_missing_url(self, int_client):
         """WebDAV test with missing URL returns failure."""
         payload = {