🛡️ Sentinel: [HIGH] Fix SSRF in integrations connection test
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>
This commit is contained in:
@@ -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.
|
||||
|
||||
+18
-1
@@ -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"}
|
||||
|
||||
Reference in New Issue
Block a user