Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 30 additions & 4 deletions py/selenium/webdriver/remote/client_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,33 @@ class AuthType(Enum):
X_API_KEY = "X-API-Key"


def _no_proxy_entry_matches(entry: str, hostname: str, netloc: str) -> bool:
"""Whether one ``no_proxy`` entry covers the host being connected to.

Follows the semantics of :func:`urllib.request.proxy_bypass_environment`: an
entry covers the host itself and any of its sub-domains, compared without
regard to case. Empty entries, which a trailing or doubled comma produces,
match nothing rather than everything.

Args:
entry: A single entry from ``no_proxy``, either a bare host
(optionally with a port, optionally dot-prefixed) or a full URL.
hostname: Lower-cased host of the remote server address, without a port.
netloc: Lower-cased host of the remote server address, with any port.

Returns:
True if the proxy should be bypassed for this host.
"""
# A bare "host:port" entry is not a URL, and parsing it as one would read
# the host as a scheme, so only entries that name a scheme are parsed.
if "://" in entry:
entry = parse.urlparse(entry).netloc
entry = entry.strip().lstrip(".").lower()
Comment thread
qodo-code-review[bot] marked this conversation as resolved.
Outdated
if not entry:
return False
return any(host == entry or host.endswith(f".{entry}") for host in (hostname, netloc))


class _ClientConfigDescriptor:
def __init__(self, name):
self.name = name
Expand Down Expand Up @@ -136,13 +163,12 @@ def get_proxy_url(self) -> str | None:
if proxy_type is ProxyType.SYSTEM:
_no_proxy = os.environ.get("no_proxy", os.environ.get("NO_PROXY"))
if _no_proxy:
hostname = (remote_add.hostname or "").lower()
netloc = remote_add.netloc.lower()
for entry in map(str.strip, _no_proxy.split(",")):
if entry == "*":
return None
n_url = parse.urlparse(entry)
if n_url.netloc and remote_add.netloc == n_url.netloc:
return None
if n_url.path in remote_add.netloc:
if _no_proxy_entry_matches(entry, hostname, netloc):
return None
return os.environ.get(
"https_proxy" if self.remote_server_addr.startswith("https://") else "http_proxy",
Expand Down
95 changes: 95 additions & 0 deletions py/test/unit/selenium/webdriver/remote/client_config_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,14 +17,35 @@

import pytest

from selenium.webdriver.common.proxy import Proxy, ProxyType
from selenium.webdriver.remote.client_config import ClientConfig

PROXY = "http://proxy.internal:3128"


@pytest.fixture
def config():
return ClientConfig(remote_server_addr="http://localhost:4444")


@pytest.fixture
def system_proxy_env(monkeypatch):
"""Clear every proxy variable, then set only ``http_proxy``."""

def setup(no_proxy=None):
for name in ("http_proxy", "HTTP_PROXY", "https_proxy", "HTTPS_PROXY", "no_proxy", "NO_PROXY"):
monkeypatch.delenv(name, raising=False)
monkeypatch.setenv("http_proxy", PROXY)
if no_proxy is not None:
monkeypatch.setenv("no_proxy", no_proxy)

return setup


def system_config(remote_server_addr="http://localhost:4444"):
return ClientConfig(remote_server_addr=remote_server_addr, proxy=Proxy(raw={"proxyType": ProxyType.SYSTEM}))
Comment thread
navin772 marked this conversation as resolved.


def test_websocket_max_message_size_defaults_to_none(config):
assert config.websocket_max_message_size is None

Expand All @@ -37,3 +58,77 @@ def test_websocket_max_message_size_can_be_set(config):
def test_websocket_max_message_size_via_constructor():
cfg = ClientConfig(remote_server_addr="http://localhost:4444", websocket_max_message_size=2**26)
assert cfg.websocket_max_message_size == 2**26


@pytest.mark.parametrize(
"no_proxy",
[
"example.com,",
",example.com",
"example.com,,other.com",
"example.com, ,other.com",
",",
"",
],
ids=[
"trailing-comma",
"leading-comma",
"doubled-comma",
"whitespace-only-entry",
"bare-comma",
"empty-value",
],
)
def test_empty_no_proxy_entries_do_not_bypass_the_proxy(system_proxy_env, no_proxy):
"""An empty entry must be ignored, not treated as matching every host."""
system_proxy_env(no_proxy)
assert system_config().get_proxy_url() == PROXY


@pytest.mark.parametrize(
("no_proxy", "server"),
[
("foo.com", "http://myfoo.com.example.org:4444"),
("example.com", "http://notexample.common.org:4444"),
("localhost", "http://localhosting.org:4444"),
],
)
def test_no_proxy_entry_does_not_match_on_a_bare_substring(system_proxy_env, no_proxy, server):
"""A bypass entry must match a whole host or a dot-delimited suffix of it."""
system_proxy_env(no_proxy)
assert system_config(server).get_proxy_url() == PROXY


@pytest.mark.parametrize(
("no_proxy", "server"),
[
("example.com", "http://example.com:4444"),
("example.com", "http://sub.example.com:4444"),
(".example.com", "http://sub.example.com:4444"),
("localhost", "http://localhost:4444"),
("other.com,example.com", "http://example.com:4444"),
("other.com, example.com", "http://example.com:4444"),
("example.com,", "http://example.com:4444"),
("EXAMPLE.COM", "http://example.com:4444"),
("127.0.0.1", "http://127.0.0.1:4444"),
],
)
def test_matching_no_proxy_entry_bypasses_the_proxy(system_proxy_env, no_proxy, server):
system_proxy_env(no_proxy)
assert system_config(server).get_proxy_url() is None


def test_no_proxy_wildcard_bypasses_every_host(system_proxy_env):
system_proxy_env("*")
assert system_config().get_proxy_url() is None


def test_no_proxy_entry_written_as_a_url_matches_only_its_host(system_proxy_env):
system_proxy_env("http://example.com")
assert system_config("http://example.com:4444").get_proxy_url() is None
assert system_config("http://localhost:4444").get_proxy_url() == PROXY


def test_proxy_is_used_when_no_proxy_is_unset(system_proxy_env):
system_proxy_env()
assert system_config().get_proxy_url() == PROXY
43 changes: 28 additions & 15 deletions py/test/unit/selenium/webdriver/remote/remote_connection_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -231,7 +231,9 @@ def test_get_proxy_url_https_auth(mock_proxy_auth_settings):
def test_get_connection_manager_without_proxy(mock_proxy_settings_missing):
remote_connection = RemoteConnection("http://remote", keep_alive=False)
conn = remote_connection._get_connection_manager()
assert isinstance(conn, PoolManager)
# ProxyManager and SOCKSProxyManager both subclass PoolManager, so isinstance
# cannot tell a direct connection from a proxied one.
assert type(conn) is PoolManager


def test_get_connection_manager_for_certs_and_timeout():
Expand Down Expand Up @@ -293,31 +295,42 @@ def test_get_connection_manager_with_auth_https_proxy(mock_proxy_auth_settings):
@pytest.mark.parametrize(
"url",
[
"*",
".localhost",
"localhost:80",
"localhost",
"LOCALHOST",
"LOCALHOST:80",
"http://localhost",
"http://localhost:80",
"https://localhost",
"test.localhost",
" localhost",
"127.0.0.1",
"127.0.0.2",
"::1",
"http://LOCALHOST",
"http://LOCALHOST:80",
"http://test.localhost",
"http://127.0.0.1",
"http://65.253.214.253",
"http://[::1]",
],
)
def test_get_connection_manager_when_no_proxy_set(mock_no_proxy_settings, url):
remote_connection = RemoteConnection(url)
conn = remote_connection._get_connection_manager()
assert isinstance(conn, PoolManager)
assert remote_connection.client_config.get_proxy_url() is None
assert type(remote_connection._get_connection_manager()) is PoolManager


@pytest.mark.parametrize(
"url",
[
"http://127.0.0.2",
"http://notlocalhost.com",
"http://localhost.evil.com",
],
)
def test_get_connection_manager_when_no_proxy_does_not_match(mock_no_proxy_settings, url):
"""A host that no_proxy does not cover must still be reached through the proxy."""
remote_connection = RemoteConnection(url)
assert remote_connection.client_config.get_proxy_url() == "http://http_proxy.com:8080"
assert isinstance(remote_connection._get_connection_manager(), ProxyManager)


def test_ignore_proxy_env_vars(mock_proxy_settings):
remote_connection = RemoteConnection("http://remote", ignore_proxy=True)
conn = remote_connection._get_connection_manager()
assert isinstance(conn, PoolManager)
assert type(conn) is PoolManager


def test_get_socks_proxy_when_set(mock_socks_proxy_settings):
Expand Down
Loading