diff --git a/security/url_sanitizer.py b/security/url_sanitizer.py index a399b11a..099c7ad3 100644 --- a/security/url_sanitizer.py +++ b/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.""" diff --git a/tests/test_security.py b/tests/test_security.py index 197f4c5f..6dca5802 100644 --- a/tests/test_security.py +++ b/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])