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

Author: Kirtan-pc

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
@@ -29,18 +30,32 @@ class URLSanitizer:
         "file", "gopher", "data", "javascript", "vbscript",
         "about", "view-source", "ws", "wss"
     }
-    
+
+    BLOCKED_PORTS: Set[int] = {
+        7, 9, 11, 13, 15, 17, 19, 20, 21, 22, 23, 25, 37, 42,
+        43, 53, 77, 79, 87, 95, 101, 102, 103, 104, 109, 110,
+        111, 113, 115, 117, 119, 123, 135, 139, 143, 179, 389,
+        443, 445, 465, 512, 513, 514, 515, 526, 530, 531, 532,
+        540, 548, 554, 556, 563, 587, 601, 636, 993, 995,
+        1433, 1434, 1521, 2049, 2375, 2376, 3128, 3306, 3389,
+        4333, 5432, 5900, 5901, 6000, 6001, 6379, 6666, 6667,
+        6668, 6669, 7001, 7002, 8000, 8080, 8081, 8443, 9000,
+        9001, 9090, 9200, 9300, 11211, 27017, 27018, 27019,
+    }
+
     def __init__(
         self,
         allowed_schemes: Optional[List[str]] = None,
         allow_localhost: bool = False,
         allow_private: bool = False,
-        max_length: int = 2048
+        max_length: int = 2048,
+        allowed_ports: Optional[List[int]] = None,
     ):
         self.allowed_schemes = allowed_schemes or ['http', 'https']
         self.allow_localhost = allow_localhost
         self.allow_private = allow_private
         self.max_length = max_length
+        self.allowed_ports = allowed_ports
         self._blocked_networks = [ip_network(net) for net in self.BLOCKED_NETWORKS]
 
     def validate_url(self, url: str) -> str:
@@ -79,7 +94,8 @@ def validate_url(self, url: str) -> str:
         host = parsed.hostname
         if not host:
             raise InvalidURLError("URL must have a valid hostname")
-        
+
+        self._validate_port(parsed.port)
         self._validate_ip_address(host)
         self._validate_characters(url)
 
@@ -95,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_characters(self, url: str) -> None:
         """Check for unsafe or malicious characters."""
@@ -134,8 +182,13 @@ def is_safe_url(self, url: str) -> bool:
 def validate_url(
     url: str,
     allowed_schemes: Optional[List[str]] = None,
+    allowed_ports: Optional[List[int]] = None,
     **kwargs
 ) -> str:
     """Convenience function to validate a URL."""
-    sanitizer = URLSanitizer(allowed_schemes=allowed_schemes, **kwargs)
+    sanitizer = URLSanitizer(
+        allowed_schemes=allowed_schemes,
+        allowed_ports=allowed_ports,
+        **kwargs
+    )
     return sanitizer.validate_url(url)

tests/test_security.py

@@ -78,6 +78,52 @@ 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_with_hostname_private_ip(self):
+        sanitizer = URLSanitizer(allow_localhost=True)
+        with pytest.raises(InvalidURLError):
+            sanitizer.validate_url("http://localhost:22")
+
+    def test_allowed_ports_allowlist(self):
+        sanitizer = URLSanitizer(allowed_ports=[80, 443])
+        result = sanitizer.validate_url("http://example.com:80")
+        assert result.startswith("http://example.com:80")
+
+    def test_allowed_ports_rejects_non_allowed(self):
+        sanitizer = URLSanitizer(allowed_ports=[80, 443])
+        with pytest.raises(InvalidURLError):
+            sanitizer.validate_url("http://example.com:8080")
+
+    def test_valid_port_not_blocked(self):
+        sanitizer = URLSanitizer()
+        result = sanitizer.validate_url("http://example.com:8888")
+        assert result.startswith("http://example.com:8888")
+
     def test_blocked_schemes(self):
         # These should be blocked because they're in BLOCKED_SCHEMES
         sanitizer = URLSanitizer()