Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use is_ssl_socket method in localstack

Best Python code snippet using localstack_python

generic_proxy.py

Source:generic_proxy.py

...362class DuplexSocket(ssl.SSLSocket):363    """Simple duplex socket wrapper that allows serving HTTP/HTTPS over the same port."""364    def accept(self):365        newsock, addr = socket.socket.accept(self)366        if DuplexSocket.is_ssl_socket(newsock) is not False:367            newsock = self.context.wrap_socket(368                newsock,369                do_handshake_on_connect=self.do_handshake_on_connect,370                suppress_ragged_eofs=self.suppress_ragged_eofs,371                server_side=True,372            )373        return newsock, addr374    @staticmethod375    def is_ssl_socket(newsock):376        """Returns True/False if the socket uses SSL or not, or None if the status cannot be determined"""377        def peek_ssl_header():378            peek_bytes = 5379            first_bytes = newsock.recv(peek_bytes, socket.MSG_PEEK)380            if len(first_bytes or "") != peek_bytes:381                return382            first_byte = first_bytes[0]383            return first_byte < 32 or first_byte >= 127384        try:385            return peek_ssl_header()386        except Exception:387            # Fix for "[Errno 11] Resource temporarily unavailable" - This can388            #   happen if we're using a non-blocking socket in a blocking thread.389            newsock.setblocking(1)390            newsock.settimeout(1)391            try:392                return peek_ssl_header()393            except Exception:394                return False395# set globally defined SSL socket implementation class396ssl.SSLContext.sslsocket_class = DuplexSocket397class GenericProxy(object):398    # TODO: move methods to different class?399    @classmethod400    def create_ssl_cert(cls, serial_number=None):401        cert_pem_file = get_cert_pem_file_path()402        return generate_ssl_cert(cert_pem_file, serial_number=serial_number)403    @classmethod404    def get_flask_ssl_context(cls, serial_number=None):405        if config.USE_SSL:406            _, cert_file_name, key_file_name = cls.create_ssl_cert(serial_number=serial_number)407            return cert_file_name, key_file_name408        return None409async def _accept_connection2(self, protocol_factory, conn, extra, sslcontext, *args, **kwargs):410    is_ssl_socket = DuplexSocket.is_ssl_socket(conn)411    if is_ssl_socket is False:412        sslcontext = None413    result = await _accept_connection2_orig(414        self, protocol_factory, conn, extra, sslcontext, *args, **kwargs415    )416    return result417# patch asyncio server to accept SSL and non-SSL traffic over same port418if hasattr(BaseSelectorEventLoop, "_accept_connection2") and not hasattr(419    BaseSelectorEventLoop, "_ls_patched"420):421    _accept_connection2_orig = BaseSelectorEventLoop._accept_connection2422    BaseSelectorEventLoop._accept_connection2 = _accept_connection2423    BaseSelectorEventLoop._ls_patched = True424def get_cert_pem_file_path():...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.