Merge pull request steam-bell-92#1660 from Kirtan-pc/fix/URLSanitizer

Fix: URLSanitizer allows SSRF via hostnames without DNS/IP revalidation

Author: steam-bell-92

security/url_sanitizer.py

@@ -5,6 +5,7 @@
 """
 
 import re
+import socket
 from urllib.parse import urlparse, urlunparse
 from typing import Optional, List, Set
 from ipaddress import ip_address, ip_network
@@ -110,17 +111,49 @@ def validate_url(self, url: str) -> str:
         return sanitized
 
     def _validate_ip_address(self, host: str) -> None:
-        """Validate IP address against blocked networks."""
+        """Validate IP address against blocked networks, resolving hostnames via DNS."""
         try:
             ip = ip_address(host)
-        except ValueError:
+            self._check_blocked_ip(ip, host)
             return
-        
+        except ValueError:
+            pass
+
+        try:
+            addrinfo = socket.getaddrinfo(host, None)
+        except socket.gaierror:
+            raise InvalidURLError(f"Failed to resolve hostname: {host}")
+
+        seen = set()
+        for _, _, _, _, sockaddr in addrinfo:
+            ip = ip_address(sockaddr[0])
+            if ip in seen:
+                continue
+            seen.add(ip)
+            self._check_blocked_ip(ip, host)
+
+    def _check_blocked_ip(self, ip, host):
+        """Check a single IP against blocked networks."""
         for network in self._blocked_networks:
             if ip in network:
                 if not (self.allow_localhost and ip.is_loopback):
                     if not (self.allow_private and ip.is_private):
                         raise InvalidURLError(f"Blocked IP address: {host}")
+
+    def _validate_port(self, port: Optional[int]) -> None:
+        """Validate port number against allowed/blocked lists."""
+        if port is None:
+            return
+        if port < 1 or port > 65535:
+            raise InvalidURLError(f"Invalid port number: {port}")
+        if self.allowed_ports is not None:
+            if port not in self.allowed_ports:
+                raise InvalidURLError(
+                    f"Port {port} is not allowed. "
+                    f"Allowed ports: {', '.join(str(p) for p in self.allowed_ports)}"
+                )
+        elif port in self.BLOCKED_PORTS:
+            raise InvalidURLError(f"Blocked port: {port}")
 
     def _validate_port(self, port: Optional[int]) -> None:
         """Validate port number against allowed/blocked lists."""

tests/test_security.py

@@ -78,15 +78,36 @@ def test_allow_private_ips(self):
         result = sanitizer.validate_url(url)
         assert result.startswith("http://10.0.0.1")
 
+    def test_hostname_resolves_to_private_ip(self):
+        """Test that hostnames resolving to private IPs are blocked."""
+        sanitizer = URLSanitizer(allow_localhost=False, allow_private=False)
+        with pytest.raises(InvalidURLError):
+            sanitizer.validate_url("http://localhost")
+
+    def test_hostname_resolves_to_private_ip_with_allow_localhost(self):
+        sanitizer = URLSanitizer(allow_localhost=True)
+        result = sanitizer.validate_url("http://localhost")
+        assert result.startswith("http://localhost")
+
+    def test_hostname_resolves_to_private_ip_with_allow_private(self):
+        sanitizer = URLSanitizer(allow_private=True)
+        result = sanitizer.validate_url("http://localhost")
+        assert result.startswith("http://localhost")
+
+    def test_unresolvable_hostname_raises_error(self):
+        sanitizer = URLSanitizer()
+        with pytest.raises(InvalidURLError):
+            sanitizer.validate_url("http://this-domain-does-not-exist-12345.com")
+
     def test_blocked_port(self):
         sanitizer = URLSanitizer()
         with pytest.raises(InvalidURLError):
             sanitizer.validate_url("http://example.com:22")
 
-    def test_blocked_port_on_ip_literal(self):
+    def test_blocked_port_with_hostname_private_ip(self):
         sanitizer = URLSanitizer(allow_localhost=True)
         with pytest.raises(InvalidURLError):
-            sanitizer.validate_url("http://127.0.0.1:22")
+            sanitizer.validate_url("http://localhost:22")
 
     def test_allowed_ports_allowlist(self):
         sanitizer = URLSanitizer(allowed_ports=[80, 443])