Add tests and function cleanup

.changes/next-release/enhancement-Exceptions-94766.json

@@ -0,0 +1,5 @@
+{
+  "type": "enhancement",
+  "category": "Exceptions",
+  "description": "ProxyConnectionError previously provided the full proxy URL. User info will now be appropriately masked if needed."
+}

botocore/httpsession.py

@@ -110,19 +110,20 @@ def ensure_boolean(val):
 
 def mask_proxy_url(proxy_url):
     """
-    Mask proxy url credentials
+    Mask proxy url credentials.
 
     :type proxy_url: str
     :param proxy_url: The proxy url, i.e. https://username:password@proxy.com
 
-    :return: Masked proxy url
+    :return: Masked proxy url, i.e. https://***:***@proxy.com
     """
-    mask = '*' * 6
+    mask = '*' * 3
     parsed_url = urlparse(proxy_url)
-    if parsed_url.username and parsed_url.password:
-        return proxy_url.replace(parsed_url.username, mask, 1).replace(parsed_url.password, mask, 1)
-    else:
-        return proxy_url
+    if parsed_url.username:
+        proxy_url = proxy_url.replace(parsed_url.username, mask, 1)
+    if parsed_url.password:
+        proxy_url = proxy_url.replace(parsed_url.password, mask, 1)
+    return proxy_url
 
 
 class ProxyConfiguration(object):

tests/unit/test_http_session.py

@@ -1,15 +1,30 @@
 import socket
 
 import pytest
-from urllib3.exceptions import NewConnectionError, ProtocolError
+from urllib3.exceptions import (
+    NewConnectionError,
+    ProtocolError,
+    ProxyError,
+)
 
 from tests import mock, unittest
 
-from botocore.awsrequest import AWSRequest
-from botocore.awsrequest import AWSHTTPConnectionPool, AWSHTTPSConnectionPool
-from botocore.httpsession import get_cert_path
-from botocore.httpsession import URLLib3Session, ProxyConfiguration
-from botocore.exceptions import ConnectionClosedError, EndpointConnectionError
+from botocore.awsrequest import (
+    AWSRequest,
+    AWSHTTPConnectionPool,
+    AWSHTTPSConnectionPool,
+)
+from botocore.httpsession import (
+    get_cert_path,
+    mask_proxy_url,
+    URLLib3Session,
+    ProxyConfiguration,
+)
+from botocore.exceptions import (
+    ConnectionClosedError,
+    EndpointConnectionError,
+    ProxyConnectionError,
+)
 
 
 class TestProxyConfiguration(unittest.TestCase):
@@ -63,6 +78,56 @@ def test_get_cert_path_certifi_or_default(self):
             self.assertEqual(path, cert_path)
 
 
+@pytest.mark.parametrize(
+    'proxy_url, expected_mask_url',
+    (
+
+        (
+            'http://myproxy.amazonaws.com',
+            'http://myproxy.amazonaws.com'
+        ),
+        (
+            'http://user@myproxy.amazonaws.com',
+            'http://***@myproxy.amazonaws.com'
+        ),
+        (
+            'http://user:pass@myproxy.amazonaws.com',
+            'http://***:***@myproxy.amazonaws.com'
+        ),
+        (
+            'https://user:pass@myproxy.amazonaws.com',
+            'https://***:***@myproxy.amazonaws.com'
+        ),
+        (
+            'http://user:pass@localhost',
+            'http://***:***@localhost'
+        ),
+        (
+            'http://user:pass@localhost:80',
+            'http://***:***@localhost:80'
+        ),
+        (
+            'http://user:pass@userpass.com',
+            'http://***:***@userpass.com'
+        ),
+        (
+            'http://user:pass@192.168.1.1',
+            'http://***:***@192.168.1.1'
+        ),
+        (
+            'http://user:pass@[::1]',
+            'http://***:***@[::1]'
+        ),
+        (
+            'http://user:pass@[::1]:80',
+            'http://***:***@[::1]:80'
+        ),
+    )
+)
+def test_mask_proxy_url(proxy_url, expected_mask_url):
+    assert mask_proxy_url(proxy_url) == expected_mask_url
+
+
 class TestURLLib3Session(unittest.TestCase):
     def setUp(self):
         self.request = AWSRequest(
@@ -399,6 +464,14 @@ def test_catches_bad_status_line(self):
         with pytest.raises(ConnectionClosedError):
             self.make_request_with_error(error)
 
+    def test_catches_proxy_error(self):
+        self.connection.urlopen.side_effect = ProxyError('test', None)
+        session = URLLib3Session(proxies={'http': 'http://user:pass@proxy.com'})
+        with pytest.raises(ProxyConnectionError) as e:
+            session.send(self.request.prepare())
+        assert 'user:pass' not in str(e.value)
+        assert 'http://***:***@proxy.com' in str(e.value)
+
     def test_aws_connection_classes_are_used(self):
         session = URLLib3Session() # noqa
         # ensure the pool manager is using the correct classes