From a57766ed7e2f8b8c2563d7a9724c793638591b07 Mon Sep 17 00:00:00 2001
From: "google-labs-jules[bot]"
 <161369871+google-labs-jules[bot]@users.noreply.github.com>
Date: Thu, 26 Mar 2026 04:00:55 +0000
Subject: [PATCH] =?UTF-8?q?=F0=9F=9B=A1=EF=B8=8F=20Sentinel:=20[HIGH]=20Fi?=
 =?UTF-8?q?x=20SSRF=20in=20integrations=20connection=20test?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Adds validation using `is_private_ip()` for user-provided hosts in `_test_imap_connection` and `_test_s3_connection` to prevent Server-Side Request Forgery vulnerabilities.

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
---
 .jules/sentinel.md      |  4 ++++
 app/api/integrations.py | 19 ++++++++++++++++++-
 2 files changed, 22 insertions(+), 1 deletion(-)

diff --git a/.jules/sentinel.md b/.jules/sentinel.md
index f3b5481a..6a6144c5 100644
--- a/.jules/sentinel.md
+++ b/.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.
+## 2026-03-26 - SSRF in Integration Connection Tests
+**Vulnerability:** The `_test_imap_connection` and `_test_s3_connection` functions in `app/api/integrations.py` did not validate user-provided `host` and `endpoint_url` variables against `is_private_ip()`. This allowed an attacker to test the presence of internal IMAP servers or direct S3 SDK API calls to internal infrastructure via SSRF.
+**Learning:** Any time a new generic connection or integration test is added, SSRF validation may be forgotten if the core network utility (`is_private_ip`) is not systematically applied to all outbound network operations, regardless of the protocol (e.g., IMAP, S3).
+**Prevention:** Establish a pattern where any user-configurable host or endpoint URL is immediately passed through the centralized `is_private_ip` validation function before any network call or third-party client initialization.
diff --git a/app/api/integrations.py b/app/api/integrations.py
index f5b11a16..99e83bc2 100644
--- a/app/api/integrations.py
+++ b/app/api/integrations.py
@@ -515,6 +515,12 @@ def _test_imap_connection(config: dict[str, Any] | None, credentials: dict[str,
     if not host or not username or not password:
         return {"success": False, "message": "Missing required fields: host, username, and password"}
 
+    from app.utils.network import is_private_ip
+
+    if is_private_ip(host):
+        logger.warning("SSRF blocked: Attempt to connect to private IP %s", host)
+        return {"success": False, "message": "Connection error: Invalid hostname or IP address"}
+
     try:
         if use_ssl:
             mail = imaplib.IMAP4_SSL(host, port)
@@ -543,17 +549,28 @@ def _test_s3_connection(config: dict[str, Any] | None, credentials: dict[str, An
     creds = credentials or {}
     bucket = cfg.get("bucket", "")
     region = cfg.get("region", "us-east-1")
+    endpoint_url = cfg.get("endpoint_url")
 
     if not bucket:
         return {"success": False, "message": "Missing required field: bucket"}
 
+    if endpoint_url:
+        from urllib.parse import urlparse
+
+        from app.utils.network import is_private_ip
+
+        parsed_url = urlparse(endpoint_url)
+        if parsed_url.hostname and is_private_ip(parsed_url.hostname):
+            logger.warning("SSRF blocked: Attempt to connect to private IP via S3 endpoint %s", endpoint_url)
+            return {"success": False, "message": "Connection error: Invalid endpoint URL or private IP"}
+
     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"}