Added no_proxy os variable bypass so urllib.requests acts as expected

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,
+                "http": self._session._proxy,
-                "https_proxy": 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))

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)