🛡️ Sentinel: [HIGH] Fix Server-Side Request Forgery in IMAP connections
🚨 Severity: HIGH 💡 Vulnerability: User-provided IMAP `host` in `_test_imap_connection` and `pull_inbox` was not validated against private IPs, creating an SSRF risk. 🎯 Impact: Attackers could abuse the endpoints to port-scan or interact with internal/private network services. 🔧 Fix: Integrated `is_private_ip` from `app.utils.network` to block connections resolving to private, loopback, link-local, or reserved IPs. ✅ Verification: Ran `test_imap_tasks.py` and `test_api_imap_accounts.py` successfully. Checked `ruff` output and diffs. Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
@@ -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-19 - Prevent SSRF in IMAP Connection Testing
|
||||
**Vulnerability:** Server-Side Request Forgery (SSRF) allowed users to port-scan or connect to internal services via user-provided `host` and `port` inputs in IMAP endpoints (`/test` and `pull_inbox`).
|
||||
**Learning:** Endpoints testing outbound connections with user-provided configurations must validate the destination host before attempting the connection to prevent exploitation of the server's network position.
|
||||
**Prevention:** Use network utilities like `is_private_ip` that resolve hostnames and block private/loopback/reserved IPs before establishing outbound connections.
|
||||
|
||||
@@ -17,6 +17,7 @@ from sqlalchemy.orm import Session
|
||||
from app.database import get_db
|
||||
from app.models import UserImapAccount
|
||||
from app.utils.encryption import decrypt_value, encrypt_value
|
||||
from app.utils.network import is_private_ip
|
||||
from app.utils.subscription import get_tier, get_user_tier_id
|
||||
from app.utils.user_scope import get_current_owner_id
|
||||
|
||||
@@ -187,6 +188,11 @@ def _test_imap_connection(host: str, port: int, username: str, password: str, us
|
||||
|
||||
Returns a dict with ``{"success": bool, "message": str}``.
|
||||
"""
|
||||
|
||||
# Security: Prevent SSRF by blocking connections to internal IPs
|
||||
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)
|
||||
|
||||
@@ -18,6 +18,7 @@ from app.utils.allowed_types import (
|
||||
DEFAULT_CATEGORIES,
|
||||
get_allowed_types_for_categories,
|
||||
)
|
||||
from app.utils.network import is_private_ip
|
||||
|
||||
# Database session for per-user IMAP accounts (imported lazily to avoid circular imports)
|
||||
_db_session_factory = None
|
||||
@@ -405,6 +406,11 @@ def pull_inbox(
|
||||
)
|
||||
processed_emails = load_processed_emails()
|
||||
|
||||
# Security: Prevent SSRF by blocking connections to internal IPs
|
||||
if is_private_ip(host):
|
||||
logger.warning("SSRF blocked: Attempt to pull mailbox from private IP %s", host)
|
||||
return
|
||||
|
||||
try:
|
||||
mail = imaplib.IMAP4_SSL(host, port) if use_ssl else imaplib.IMAP4(host, port)
|
||||
mail.login(username, password)
|
||||
|
||||
+10
@@ -0,0 +1,10 @@
|
||||
import re
|
||||
|
||||
with open("app/api/imap_accounts.py", "r") as f:
|
||||
content = f.read()
|
||||
|
||||
# Remove duplicate imports
|
||||
content = re.sub(r"from app\.utils\.network import is_private_ip\nfrom app\.utils\.network import is_private_ip\n", "from app.utils.network import is_private_ip\n", content)
|
||||
|
||||
with open("app/api/imap_accounts.py", "w") as f:
|
||||
f.write(content)
|
||||
@@ -0,0 +1,29 @@
|
||||
import re
|
||||
|
||||
with open("app/api/imap_accounts.py", "r") as f:
|
||||
content = f.read()
|
||||
|
||||
# Add import
|
||||
import_stmt = "from app.utils.network import is_private_ip\n"
|
||||
content = re.sub(
|
||||
r"(from app\.utils\.user_scope import get_current_owner_id\n)",
|
||||
r"\1" + import_stmt,
|
||||
content
|
||||
)
|
||||
|
||||
# Add SSRF check
|
||||
ssrf_check = """
|
||||
# Security: Prevent SSRF by blocking connections to internal IPs
|
||||
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"}
|
||||
"""
|
||||
|
||||
content = re.sub(
|
||||
r"(def _test_imap_connection\(host: str, port: int, username: str, password: str, use_ssl: bool\) -> dict\[str, Any\]:\n \"\"\"Attempt to connect and log in to the IMAP server.\n\n Returns a dict with ``\{\"success\": bool, \"message\": str\}``\.\n \"\"\"\n)",
|
||||
r"\1" + ssrf_check,
|
||||
content
|
||||
)
|
||||
|
||||
with open("app/api/imap_accounts.py", "w") as f:
|
||||
f.write(content)
|
||||
@@ -0,0 +1,9 @@
|
||||
import re
|
||||
|
||||
with open("app/api/imap_accounts.py", "r") as f:
|
||||
content = f.read()
|
||||
|
||||
if "is_private_ip" in content:
|
||||
print("Patch successful!")
|
||||
else:
|
||||
print("Patch failed!")
|
||||
Reference in New Issue
Block a user