🛡️ Sentinel: [HIGH] Fix SSRF bypass via DNS resolution

.jules/sentinel.md

@@ -39,3 +39,10 @@
 **Prevention:**
 1. Parse URLs and check hostnames against `localhost` and private IP ranges using `ipaddress` module.
 2. Enforce strict length limits on user inputs (e.g., profile IDs) to prevent resource exhaustion or buffer abuse.
+
+## 2025-01-27 - [SSRF Protection via DNS Resolution]
+**Vulnerability:** The application's SSRF protection checked `hostname` strings for private IP literals but did not resolve domains. This allowed an attacker to bypass the check using a public domain that resolves to a private IP (e.g., `localtest.me` -> `127.0.0.1`).
+**Learning:** Checking hostnames only as strings is insufficient for SSRF protection because of DNS rebinding and public domains pointing to private IPs.
+**Prevention:**
+1. Resolve domains using `socket.getaddrinfo` and check all returned IPs against private ranges.
+2. Implement this check before making the connection.

main.py

@@ -23,6 +23,7 @@
 import concurrent.futures
 import threading
 import ipaddress
+import socket
 from urllib.parse import urlparse
 from typing import Dict, List, Optional, Any, Set, Sequence
 
@@ -209,8 +210,24 @@
                 log.warning(f"Skipping unsafe URL (private IP): {sanitize_for_log(url)}")
                 return False
         except ValueError:
-            # Not an IP literal, it's a domain.
-            pass
+            # Not an IP literal, it's a domain. Resolve it to check for private IPs.
+            try:
+                # Resolve hostname to check if it points to private IPs
+                # socket.getaddrinfo returns a list of tuples. sockaddr is the 5th element.
+                # For AF_INET/AF_INET6, sockaddr[0] is the IP address string.
+                addr_infos = socket.getaddrinfo(hostname, None)
+                for family, type_, proto, canonname, sockaddr in addr_infos:
+                    ip_str = sockaddr[0]
+                    ip = ipaddress.ip_address(ip_str)
+                    if ip.is_private or ip.is_loopback:
+                        log.warning(f"Skipping unsafe URL (domain resolves to private IP {ip_str}): {sanitize_for_log(url)}")
+                        return False
+            except (socket.gaierror, ValueError):
+                # If DNS resolution fails, or IP parsing fails, we pass.
+                # If it's a real DNS error, httpx will fail later.
+                pass
+            except Exception as e:
+                log.warning(f"DNS resolution check failed for {sanitize_for_log(url)}: {e}")
 
     except Exception as e:
         log.warning(f"Failed to validate URL {sanitize_for_log(url)}: {e}")
@@ -373,7 +390,7 @@
         log.error(f"Failed to get existing rules: {sanitize_for_log(e)}")
         return set()
 
 def fetch_folder_data(url: str) -> Dict[str, Any]:
     js = _gh_get(url)
     if not validate_folder_data(js, url):
         raise KeyError(f"Invalid folder data from {sanitize_for_log(url)}")
@@ -668,7 +685,7 @@
                         profile_id,
                         existing_rules,
                         existing_rules_lock,
                         client  # Pass the persistent client
                     ): folder_data
                     for folder_data in folder_data_list
                 }