Implementation to honor no_proxy environment variable (#1098)

dulwich/client.py

@@ -71,7 +71,6 @@ from urllib.parse import (
 if TYPE_CHECKING:
     import urllib3
 
-
 import dulwich
 from dulwich.config import get_xdg_config_home_path, Config, apply_instead_of
 from dulwich.errors import (
@@ -1811,7 +1810,7 @@ def default_user_agent_string():
 
 
 def default_urllib3_manager(   # noqa: C901
-    config, pool_manager_cls=None, proxy_manager_cls=None, **override_kwargs
+    config, pool_manager_cls=None, proxy_manager_cls=None, base_url=None, **override_kwargs
 ) -> Union["urllib3.ProxyManager", "urllib3.PoolManager"]:
     """Return urllib3 connection pool manager.
 
@@ -1836,6 +1835,10 @@ def default_urllib3_manager(   # noqa: C901
             if proxy_server:
                 break
 
+    if proxy_server:
+        if check_for_proxy_bypass(base_url):
+            proxy_server = None
+    
     if config is not None:
         if proxy_server is None:
             try:
@@ -1892,6 +1895,54 @@ def default_urllib3_manager(   # noqa: C901
     return manager
 
 
+def check_for_proxy_bypass(base_url):
+    # Check if a proxy bypass is defined with the no_proxy environment variable
+    if base_url:  # only check if base_url is provided
+        no_proxy_str = os.environ.get("no_proxy")
+        if no_proxy_str:
+            # implementation based on curl behavior: https://curl.se/libcurl/c/CURLOPT_NOPROXY.html
+            # get hostname of provided parsed url
+            parsed_url = urlparse(base_url)
+            hostname = parsed_url.hostname
+
+            if hostname:
+                import ipaddress
+
+                # check if hostname is an ip address
+                try:
+                    hostname_ip = ipaddress.ip_address(hostname)
+                except ValueError:
+                    hostname_ip = None
+
+                no_proxy_values = no_proxy_str.split(',')
+                for no_proxy_value in no_proxy_values:
+                    no_proxy_value = no_proxy_value.strip()
+                    if no_proxy_value:
+                        no_proxy_value = no_proxy_value.lower()
+                        no_proxy_value = no_proxy_value.lstrip('.')  # ignore leading dots
+
+                        if hostname_ip:
+                            # check if no_proxy_value is a ip network
+                            try:
+                                no_proxy_value_network = ipaddress.ip_network(no_proxy_value, strict=False)
+                            except ValueError:
+                                no_proxy_value_network = None
+                            if no_proxy_value_network:
+                                # if hostname is a ip address and no_proxy_value is a ip network -> check if ip address is part of network
+                                if hostname_ip in no_proxy_value_network:
+                                    return True
+                                
+                        if no_proxy_value == '*':
+                            # '*' is special case for always bypass proxy
+                            return True
+                        if hostname == no_proxy_value:
+                            return True
+                        no_proxy_value = '.' + no_proxy_value   # add a dot to only match complete domains
+                        if hostname.endswith(no_proxy_value):
+                            return True
+    return False
+
+
 class AbstractHttpGitClient(GitClient):
     """Abstract base class for HTTP Git Clients.
 
@@ -2173,7 +2224,7 @@ class Urllib3HttpGitClient(AbstractHttpGitClient):
         self._password = password
 
         if pool_manager is None:
-            self.pool_manager = default_urllib3_manager(config)
+            self.pool_manager = default_urllib3_manager(config, base_url=base_url)
         else:
             self.pool_manager = pool_manager
 

dulwich/tests/test_client.py

@@ -1248,6 +1248,168 @@ class DefaultUrllib3ManagerTest(TestCase):
         self.assertIsInstance(manager, urllib3.PoolManager)
         del os.environ["http_proxy"]
 
+    def test_environment_no_proxy_1(self):
+        import urllib3
+
+        config = ConfigDict()
+        os.environ["http_proxy"] = "http://myproxy:8080"
+        os.environ["no_proxy"] = "xyz,abc.def.gh,abc.gh"
+        base_url = "http://xyz.abc.def.gh:8080/path/port"
+        manager = default_urllib3_manager(config=config, base_url=base_url)
+        self.assertNotIsInstance(manager, urllib3.ProxyManager)
+        self.assertIsInstance(manager, urllib3.PoolManager)
+        del os.environ["http_proxy"]
+        del os.environ["no_proxy"]
+
+    def test_environment_no_proxy_2(self):
+        import urllib3
+
+        config = ConfigDict()
+        os.environ["http_proxy"] = "http://myproxy:8080"
+        os.environ["no_proxy"] = "xyz,abc.def.gh,abc.gh,ample.com"
+        base_url = "http://ample.com/path/port"
+        manager = default_urllib3_manager(config=config, base_url=base_url)
+        self.assertNotIsInstance(manager, urllib3.ProxyManager)
+        self.assertIsInstance(manager, urllib3.PoolManager)
+        del os.environ["http_proxy"]
+        del os.environ["no_proxy"]
+
+    def test_environment_no_proxy_3(self):
+        import urllib3
+
+        config = ConfigDict()
+        os.environ["http_proxy"] = "http://myproxy:8080"
+        os.environ["no_proxy"] = "xyz,abc.def.gh,abc.gh,ample.com"
+        base_url = "http://ample.com:80/path/port"
+        manager = default_urllib3_manager(config=config, base_url=base_url)
+        self.assertNotIsInstance(manager, urllib3.ProxyManager)
+        self.assertIsInstance(manager, urllib3.PoolManager)
+        del os.environ["http_proxy"]
+        del os.environ["no_proxy"]
+
+    def test_environment_no_proxy_4(self):
+        import urllib3
+
+        config = ConfigDict()
+        os.environ["http_proxy"] = "http://myproxy:8080"
+        os.environ["no_proxy"] = "xyz,abc.def.gh,abc.gh,ample.com"
+        base_url = "http://www.ample.com/path/port"
+        manager = default_urllib3_manager(config=config, base_url=base_url)
+        self.assertNotIsInstance(manager, urllib3.ProxyManager)
+        self.assertIsInstance(manager, urllib3.PoolManager)
+        del os.environ["http_proxy"]
+        del os.environ["no_proxy"]
+
+    def test_environment_no_proxy_5(self):
+        import urllib3
+
+        config = ConfigDict()
+        os.environ["http_proxy"] = "http://myproxy:8080"
+        os.environ["no_proxy"] = "xyz,abc.def.gh,abc.gh,ample.com"
+        base_url = "http://www.example.com/path/port"
+        manager = default_urllib3_manager(config=config, base_url=base_url)
+        self.assertIsInstance(manager, urllib3.ProxyManager)
+        self.assertTrue(hasattr(manager, "proxy"))
+        self.assertEqual(manager.proxy.scheme, "http")
+        self.assertEqual(manager.proxy.host, "myproxy")
+        self.assertEqual(manager.proxy.port, 8080)
+        del os.environ["http_proxy"]
+        del os.environ["no_proxy"]
+
+    def test_environment_no_proxy_6(self):
+        import urllib3
+
+        config = ConfigDict()
+        os.environ["http_proxy"] = "http://myproxy:8080"
+        os.environ["no_proxy"] = "xyz,abc.def.gh,abc.gh,ample.com"
+        base_url = "http://ample.com.org/path/port"
+        manager = default_urllib3_manager(config=config, base_url=base_url)
+        self.assertIsInstance(manager, urllib3.ProxyManager)
+        self.assertTrue(hasattr(manager, "proxy"))
+        self.assertEqual(manager.proxy.scheme, "http")
+        self.assertEqual(manager.proxy.host, "myproxy")
+        self.assertEqual(manager.proxy.port, 8080)
+        del os.environ["http_proxy"]
+        del os.environ["no_proxy"]
+
+    def test_environment_no_proxy_ipv4_address_1(self):
+        import urllib3
+
+        config = ConfigDict()
+        os.environ["http_proxy"] = "http://myproxy:8080"
+        os.environ["no_proxy"] = "xyz,abc.def.gh,192.168.0.10,ample.com"
+        base_url = "http://192.168.0.10/path/port"
+        manager = default_urllib3_manager(config=config, base_url=base_url)
+        self.assertNotIsInstance(manager, urllib3.ProxyManager)
+        self.assertIsInstance(manager, urllib3.PoolManager)
+        del os.environ["http_proxy"]
+        del os.environ["no_proxy"]
+
+    def test_environment_no_proxy_ipv4_address_2(self):
+        import urllib3
+
+        config = ConfigDict()
+        os.environ["http_proxy"] = "http://myproxy:8080"
+        os.environ["no_proxy"] = "xyz,abc.def.gh,192.168.0.10,ample.com"
+        base_url = "http://192.168.0.10:8888/path/port"
+        manager = default_urllib3_manager(config=config, base_url=base_url)
+        self.assertNotIsInstance(manager, urllib3.ProxyManager)
+        self.assertIsInstance(manager, urllib3.PoolManager)
+        del os.environ["http_proxy"]
+        del os.environ["no_proxy"]
+    
+    def test_environment_no_proxy_ipv4_address_3(self):
+        import urllib3
+
+        config = ConfigDict()
+        os.environ["http_proxy"] = "http://myproxy:8080"
+        os.environ["no_proxy"] = "xyz,abc.def.gh,ff80:1::/64,192.168.0.0/24,ample.com"
+        base_url = "http://192.168.0.10/path/port"
+        manager = default_urllib3_manager(config=config, base_url=base_url)
+        self.assertNotIsInstance(manager, urllib3.ProxyManager)
+        self.assertIsInstance(manager, urllib3.PoolManager)
+        del os.environ["http_proxy"]
+        del os.environ["no_proxy"]
+
+    def test_environment_no_proxy_ipv6_address_1(self):
+        import urllib3
+
+        config = ConfigDict()
+        os.environ["http_proxy"] = "http://myproxy:8080"
+        os.environ["no_proxy"] = "xyz,abc.def.gh,ff80:1::affe,ample.com"
+        base_url = "http://[ff80:1::affe]/path/port"
+        manager = default_urllib3_manager(config=config, base_url=base_url)
+        self.assertNotIsInstance(manager, urllib3.ProxyManager)
+        self.assertIsInstance(manager, urllib3.PoolManager)
+        del os.environ["http_proxy"]
+        del os.environ["no_proxy"]
+
+    def test_environment_no_proxy_ipv6_address_2(self):
+        import urllib3
+
+        config = ConfigDict()
+        os.environ["http_proxy"] = "http://myproxy:8080"
+        os.environ["no_proxy"] = "xyz,abc.def.gh,ff80:1::affe,ample.com"
+        base_url = "http://[ff80:1::affe]:1234/path/port"
+        manager = default_urllib3_manager(config=config, base_url=base_url)
+        self.assertNotIsInstance(manager, urllib3.ProxyManager)
+        self.assertIsInstance(manager, urllib3.PoolManager)
+        del os.environ["http_proxy"]
+        del os.environ["no_proxy"]
+
+    def test_environment_no_proxy_ipv6_address_3(self):
+        import urllib3
+
+        config = ConfigDict()
+        os.environ["http_proxy"] = "http://myproxy:8080"
+        os.environ["no_proxy"] = "xyz,abc.def.gh,192.168.0.0/24,ff80:1::/64,ample.com"
+        base_url = "http://[ff80:1::affe]/path/port"
+        manager = default_urllib3_manager(config=config, base_url=base_url)
+        self.assertNotIsInstance(manager, urllib3.ProxyManager)
+        self.assertIsInstance(manager, urllib3.PoolManager)
+        del os.environ["http_proxy"]
+        del os.environ["no_proxy"]
+
     def test_config_proxy_custom_cls(self):
         import urllib3