diff --git a/src/meshctrl/files.py b/src/meshctrl/files.py index 3c386df..b8958ac 100644 --- a/src/meshctrl/files.py +++ b/src/meshctrl/files.py @@ -4,9 +4,27 @@ from . import exceptions from . import util import asyncio import json -import urllib +import importlib +import importlib.util import shutil +# import urllib +# import urllib.request +import urllib.parse +old_parse = urllib.parse +# Default proxy handler uses OS defined no_proxy in order to be helpful. This is unhelpful for our usecase. Monkey patch out proxy getting functions, but don't effect the user's urllib instance. +spec = importlib.util.find_spec('urllib') +urllib = importlib.util.module_from_spec(spec) +spec.loader.exec_module(urllib) +spec = importlib.util.find_spec('urllib.request') +urllib.request = importlib.util.module_from_spec(spec) +spec.loader.exec_module(urllib.request) +urllib.parse = old_parse +urllib.request.getproxies_environment = lambda: {} +urllib.request.getproxies_registry = lambda: {} +urllib.request.getproxies_macosx_sysconf = lambda: {} +urllib.request.getproxies = lambda: {} + class Files(tunnel.Tunnel): def __init__(self, session, node): super().__init__(session, node.nodeid, constants.Protocol.FILES) @@ -23,8 +41,9 @@ class Files(tunnel.Tunnel): if self._session._proxy is not None: # We don't know which protocol the user is going to use, but we only need support one at a time, so just assume both proxies = { - "http_proxy": self._session._proxy, - "https_proxy": self._session._proxy + "http": self._session._proxy, + "https": self._session._proxy, + "no": "" } self._proxy_handler = urllib.request.ProxyHandler(proxies=proxies) self._http_opener = urllib.request.build_opener(self._proxy_handler, urllib.request.HTTPSHandler(context=self._session._ssl_context)) diff --git a/tests/test_files.py b/tests/test_files.py index ed888f9..e6d760c 100644 --- a/tests/test_files.py +++ b/tests/test_files.py @@ -53,6 +53,18 @@ async def test_commands(env): finally: assert (await admin_session.remove_device_group(mesh.meshid, timeout=10)), "Failed to remove device group" +async def test_os_proxy_bypass(): + os.environ["no_proxy"] = "*" + import urllib + import urllib.request + os_proxies = urllib.request.getproxies() + meshctrl_proxies = meshctrl.files.urllib.request.getproxies() + print(f"os_proxies: {os_proxies}") + print(f"meshctrl_proxies: {meshctrl_proxies}") + assert meshctrl_proxies.get("no", None) == None, "Meshctrl is using system proxies" + assert os_proxies.get("no", None) == "*", "System is using meshctrl proxies" + assert os_proxies != meshctrl_proxies, "Override didn't work" + async def test_upload_download(env): async with meshctrl.Session("wss://" + env.dockerurl, user="admin", password=env.users["admin"], ignore_ssl=True, proxy=env.proxyurl) as admin_session: mesh = await admin_session.add_device_group("test", description="This is a test group", amtonly=False, features=0, consent=0, timeout=10)