-
Notifications
You must be signed in to change notification settings - Fork 278
Handle proxy variables #2789
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
kairoaraujo
merged 6 commits into
theupdateframework:develop
from
jku:handle-proxy-variables
Mar 4, 2025
Merged
Handle proxy variables #2789
Changes from all commits
Commits
Show all changes
6 commits
Select commit
Hold shift + click to select a range
396ba07
ngclient: Add proxy environment variable handling
jku 5f9fefb
tests: Add tests for ProxyEnvironment
jku 80b6290
Use __future__ to make old python happy
jku 9a4e749
ngclient: Add docs on HTTP in general
jku 265e772
ProxyEnvironment: Handle no_proxy="*"
jku 98fcd71
Changelog: Add missing entries
jku File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,217 @@ | ||
# Copyright 2025, the TUF contributors | ||
# SPDX-License-Identifier: MIT OR Apache-2.0 | ||
|
||
"""Test ngclient ProxyEnvironment""" | ||
|
||
from __future__ import annotations | ||
|
||
import sys | ||
import unittest | ||
from unittest.mock import Mock, patch | ||
|
||
from urllib3 import PoolManager, ProxyManager | ||
|
||
from tests import utils | ||
from tuf.ngclient._internal.proxy import ProxyEnvironment | ||
|
||
|
||
class TestProxyEnvironment(unittest.TestCase): | ||
"""Test ngclient ProxyEnvironment implementation | ||
|
||
These tests use the ProxyEnvironment.get_pool_manager() endpoint and then | ||
look at the ProxyEnvironment._poolmanagers dict keys to decide if the result | ||
is correct. | ||
|
||
The test environment is changed via mocking getproxies(): this is a urllib | ||
method that returns a dict with the proxy environment variable contents. | ||
|
||
Testing ProxyEnvironment.request() would possibly be better but far more | ||
difficult: the current test implementation does not require actually setting up | ||
all of the different proxies. | ||
""" | ||
|
||
def assert_pool_managers( | ||
self, env: ProxyEnvironment, expected: list[str | None] | ||
) -> None: | ||
# Pool managers have the expected proxy urls | ||
self.assertEqual(list(env._pool_managers.keys()), expected) | ||
|
||
# Pool manager types are as expected | ||
for proxy_url, pool_manager in env._pool_managers.items(): | ||
self.assertIsInstance(pool_manager, PoolManager) | ||
if proxy_url is not None: | ||
self.assertIsInstance(pool_manager, ProxyManager) | ||
|
||
@patch("tuf.ngclient._internal.proxy.getproxies") | ||
def test_no_variables(self, mock_getproxies: Mock) -> None: | ||
mock_getproxies.return_value = {} | ||
|
||
env = ProxyEnvironment() | ||
env.get_pool_manager("http", "example.com") | ||
env.get_pool_manager("https", "example.com") | ||
env.get_pool_manager("https", "example.com") | ||
env.get_pool_manager("https", "subdomain.example.com") | ||
env.get_pool_manager("https", "differentsite.com") | ||
|
||
# There is a single pool manager (no proxies) | ||
self.assert_pool_managers(env, [None]) | ||
|
||
@patch("tuf.ngclient._internal.proxy.getproxies") | ||
def test_proxy_set(self, mock_getproxies: Mock) -> None: | ||
mock_getproxies.return_value = { | ||
"https": "http://localhost:8888", | ||
} | ||
|
||
env = ProxyEnvironment() | ||
env.get_pool_manager("http", "example.com") | ||
env.get_pool_manager("https", "example.com") | ||
env.get_pool_manager("https", "example.com") | ||
env.get_pool_manager("https", "differentsite.com") | ||
|
||
# There are two pool managers: A plain poolmanager and https proxymanager | ||
self.assert_pool_managers(env, [None, "http://localhost:8888"]) | ||
|
||
@patch("tuf.ngclient._internal.proxy.getproxies") | ||
def test_proxies_set(self, mock_getproxies: Mock) -> None: | ||
mock_getproxies.return_value = { | ||
"http": "http://localhost:8888", | ||
"https": "http://localhost:9999", | ||
} | ||
|
||
env = ProxyEnvironment() | ||
env.get_pool_manager("http", "example.com") | ||
env.get_pool_manager("https", "example.com") | ||
env.get_pool_manager("https", "example.com") | ||
env.get_pool_manager("https", "subdomain.example.com") | ||
env.get_pool_manager("https", "differentsite.com") | ||
|
||
# There are two pool managers: A http proxymanager and https proxymanager | ||
self.assert_pool_managers( | ||
env, ["http://localhost:8888", "http://localhost:9999"] | ||
) | ||
|
||
@patch("tuf.ngclient._internal.proxy.getproxies") | ||
def test_no_proxy_set(self, mock_getproxies: Mock) -> None: | ||
mock_getproxies.return_value = { | ||
"http": "http://localhost:8888", | ||
"https": "http://localhost:9999", | ||
"no": "somesite.com, example.com, another.site.com", | ||
} | ||
|
||
env = ProxyEnvironment() | ||
env.get_pool_manager("http", "example.com") | ||
env.get_pool_manager("https", "example.com") | ||
env.get_pool_manager("https", "example.com") | ||
|
||
# There is a single pool manager (no proxies) | ||
self.assert_pool_managers(env, [None]) | ||
|
||
env.get_pool_manager("http", "differentsite.com") | ||
env.get_pool_manager("https", "differentsite.com") | ||
|
||
# There are three pool managers: plain poolmanager for no_proxy domains, | ||
# http proxymanager and https proxymanager | ||
self.assert_pool_managers( | ||
env, [None, "http://localhost:8888", "http://localhost:9999"] | ||
) | ||
|
||
@patch("tuf.ngclient._internal.proxy.getproxies") | ||
def test_no_proxy_subdomain_match(self, mock_getproxies: Mock) -> None: | ||
mock_getproxies.return_value = { | ||
"https": "http://localhost:9999", | ||
"no": "somesite.com, example.com, another.site.com", | ||
} | ||
|
||
env = ProxyEnvironment() | ||
|
||
# this should match example.com in no_proxy | ||
env.get_pool_manager("https", "subdomain.example.com") | ||
|
||
# There is a single pool manager (no proxies) | ||
self.assert_pool_managers(env, [None]) | ||
|
||
# this should not match example.com in no_proxy | ||
env.get_pool_manager("https", "xexample.com") | ||
|
||
# There are two pool managers: plain poolmanager for no_proxy domains, | ||
# and a https proxymanager | ||
self.assert_pool_managers(env, [None, "http://localhost:9999"]) | ||
|
||
@patch("tuf.ngclient._internal.proxy.getproxies") | ||
def test_no_proxy_wildcard(self, mock_getproxies: Mock) -> None: | ||
mock_getproxies.return_value = { | ||
"https": "http://localhost:8888", | ||
"no": "*", | ||
} | ||
|
||
env = ProxyEnvironment() | ||
env.get_pool_manager("https", "example.com") | ||
env.get_pool_manager("https", "differentsite.com") | ||
env.get_pool_manager("https", "subdomain.example.com") | ||
|
||
# There is a single pool manager, no proxies | ||
self.assert_pool_managers(env, [None]) | ||
|
||
@patch("tuf.ngclient._internal.proxy.getproxies") | ||
def test_no_proxy_leading_dot(self, mock_getproxies: Mock) -> None: | ||
mock_getproxies.return_value = { | ||
"https": "http://localhost:8888", | ||
"no": ".example.com", | ||
} | ||
|
||
env = ProxyEnvironment() | ||
env.get_pool_manager("https", "example.com") | ||
env.get_pool_manager("https", "subdomain.example.com") | ||
|
||
# There is a single pool manager, no proxies | ||
self.assert_pool_managers(env, [None]) | ||
|
||
@patch("tuf.ngclient._internal.proxy.getproxies") | ||
def test_all_proxy_set(self, mock_getproxies: Mock) -> None: | ||
mock_getproxies.return_value = { | ||
"all": "http://localhost:8888", | ||
} | ||
|
||
env = ProxyEnvironment() | ||
env.get_pool_manager("http", "example.com") | ||
env.get_pool_manager("https", "example.com") | ||
env.get_pool_manager("https", "example.com") | ||
env.get_pool_manager("https", "subdomain.example.com") | ||
env.get_pool_manager("https", "differentsite.com") | ||
|
||
# There is a single proxy manager | ||
self.assert_pool_managers(env, ["http://localhost:8888"]) | ||
|
||
# This urllib3 currently only handles http and https but let's test anyway | ||
env.get_pool_manager("file", None) | ||
|
||
# proxy manager and a plain pool manager | ||
self.assert_pool_managers(env, ["http://localhost:8888", None]) | ||
|
||
@patch("tuf.ngclient._internal.proxy.getproxies") | ||
def test_all_proxy_and_no_proxy_set(self, mock_getproxies: Mock) -> None: | ||
mock_getproxies.return_value = { | ||
"all": "http://localhost:8888", | ||
"no": "somesite.com, example.com, another.site.com", | ||
} | ||
|
||
env = ProxyEnvironment() | ||
env.get_pool_manager("http", "example.com") | ||
env.get_pool_manager("https", "example.com") | ||
env.get_pool_manager("https", "example.com") | ||
env.get_pool_manager("https", "subdomain.example.com") | ||
|
||
# There is a single pool manager (no proxies) | ||
self.assert_pool_managers(env, [None]) | ||
|
||
env.get_pool_manager("http", "differentsite.com") | ||
env.get_pool_manager("https", "differentsite.com") | ||
|
||
# There are two pool managers: plain poolmanager for no_proxy domains and | ||
# one proxymanager | ||
self.assert_pool_managers(env, [None, "http://localhost:8888"]) | ||
|
||
|
||
if __name__ == "__main__": | ||
utils.configure_test_logging(sys.argv) | ||
unittest.main() |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,101 @@ | ||
# Copyright New York University and the TUF contributors | ||
# SPDX-License-Identifier: MIT OR Apache-2.0 | ||
|
||
"""Proxy environment variable handling with Urllib3""" | ||
|
||
from __future__ import annotations | ||
|
||
from typing import Any | ||
from urllib.request import getproxies | ||
|
||
from urllib3 import BaseHTTPResponse, PoolManager, ProxyManager | ||
from urllib3.util.url import parse_url | ||
|
||
|
||
# TODO: ProxyEnvironment could implement the whole PoolManager.RequestMethods | ||
# Mixin: We only need request() so nothing else is currently implemented | ||
class ProxyEnvironment: | ||
"""A PoolManager manager for automatic proxy handling based on env variables | ||
|
||
Keeps track of PoolManagers for different proxy urls based on proxy | ||
environment variables. Use `get_pool_manager()` or `request()` to access | ||
the right manager for a scheme/host. | ||
|
||
Supports '*_proxy' variables, with special handling for 'no_proxy' and | ||
'all_proxy'. | ||
""" | ||
|
||
def __init__( | ||
self, | ||
**kw_args: Any, # noqa: ANN401 | ||
) -> None: | ||
self._pool_managers: dict[str | None, PoolManager] = {} | ||
self._kw_args = kw_args | ||
|
||
self._proxies = getproxies() | ||
self._all_proxy = self._proxies.pop("all", None) | ||
no_proxy = self._proxies.pop("no", None) | ||
if no_proxy is None: | ||
self._no_proxy_hosts = [] | ||
else: | ||
# split by comma, remove leading periods | ||
self._no_proxy_hosts = [ | ||
h.lstrip(".") for h in no_proxy.replace(" ", "").split(",") if h | ||
] | ||
|
||
def _get_proxy(self, scheme: str | None, host: str | None) -> str | None: | ||
"""Get a proxy url for scheme and host based on proxy env variables""" | ||
|
||
if host is None: | ||
# urllib3 only handles http/https but we can do something reasonable | ||
# even for schemes that don't require host (like file) | ||
return None | ||
|
||
# does host match any of the "no_proxy" hosts? | ||
for no_proxy_host in self._no_proxy_hosts: | ||
# wildcard match, exact hostname match, or parent domain match | ||
if no_proxy_host in ("*", host) or host.endswith( | ||
f".{no_proxy_host}" | ||
): | ||
return None | ||
|
||
if scheme in self._proxies: | ||
return self._proxies[scheme] | ||
if self._all_proxy is not None: | ||
return self._all_proxy | ||
|
||
return None | ||
|
||
def get_pool_manager( | ||
self, scheme: str | None, host: str | None | ||
) -> PoolManager: | ||
"""Get a poolmanager for scheme and host. | ||
|
||
Returns a ProxyManager if that is correct based on current proxy env | ||
variables, otherwise returns a PoolManager | ||
""" | ||
|
||
proxy = self._get_proxy(scheme, host) | ||
if proxy not in self._pool_managers: | ||
if proxy is None: | ||
self._pool_managers[proxy] = PoolManager(**self._kw_args) | ||
else: | ||
self._pool_managers[proxy] = ProxyManager( | ||
proxy, | ||
**self._kw_args, | ||
) | ||
|
||
return self._pool_managers[proxy] | ||
|
||
def request( | ||
self, | ||
method: str, | ||
url: str, | ||
**request_kw: Any, # noqa: ANN401 | ||
) -> BaseHTTPResponse: | ||
"""Make a request using a PoolManager chosen based on url and | ||
proxy environment variables. | ||
""" | ||
u = parse_url(url) | ||
manager = self.get_pool_manager(u.scheme, u.host) | ||
return manager.request(method, url, **request_kw) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
we could implement this method ourselves as well but I tried to keep this minimal