Method proxy_ssl_one

test/with_dummyserver/test_socketlevel.py:1238–1276

(listener: socket.socket)

    def test_connect_reconn(self) -> None:
        def proxy_ssl_one(listener: socket.socket) -> None:
            sock = listener.accept()[0]

            buf = b""
            while not buf.endswith(b"\r\n\r\n"):
                buf += sock.recv(65536)
            s = buf.decode("utf-8")
            if not s.startswith("CONNECT "):
                sock.send(b"HTTP/1.1 405 Method not allowed\r\nAllow: CONNECT\r\n\r\n")
                sock.close()
                return

            if not s.startswith(f"CONNECT {self.host}:443"):
                sock.send(b"HTTP/1.1 403 Forbidden\r\n\r\n")
                sock.close()
                return

            sock.send(b"HTTP/1.1 200 Connection Established\r\n\r\n")
            ssl_sock = original_ssl_wrap_socket(
                sock,
                server_side=True,
                keyfile=DEFAULT_CERTS["keyfile"],
                certfile=DEFAULT_CERTS["certfile"],
                ca_certs=DEFAULT_CA,
            )

            buf = b""
            while not buf.endswith(b"\r\n\r\n"):
                buf += ssl_sock.recv(65536)

            ssl_sock.send(
                b"HTTP/1.1 200 OK\r\n"
                b"Content-Type: text/plain\r\n"
                b"Content-Length: 2\r\n"
                b"Connection: close\r\n"
                b"\r\n"
                b"Hi"
            )
            ssl_sock.close()

        def echo_socket_handler(listener: socket.socket) -> None:
            proxy_ssl_one(listener)