class ProxyConnectError(ConnectError):
    """
    Exception raised when we are unable to start a connection using a HTTP proxy
    This indicates an issue with the HTTP Proxy in use rather than the final
    endpoint we wanted to contact.
    """

    pass class HttpConnectProtocol(asyncio.Protocol):
    """
    This is for use with asyncio's Protocol and Transport API.

    It performs the setup of a HTTP CONNECT proxy connection, then the calling
     code is responsible for handing over to another asyncio.Protocol.

    For Twisted, see twisted_connectproxyclient.py instead.

    The intended usage of this class is to use it in a protocol factory in
    `AbstractEventLoop.create_connection`, then await `wait_for_establishment`
    and hand the transport over to another protocol, potentially wrapping it in
    TLS with `AbstractEventLoop.start_tls`.

    Once the connection is made, the `HttpConnectProtocol` is redundant and can
    be forgotten about; the protocol stack might look like:

                before                           after
                                                 +------------------+
                                                 |  HTTP Protocol   |
                +---------------------+          +------------------+
                | HTTP Proxy Protocol |   ===>   | SSL/TLS Protocol |
                +---------------------+----------+------------------+
                |              Underlying TCP Transport             |
                +---------------------------------------------------+
        (assuming a proxied HTTPS connection is what you were after)
    """

    def __init__(
        self,
        target_hostport: Tuple[str, int],
        proxy_credentials: Optional[Tuple[str, str]],
        protocol_factory: Callable[[], Protocol],
        sslcontext: Optional[SSLContext],
        loop: Optional[AbstractEventLoop] = None,
    ):
        """
        Args:
            target_hostport:
                The host & port of the destination that the proxy should connect
                to on your behalf.
                Examples: ('example.org', 443)

            proxy_credentials:
                An optional (username, password) tuple of strings to pass to the proxy.

            protocol_factory:
                A 0-argument function which, when called, returns a Protocol
                to switch over to.

            sslcontext:
                If TLS is desired after the connection is completed, pass an
                SSLContext here, making sure it is safe for your purposes —
                see ssl.create_default_context's documentation as a starting
                point.

            loop (optional):
                An asyncio EventLoop to use; if not provided, the default will
                be used.
        """
        # set to True when we have called `switch_over_when_ready`.
        self._switch_over_called = False

        # (host, port) of the target that we want a tunnel to
        self._target_hostport = target_hostport

        # buffer for the HTTP response that comes back from the HTTP proxy
        self._response_buffer = b""

        # underlying transport
        self._transport: Transport = None  # type: ignore

        # the proxy's credentials as a string pair, or None
        self._proxy_credentials = proxy_credentials

        # function of () -> Protocol, to be called once when we switch over to
        # this protocol
        self._protocol_factory = protocol_factory

        # optional SSLContext if TLS is desired, None otherwise
        self._sslcontext = sslcontext

        # asyncio EventLoop
        self._event_loop = loop or asyncio.get_event_loop()

        # This future is completed when it is safe to take back control of the
        # transport.
        # It completes with leftover bytes for the next protocol.
        self._tunnel_established_future: Future[bytes] = Future()

    async def switch_over_when_ready(self) -> Tuple[BaseTransport, Protocol]:
        """
        Waits until we are connected to the remote (i.e. that our CONNECT
        request succeeds).
        Then constructs the requested protocol and attaches it to the transport,
        potentially wrapping it in TLS first.
        Returns:
            the transport followed by the constructed protocol that uses it
            Note: the transport may be an SSLTransport; it is not necessarily
                the same one used to communicate with the proxy directly.
        """

        if self._switch_over_called:
            raise RuntimeError(
                "Can only use `HttpConnectProtocol.switch_over_when_ready` once."
            )
        self._switch_over_called = True

        left_over_bytes = await self._tunnel_established_future
        # construct the desired protocol and hand over the transport to it
        new_protocol = self._protocol_factory()

        if self._sslcontext:
            if left_over_bytes:
                # in TLS, the client transmits first, so this is theoretically
                # unreachable
                raise RuntimeError("Left over bytes should not occur with TLS")

            # There is a race where the `new_protocol` may get given data before
            # we manage to call `connection_made` on it, which can lead to
            # exceptions if the protocol then tries to write to the transport
            # that is has been given yet.
            buffered_protocol = _BufferedWrapperProtocol(new_protocol)

            # be careful not to use the `transport` ever again after passing it
            # to start_tls — we overwrite our variable with the TLS-wrapped
            # transport to avoid that!
            transport = await self._event_loop.start_tls(
                self._transport,
                buffered_protocol,
                self._sslcontext,
                server_hostname=self._target_hostport[0],
            )

            # start_tls does NOT call connection_made on new_protocol, so we
            # must do it ourselves
            buffered_protocol.connection_made(transport)
        else:
            # no wrapping required for non-TLS
            transport = self._transport
            # wire up transport to call `data_received` etc. on the new transport
            transport.set_protocol(new_protocol)
            # let the protocol know it has been connected to the transport
            new_protocol.connection_made(transport)

            if left_over_bytes:
                # pass over dangling bytes if applicable
                new_protocol.data_received(left_over_bytes)

        logger.debug("Finished switching protocol")

        return transport, new_protocol

    def data_received(self, data: bytes) -> None:
        super().data_received(data)
        self._response_buffer += data
        if b"\r\n\r\n" not in self._response_buffer:
            # we haven't finished the headers yet
            return

        # The response headers are terminated by a double CRLF.
        # NB we want want 'in' instead of 'endswith'
        #  as no guarantee error page (or even bytes from the target server)
        #  won't come immediately.

        # All HTTP header lines are terminated by CRLF.
        # the first line of the response headers is the Status Line
        try:
            response_header, dangling_bytes = self._response_buffer.split(
                b"\r\n\r\n", maxsplit=1
            )
            lines = response_header.split(b"\r\n")
            status_line = lines[0]
            # maxsplit=2 denotes the number of separators, not the № items
            # StatusLine ← HTTPVersion SP StatusCode SP ReasonPhrase
            # None of the fields may contain CRLF, and only ReasonPhrase may
            # contain SP.
            [http_version, status, reason_phrase] = status_line.split(b" ", maxsplit=2)
            logger.debug(
                "CONNECT response from proxy: hv=%s, r=%s, rp=%s",
                http_version,
                status,
                reason_phrase,
            )
            if status != b"200":
                # 200 Successful (aka Connection Established) is what we want
                # if it is not what we have, then we don't have a tunnel
                self._transport.close()
                raise ProxyConnectError(
                    "Error from HTTP Proxy"
                    f" whilst attempting CONNECT: {status.decode()}"
                    f" ({reason_phrase.decode()}); aborting connection."
                )

            logger.debug("Ready to switch over protocol")

            self._response_buffer = None  # type: ignore
            # TLS doesn't seem to allow the server to talk before the client begins
            # the handshake, but plain HTTP/2 seems like the server can talk first
            # and who knows what the future holds?
            # (we may wish to use other protocols or TLS might change)
            # So we must also keep the left-over bytes to hand to the next Protocol
            self._tunnel_established_future.set_result(dangling_bytes)
        except Exception as exc:
            self._tunnel_established_future.set_exception(exc)

    def connection_made(self, transport: BaseTransport) -> None:
        if not isinstance(transport, Transport):
            raise ValueError("transport must be a proper Transport")

        super().connection_made(transport)
        # when we get a TCP connection to the HTTP proxy, we invoke the CONNECT
        # method on it to open a tunnelled TCP connection through the proxy to
        # the other side
        host, port = self._target_hostport
        transport.write(f"CONNECT {host}:{port} HTTP/1.0\r\n".encode())
        if self._proxy_credentials:
            username, password = self._proxy_credentials
            # a credential pair is a urlsafe-base64-encoded pair separated by colon
            encoded_credentials = urlsafe_b64encode(f"{username}:{password}".encode())
            transport.write(
                b"Proxy-Authorization: basic " + encoded_credentials + b"\r\n"
            )
        # a blank line terminates the request headers
        transport.write(b"\r\n")

        logger.debug("Initiating proxy CONNECT")

        # now we wait ...
        self._transport = transport REQ_PATH = b"/_matrix/push/v1/notify"
class TestCase(unittest.TestCase):
class ExtendedMemoryReactorClock(MemoryReactorClock):
        class FakeResolver:
class DummyResponse:
class HTTPResult:
class FakeChannel:
    def config_setup(self, config):
    def setUp(self):
    def _make_dummy_notification(self, devices):
    def _make_dummy_notification_event_id_only(self, devices):
    def _make_dummy_notification_badge_only(self, devices):
    def _request(self, payload: Union[str, dict]) -> Union[dict, int]:
    def _multi_requests(
        self, payloads: List[Union[str, dict]]
    ) -> List[Union[dict, int]]:
        def dump_if_needed(payload):
        def all_channels_done():
        def channel_result(channel):
    def __init__(self):
            def getHostByName(name, timeout=None):
    def installNameResolver(self, resolver):
    def callFromThread(self, function, *args):
    def callLater(self, when, what, *a, **kw):
    def wait_for_work(self, early_stop=lambda: False):
    def __init__(self, code):
def make_async_magic_mock(ret_val):
    async def dummy(*_args, **_kwargs):
    def code(self):
    def writeHeaders(self, version, code, reason, headers):
    def write(self, content):
    def requestDone(self, _self):
    def getPeer(self):
    def getHost(self):
    def transport(self):
    def process_request(self, method: bytes, request_path: bytes, content: BinaryIO): class EchoProtocol(Protocol):
    """A protocol that immediately echoes all data it receives"""

    def __init__(self):
        self._to_transmit = b""
        self.received_bytes = b""
        self.transport = None

    def data_received(self, data: bytes) -> None:
        self.received_bytes += data
        assert self.transport
        self.transport.write(data)

    def connection_made(self, transport: transports.BaseTransport) -> None:
        assert isinstance(transport, Transport)
        self.transport = transport
        if self._to_transmit:
            transport.write(self._to_transmit)

    def write(self, data: bytes) -> None:
        if self.transport:
            self.transport.write(data)
        else:
            self._to_transmit += data class MockProtocol(Protocol):
    """
    A protocol intended to be driven by tests.
    Stores received data into a buffer.
    """

    def __init__(self):
        self._to_transmit = b""
        self.received_bytes = b""
        self.transport = None

    def data_received(self, data: bytes) -> None:
        self.received_bytes += data

    def connection_made(self, transport: transports.BaseTransport) -> None:
        assert isinstance(transport, Transport)
        self.transport = transport
        if self._to_transmit:
            transport.write(self._to_transmit)

    def write(self, data: bytes) -> None:
        if self.transport:
            self.transport.write(data)
        else:
            self._to_transmit += data class MockTransport(Transport):
    """
    A transport intended to be driven by tests.
    Stores received data into a buffer.
    """

    def __init__(self):
        # Holds bytes received
        self.buffer = b""

        # Whether we reached the end of file/stream
        self.eofed = False

        # Whether the connection was aborted
        self.aborted = False

        # The protocol attached to this transport
        self.protocol = None

        # Whether this transport was closed
        self.closed = False

        # We need to explicitly mark that this connection allows start tls,
        # otherwise `loop.start_tls` will raise an exception.
        self._start_tls_compatible = True

    def reset_mock(self) -> None:
        self.buffer = b""
        self.eofed = False
        self.aborted = False
        self.closed = False

    def is_reading(self) -> bool:
        return True

    def pause_reading(self) -> None:
        pass  # NOP

    def resume_reading(self) -> None:
        pass  # NOP

    def set_write_buffer_limits(self, high: int = None, low: int = None) -> None:
        pass  # NOP

    def get_write_buffer_size(self) -> int:
        """Return the current size of the write buffer."""
        raise NotImplementedError

    def write(self, data: bytes) -> None:
        self.buffer += data

    def write_eof(self) -> None:
        self.eofed = True

    def can_write_eof(self) -> bool:
        return True

    def abort(self) -> None:
        self.aborted = True

    def pretend_to_receive(self, data: bytes) -> None:
        proto = self.get_protocol()
        assert isinstance(proto, Protocol)
        proto.data_received(data)

    def set_protocol(self, protocol: BaseProtocol) -> None:
        self.protocol = protocol

    def get_protocol(self) -> BaseProtocol:
        assert isinstance(self.protocol, BaseProtocol)
        return self.protocol

    def close(self) -> None:
        self.closed = True class TimelessEventLoopWrapper:
    @property  # type: ignore
    def __class__(self):
        """
        Fakes isinstance(this, AbstractEventLoop) so we can set_event_loop
        without fail.
        """
        return self._wrapped_loop.__class__

    def __init__(self, wrapped_loop: AbstractEventLoop):
        self._wrapped_loop = wrapped_loop
        self._time = 0.0
        self._to_be_called: List[Tuple[float, Any, Any, Any]] = []

    def advance(self, time_delta: float):
        target_time = self._time + time_delta
        logger.debug(
            "advancing from %f by %f (%d in queue)",
            self._time,
            time_delta,
            len(self._to_be_called),
        )
        while self._time < target_time and self._to_be_called:
            # pop off the next callback from the queue
            next_time, next_callback, args, _context = self._to_be_called[0]
            if next_time > target_time:
                # this isn't allowed to run yet
                break
            logger.debug("callback at %f on %r", next_time, next_callback)
            self._to_be_called = self._to_be_called[1:]
            self._time = next_time
            next_callback(*args)

        # no more tasks can run now but advance to the time anyway
        self._time = target_time

    def __getattr__(self, item: str):
        """
        We use this to delegate other method calls to the real EventLoop.
        """
        value = getattr(self._wrapped_loop, item)
        if isinstance(value, types.MethodType):
            # rebind this method to be called on us
            # this makes the wrapped class use our overridden methods when
            # available.
            # we have to do this because methods are bound to the underlying
            # event loop, which will call `self.call_later` or something
            # which won't normally hit us because we are not an actual subtype.
            return types.MethodType(value.__func__, self)
        else:
            return value

    def call_later(
        self,
        delay: float,
        callback: Callable,
        *args: Any,
        context: Optional[Context] = None,
    ):
        self.call_at(self._time + delay, callback, *args, context=context)

        # We're meant to return a canceller, but can cheat and return a no-op one
        # instead.
        class _Canceller:
            def cancel(self):
                pass

        return _Canceller()

    def call_at(
        self,
        when: float,
        callback: Callable,
        *args: Any,
        context: Optional[Context] = None,
    ):
        logger.debug(f"Calling {callback} at %f...", when)
        self._to_be_called.append((when, callback, args, context))

        # re-sort list in ascending time order
        self._to_be_called.sort(key=lambda x: x[0])

    def call_soon(
        self, callback: Callable, *args: Any, context: Optional[Context] = None
    ):
        return self.call_later(0, callback, *args, context=context)

    def time(self) -> float:
        return self._time def create_test_cert_file(sanlist):
    """build an x509 certificate file

    Args:
        sanlist: list[bytes]: a list of subjectAltName values for the cert

    Returns:
        str: the path to the file
    """
    global cert_file_count
    csr_filename = "server.csr"
    cnf_filename = "server.%i.cnf" % (cert_file_count,)
    cert_filename = "server.%i.crt" % (cert_file_count,)
    cert_file_count += 1

    # first build a CSR
    subprocess.check_call(
        [
            "openssl",
            "req",
            "-new",
            "-key",
            get_test_key_file(),
            "-subj",
            "/",
            "-out",
            csr_filename,
        ]
    )

    # now a config file describing the right SAN entries
    sanentries = b",".join(sanlist)
    with open(cnf_filename, "wb") as f:
        f.write(CONFIG_TEMPLATE % {b"sanentries": sanentries})

    # finally the cert
    ca_key_filename = os.path.join(os.path.dirname(__file__), "tls/ca.key")
    ca_cert_filename = get_test_ca_cert_file()
    subprocess.check_call(
        [
            "openssl",
            "x509",
            "-req",
            "-in",
            csr_filename,
            "-CA",
            ca_cert_filename,
            "-CAkey",
            ca_key_filename,
            "-set_serial",
            "1",
            "-extfile",
            cnf_filename,
            "-out",
            cert_filename,
        ]
    )

    return cert_filename def get_test_ca_cert_file():
    """Get the path to the test CA cert

    The keypair is generated with:

        openssl genrsa -out ca.key 2048
        openssl req -new -x509 -key ca.key -days 3650 -out ca.crt \
            -subj '/CN=synapse test CA'
    """
    return os.path.join(os.path.dirname(__file__), "tls/ca.crt") def get_test_key_file():
    """get the path to the test key

    The key file is made with:

        openssl genrsa -out server.key 2048
    """
    return os.path.join(os.path.dirname(__file__), "tls/server.key")
import asyncio
import ssl
from asyncio import AbstractEventLoop, BaseTransport, Protocol, Task
from typing import Optional, Tuple, cast
from sygnal.exceptions import ProxyConnectError
from sygnal.helper.proxy.proxy_asyncio import HttpConnectProtocol
from tests import testutils
from tests.asyncio_test_helpers import (
    EchoProtocol,
    MockProtocol,
    MockTransport,
    TimelessEventLoopWrapper,
)
from tests.twisted_test_helpers import (
    create_test_cert_file,
    get_test_ca_cert_file,
    get_test_key_file,
)
            fake_proxy.buffer,
            b"CONNECT example.org:443 HTTP/1.0\r\n"
            b"Proxy-Authorization: basic dXNlcjpzZWNyZXQ=\r\n\r\n",
        )
        # Reset the proxy mock
        fake_proxy.reset_mock()

        # For the sake of this test, pretend the credentials are incorrect so
        # send a sad response with a HTML error page
        fake_proxy.pretend_to_receive(
            b"HTTP/1.0 401 Unauthorised\r\n\r\n<HTML>... some error here ...</HTML>"
        )

        # advance event loop because we have to let coroutines be executed
        self.loop.advance(1.0)

        # *now* this future should have completed
        self.assertTrue(switch_over_task.done())

        # but we should have failed
        self.assertIsInstance(switch_over_task.exception(), ProxyConnectError)

        # check our protocol did not receive anything, because it was an HTTP-
        # level error, not actually a connection to our target.
        self.assertEqual(fake_protocol.received_bytes, b"")


class AsyncioHttpProxyTLSTest(testutils.TestCase):
    """Test that using a HTTPS proxy works.

    This is a bit convoluted to try and test that we don't hit a race where the
    new client protocol can receive data before `connection_made` is called,
    which can cause problems if it tries to write to the connection that it
    hasn't been given yet.
    """

    def config_setup(self, config):
        super().config_setup(config)
        config["apps"]["com.example.spqr"] = {
            "type": "tests.test_pushgateway_api_v1.TestPushkin"
        }
        self.base_loop = asyncio.new_event_loop()
        augmented_loop = TimelessEventLoopWrapper(self.base_loop)  # type: ignore
        asyncio.set_event_loop(cast(AbstractEventLoop, augmented_loop))

        self.loop = augmented_loop

        self.proxy_context = ssl.create_default_context()
        self.proxy_context.load_verify_locations(get_test_ca_cert_file())
        self.proxy_context.set_ciphers("DEFAULT")

    def make_fake_proxy(
        self,
        host: str,
        port: int,
        proxy_credentials: Optional[Tuple[str, str]],
    ) -> Tuple[EchoProtocol, MockTransport, "Task[Tuple[BaseTransport, Protocol]]"]:
        # Task[Tuple[MockTransport, MockProtocol]]

        # make a fake proxy
        fake_proxy = MockTransport()

        # We connect with an echo protocol to test that we can always write when
        # we receive data.
        fake_protocol = EchoProtocol()

        # create a HTTP CONNECT proxy client protocol
        http_connect_protocol = HttpConnectProtocol(
            target_hostport=(host, port),
            proxy_credentials=proxy_credentials,
            protocol_factory=lambda: fake_protocol,
            sslcontext=self.proxy_context,
            loop=None,
        )
        switch_over_task = self.loop.create_task(
            http_connect_protocol.switch_over_when_ready()
        )
        # check the task is not somehow already marked as done before we even
        # receive anything.
        self.assertFalse(switch_over_task.done())
        # connect the proxy client to the proxy
        fake_proxy.set_protocol(http_connect_protocol)
        http_connect_protocol.connection_made(fake_proxy)
        return fake_protocol, fake_proxy, switch_over_task

    def test_connect_no_credentials(self):
        """
        Tests the proxy connection procedure when there is no basic auth.
        """
        host = "example.org"
        port = 443
        proxy_credentials = None
        fake_protocol, fake_proxy, switch_over_task = self.make_fake_proxy(
            host, port, proxy_credentials
        )

        # Check that the proxy got the proper CONNECT request.
        self.assertEqual(fake_proxy.buffer, b"CONNECT example.org:443 HTTP/1.0\r\n\r\n")
        # Reset the proxy mock
        fake_proxy.reset_mock()

        # pretend we got a happy response
        fake_proxy.pretend_to_receive(b"HTTP/1.0 200 Connection Established\r\n\r\n")

        # Since we're talking TLS we need to create a server TLS connection that
        # we can use to talk to each other.
        context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
        context.load_cert_chain(
            create_test_cert_file([b"DNS:example.org"]), keyfile=get_test_key_file()
        )
        context.set_ciphers("DEFAULT")

        # Note that we have to use a different event loop wrapper here as we
        # want that server side setup to finish before the client side setup, so
        # that we can trigger any races.
        server_loop = TimelessEventLoopWrapper(self.base_loop)  # type: ignore
        server_transport = MockTransport()
        proxy_ft = server_loop.create_task(
            server_loop.start_tls(
                server_transport,

                MockProtocol(),