Properly parse proxy urls with credentials in them

cloudfoundry_client/doppler/client.py

@@ -19,16 +19,20 @@ class DopplerClient(object):
     def __init__(self, doppler_endpoint: str, proxy: str, verify_ssl: bool, credentials_manager: CredentialManager):
         self.proxy_host = None
         self.proxy_port = None
+        self.proxy_auth = None
         self.ws_doppler_endpoint = doppler_endpoint
         self.http_doppler_endpoint = re.sub("^ws", "http", doppler_endpoint)
         self.verify_ssl = verify_ssl
         self.credentials_manager = credentials_manager
         if proxy is not None and len(proxy) > 0:
-            proxy_domain = urlparse(proxy).netloc
-            idx = proxy_domain.find(":")
-            if 0 < idx < len(proxy_domain) - 2:
-                self.proxy_host = proxy_domain[:idx]
-                self.proxy_port = int(proxy_domain[idx + 1 :])
+            proxy_parsed = urlparse(proxy)
+            self.proxy_host = proxy_parsed.hostname
+            if proxy_parsed.port is not None:
+                self.proxy_port = proxy_parsed.port
+            else:
+                self.proxy_port = 443 if proxy_parsed.scheme == "https" else 80
+            if proxy_parsed.username is not None and proxy_parsed.password is not None:
+                self.proxy_auth = (proxy_parsed.username, proxy_parsed.password)
 
     def recent_logs(self, app_guid: str) -> EnvelopeStream:
         url = "%s/apps/%s/recentlogs" % (self.http_doppler_endpoint, app_guid)
@@ -46,6 +50,7 @@ def stream_logs(self, app_guid: str) -> EnvelopeStream:
             verify_ssl=self.verify_ssl,
             proxy_host=self.proxy_host,
             proxy_port=self.proxy_port,
+            proxy_auth=self.proxy_auth,
         ) as websocket:
             for message in websocket:
                 yield DopplerClient._parse_envelope(message)

cloudfoundry_client/doppler/websocket_envelope_reader.py

@@ -1,5 +1,5 @@
 import ssl
-from typing import Callable, Optional
+from typing import Callable, Optional, Tuple
 
 import websocket
 
@@ -12,6 +12,7 @@ def __init__(
         verify_ssl: bool = True,
         proxy_host: Optional[str] = None,
         proxy_port: Optional[int] = None,
+        proxy_auth: Optional[Tuple[str, str]] = None,
     ):
         if not verify_ssl:
             self._ws = websocket.WebSocket(sslopt=dict(cert_reqs=ssl.CERT_NONE))
@@ -20,13 +21,15 @@ def __init__(
         self._url = url
         self._proxy_host = proxy_host
         self._proxy_port = proxy_port
+        self._proxy_auth = proxy_auth
         self._access_token_provider = access_token_provider
 
     def connect(self):
         kw_args = dict(header=dict(Authorization="Bearer %s" % self._access_token_provider()))
         if self._proxy_host is not None and self._proxy_port is not None:
             kw_args["http_proxy_host"] = self._proxy_host
             kw_args["http_proxy_port"] = str(self._proxy_port)
+            kw_args["http_proxy_auth"] = self._proxy_auth
         self._ws.connect(self._url, **kw_args)
 
     def close(self):

cloudfoundry_client/rlpgateway/client.py

@@ -1,5 +1,4 @@
 import logging
-from urllib.parse import urlparse
 
 import aiohttp
 
@@ -15,18 +14,13 @@ class RLPGatewayClient(object):
     """
 
     def __init__(self, rlp_gateway_endpoint, proxy, verify_ssl, credentials_manager):
-        self.proxy_host = None
-        self.proxy_port = None
+        self.proxy = None
         self.rlp_gateway_endpoint = rlp_gateway_endpoint
         self.verify_ssl = verify_ssl
         self.credentials_manager = credentials_manager
 
         if proxy is not None and len(proxy) > 0:
-            proxy_domain = urlparse(proxy).netloc
-            idx = proxy_domain.find(":")
-            if 0 < idx < len(proxy_domain) - 2:
-                self.proxy_host = proxy_domain[:idx]
-                self.proxy_port = int(proxy_domain[idx + 1 :])
+            self.proxy = proxy
 
     async def stream_logs(self, app_guid, **kwargs):
         url = f"{self.rlp_gateway_endpoint}/v2/read"
@@ -40,7 +34,7 @@ async def stream_logs(self, app_guid, **kwargs):
             headers.update(kwargs["headers"])
         if "params" in kwargs:
             params.update(kwargs["params"])
-        async with aiohttp.ClientSession(headers=headers) as session:
+        async with aiohttp.ClientSession(headers=headers, proxy=self.proxy) as session:
             async with session.get(url=url, params=params) as response:
                 if response.status == 204:
                     yield {}