🛡️ Sentinel: [HIGH] Fix DNS Rebinding / SSRF Vulnerability

.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.
+
+## 2025-02-23 - [DNS Rebinding / SSRF via Domain Resolution]
+**Vulnerability:** The `validate_folder_url` function checked for private IPs but only when the hostname was an IP literal. It failed to resolve domain names to check if they pointed to private IPs (DNS Rebinding risk).
+**Learning:** Validating hostnames solely by string inspection is insufficient. A public-looking domain can resolve to `127.0.0.1`. Defense-in-depth requires resolving the hostname and validating the IP address before connection.
+**Prevention:**
+1. Perform DNS resolution (`socket.getaddrinfo`) on hostnames.
+2. Check all resolved IPs against private/loopback ranges.
+3. Fail safe if DNS resolution fails.

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
 
@@ -210,7 +211,22 @@
                 return False
         except ValueError:
             # Not an IP literal, it's a domain.
-            pass
+            # Perform DNS resolution to check for private IPs (DNS Rebinding protection)
+            try:
+                # Resolve hostname to check if it points to a private IP
+                addr_infos = socket.getaddrinfo(hostname, None)
+                for family, kind, proto, canonname, sockaddr in addr_infos:
+                    ip_str = sockaddr[0]
+                    resolved_ip = ipaddress.ip_address(ip_str)
+                    if resolved_ip.is_private or resolved_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) as e:
+                log.warning(f"Skipping unsafe URL (DNS resolution failed): {sanitize_for_log(url)}")
+                return False
+            except Exception as e:
+                log.warning(f"Skipping unsafe URL (DNS check error): {e}")
+                return False
 
     except Exception as e:
         log.warning(f"Failed to validate URL {sanitize_for_log(url)}: {e}")
@@ -323,7 +339,7 @@
         }
     except (httpx.HTTPError, KeyError) as e:
         log.error(f"Failed to list existing folders: {sanitize_for_log(e)}")
         return {}
 
 def get_all_existing_rules(client: httpx.Client, profile_id: str) -> Set[str]:
     all_rules = set()
@@ -663,7 +679,7 @@
             with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
                 future_to_folder = {
                     executor.submit(
                         _process_single_folder,
                         folder_data,
                         profile_id,
                         existing_rules,

tests/test_security.py

@@ -0,0 +1,35 @@
+import unittest
+from unittest.mock import patch
+import socket
+import logging
+import sys
+import os
+
+# Add parent directory to path so we can import main
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+
+# Configure logging to avoid polluting output
+logging.basicConfig(level=logging.CRITICAL)
+
+import main
+
+class TestSecurity(unittest.TestCase):
+    def test_validate_folder_url_prevents_dns_rebinding(self):
+        """
+        Verify that the implementation prevents domains that resolve to private IPs.
+        """
+        suspicious_url = "https://internal.example.com/list.json"
+
+        # Mock socket.getaddrinfo to return 127.0.0.1
+        with patch("socket.getaddrinfo") as mock_dns:
+            mock_dns.return_value = [
+                (socket.AF_INET, socket.SOCK_STREAM, 6, '', ('127.0.0.1', 443))
+            ]
+
+            result = main.validate_folder_url(suspicious_url)
+
+            # Should be False (Secure)
+            self.assertFalse(result, "validate_folder_url should return False for domains resolving to private IPs")
+
+if __name__ == "__main__":
+    unittest.main()