Skip to content

Commit

Permalink
http: Support CIDR Blocks in no_proxy Env Variable
Browse files Browse the repository at this point in the history
The behavior should now match that of curl. You can specify
no_proxy='192.168.0.0/16' to exclude that IP range. This is especially
useful for not-my-board, because Exporters are accessed with their IP
address, not with a hostname.
  • Loading branch information
holesch committed Apr 14, 2024
1 parent cc26121 commit 7ddd20c
Show file tree
Hide file tree
Showing 2 changed files with 148 additions and 3 deletions.
90 changes: 87 additions & 3 deletions not_my_board/_http.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import asyncio
import codecs
import contextlib
import ipaddress
import json
import logging
import ssl
Expand Down Expand Up @@ -124,9 +125,7 @@ async def _connect(self, url):

def _get_proxy(self, url):
proxy = self._proxies.get(url.scheme)
if proxy and not urllib.request.proxy_bypass_environment(
url.host, self._proxies
):
if proxy and not is_proxy_disabled(url.host, self._proxies.get("no", "")):
return proxy
return None

Expand Down Expand Up @@ -349,6 +348,91 @@ class _ParsedURL:
ssl: Union[bool, ssl.SSLContext]


def is_proxy_disabled(host, no_proxy_env):
if not host or not no_proxy_env:
return False

if no_proxy_env == "*":
return True

def patterns(network_type=None):
for pattern in no_proxy_env.split(","):
pattern = pattern.strip()
if pattern:
if network_type is not None:
try:
pattern = network_type(pattern, strict=False)
except ValueError:
continue
yield pattern

is_disabled = False

if host[0] == "[":
# match IPv6
return _is_proxy_disabled_ipv6(host, patterns(ipaddress.IPv6Network))
else:
try:
addr = ipaddress.IPv4Address(host)
except ValueError:
# neither IPv4 nor IPv6 address, match hostname
is_disabled = _is_proxy_disabled_host(host, patterns())
else:
# match IPv4
for net in patterns(ipaddress.IPv4Network):
if addr in net:
is_disabled = True
break

return is_disabled


def _is_proxy_disabled_ipv6(host, disabled_networks):
end = host.find("]")
if end > 0:
try:
addr = ipaddress.IPv6Address(host[1:end])
except ValueError:
pass
else:
for net in disabled_networks:
if addr in net:
return True
return False


def _is_proxy_disabled_host(host, patterns):
# ignore trailing dots in the host name
if host[-1] == ".":
host = host[:-1]

# ignore case
host = host.lower()

for pattern in patterns:
# ignore trailing dots in the pattern to check
if pattern[-1] == ".":
pattern = pattern[:-1]

if pattern and pattern[0] == ".":
# ignore leading pattern dot as well
pattern = pattern[1:]

if not pattern:
continue

# exact match: example.com matches 'example.com'
if host == pattern.lower():
return True

# tail match: www.example.com matches 'example.com'
# note: nonexample.com does not match 'example.com'
if len(pattern) < len(host) and host.endswith(f".{pattern}"):
return True

return False


# pylint: disable=protected-access
# remove, if Python version < 3.11 is no longer supported
async def _start_tls(writer, url):
Expand Down
61 changes: 61 additions & 0 deletions tests/test_http.py
Original file line number Diff line number Diff line change
Expand Up @@ -77,3 +77,64 @@ async def test_proxy_connect_https(tinyproxy):
client = http.Client(ca_files=[root_cert], proxies={"https": tinyproxy})
response = await client.get_json("https://127.0.0.1:2092/api/v1/places")
assert response == {"places": []}


@pytest.mark.parametrize(
"host,no_proxy_env,expected",
[
# Test cases taken from curl
("www.example.com", "localhost,.example.com,.example.de", True),
("www.example.com.", "localhost,.example.com,.example.de", True),
("example.com", "localhost,.example.com,.example.de", True),
("example.com.", "localhost,.example.com,.example.de", True),
("www.example.com", "localhost,.example.com.,.example.de", True),
("www.example.com", "localhost,www.example.com.,.example.de", True),
("example.com", "localhost,example.com,.example.de", True),
("example.com.", "localhost,example.com,.example.de", True),
("nexample.com", "localhost,example.com,.example.de", False),
("www.example.com", "localhost,example.com,.example.de", True),
("127.0.0.1", "127.0.0.1,localhost", True),
("127.0.0.1", "127.0.0.1,localhost,", True),
("127.0.0.1", "127.0.0.1/8,localhost,", True),
("127.0.0.1", "127.0.0.1/28,localhost,", True),
("127.0.0.1", "127.0.0.1/31,localhost,", True),
("127.0.0.1", "localhost,127.0.0.1", True),
("localhost", "localhost,127.0.0.1", True),
("localhost", "127.0.0.1,localhost", True),
("foobar", "barfoo", False),
("foobar", "foobar", True),
("192.168.0.1", "foobar", False),
("192.168.0.1", "192.168.0.0/16", True),
("192.168.0.1", "192.168.0.0/24", True),
("192.168.0.1", "192.168.0.0/32", False),
("192.168.0.1", "192.168.0.0", False),
("192.168.1.1", "192.168.0.0/24", False),
("192.168.1.1", "foo, bar, 192.168.0.0/24", False),
("192.168.1.1", "foo, bar, 192.168.0.0/16", True),
("[::1]", "foo, bar, 192.168.0.0/16", False),
("[::1]", "foo, bar, ::1/64", True),
("bar", "foo, bar, ::1/64", True),
("BAr", "foo, bar, ::1/64", True),
("BAr", "foo,,,,, bar, ::1/64", True),
("www.example.com", "foo, .example.com", True),
("www.example.com", "www2.example.com, .example.net", False),
("example.com", ".example.com, .example.net", True),
("nonexample.com", ".example.com, .example.net", False),
# Test cases taken from CPython without host:port cases
("anotherdomain.com", "localhost, anotherdomain.com", True),
("localhost", "localhost, anotherdomain.com, .d.o.t", True),
("LocalHost", "localhost, anotherdomain.com, .d.o.t", True),
("LOCALHOST", "localhost, anotherdomain.com, .d.o.t", True),
(".localhost", "localhost, anotherdomain.com, .d.o.t", True),
("foo.d.o.t", "localhost, anotherdomain.com, .d.o.t", True),
("d.o.t", "localhost, anotherdomain.com, .d.o.t", True),
("prelocalhost", "localhost, anotherdomain.com, .d.o.t", False),
("newdomain.com", "localhost, anotherdomain.com, .d.o.t", False),
("newdomain.com", "*", True),
("anotherdomain.com", "*, anotherdomain.com", True),
("newdomain.com", "*, anotherdomain.com", False),
("localhost\n", "localhost, anotherdomain.com", False),
],
)
def test_is_proxy_disabled(host, no_proxy_env, expected):
assert http.is_proxy_disabled(host, no_proxy_env) == expected

0 comments on commit 7ddd20c

Please sign in to comment.