🔒️(ssrf) factorize SSRF code, allow redirects in image proxy (#631)

Also use the SSRF code in IMAP imports.

Author: sylvinus

src/backend/core/api/viewsets/blob.py

@@ -4,6 +4,7 @@
 
 from django.http import HttpResponse
 from django.utils.decorators import method_decorator
+from django.utils.http import content_disposition_header
 from django.views.decorators.csrf import csrf_exempt
 
 from drf_spectacular.utils import OpenApiParameter, OpenApiResponse, extend_schema
@@ -177,8 +178,8 @@ def download(self, request, pk=None):
                 )
 
                 # Add appropriate headers for download
-                response["Content-Disposition"] = (
-                    f'attachment; filename="{attachment["name"]}"'
+                response["Content-Disposition"] = content_disposition_header(
+                    True, attachment["name"]
                 )
                 response["Content-Length"] = attachment["size"]
                 # Enable browser caching for 30 days (inline images benefit from this)
@@ -207,7 +208,9 @@ def download(self, request, pk=None):
                 )
 
                 # Add appropriate headers for download
-                response["Content-Disposition"] = f'attachment; filename="{filename}"'
+                response["Content-Disposition"] = content_disposition_header(
+                    True, filename
+                )
                 response["Content-Length"] = blob.size
                 # Enable browser caching for 30 days (inline images benefit from this)
                 response["Cache-Control"] = "private, max-age=2592000"

src/backend/core/api/viewsets/image_proxy.py

@@ -1,271 +1,25 @@
 """API ViewSet for proxying external images."""
 
-import ipaddress
 import logging
-import socket
-from urllib.parse import ParseResult, unquote, urlparse, urlunparse
+from urllib.parse import unquote
 
 from django.conf import settings
 from django.http import HttpResponse
 
 import magic
 import requests
 from drf_spectacular.utils import OpenApiParameter, OpenApiResponse, extend_schema
-from requests.adapters import HTTPAdapter
 from rest_framework import status as http_status
 from rest_framework.response import Response
 from rest_framework.viewsets import ViewSet
 
 from core import enums, models
 from core.api import permissions
+from core.services.ssrf import SSRFSafeSession, SSRFValidationError
 
 logger = logging.getLogger(__name__)
 
 
-class SSRFValidationError(Exception):
-    """Exception raised when URL validation fails due to SSRF protection."""
-
-
-class SSRFProtectedAdapter(HTTPAdapter):
-    """
-    HTTPAdapter that connects to a pre-validated IP address while maintaining
-    proper TLS certificate verification against the original hostname.
-
-    This prevents TOCTOU DNS rebinding attacks by:
-    1. Connecting to the IP address that was validated (not re-resolving DNS)
-    2. Verifying TLS certificates against the original hostname (for HTTPS)
-    3. Setting the Host header correctly for virtual hosting
-    """
-
-    def __init__(
-        self,
-        dest_ip: str,
-        dest_port: int,
-        original_hostname: str,
-        original_scheme: str,
-        **kwargs,
-    ):
-        self.dest_ip = dest_ip
-        self.dest_port = dest_port
-        self.original_hostname = original_hostname
-        self.original_scheme = original_scheme
-        super().__init__(**kwargs)
-
-    def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs):
-        """Initialize pool manager with TLS hostname verification settings."""
-        if self.original_scheme == "https":
-            # Ensure TLS certificate is verified against the original hostname
-            # even though we're connecting to an IP address
-            pool_kwargs["assert_hostname"] = self.original_hostname
-            pool_kwargs["server_hostname"] = self.original_hostname
-        super().init_poolmanager(connections, maxsize, block, **pool_kwargs)
-
-    def send(
-        self, request, stream=False, timeout=None, verify=True, cert=None, proxies=None
-    ):
-        """Send request, rewriting URL to connect to the validated IP address."""
-        parsed = urlparse(request.url)
-
-        # Build URL with validated IP instead of hostname
-        # IPv6 addresses need brackets in URLs
-        if ":" in self.dest_ip:
-            ip_netloc = f"[{self.dest_ip}]:{self.dest_port}"
-        else:
-            ip_netloc = f"{self.dest_ip}:{self.dest_port}"
-
-        # Reconstruct URL with IP address
-        request.url = urlunparse(
-            (
-                parsed.scheme,
-                ip_netloc,
-                parsed.path,
-                parsed.params,
-                parsed.query,
-                parsed.fragment,
-            )
-        )
-
-        # Set Host header to original hostname for virtual hosting
-        # Include port only if non-standard
-        if parsed.port and parsed.port not in (80, 443):
-            request.headers["Host"] = f"{self.original_hostname}:{parsed.port}"
-        else:
-            request.headers["Host"] = self.original_hostname
-
-        return super().send(
-            request,
-            stream=stream,
-            timeout=timeout,
-            verify=verify,
-            cert=cert,
-            proxies=proxies,
-        )
-
-
-class SSRFSafeSession:
-    """
-    HTTP Session with built-in SSRF protection.
-
-    This class provides a safe way to make HTTP requests by:
-    1. Validating URL scheme (only http/https allowed)
-    2. Blocking direct IP addresses (legitimate services use domain names)
-    3. Resolving hostnames and blocking private/internal IPs
-    4. Pinning resolved IPs to prevent DNS rebinding attacks (TOCTOU)
-
-    Usage:
-        try:
-            response = SSRFSafeSession().get("https://example.com/image.png", timeout=10)
-        except SSRFValidationError:
-            # URL was blocked for security reasons
-            pass
-    """
-
-    def _validate_url(self, parsed_url: ParseResult) -> list[str]:
-        """
-        Validate that a URL is safe to fetch (SSRF protection).
-
-        This function prevents Server-Side Request Forgery (SSRF) attacks by
-        validating URLs before making HTTP requests. It implements a defense-in-depth
-        approach:
-
-        1. Only allows http/https schemes
-        2. Blocks all IP addresses (legitimate emails use domain names)
-        3. Resolves hostnames and blocks if they resolve to private/internal IPs
-        (prevents DNS rebinding attacks where attacker-controlled DNS returns
-        127.0.0.1 or internal IPs)
-
-        Blocked addresses include:
-        - Any direct IP address (e.g., http://192.168.1.1/)
-        - Private IP ranges (RFC1918: 10.x.x.x, 172.16-31.x.x, 192.168.x.x)
-        - Loopback addresses (127.x.x.x, ::1)
-        - Link-local addresses (169.254.x.x, fe80::/10)
-        - Multicast and reserved addresses
-        - Cloud provider metadata endpoints (169.254.169.254, fd00:ec2::254)
-
-        Args:
-            parsed_url: The parsed URL to validate
-
-        Returns:
-            List of validated IP addresses that the hostname resolves to
-
-        Raises:
-            SSRFValidationError: If the URL is unsafe
-        """
-        # Only allow http and https schemes
-        if parsed_url.scheme not in {"http", "https"}:
-            raise SSRFValidationError("Invalid URL scheme (only http/https allowed)")
-
-        # Require a hostname
-        if not parsed_url.hostname:
-            raise SSRFValidationError("Invalid URL (missing hostname)")
-
-        # Block all IP addresses (legitimate services use domain names)
-        try:
-            ipaddress.ip_address(parsed_url.hostname)
-            raise SSRFValidationError(
-                "IP addresses are not allowed (domain name required)"
-            )
-        except ValueError:
-            # Not an IP address, continue validation
-            pass
-
-        # Resolve hostname to IP addresses
-        try:
-            addr_info = socket.getaddrinfo(
-                parsed_url.hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM
-            )
-        except socket.gaierror as exc:
-            raise SSRFValidationError("Unable to resolve hostname") from exc
-
-        # Check all resolved IP addresses
-        valid_ips = []
-        for _, _, _, _, sockaddr in addr_info:
-            ip_str = sockaddr[0]
-            try:
-                ip_addr = ipaddress.ip_address(ip_str)
-
-                if ip_addr.is_private:
-                    raise SSRFValidationError("Domain resolves to private IP address")
-
-                if ip_addr.is_loopback:
-                    raise SSRFValidationError("Domain resolves to loopback address")
-
-                if ip_addr.is_link_local:
-                    raise SSRFValidationError("Domain resolves to link-local address")
-
-                if ip_addr.is_multicast:
-                    raise SSRFValidationError("Domain resolves to multicast address")
-
-                if ip_addr.is_reserved:
-                    raise SSRFValidationError("Domain resolves to reserved address")
-
-                # Block known cloud metadata IPs
-                if ip_str in ("169.254.169.254", "fd00:ec2::254"):
-                    raise SSRFValidationError(
-                        "Domain resolves to cloud metadata endpoint"
-                    )
-
-                valid_ips.append(ip_str)
-
-            except ValueError as exc:
-                raise SSRFValidationError("Invalid IP address in DNS response") from exc
-
-        if not valid_ips:
-            raise SSRFValidationError("No valid IP addresses found")
-
-        return valid_ips
-
-    def get(self, url: str, timeout: int, **kwargs) -> requests.Response:
-        """
-        Perform a safe HTTP GET request with SSRF protection and IP pinning.
-
-        This method:
-        1. Parses and validates the URL
-        2. Resolves DNS and validates all returned IPs
-        3. Creates a requests Session with a custom HTTPAdapter that:
-           - Connects directly to the validated IP (preventing DNS rebinding)
-           - Maintains proper TLS certificate verification against the hostname
-           - Sets the Host header correctly for virtual hosting
-
-        Args:
-            url: The URL to fetch
-            timeout: Request timeout in seconds
-            **kwargs: Additional arguments passed to requests.Session.get()
-
-        Returns:
-            requests.Response object
-
-        Raises:
-            SSRFValidationError: If the URL fails security validation
-            requests.RequestException: If the HTTP request fails
-        """
-        parsed_url = urlparse(url)
-        valid_ips = self._validate_url(parsed_url)
-
-        # Determine the port (explicit or default based on scheme)
-        if parsed_url.port:
-            port = parsed_url.port
-        elif parsed_url.scheme == "http":
-            port = 80
-        else:
-            port = 443
-
-        # Create a session with our SSRF-protected adapter that pins to the validated IP
-        session = requests.Session()
-        adapter = SSRFProtectedAdapter(
-            dest_ip=valid_ips[0],
-            dest_port=port,
-            original_hostname=parsed_url.hostname,
-            original_scheme=parsed_url.scheme,
-        )
-
-        # Mount the adapter for both http and https schemes
-        session.mount("http://", adapter)
-        session.mount("https://", adapter)
-
-        return session.get(url, timeout=timeout, **kwargs)
-
-
 class ImageProxySuspiciousResponse(HttpResponse):
     """
     Response for suspicious content that has been blocked by our image proxy.
@@ -356,7 +110,6 @@ def list(self, request, mailbox_id=None):
                 timeout=10,
                 stream=True,
                 headers={"User-Agent": "Messages-ImageProxy/1.0"},
-                allow_redirects=False,
             )
             response.raise_for_status()
 

src/backend/core/api/viewsets/inbound/widget.py

@@ -35,9 +35,9 @@ def authenticate(self, request):
         if not channel_id:
             raise AuthenticationFailed("Missing channel_id")
 
-        # API key authentication for check endpoint
+        # Only allow widget-type channels
         try:
-            channel = models.Channel.objects.get(id=channel_id)
+            channel = models.Channel.objects.get(id=channel_id, type="widget")
         except models.Channel.DoesNotExist as e:
             raise AuthenticationFailed("Invalid channel_id") from e
 

src/backend/core/services/importer/imap.py

@@ -22,6 +22,7 @@
 
 from core.mda.inbound import deliver_inbound_message
 from core.mda.rfc5322 import parse_email_message
+from core.services.ssrf import SSRFValidationError, validate_hostname
 
 logger = get_task_logger(__name__)
 
@@ -62,6 +63,21 @@ def decode_match(match):
     return re.sub(r"&([^-]*)-", decode_match, s)
 
 
+def _validate_imap_host(server: str) -> None:
+    """Validate that the IMAP server hostname is not a private/internal address.
+
+    Wraps the shared SSRF validator but allows public IP literals, which are
+    legitimate addresses for customer-supplied IMAP servers.
+
+    Raises:
+        ValueError: If the hostname resolves to a blocked IP address.
+    """
+    try:
+        validate_hostname(server, allow_ip_literal=True)
+    except SSRFValidationError as exc:
+        raise ValueError(f"IMAP server {server} is not allowed: {exc}") from exc
+
+
 class IMAPConnectionManager:
     """Context manager for IMAP connections with proper cleanup."""
 
@@ -76,6 +92,9 @@ def __init__(
         self.connection = None
 
     def __enter__(self):
+        # Validate the server hostname to prevent SSRF
+        _validate_imap_host(self.server)
+
         # Port 143 typically uses STARTTLS, port 993 uses SSL direct
         # If use_ssl=True and port is 143, use STARTTLS instead of SSL direct
         use_starttls = self.use_ssl and self.port == 143