diff --git a/src/requests/utils.py b/src/requests/utils.py index 8ab55852cc..7f2ef3c17d 100644 --- a/src/requests/utils.py +++ b/src/requests/utils.py @@ -9,6 +9,7 @@ import codecs import contextlib import io +import ipaddress import os import re import socket @@ -669,19 +670,27 @@ def requote_uri(uri): return quote(uri, safe=safe_without_percent) -def address_in_network(ip, net): - """This function allows you to check if an IP belongs to a network subnet +def address_in_network(ip: str, net: str) -> bool: + """Check if an IP address belongs to a network subnet. - Example: returns True if ip = 192.168.1.1 and net = 192.168.1.0/24 - returns False if ip = 192.168.1.1 and net = 192.168.100.0/24 + Supports both IPv4 and IPv6 addresses and networks. + Returns False if IP version doesn't match network version. + Example: returns True if ip = '192.168.1.1' and net = '192.168.1.0/24' + returns False if ip = '192.168.1.1' and net = '192.168.100.0/24' + returns True if ip = '2001:db8::1' and net = '2001:db8::/32' + returns False if ip = '192.168.1.1' and net = '2001:db8::/32' + + :param ip: IP address string (IPv4 or IPv6) + :param net: Network in CIDR notation (e.g., '192.168.1.0/24' or '2001:db8::/32') :rtype: bool """ - ipaddr = struct.unpack("=L", socket.inet_aton(ip))[0] - netaddr, bits = net.split("/") - netmask = struct.unpack("=L", socket.inet_aton(dotted_netmask(int(bits))))[0] - network = struct.unpack("=L", socket.inet_aton(netaddr))[0] & netmask - return (ipaddr & netmask) == (network & netmask) + try: + ip_obj = ipaddress.ip_address(ip) + network_obj = ipaddress.ip_network(net, strict=False) + return ip_obj in network_obj + except (ipaddress.AddressValueError, ipaddress.NetmaskValueError, ValueError): + return False def dotted_netmask(mask): @@ -695,39 +704,87 @@ def dotted_netmask(mask): return socket.inet_ntoa(struct.pack(">I", bits)) -def is_ipv4_address(string_ip): +def is_ipv4_address(string_ip: str) -> bool: + """Check if a string is a valid IPv4 address. + + Example: returns True if string_ip = '192.168.1.1' + returns False if string_ip = 'localhost' + + :param string_ip: The string to check + :rtype: bool """ + try: + ipaddress.IPv4Address(string_ip) + except (ipaddress.AddressValueError, ValueError): + return False + return True + + +def is_ipv6_address(string_ip: str) -> bool: + """Check if a string is a valid IPv6 address. + + Example: returns True if string_ip = '::1' + returns True if string_ip = '2001:db8::1' + returns False if string_ip = '192.168.1.1' + + :param string_ip: The string to check :rtype: bool """ try: - socket.inet_aton(string_ip) - except OSError: + ipaddress.IPv6Address(string_ip) + except (ipaddress.AddressValueError, ValueError): return False return True -def is_valid_cidr(string_network): +def compare_ips(a: str, b: str) -> bool: + """Compare two IP addresses for equality. + + Normalizes IPv6 addresses to handle different compression formats. + Works with both IPv4 and IPv6 addresses. + + Example: returns True if a = '::1' and b = '0:0:0:0:0:0:0:1' + returns True if a = '192.168.1.1' and b = '192.168.1.1' + returns False if a = '::1' and b = '::2' + returns False if a = '192.168.1.1' and b = '::1' + + :param a: First IP address string + :param b: Second IP address string + :rtype: bool """ - Very simple check of the cidr format in no_proxy variable. + try: + return ipaddress.ip_address(a) == ipaddress.ip_address(b) + except (ipaddress.AddressValueError, ValueError): + return False + + +def is_valid_cidr(string_network: str) -> bool: + """Check if a string is valid CIDR notation for IPv4 or IPv6. + Example: returns True if string_network = '192.168.1.0/24' + returns True if string_network = '2001:db8::/32' + returns False if string_network = '192.168.1.0/33' + + :param string_network: The CIDR string to check :rtype: bool """ - if string_network.count("/") == 1: - try: - mask = int(string_network.split("/")[1]) - except ValueError: - return False + if string_network.count("/") != 1: + return False - if mask < 1 or mask > 32: - return False + try: + address, mask_str = string_network.split("/") + mask = int(mask_str) + except ValueError: + return False - try: - socket.inet_aton(string_network.split("/")[0]) - except OSError: - return False + # Check if it's IPv4 + if is_ipv4_address(address): + return 1 <= mask <= 32 + # Check if it's IPv6 + elif is_ipv6_address(address): + return 0 <= mask <= 128 else: return False - return True @contextlib.contextmanager @@ -780,12 +837,12 @@ def get_proxy(key): # the end of the hostname, both with and without the port. no_proxy = (host for host in no_proxy.replace(" ", "").split(",") if host) - if is_ipv4_address(parsed.hostname): + if is_ipv4_address(parsed.hostname) or is_ipv6_address(parsed.hostname): for proxy_ip in no_proxy: if is_valid_cidr(proxy_ip): if address_in_network(parsed.hostname, proxy_ip): return True - elif parsed.hostname == proxy_ip: + elif compare_ips(parsed.hostname, proxy_ip): # If no_proxy ip was defined in plain IP notation instead of cidr notation & # matches the IP of the index return True diff --git a/tests/test_utils.py b/tests/test_utils.py index f9a287af1b..cc05bb476d 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -17,6 +17,7 @@ _parse_content_type_header, add_dict_to_cookiejar, address_in_network, + compare_ips, dotted_netmask, extract_zipped_paths, get_auth_from_url, @@ -27,6 +28,7 @@ guess_filename, guess_json_utf, is_ipv4_address, + is_ipv6_address, is_valid_cidr, iter_slices, parse_dict_header, @@ -281,18 +283,76 @@ def test_invalid(self, value): assert not is_ipv4_address(value) +class TestIsIPv6Address: + @pytest.mark.parametrize( + "value", + ( + "::1", # loopback + "::0", # all zeros + "2001:db8::1", # compressed + "2001:0db8:0000:0000:0000:0000:0000:0001", # full notation + "fe80::1", # link-local + "ff02::1", # multicast + "2001:db8:85a3::8a2e:370:7334", # compressed with multiple segments + "::", # all zeros compressed + "::ffff:192.0.2.1", # IPv4-mapped IPv6 + ), + ) + def test_valid(self, value): + assert is_ipv6_address(value) + + @pytest.mark.parametrize( + "value", + ( + "192.168.1.1", # IPv4 address + "localhost", # hostname + "gggg::1", # invalid hex + "::1::2", # double compression + "2001:db8::1::2", # double compression + "2001:db8:85a3::8a2e:370g:7334", # invalid character + "", # empty string + "not-an-ip", # random string + ), + ) + def test_invalid(self, value): + assert not is_ipv6_address(value) + + class TestIsValidCIDR: - def test_valid(self): - assert is_valid_cidr("192.168.1.0/24") + @pytest.mark.parametrize( + "value", + ( + "192.168.1.0/24", # IPv4 CIDR + "192.168.1.0/32", # IPv4 single host + "10.0.0.0/8", # IPv4 large network + "192.168.1.0/1", # IPv4 minimal mask + "2001:db8::/32", # IPv6 CIDR + "2001:db8::/64", # IPv6 typical subnet + "2001:db8::/128", # IPv6 single host + "fe80::/10", # IPv6 link-local + "::/0", # IPv6 all networks + "::1/128", # IPv6 loopback + ), + ) + def test_valid(self, value): + assert is_valid_cidr(value) @pytest.mark.parametrize( "value", ( - "8.8.8.8", - "192.168.1.0/a", - "192.168.1.0/128", - "192.168.1.0/-1", - "192.168.1.999/24", + "8.8.8.8", # no mask + "192.168.1.0/a", # non-numeric mask + "192.168.1.0/128", # mask too large for IPv4 + "192.168.1.0/-1", # negative mask + "192.168.1.999/24", # invalid IPv4 + "192.168.1.0/0", # mask is 0 (invalid) + "192.168.1.0/33", # mask too large for IPv4 + "2001:db8::/129", # mask too large for IPv6 + "2001:db8::/-1", # negative mask for IPv6 + "2001:db8::/a", # non-numeric mask for IPv6 + "2001:db8::1", # no mask + "gggg::/64", # invalid IPv6 + "192.168.1.0/24/32", # multiple slashes ), ) def test_invalid(self, value): @@ -300,11 +360,93 @@ def test_invalid(self, value): class TestAddressInNetwork: - def test_valid(self): - assert address_in_network("192.168.1.1", "192.168.1.0/24") + @pytest.mark.parametrize( + "ip, network", + ( + ("192.168.1.1", "192.168.1.0/24"), # IPv4 basic + ("192.168.1.0", "192.168.1.0/24"), # network address + ("192.168.1.255", "192.168.1.0/24"), # broadcast address + ("10.0.0.1", "10.0.0.0/8"), # large IPv4 network + ("2001:db8::1", "2001:db8::/32"), # IPv6 basic + ("2001:db8:0:0:0:0:0:1", "2001:db8::/32"), # IPv6 full notation + ("2001:db8::ffff", "2001:db8::/32"), # IPv6 in network + ("fe80::1", "fe80::/10"), # link-local + ("::1", "::1/128"), # loopback single host + ("::", "::/128"), # all zeros single host + ), + ) + def test_valid(self, ip, network): + assert address_in_network(ip, network) - def test_invalid(self): - assert not address_in_network("172.16.0.1", "192.168.1.0/24") + @pytest.mark.parametrize( + "ip, network", + ( + ("172.16.0.1", "192.168.1.0/24"), # IPv4 not in network + ("192.168.2.1", "192.168.1.0/24"), # IPv4 adjacent network + ("192.168.1.1", "192.168.1.0/32"), # host doesn't match + ("2001:db9::1", "2001:db8::/32"), # IPv6 not in network + ("2001:db8:1::1", "2001:db8::/64"), # IPv6 different subnet + ("192.168.1.1", "2001:db8::/32"), # IPv4 vs IPv6 + ("2001:db8::1", "192.168.1.0/24"), # IPv6 vs IPv4 + ("::1", "fe80::/10"), # loopback not in link-local + ), + ) + def test_invalid(self, ip, network): + assert not address_in_network(ip, network) + + @pytest.mark.parametrize( + "ip, network", + ( + ("not-an-ip", "192.168.1.0/24"), # invalid IP + ("192.168.1.1", "not-a-network"), # invalid network + ("192.168.1.1", "192.168.1.0/abc"), # invalid CIDR + ), + ) + def test_invalid_input(self, ip, network): + assert not address_in_network(ip, network) + + +class TestCompareIPs: + @pytest.mark.parametrize( + "a, b", + ( + ("192.168.1.1", "192.168.1.1"), # IPv4 identical + ("::1", "::1"), # IPv6 identical + ("::1", "0:0:0:0:0:0:0:1"), # IPv6 compressed vs full + ("2001:db8::1", "2001:0db8:0000:0000:0000:0000:0000:0001"), # IPv6 variations + ("::ffff:192.0.2.1", "::ffff:c000:201"), # IPv4-mapped IPv6 + ("::", "0:0:0:0:0:0:0:0"), # all zeros + ("fe80::1", "fe80:0:0:0:0:0:0:1"), # link-local + ), + ) + def test_equal(self, a, b): + assert compare_ips(a, b) + + @pytest.mark.parametrize( + "a, b", + ( + ("192.168.1.1", "192.168.1.2"), # IPv4 different + ("::1", "::2"), # IPv6 different + ("192.168.1.1", "::1"), # IPv4 vs IPv6 + ("2001:db8::1", "2001:db8::2"), # IPv6 different + ("10.0.0.1", "192.168.1.1"), # different IPv4 + ), + ) + def test_not_equal(self, a, b): + assert not compare_ips(a, b) + + @pytest.mark.parametrize( + "a, b", + ( + ("not-an-ip", "192.168.1.1"), # invalid first + ("192.168.1.1", "not-an-ip"), # invalid second + ("not-an-ip", "not-an-ip"), # both invalid + ("", "192.168.1.1"), # empty first + ("192.168.1.1", ""), # empty second + ), + ) + def test_invalid_input(self, a, b): + assert not compare_ips(a, b) class TestGuessFilename: @@ -349,6 +491,13 @@ def test_zipped_paths_extracted(self, tmpdir): _, name = os.path.splitdrive(__file__) zipped_path = os.path.join(zipped_py.strpath, name.lstrip(r"\/")) + + # Clean up any previously cached extracted file to ensure fresh extraction + import tempfile + expected_extracted_path = os.path.join(tempfile.gettempdir(), name.split(os.sep)[-1]) + if os.path.exists(expected_extracted_path): + os.remove(expected_extracted_path) + extracted_path = extract_zipped_paths(zipped_path) assert extracted_path != zipped_path @@ -731,6 +880,7 @@ def test_urldefragauth(url, expected): @pytest.mark.parametrize( "url, expected", ( + # IPv4 tests ("http://192.168.0.1:5000/", True), ("http://192.168.0.1/", True), ("http://172.16.1.1/", True), @@ -741,6 +891,21 @@ def test_urldefragauth(url, expected): ("http://172.16.1.12:5000/", False), ("http://google.com:5000/v1.0/", False), ("file:///some/path/on/disk", True), + # IPv6 tests - exact match + ("http://[::1]/", True), + ("http://[0:0:0:0:0:0:0:1]/", True), # full notation of ::1 + ("http://[::1]:8080/", True), + ("http://[fe80::1]/", True), + ("http://[fe80:0:0:0:0:0:0:1]/", True), # full notation of fe80::1 + # IPv6 tests - CIDR match + ("http://[2001:db8::1]/", True), + ("http://[2001:db8::ffff]/", True), + ("http://[2001:db8:0:0:0:0:0:1]/", True), # full notation + ("http://[2001:db8::1]:9000/", True), + # IPv6 tests - should not bypass + ("http://[2001:db9::1]/", False), # not in no_proxy + ("http://[fe80::2]/", False), # not exact match + ("http://[::2]/", False), # not in no_proxy ), ) def test_should_bypass_proxies(url, expected, monkeypatch): @@ -749,11 +914,11 @@ def test_should_bypass_proxies(url, expected, monkeypatch): """ monkeypatch.setenv( "no_proxy", - "192.168.0.0/24,127.0.0.1,localhost.localdomain,172.16.1.1, google.com:6000", + "192.168.0.0/24,127.0.0.1,localhost.localdomain,172.16.1.1,::1,fe80::1,2001:db8::/32, google.com:6000", ) monkeypatch.setenv( "NO_PROXY", - "192.168.0.0/24,127.0.0.1,localhost.localdomain,172.16.1.1, google.com:6000", + "192.168.0.0/24,127.0.0.1,localhost.localdomain,172.16.1.1,::1,fe80::1,2001:db8::/32, google.com:6000", ) assert should_bypass_proxies(url, no_proxy=None) == expected