From 1feefcf86a858f4e9fc08cd28d712061fe5eb6e9 Mon Sep 17 00:00:00 2001 From: Kashif Khan Date: Wed, 14 Sep 2022 18:57:01 -0500 Subject: [PATCH 01/11] init changes for aiohttp ws --- .../eventhub/_pyamqp/aio/_transport_async.py | 40 ++++++++++--------- 1 file changed, 21 insertions(+), 19 deletions(-) diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py index 33fe7e812315..327151312559 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py @@ -427,6 +427,7 @@ def __init__(self, host, port=WEBSOCKET_PORT, connect_timeout=None, ssl=None, ** self._custom_endpoint = kwargs.get("custom_endpoint") self.host = host self.ws = None + self.session = None self._http_proxy = kwargs.get('http_proxy', None) async def connect(self): @@ -439,23 +440,27 @@ async def connect(self): if username or password: http_proxy_auth = (username, password) try: - from websocket import create_connection - self.ws = create_connection( - url="wss://{}".format(self._custom_endpoint or self.host), - subprotocols=[AMQP_WS_SUBPROTOCOL], + from aiohttp import ClientSession + self.session = ClientSession() + + if username or password: + from aiohttp import BasicAuth + http_proxy_auth = BasicAuth(login=username,password=password) + + self.ws = self.session.ws_connect( + url = "wss://{}".format(self._custom_endpoint or self.host), + protocols=[AMQP_WS_SUBPROTOCOL], timeout=self._connect_timeout, - skip_utf8_validation=True, - sslopt=self.sslopts, - http_proxy_host=http_proxy_host, - http_proxy_port=http_proxy_port, - http_proxy_auth=http_proxy_auth - ) + proxy=http_proxy_host + f":{http_proxy_port}" if http_proxy_port else "" + proxy_auth=http_proxy_auth, + #figure out ssl + ) except ImportError: - raise ValueError("Please install websocket-client library to use websocket transport.") + raise ValueError("Please install aiohttp library to use websocket transport.") async def _read(self, n, buffer=None, **kwargs): # pylint: disable=unused-arguments """Read exactly n bytes from the peer.""" - from websocket import WebSocketTimeoutException + from aiohttp import length = 0 view = buffer or memoryview(bytearray(n)) @@ -464,9 +469,7 @@ async def _read(self, n, buffer=None, **kwargs): # pylint: disable=unused-argume n -= nbytes try: while n: - data = await self.loop.run_in_executor( - None, self.ws.recv - ) + data = await self.ws.receive_bytes() if len(data) <= n: view[length: length + len(data)] = data @@ -477,12 +480,13 @@ async def _read(self, n, buffer=None, **kwargs): # pylint: disable=unused-argume n = 0 return view - except WebSocketTimeoutException as wex: + except asyncio.TimeoutError as wex: raise TimeoutError() def close(self): """Do any preliminary work in shutting down the connection.""" self.ws.close() + self.session.close() self.connected = False async def write(self, s): @@ -491,6 +495,4 @@ async def write(self, s): See http://tools.ietf.org/html/rfc5234 http://tools.ietf.org/html/rfc6455#section-5.2 """ - await self.loop.run_in_executor( - None, self.ws.send_binary, s - ) + await self.ws.send_bytes(s) From 42f9cec69a3cdf44195e3a50e428e216d9bf1e50 Mon Sep 17 00:00:00 2001 From: Kashif Khan Date: Thu, 15 Sep 2022 08:49:33 -0500 Subject: [PATCH 02/11] fixes + formatting --- .../eventhub/_pyamqp/aio/_transport_async.py | 183 +++++++++++------- 1 file changed, 118 insertions(+), 65 deletions(-) diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py index 327151312559..2d6438ad3491 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py @@ -1,4 +1,4 @@ -#------------------------------------------------------------------------- +# ------------------------------------------------------------------------- # This is a fork of the transport.py which was originally written by Barry Pederson and # maintained by the Celery project: https://github.com/celery/py-amqp. # @@ -30,7 +30,7 @@ # CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF # THE POSSIBILITY OF SUCH DAMAGE. -#------------------------------------------------------------------------- +# ------------------------------------------------------------------------- import asyncio import errno @@ -38,11 +38,12 @@ import socket import ssl import struct -from ssl import SSLError +from ssl import SSLContext, SSLError from contextlib import contextmanager from io import BytesIO import logging from threading import Lock +from typing import Optional import certifi @@ -61,7 +62,7 @@ set_cloexec, AMQP_PORT, TIMEOUT_INTERVAL, - WebSocketTransport + WebSocketTransport, ) @@ -71,20 +72,21 @@ def get_running_loop(): try: import asyncio # pylint: disable=import-error + return asyncio.get_running_loop() except AttributeError: # 3.6 loop = None try: loop = asyncio._get_running_loop() # pylint: disable=protected-access except AttributeError: - _LOGGER.warning('This version of Python is deprecated, please upgrade to >= v3.6') + _LOGGER.warning("This version of Python is deprecated, please upgrade to >= v3.6") if loop is None: - _LOGGER.warning('No running event loop') + _LOGGER.warning("No running event loop") loop = asyncio.get_event_loop() return loop -class AsyncTransportMixin(): +class AsyncTransportMixin: async def receive_frame(self, timeout=None, *args, **kwargs): try: header, channel, payload = await asyncio.wait_for(self.read(**kwargs), timeout=timeout) @@ -104,11 +106,11 @@ async def read(self, verify_frame_type=0, **kwargs): frame_header = memoryview(bytearray(8)) read_frame_buffer.write(await self._read(8, buffer=frame_header, initial=True)) - channel = struct.unpack('>H', frame_header[6:])[0] + channel = struct.unpack(">H", frame_header[6:])[0] size = frame_header[0:4] if size == AMQP_FRAME: # Empty frame or AMQP header negotiation return frame_header, channel, None - size = struct.unpack('>I', size)[0] + size = struct.unpack(">I", size)[0] offset = frame_header[4] frame_type = frame_header[5] @@ -121,7 +123,7 @@ async def read(self, verify_frame_type=0, **kwargs): read_frame_buffer.write(await self._read(size - SIGNED_INT_MAX, buffer=payload[SIGNED_INT_MAX:])) else: read_frame_buffer.write(await self._read(payload_size, buffer=payload)) - except (TimeoutError, socket.timeout, asyncio.IncompleteReadError): + except (TimeoutError, socket.timeout, asyncio.IncompleteReadError): read_frame_buffer.write(self._read_buffer.getvalue()) self._read_buffer = read_frame_buffer self._read_buffer.seek(0) @@ -129,7 +131,7 @@ async def read(self, verify_frame_type=0, **kwargs): except (OSError, IOError, SSLError, socket.error) as exc: # Don't disconnect for ssl read time outs # http://bugs.python.org/issue10272 - if isinstance(exc, SSLError) and 'timed out' in str(exc): + if isinstance(exc, SSLError) and "timed out" in str(exc): raise socket.timeout() if get_errno(exc) not in _UNAVAIL: self.connected = False @@ -142,18 +144,28 @@ async def send_frame(self, channel, frame, **kwargs): if performative is None: data = header else: - encoded_channel = struct.pack('>H', channel) + encoded_channel = struct.pack(">H", channel) data = header + encoded_channel + performative await self.write(data) - #_LOGGER.info("OCH%d -> %r", channel, frame) + # _LOGGER.info("OCH%d -> %r", channel, frame) + class AsyncTransport(AsyncTransportMixin): """Common superclass for TCP and SSL transports.""" - def __init__(self, host, port=AMQP_PORT, connect_timeout=None, - read_timeout=None, write_timeout=None, ssl=False, - socket_settings=None, raise_on_initial_eintr=True, **kwargs): + def __init__( + self, + host, + port=AMQP_PORT, + connect_timeout=None, + read_timeout=None, + write_timeout=None, + ssl=False, + socket_settings=None, + raise_on_initial_eintr=True, + **kwargs, + ): self.connected = False self.sock = None self.reader = None @@ -161,7 +173,7 @@ def __init__(self, host, port=AMQP_PORT, connect_timeout=None, self.raise_on_initial_eintr = raise_on_initial_eintr self._read_buffer = BytesIO() self.host, self.port = to_host_port(host, port) - + self.connect_timeout = connect_timeout self.read_timeout = read_timeout self.write_timeout = write_timeout @@ -174,19 +186,23 @@ def _build_ssl_opts(self, sslopts): if sslopts in [True, False, None, {}]: return sslopts try: - if 'context' in sslopts: - return self._build_ssl_context(sslopts, **sslopts.pop('context')) - ssl_version = sslopts.get('ssl_version') + if "context" in sslopts: + return self._build_ssl_context(sslopts, **sslopts.pop("context")) + ssl_version = sslopts.get("ssl_version") if ssl_version is None: ssl_version = ssl.PROTOCOL_TLS # Set SNI headers if supported - server_hostname = sslopts.get('server_hostname') - if (server_hostname is not None) and (hasattr(ssl, 'HAS_SNI') and ssl.HAS_SNI) and (hasattr(ssl, 'SSLContext')): + server_hostname = sslopts.get("server_hostname") + if ( + (server_hostname is not None) + and (hasattr(ssl, "HAS_SNI") and ssl.HAS_SNI) + and (hasattr(ssl, "SSLContext")) + ): context = ssl.SSLContext(ssl_version) - cert_reqs = sslopts.get('cert_reqs', ssl.CERT_REQUIRED) - certfile = sslopts.get('certfile') - keyfile = sslopts.get('keyfile') + cert_reqs = sslopts.get("cert_reqs", ssl.CERT_REQUIRED) + certfile = sslopts.get("certfile") + keyfile = sslopts.get("keyfile") context.verify_mode = cert_reqs if cert_reqs != ssl.CERT_NONE: context.check_hostname = True @@ -195,7 +211,7 @@ def _build_ssl_opts(self, sslopts): return context return True except TypeError: - raise TypeError('SSL configuration must be a dictionary, or the value True.') + raise TypeError("SSL configuration must be a dictionary, or the value True.") def _build_ssl_context(self, sslopts, check_hostname=None, **ctx_options): ctx = ssl.create_default_context(**ctx_options) @@ -211,12 +227,12 @@ async def connect(self): return await self._connect(self.host, self.port, self.connect_timeout) self._init_socket( - self.socket_settings, self.read_timeout, self.write_timeout, + self.socket_settings, + self.read_timeout, + self.write_timeout, ) self.reader, self.writer = await asyncio.open_connection( - sock=self.sock, - ssl=self.sslopts, - server_hostname=self.host if self.sslopts else None + sock=self.sock, ssl=self.sslopts, server_hostname=self.host if self.sslopts else None ) # we've sent the banner; signal connect # EINTR, EAGAIN, EWOULDBLOCK would signal that the banner @@ -245,8 +261,7 @@ async def _connect(self, host, port, timeout): for n, family in enumerate(addr_types): # first, resolve the address for a single address family try: - entries = await self.loop.getaddrinfo( - host, port, family=family, type=socket.SOCK_STREAM, proto=SOL_TCP) + entries = await self.loop.getaddrinfo(host, port, family=family, type=socket.SOCK_STREAM, proto=SOL_TCP) entries_num = len(entries) except socket.gaierror: # we may have depleted all our options @@ -254,10 +269,7 @@ async def _connect(self, host, port, timeout): # if getaddrinfo succeeded before for another address # family, reraise the previous socket.error since it's more # relevant to users - raise (e - if e is not None - else socket.error( - "failed to resolve broker hostname")) + raise (e if e is not None else socket.error("failed to resolve broker hostname")) continue # pragma: no cover # now that we have address(es) for the hostname, connect to broker @@ -305,7 +317,7 @@ def _get_tcp_socket_defaults(self, sock): tcp_opts = {} for opt in KNOWN_TCP_OPTS: enum = None - if opt == 'TCP_USER_TIMEOUT': + if opt == "TCP_USER_TIMEOUT": try: from socket import TCP_USER_TIMEOUT as enum except ImportError: @@ -318,8 +330,7 @@ def _get_tcp_socket_defaults(self, sock): if opt in DEFAULT_SOCKET_SETTINGS: tcp_opts[enum] = DEFAULT_SOCKET_SETTINGS[opt] elif hasattr(socket, opt): - tcp_opts[enum] = sock.getsockopt( - SOL_TCP, getattr(socket, opt)) + tcp_opts[enum] = sock.getsockopt(SOL_TCP, getattr(socket, opt)) return tcp_opts def _set_socket_options(self, socket_settings): @@ -329,8 +340,7 @@ def _set_socket_options(self, socket_settings): for opt, val in tcp_opts.items(): self.sock.setsockopt(SOL_TCP, opt, val) - async def _read(self, toread, initial=False, buffer=None, - _errnos=(errno.ENOENT, errno.EAGAIN, errno.EINTR)): + async def _read(self, toread, initial=False, buffer=None, _errnos=(errno.ENOENT, errno.EAGAIN, errno.EINTR)): # According to SSL_read(3), it can at most return 16kb of data. # Thus, we use an internal read buffer like TCPTransport._read # to get the exact number of bytes wanted. @@ -342,16 +352,16 @@ async def _read(self, toread, initial=False, buffer=None, try: while toread: try: - view[nbytes:nbytes + toread] = await self.reader.readexactly(toread) + view[nbytes : nbytes + toread] = await self.reader.readexactly(toread) nbytes = toread except asyncio.IncompleteReadError as exc: pbytes = len(exc.partial) - view[nbytes:nbytes + pbytes] = exc.partial + view[nbytes : nbytes + pbytes] = exc.partial nbytes = pbytes except socket.error as exc: # ssl.sock.read may cause a SSLerror without errno # http://bugs.python.org/issue10272 - if isinstance(exc, SSLError) and 'timed out' in str(exc): + if isinstance(exc, SSLError) and "timed out" in str(exc): raise socket.timeout() # ssl.sock.read may cause ENOENT if the # operation couldn't be performed (Issue celery#1414). @@ -361,7 +371,7 @@ async def _read(self, toread, initial=False, buffer=None, continue raise if not nbytes: - raise IOError('Server unexpectedly closed connection') + raise IOError("Server unexpectedly closed connection") length += nbytes toread -= nbytes @@ -412,13 +422,15 @@ async def negotiate(self): await self.write(TLS_HEADER_FRAME) channel, returned_header = await self.receive_frame(verify_frame_type=None) if returned_header[1] == TLS_HEADER_FRAME: - raise ValueError("Mismatching TLS header protocol. Excpected: {}, received: {}".format( - TLS_HEADER_FRAME, returned_header[1])) + raise ValueError( + "Mismatching TLS header protocol. Excpected: {}, received: {}".format( + TLS_HEADER_FRAME, returned_header[1] + ) + ) class WebSocketTransportAsync(AsyncTransportMixin): - def __init__(self, host, port=WEBSOCKET_PORT, connect_timeout=None, ssl=None, **kwargs - ): + def __init__(self, host, port=WEBSOCKET_PORT, connect_timeout=None, ssl=None, **kwargs): self._read_buffer = BytesIO() self.loop = get_running_loop() self.socket_lock = asyncio.Lock() @@ -428,39 +440,80 @@ def __init__(self, host, port=WEBSOCKET_PORT, connect_timeout=None, ssl=None, ** self.host = host self.ws = None self.session = None - self._http_proxy = kwargs.get('http_proxy', None) + self._http_proxy = kwargs.get("http_proxy", None) async def connect(self): http_proxy_host, http_proxy_port, http_proxy_auth = None, None, None + ssl_opts: Optional[SSLContext] = self._build_ssl_opts(self.sslopts) if len(self.ssl_opts) > 0 else None + if self._http_proxy: - http_proxy_host = self._http_proxy['proxy_hostname'] - http_proxy_port = self._http_proxy['proxy_port'] - username = self._http_proxy.get('username', None) - password = self._http_proxy.get('password', None) + http_proxy_host = self._http_proxy["proxy_hostname"] + http_proxy_port = self._http_proxy["proxy_port"] + username = self._http_proxy.get("username", None) + password = self._http_proxy.get("password", None) if username or password: http_proxy_auth = (username, password) try: from aiohttp import ClientSession + self.session = ClientSession() if username or password: from aiohttp import BasicAuth - http_proxy_auth = BasicAuth(login=username,password=password) + + http_proxy_auth = BasicAuth(login=username, password=password) self.ws = self.session.ws_connect( - url = "wss://{}".format(self._custom_endpoint or self.host), + url="wss://{}".format(self._custom_endpoint or self.host), protocols=[AMQP_WS_SUBPROTOCOL], timeout=self._connect_timeout, - proxy=http_proxy_host + f":{http_proxy_port}" if http_proxy_port else "" + proxy=http_proxy_host + f":{http_proxy_port}" if http_proxy_port else "", proxy_auth=http_proxy_auth, - #figure out ssl - ) + ssl=ssl_opts, + ) except ImportError: raise ValueError("Please install aiohttp library to use websocket transport.") - async def _read(self, n, buffer=None, **kwargs): # pylint: disable=unused-arguments + def _build_ssl_opts(self, sslopts): + if sslopts in [True, False, None, {}]: + return sslopts + try: + if "context" in sslopts: + return self._build_ssl_context(sslopts, **sslopts.pop("context")) + ssl_version = sslopts.get("ssl_version") + if ssl_version is None: + ssl_version = ssl.PROTOCOL_TLS + + # Set SNI headers if supported + server_hostname = sslopts.get("server_hostname") + if ( + (server_hostname is not None) + and (hasattr(ssl, "HAS_SNI") and ssl.HAS_SNI) + and (hasattr(ssl, "SSLContext")) + ): + context = ssl.SSLContext(ssl_version) + cert_reqs = sslopts.get("cert_reqs", ssl.CERT_REQUIRED) + certfile = sslopts.get("certfile") + keyfile = sslopts.get("keyfile") + context.verify_mode = cert_reqs + if cert_reqs != ssl.CERT_NONE: + context.check_hostname = True + if (certfile is not None) and (keyfile is not None): + context.load_cert_chain(certfile, keyfile) + return context + return True + except TypeError: + raise TypeError("SSL configuration must be a dictionary, or the value True.") + + def _build_ssl_context(self, sslopts, check_hostname=None, **ctx_options): + ctx = ssl.create_default_context(**ctx_options) + ctx.verify_mode = ssl.CERT_REQUIRED + ctx.load_verify_locations(cafile=certifi.where()) + ctx.check_hostname = check_hostname + return ctx + + async def _read(self, n, buffer=None, **kwargs): # pylint: disable=unused-arguments """Read exactly n bytes from the peer.""" - from aiohttp import length = 0 view = buffer or memoryview(bytearray(n)) @@ -472,15 +525,15 @@ async def _read(self, n, buffer=None, **kwargs): # pylint: disable=unused-argume data = await self.ws.receive_bytes() if len(data) <= n: - view[length: length + len(data)] = data + view[length : length + len(data)] = data n -= len(data) else: - view[length: length + n] = data[0:n] + view[length : length + n] = data[0:n] self._read_buffer = BytesIO(data[n:]) n = 0 - return view - except asyncio.TimeoutError as wex: + return view + except (asyncio.CancelledError, asyncio.TimeoutError) as wex: raise TimeoutError() def close(self): From 1317711a5d6fff4ecd2d1780ee6c4da44fadb92a Mon Sep 17 00:00:00 2001 From: Kashif Khan Date: Thu, 15 Sep 2022 11:00:06 -0500 Subject: [PATCH 03/11] fixes for context manager --- .../eventhub/_pyamqp/aio/_transport_async.py | 96 ++++++++++++------- 1 file changed, 60 insertions(+), 36 deletions(-) diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py index 2d6438ad3491..358f13d0592f 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py @@ -441,36 +441,35 @@ def __init__(self, host, port=WEBSOCKET_PORT, connect_timeout=None, ssl=None, ** self.ws = None self.session = None self._http_proxy = kwargs.get("http_proxy", None) + self.http_proxy_auth = None + self.http_proxy_host = None + self.http_proxy_port = None async def connect(self): - http_proxy_host, http_proxy_port, http_proxy_auth = None, None, None - ssl_opts: Optional[SSLContext] = self._build_ssl_opts(self.sslopts) if len(self.ssl_opts) > 0 else None + username, password = None, None + ssl_opts: Optional[SSLContext] = self._build_ssl_opts(self.sslopts) if len(self.sslopts) > 0 else None + if ssl_opts: + self.sslopts = ssl_opts if self._http_proxy: - http_proxy_host = self._http_proxy["proxy_hostname"] - http_proxy_port = self._http_proxy["proxy_port"] + self.http_proxy_host = self._http_proxy["proxy_hostname"] + self.http_proxy_port = self._http_proxy["proxy_port"] username = self._http_proxy.get("username", None) password = self._http_proxy.get("password", None) - if username or password: - http_proxy_auth = (username, password) - try: - from aiohttp import ClientSession - - self.session = ClientSession() + try: if username or password: from aiohttp import BasicAuth - - http_proxy_auth = BasicAuth(login=username, password=password) - - self.ws = self.session.ws_connect( - url="wss://{}".format(self._custom_endpoint or self.host), - protocols=[AMQP_WS_SUBPROTOCOL], - timeout=self._connect_timeout, - proxy=http_proxy_host + f":{http_proxy_port}" if http_proxy_port else "", - proxy_auth=http_proxy_auth, - ssl=ssl_opts, - ) + self.http_proxy_auth = BasicAuth(login=username, password=password) + + # self.ws = self.session._ws_connect( + # url="wss://{}".format(self._custom_endpoint or self.host), + # protocols=[AMQP_WS_SUBPROTOCOL], + # timeout=self._connect_timeout, + # proxy=http_proxy_host + f":{http_proxy_port}" if http_proxy_port else "", + # proxy_auth=http_proxy_auth, + # ssl=ssl_opts, + # ) except ImportError: raise ValueError("Please install aiohttp library to use websocket transport.") @@ -520,21 +519,32 @@ async def _read(self, n, buffer=None, **kwargs): # pylint: disable=unused-argum nbytes = self._read_buffer.readinto(view) length += nbytes n -= nbytes - try: - while n: - data = await self.ws.receive_bytes() + from aiohttp import ClientSession + async with ClientSession() as client: + async with client.ws_connect( + url="wss://{}".format(self._custom_endpoint or self.host), + autoclose=False, + protocols=[AMQP_WS_SUBPROTOCOL], + #timeout=self._connect_timeout, + #proxy=(self.http_proxy_host + f":{self.http_proxy_port}" if self.http_proxy_port else "") if self.http_proxy_host else None, + #auth=self.http_proxy_auth, + #ssl = self.sslopts + ) as ws: + try: + while n: + data = await ws.receive_bytes() - if len(data) <= n: - view[length : length + len(data)] = data - n -= len(data) - else: - view[length : length + n] = data[0:n] - self._read_buffer = BytesIO(data[n:]) - n = 0 + if len(data) <= n: + view[length : length + len(data)] = data + n -= len(data) + else: + view[length : length + n] = data[0:n] + self._read_buffer = BytesIO(data[n:]) + n = 0 - return view - except (asyncio.CancelledError, asyncio.TimeoutError) as wex: - raise TimeoutError() + return view + except (asyncio.CancelledError, asyncio.TimeoutError) as wex: + raise TimeoutError() def close(self): """Do any preliminary work in shutting down the connection.""" @@ -547,5 +557,19 @@ async def write(self, s): ABNF, OPCODE_BINARY = 0x2 See http://tools.ietf.org/html/rfc5234 http://tools.ietf.org/html/rfc6455#section-5.2 - """ - await self.ws.send_bytes(s) + """ + from aiohttp import ClientSession + + async with ClientSession() as client: + async with client.ws_connect( + url="wss://{}".format(self._custom_endpoint or self.host), + autoclose=False, + protocols=[AMQP_WS_SUBPROTOCOL], + #timeout=self._connect_timeout, + #proxy=(self.http_proxy_host + f":{self.http_proxy_port}" if self.http_proxy_port else "") if self.http_proxy_host else None, + #auth=self.http_proxy_auth, + #ssl = self.sslopts + ) as ws: + await ws.send_bytes(s) + + From 9793fbc061693349905afdd0e2191994c64da57a Mon Sep 17 00:00:00 2001 From: Kashif Khan Date: Thu, 15 Sep 2022 14:25:09 -0500 Subject: [PATCH 04/11] changes --- .../eventhub/_pyamqp/aio/_connection_async.py | 2 +- .../eventhub/_pyamqp/aio/_transport_async.py | 126 +++++------------- 2 files changed, 32 insertions(+), 96 deletions(-) diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_connection_async.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_connection_async.py index 2c3072143591..354c733bed19 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_connection_async.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_connection_async.py @@ -189,7 +189,7 @@ async def _disconnect(self, *args) -> None: if self.state == ConnectionState.END: return await self._set_state(ConnectionState.END) - self._transport.close() + await self._transport.close() def _can_read(self): # type: () -> bool diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py index 358f13d0592f..e84f28fb72e9 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py @@ -384,7 +384,7 @@ async def _write(self, s): """Write a string out to the SSL socket fully.""" self.writer.write(s) - def close(self): + async def close(self): if self.writer is not None: if self.sslopts: # see issue: https://github.com/encode/httpx/issues/914 @@ -442,14 +442,9 @@ def __init__(self, host, port=WEBSOCKET_PORT, connect_timeout=None, ssl=None, ** self.session = None self._http_proxy = kwargs.get("http_proxy", None) self.http_proxy_auth = None - self.http_proxy_host = None - self.http_proxy_port = None - + async def connect(self): username, password = None, None - ssl_opts: Optional[SSLContext] = self._build_ssl_opts(self.sslopts) if len(self.sslopts) > 0 else None - if ssl_opts: - self.sslopts = ssl_opts if self._http_proxy: self.http_proxy_host = self._http_proxy["proxy_hostname"] @@ -458,59 +453,24 @@ async def connect(self): password = self._http_proxy.get("password", None) try: + from aiohttp import ClientSession + if username or password: from aiohttp import BasicAuth self.http_proxy_auth = BasicAuth(login=username, password=password) - - # self.ws = self.session._ws_connect( - # url="wss://{}".format(self._custom_endpoint or self.host), - # protocols=[AMQP_WS_SUBPROTOCOL], - # timeout=self._connect_timeout, - # proxy=http_proxy_host + f":{http_proxy_port}" if http_proxy_port else "", - # proxy_auth=http_proxy_auth, - # ssl=ssl_opts, - # ) + + self.session = ClientSession() + self.ws = await self.session.ws_connect( + url="wss://{}".format(self._custom_endpoint or self.host), + timeout=self._connect_timeout, + protocols=[AMQP_WS_SUBPROTOCOL], + autoclose=False, + proxy_auth=self.http_proxy_auth + ) + except ImportError: raise ValueError("Please install aiohttp library to use websocket transport.") - def _build_ssl_opts(self, sslopts): - if sslopts in [True, False, None, {}]: - return sslopts - try: - if "context" in sslopts: - return self._build_ssl_context(sslopts, **sslopts.pop("context")) - ssl_version = sslopts.get("ssl_version") - if ssl_version is None: - ssl_version = ssl.PROTOCOL_TLS - - # Set SNI headers if supported - server_hostname = sslopts.get("server_hostname") - if ( - (server_hostname is not None) - and (hasattr(ssl, "HAS_SNI") and ssl.HAS_SNI) - and (hasattr(ssl, "SSLContext")) - ): - context = ssl.SSLContext(ssl_version) - cert_reqs = sslopts.get("cert_reqs", ssl.CERT_REQUIRED) - certfile = sslopts.get("certfile") - keyfile = sslopts.get("keyfile") - context.verify_mode = cert_reqs - if cert_reqs != ssl.CERT_NONE: - context.check_hostname = True - if (certfile is not None) and (keyfile is not None): - context.load_cert_chain(certfile, keyfile) - return context - return True - except TypeError: - raise TypeError("SSL configuration must be a dictionary, or the value True.") - - def _build_ssl_context(self, sslopts, check_hostname=None, **ctx_options): - ctx = ssl.create_default_context(**ctx_options) - ctx.verify_mode = ssl.CERT_REQUIRED - ctx.load_verify_locations(cafile=certifi.where()) - ctx.check_hostname = check_hostname - return ctx - async def _read(self, n, buffer=None, **kwargs): # pylint: disable=unused-arguments """Read exactly n bytes from the peer.""" @@ -519,37 +479,25 @@ async def _read(self, n, buffer=None, **kwargs): # pylint: disable=unused-argum nbytes = self._read_buffer.readinto(view) length += nbytes n -= nbytes - from aiohttp import ClientSession - async with ClientSession() as client: - async with client.ws_connect( - url="wss://{}".format(self._custom_endpoint or self.host), - autoclose=False, - protocols=[AMQP_WS_SUBPROTOCOL], - #timeout=self._connect_timeout, - #proxy=(self.http_proxy_host + f":{self.http_proxy_port}" if self.http_proxy_port else "") if self.http_proxy_host else None, - #auth=self.http_proxy_auth, - #ssl = self.sslopts - ) as ws: - try: - while n: - data = await ws.receive_bytes() - - if len(data) <= n: - view[length : length + len(data)] = data - n -= len(data) - else: - view[length : length + n] = data[0:n] - self._read_buffer = BytesIO(data[n:]) - n = 0 - - return view - except (asyncio.CancelledError, asyncio.TimeoutError) as wex: + + try: + while n: + data = await self.ws.receive_bytes() + if len(data) <= n: + view[length : length + len(data)] = data + n -= len(data) + else: + view[length : length + n] = data[0:n] + self._read_buffer = BytesIO(data[n:]) + n = 0 + return view + except (asyncio.TimeoutError) as wex: raise TimeoutError() - def close(self): + async def close(self): """Do any preliminary work in shutting down the connection.""" - self.ws.close() - self.session.close() + await self.ws.close() + await self.session.close() self.connected = False async def write(self, s): @@ -558,18 +506,6 @@ async def write(self, s): See http://tools.ietf.org/html/rfc5234 http://tools.ietf.org/html/rfc6455#section-5.2 """ - from aiohttp import ClientSession - - async with ClientSession() as client: - async with client.ws_connect( - url="wss://{}".format(self._custom_endpoint or self.host), - autoclose=False, - protocols=[AMQP_WS_SUBPROTOCOL], - #timeout=self._connect_timeout, - #proxy=(self.http_proxy_host + f":{self.http_proxy_port}" if self.http_proxy_port else "") if self.http_proxy_host else None, - #auth=self.http_proxy_auth, - #ssl = self.sslopts - ) as ws: - await ws.send_bytes(s) + await self.ws.send_bytes(s) From 3edba98bd8518aed929d68a0dd5aceac8ed5acbf Mon Sep 17 00:00:00 2001 From: Kashif Khan Date: Thu, 15 Sep 2022 14:47:00 -0500 Subject: [PATCH 05/11] ssl options --- .../eventhub/_pyamqp/aio/_transport_async.py | 41 ++++++++++++++++++- 1 file changed, 40 insertions(+), 1 deletion(-) diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py index e84f28fb72e9..1c86892718ea 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py @@ -465,7 +465,8 @@ async def connect(self): timeout=self._connect_timeout, protocols=[AMQP_WS_SUBPROTOCOL], autoclose=False, - proxy_auth=self.http_proxy_auth + proxy_auth=self.http_proxy_auth, + ssl=self._build_ssl_opts(self.sslopts) ) except ImportError: @@ -507,5 +508,43 @@ async def write(self, s): http://tools.ietf.org/html/rfc6455#section-5.2 """ await self.ws.send_bytes(s) + + def _build_ssl_opts(self, sslopts): + if sslopts in [True, False, None, {}]: + return sslopts + try: + if "context" in sslopts: + return self._build_ssl_context(sslopts, **sslopts.pop("context")) + ssl_version = sslopts.get("ssl_version") + if ssl_version is None: + ssl_version = ssl.PROTOCOL_TLS + + # Set SNI headers if supported + server_hostname = sslopts.get("server_hostname") + if ( + (server_hostname is not None) + and (hasattr(ssl, "HAS_SNI") and ssl.HAS_SNI) + and (hasattr(ssl, "SSLContext")) + ): + context = ssl.SSLContext(ssl_version) + cert_reqs = sslopts.get("cert_reqs", ssl.CERT_REQUIRED) + certfile = sslopts.get("certfile") + keyfile = sslopts.get("keyfile") + context.verify_mode = cert_reqs + if cert_reqs != ssl.CERT_NONE: + context.check_hostname = True + if (certfile is not None) and (keyfile is not None): + context.load_cert_chain(certfile, keyfile) + return context + return True + except TypeError: + raise TypeError("SSL configuration must be a dictionary, or the value True.") + + def _build_ssl_context(self, sslopts, check_hostname=None, **ctx_options): + ctx = ssl.create_default_context(**ctx_options) + ctx.verify_mode = ssl.CERT_REQUIRED + ctx.load_verify_locations(cafile=certifi.where()) + ctx.check_hostname = check_hostname + return ctx From 554b9299e566b43fa810c2633188364da50f4211 Mon Sep 17 00:00:00 2001 From: Kashif Khan Date: Thu, 15 Sep 2022 15:59:14 -0500 Subject: [PATCH 06/11] clean up --- .../azure/eventhub/_pyamqp/aio/_transport_async.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py index 1c86892718ea..a782e1bbad05 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py @@ -432,9 +432,8 @@ async def negotiate(self): class WebSocketTransportAsync(AsyncTransportMixin): def __init__(self, host, port=WEBSOCKET_PORT, connect_timeout=None, ssl=None, **kwargs): self._read_buffer = BytesIO() - self.loop = get_running_loop() self.socket_lock = asyncio.Lock() - self.sslopts = ssl if isinstance(ssl, dict) else {} + self.sslopts = self._build_ssl_opts(ssl) if isinstance(ssl, dict) else None self._connect_timeout = connect_timeout or TIMEOUT_INTERVAL self._custom_endpoint = kwargs.get("custom_endpoint") self.host = host @@ -466,7 +465,7 @@ async def connect(self): protocols=[AMQP_WS_SUBPROTOCOL], autoclose=False, proxy_auth=self.http_proxy_auth, - ssl=self._build_ssl_opts(self.sslopts) + ssl=self.sslopts ) except ImportError: @@ -493,7 +492,7 @@ async def _read(self, n, buffer=None, **kwargs): # pylint: disable=unused-argum n = 0 return view except (asyncio.TimeoutError) as wex: - raise TimeoutError() + raise TimeoutError() async def close(self): """Do any preliminary work in shutting down the connection.""" @@ -506,7 +505,7 @@ async def write(self, s): ABNF, OPCODE_BINARY = 0x2 See http://tools.ietf.org/html/rfc5234 http://tools.ietf.org/html/rfc6455#section-5.2 - """ + """ await self.ws.send_bytes(s) def _build_ssl_opts(self, sslopts): From d55cf3808ad9caa33880f2527ad4ae97ac555dfb Mon Sep 17 00:00:00 2001 From: Kashif Khan Date: Thu, 15 Sep 2022 16:15:26 -0500 Subject: [PATCH 07/11] move build opts in to mixin --- .../eventhub/_pyamqp/aio/_transport_async.py | 104 ++++++------------ 1 file changed, 34 insertions(+), 70 deletions(-) diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py index a782e1bbad05..6d8ebf149407 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py @@ -150,38 +150,6 @@ async def send_frame(self, channel, frame, **kwargs): await self.write(data) # _LOGGER.info("OCH%d -> %r", channel, frame) - -class AsyncTransport(AsyncTransportMixin): - """Common superclass for TCP and SSL transports.""" - - def __init__( - self, - host, - port=AMQP_PORT, - connect_timeout=None, - read_timeout=None, - write_timeout=None, - ssl=False, - socket_settings=None, - raise_on_initial_eintr=True, - **kwargs, - ): - self.connected = False - self.sock = None - self.reader = None - self.writer = None - self.raise_on_initial_eintr = raise_on_initial_eintr - self._read_buffer = BytesIO() - self.host, self.port = to_host_port(host, port) - - self.connect_timeout = connect_timeout - self.read_timeout = read_timeout - self.write_timeout = write_timeout - self.socket_settings = socket_settings - self.loop = get_running_loop() - self.socket_lock = asyncio.Lock() - self.sslopts = self._build_ssl_opts(ssl) - def _build_ssl_opts(self, sslopts): if sslopts in [True, False, None, {}]: return sslopts @@ -220,6 +188,40 @@ def _build_ssl_context(self, sslopts, check_hostname=None, **ctx_options): ctx.check_hostname = check_hostname return ctx + +class AsyncTransport(AsyncTransportMixin): + """Common superclass for TCP and SSL transports.""" + + def __init__( + self, + host, + port=AMQP_PORT, + connect_timeout=None, + read_timeout=None, + write_timeout=None, + ssl=False, + socket_settings=None, + raise_on_initial_eintr=True, + **kwargs, + ): + self.connected = False + self.sock = None + self.reader = None + self.writer = None + self.raise_on_initial_eintr = raise_on_initial_eintr + self._read_buffer = BytesIO() + self.host, self.port = to_host_port(host, port) + + self.connect_timeout = connect_timeout + self.read_timeout = read_timeout + self.write_timeout = write_timeout + self.socket_settings = socket_settings + self.loop = get_running_loop() + self.socket_lock = asyncio.Lock() + self.sslopts = self._build_ssl_opts(ssl) + + + async def connect(self): try: # are we already connected? @@ -507,43 +509,5 @@ async def write(self, s): http://tools.ietf.org/html/rfc6455#section-5.2 """ await self.ws.send_bytes(s) - - def _build_ssl_opts(self, sslopts): - if sslopts in [True, False, None, {}]: - return sslopts - try: - if "context" in sslopts: - return self._build_ssl_context(sslopts, **sslopts.pop("context")) - ssl_version = sslopts.get("ssl_version") - if ssl_version is None: - ssl_version = ssl.PROTOCOL_TLS - - # Set SNI headers if supported - server_hostname = sslopts.get("server_hostname") - if ( - (server_hostname is not None) - and (hasattr(ssl, "HAS_SNI") and ssl.HAS_SNI) - and (hasattr(ssl, "SSLContext")) - ): - context = ssl.SSLContext(ssl_version) - cert_reqs = sslopts.get("cert_reqs", ssl.CERT_REQUIRED) - certfile = sslopts.get("certfile") - keyfile = sslopts.get("keyfile") - context.verify_mode = cert_reqs - if cert_reqs != ssl.CERT_NONE: - context.check_hostname = True - if (certfile is not None) and (keyfile is not None): - context.load_cert_chain(certfile, keyfile) - return context - return True - except TypeError: - raise TypeError("SSL configuration must be a dictionary, or the value True.") - - def _build_ssl_context(self, sslopts, check_hostname=None, **ctx_options): - ctx = ssl.create_default_context(**ctx_options) - ctx.verify_mode = ssl.CERT_REQUIRED - ctx.load_verify_locations(cafile=certifi.where()) - ctx.check_hostname = check_hostname - return ctx From 0294a13d1659c168249d7b2b07142f77cc553c55 Mon Sep 17 00:00:00 2001 From: Kashif Khan Date: Thu, 15 Sep 2022 16:37:19 -0500 Subject: [PATCH 08/11] pass in proxy information --- .../azure/eventhub/_pyamqp/aio/_transport_async.py | 1 + 1 file changed, 1 insertion(+) diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py index 6d8ebf149407..5b86d0e51c01 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py @@ -466,6 +466,7 @@ async def connect(self): timeout=self._connect_timeout, protocols=[AMQP_WS_SUBPROTOCOL], autoclose=False, + proxy=self.http_proxy_host + f":{self.http_proxy_port}" if self.http_proxy_port else "", proxy_auth=self.http_proxy_auth, ssl=self.sslopts ) From faf6d30235ee9323ea9e7ddc32466e713c54a1d6 Mon Sep 17 00:00:00 2001 From: Kashif Khan Date: Thu, 15 Sep 2022 17:12:28 -0500 Subject: [PATCH 09/11] attach port to proxy if given --- .../azure/eventhub/_pyamqp/aio/_transport_async.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py index 5b86d0e51c01..be8fd3cfda91 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py @@ -443,6 +443,7 @@ def __init__(self, host, port=WEBSOCKET_PORT, connect_timeout=None, ssl=None, ** self.session = None self._http_proxy = kwargs.get("http_proxy", None) self.http_proxy_auth = None + self.http_proxy_host = None async def connect(self): username, password = None, None @@ -450,6 +451,8 @@ async def connect(self): if self._http_proxy: self.http_proxy_host = self._http_proxy["proxy_hostname"] self.http_proxy_port = self._http_proxy["proxy_port"] + if self.http_proxy_host and self.http_proxy_port: + self.http_proxy_host += f":{self.http_proxy_port}" username = self._http_proxy.get("username", None) password = self._http_proxy.get("password", None) @@ -466,7 +469,7 @@ async def connect(self): timeout=self._connect_timeout, protocols=[AMQP_WS_SUBPROTOCOL], autoclose=False, - proxy=self.http_proxy_host + f":{self.http_proxy_port}" if self.http_proxy_port else "", + proxy=self.http_proxy_host, proxy_auth=self.http_proxy_auth, ssl=self.sslopts ) From a8cf10b84508d73c39eaf9a1b73edce327ea3f79 Mon Sep 17 00:00:00 2001 From: Kashif Khan Date: Tue, 20 Sep 2022 08:33:33 -0500 Subject: [PATCH 10/11] address comments --- .../eventhub/_pyamqp/aio/_transport_async.py | 20 +++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py index be8fd3cfda91..d193a02d8e48 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py @@ -442,17 +442,17 @@ def __init__(self, host, port=WEBSOCKET_PORT, connect_timeout=None, ssl=None, ** self.ws = None self.session = None self._http_proxy = kwargs.get("http_proxy", None) - self.http_proxy_auth = None - self.http_proxy_host = None async def connect(self): username, password = None, None + http_proxy_host, http_proxy_port = None, None + http_proxy_auth = None if self._http_proxy: - self.http_proxy_host = self._http_proxy["proxy_hostname"] - self.http_proxy_port = self._http_proxy["proxy_port"] - if self.http_proxy_host and self.http_proxy_port: - self.http_proxy_host += f":{self.http_proxy_port}" + http_proxy_host = self._http_proxy["proxy_hostname"] + http_proxy_port = self._http_proxy["proxy_port"] + if http_proxy_host and http_proxy_port: + self.http_proxy_host = f"{http_proxy_host}:{http_proxy_port}" username = self._http_proxy.get("username", None) password = self._http_proxy.get("password", None) @@ -461,7 +461,7 @@ async def connect(self): if username or password: from aiohttp import BasicAuth - self.http_proxy_auth = BasicAuth(login=username, password=password) + http_proxy_auth = BasicAuth(login=username, password=password) self.session = ClientSession() self.ws = await self.session.ws_connect( @@ -469,8 +469,8 @@ async def connect(self): timeout=self._connect_timeout, protocols=[AMQP_WS_SUBPROTOCOL], autoclose=False, - proxy=self.http_proxy_host, - proxy_auth=self.http_proxy_auth, + proxy=http_proxy_host, + proxy_auth=http_proxy_auth, ssl=self.sslopts ) @@ -507,7 +507,7 @@ async def close(self): self.connected = False async def write(self, s): - """Completely write a string to the peer. + """Completely write a string (byte array) to the peer. ABNF, OPCODE_BINARY = 0x2 See http://tools.ietf.org/html/rfc5234 http://tools.ietf.org/html/rfc6455#section-5.2 From 8a25986b69f6fd9b559c2dff4bbcfc1c2f4db410 Mon Sep 17 00:00:00 2001 From: Kashif Khan <361477+kashifkhan@users.noreply.github.com> Date: Mon, 26 Sep 2022 21:04:03 -0500 Subject: [PATCH 11/11] remove self from proxy host --- .../azure/eventhub/_pyamqp/aio/_transport_async.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py index 1d54a44e6c52..b53971ddb203 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py @@ -449,7 +449,7 @@ async def connect(self): http_proxy_host = self._http_proxy["proxy_hostname"] http_proxy_port = self._http_proxy["proxy_port"] if http_proxy_host and http_proxy_port: - self.http_proxy_host = f"{http_proxy_host}:{http_proxy_port}" + http_proxy_host = f"{http_proxy_host}:{http_proxy_port}" username = self._http_proxy.get("username", None) password = self._http_proxy.get("password", None) @@ -509,4 +509,4 @@ async def write(self, s): See http://tools.ietf.org/html/rfc5234 http://tools.ietf.org/html/rfc6455#section-5.2 """ - await self.ws.send_bytes(s) \ No newline at end of file + await self.ws.send_bytes(s)