Skip to content

Commit 6861ee4

Browse files
authored
Use ipaddress module for no_proxy parsing (#1033)
* Use ipaddress module for no_proxy parsing This enables support for IPv6 CIDRs in the no_proxy option * Apply black formatting
1 parent 729bdb8 commit 6861ee4

File tree

2 files changed

+21
-14
lines changed

2 files changed

+21
-14
lines changed

websocket/_url.py

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
1+
import ipaddress
12
import os
2-
import socket
3-
import struct
43
from typing import Optional
54
from urllib.parse import unquote, urlparse
65
from ._exceptions import WebSocketProxyException
@@ -74,29 +73,30 @@ def parse_url(url: str) -> tuple:
7473

7574

7675
def _is_ip_address(addr: str) -> bool:
76+
if not isinstance(addr, str):
77+
raise TypeError("_is_ip_address() argument 1 must be str")
7778
try:
78-
socket.inet_aton(addr)
79-
except socket.error:
79+
ipaddress.ip_address(addr)
80+
except ValueError:
8081
return False
8182
else:
8283
return True
8384

8485

8586
def _is_subnet_address(hostname: str) -> bool:
8687
try:
87-
addr, netmask = hostname.split("/")
88-
return _is_ip_address(addr) and 0 <= int(netmask) < 32
88+
ipaddress.ip_network(hostname)
8989
except ValueError:
9090
return False
91+
else:
92+
return True
9193

9294

9395
def _is_address_in_network(ip: str, net: str) -> bool:
94-
ipaddr: int = struct.unpack("!I", socket.inet_aton(ip))[0]
95-
netaddr, netmask = net.split("/")
96-
netaddr: int = struct.unpack("!I", socket.inet_aton(netaddr))[0]
97-
98-
netmask = (0xFFFFFFFF << (32 - int(netmask))) & 0xFFFFFFFF
99-
return ipaddr & netmask == netaddr
96+
try:
97+
return ipaddress.ip_network(ip).subnet_of(ipaddress.ip_network(net))
98+
except TypeError:
99+
return False
100100

101101

102102
def _is_no_proxy_host(hostname: str, no_proxy: Optional[list[str]]) -> bool:

websocket/tests/test_url.py

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,8 @@ def test_address_in_network(self):
3636
self.assertTrue(_is_address_in_network("127.0.0.1", "127.0.0.0/8"))
3737
self.assertTrue(_is_address_in_network("127.1.0.1", "127.0.0.0/8"))
3838
self.assertFalse(_is_address_in_network("127.1.0.1", "127.0.0.0/24"))
39+
self.assertTrue(_is_address_in_network("2001:db8::1", "2001:db8::/64"))
40+
self.assertFalse(_is_address_in_network("2001:db8:1::1", "2001:db8::/64"))
3941

4042
def test_parse_url(self):
4143
p = parse_url("ws://www.example.com/r")
@@ -167,11 +169,16 @@ def test_ip_address_in_range(self):
167169
self.assertTrue(_is_no_proxy_host("127.0.0.1", ["127.0.0.0/8"]))
168170
self.assertTrue(_is_no_proxy_host("127.0.0.2", ["127.0.0.0/8"]))
169171
self.assertFalse(_is_no_proxy_host("127.1.0.1", ["127.0.0.0/24"]))
170-
os.environ["no_proxy"] = "127.0.0.0/8"
172+
self.assertTrue(_is_no_proxy_host("2001:db8::1", ["2001:db8::/64"]))
173+
self.assertFalse(_is_no_proxy_host("2001:db8:1::1", ["2001:db8::/64"]))
174+
os.environ["no_proxy"] = "127.0.0.0/8,2001:db8::/64"
171175
self.assertTrue(_is_no_proxy_host("127.0.0.1", None))
172176
self.assertTrue(_is_no_proxy_host("127.0.0.2", None))
173-
os.environ["no_proxy"] = "127.0.0.0/24"
177+
self.assertTrue(_is_no_proxy_host("2001:db8::1", None))
178+
self.assertFalse(_is_no_proxy_host("2001:db8:1::1", None))
179+
os.environ["no_proxy"] = "127.0.0.0/24,2001:db8::/64"
174180
self.assertFalse(_is_no_proxy_host("127.1.0.1", None))
181+
self.assertFalse(_is_no_proxy_host("2001:db8:1::1", None))
175182

176183
def test_hostname_match(self):
177184
self.assertTrue(_is_no_proxy_host("my.websocket.org", ["my.websocket.org"]))

0 commit comments

Comments
 (0)