🛡️ Sentinel: [HIGH] Fix SSRF bypass via HTTP redirects in URL upload

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

.jules/sentinel.md

@@ -19,3 +19,7 @@
 **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.
+## 2026-03-27 - SSRF Bypass via HTTP Redirects
+**Vulnerability:** The `process_url` endpoint in `app/api/url_upload.py` used `httpx.AsyncClient` with `follow_redirects=True`. While the initial user-provided URL was validated against SSRF, if the remote server returned an HTTP redirect to an internal IP (like 127.0.0.1 or an AWS metadata endpoint), the HTTP client would automatically follow the redirect without validating the new target URL.
+**Learning:** Initial URL validation is insufficient if the HTTP client automatically follows redirects. Attackers can easily set up external servers that respond with `302 Found` pointing to internal network addresses.
+**Prevention:** If `follow_redirects=True` is required, always implement an event hook (e.g., `event_hooks={"response": [check_redirect]}`) to intercept redirect responses, extract the `Location` header, and validate the target URL using `is_private_ip` or `validate_url_safety` before the client follows it.

app/api/url_upload.py

@@ -137,6 +137,15 @@ async def process_url(
     # Validate URL safety (SSRF protection)
     validate_url_safety(url)
 
+    # Event hook to intercept and validate redirects
+    async def check_redirect(response: httpx.Response):
+        if response.is_redirect:
+            location = response.headers.get("Location")
+            if location:
+                redirect_url = str(response.url.join(location))
+                # Validate the redirect target
+                validate_url_safety(redirect_url)
+
     # Parse URL to extract filename if not provided
     if url_request.filename:
         original_filename = url_request.filename
@@ -165,6 +174,7 @@ async def process_url(
             headers={
                 "User-Agent": "DocuElevate/1.0",  # Identify ourselves
             },
+            event_hooks={"response": [check_redirect]},
         ) as client:
             async with client.stream("GET", url) as response:
                 response.raise_for_status()

reproduce.py

@@ -1,34 +0,0 @@
-import sys
-from unittest.mock import MagicMock
-from fastapi.templating import Jinja2Templates
-
-import os
-# We don't really need a real path, but let's mock it
-os.makedirs("templates", exist_ok=True)
-with open("templates/files.html", "w") as f:
-    f.write("Hello")
-
-templates = Jinja2Templates(directory="templates")
-original_template_response = templates.TemplateResponse
-
-def template_response_with_version(*args, **kwargs):
-    if len(args) == 2 and isinstance(args[0], str) and isinstance(args[1], dict):
-        context = args[1]
-        request = context.get("request")
-        if request is not None:
-            # THIS IS MY FIX
-            print("Running fix logic")
-            return original_template_response(request=request, name=args[0], context=context, **kwargs)
-
-    print("Running original fallback logic")
-    return original_template_response(*args, **kwargs)
-
-templates.TemplateResponse = template_response_with_version
-
-req = MagicMock()
-try:
-    templates.TemplateResponse("files.html", {"request": req})
-    print("SUCCESS")
-except Exception as e:
-    import traceback
-    traceback.print_exc()