🛡️ Sentinel: [HIGH] Fix SSRF protection by resolving domains

.jules/sentinel.md

@@ -39,3 +39,11 @@
 **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.
+
+## 2026-01-17 - [SSRF Protection Enhancement]
+**Vulnerability:** The `validate_folder_url` function only checked for IP literals, allowing domains resolving to private IPs (e.g., DNS rebinding or internal domains) to bypass SSRF protection.
+**Learning:** Checking `ipaddress.ip_address(hostname)` is insufficient for validation if `hostname` is a domain. DNS resolution is required to validate the actual destination.
+**Prevention:**
+1. Resolve domains using `socket.getaddrinfo` to obtain the underlying IP addresses.
+2. Check all returned IPs against private and loopback ranges.
+3. Fail closed (block the URL) if resolution fails or returns any private IP.

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,22 @@
                 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 and check IPs.
+            try:
+                # Resolve hostname to IPs (IPv4 and IPv6)
+                # We filter for AF_INET/AF_INET6 to ensure we get IP addresses
+                addr_info = socket.getaddrinfo(hostname, None, proto=socket.IPPROTO_TCP)
+                for res in addr_info:
+                    # res is (family, type, proto, canonname, sockaddr)
+                    # sockaddr is (address, port) for AF_INET/AF_INET6
+                    ip_str = res[4][0]
+                    ip = ipaddress.ip_address(ip_str)
+                    if ip.is_private or ip.is_loopback:
+                        log.warning(f"Skipping unsafe URL (domain {hostname} resolves to private IP {ip}): {sanitize_for_log(url)}")
+                        return False
+            except (socket.gaierror, ValueError, OSError) as e:
+                log.warning(f"Failed to resolve/validate domain {hostname}: {e}")
+                return False
 
     except Exception as e:
         log.warning(f"Failed to validate URL {sanitize_for_log(url)}: {e}")

tests/test_ssrf.py

@@ -0,0 +1,54 @@
+import unittest
+from unittest.mock import patch, MagicMock
+import sys
+import os
+import socket
+
+# Add root to path to import main
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+import main
+
+class TestSSRF(unittest.TestCase):
+    def test_domain_resolving_to_private_ip(self):
+        """
+        Test that a domain resolving to a private IP is blocked.
+        This simulates a DNS Rebinding attack or SSRF attempt against internal infrastructure.
+        """
+        # We need to mock socket.getaddrinfo because the fix will use it.
+        # For the current code, this mock is unused, but the test ensures
+        # that 'internal.example.com' (which is not an IP literal) passes validation currently
+        # and will fail validation (be blocked) after the fix.
+
+        with patch('socket.getaddrinfo') as mock_getaddrinfo:
+            # Simulate resolving to 192.168.1.1
+            mock_getaddrinfo.return_value = [
+                (socket.AF_INET, socket.SOCK_STREAM, 6, '', ('192.168.1.1', 443))
+            ]
+
+            url = "https://internal.example.com/config.json"
+
+            # This calls the function in main.py
+            result = main.validate_folder_url(url)
+
+            # We expect this to be False (Blocked)
+            self.assertFalse(result, "Should block domain resolving to private IP")
+
+    def test_domain_resolving_to_public_ip(self):
+        """
+        Test that a domain resolving to a public IP is allowed.
+        """
+        with patch('socket.getaddrinfo') as mock_getaddrinfo:
+            # Simulate resolving to 8.8.8.8
+            mock_getaddrinfo.return_value = [
+                (socket.AF_INET, socket.SOCK_STREAM, 6, '', ('8.8.8.8', 443))
+            ]
+
+            url = "https://public.example.com/config.json"
+
+            result = main.validate_folder_url(url)
+
+            self.assertTrue(result, "Should allow domain resolving to public IP")
+
+if __name__ == '__main__':
+    unittest.main()