codekingpro/portable-devtools

1"""2This addon determines the next protocol layer in our proxy stack.3Whenever a protocol layer in the proxy wants to pass a connection to a child layer and isn't sure which protocol comes4next, it calls the `next_layer` hook, which ends up here.5For example, if mitmproxy runs as a regular proxy, we first need to determine if6new clients start with a TLS handshake right away (Secure Web Proxy) or send a plaintext HTTP CONNECT request.7This addon here peeks at the incoming bytes and then makes a decision based on proxy mode, mitmproxy options, etc.8 9For a typical HTTPS request, this addon is called a couple of times: First to determine that we start with an HTTP layer10which processes the `CONNECT` request, a second time to determine that the client then starts negotiating TLS, and a11third time when we check if the protocol within that TLS stream is actually HTTP or something else.12 13Sometimes it's useful to hardcode specific logic in next_layer when one wants to do fancy things.14In that case it's not necessary to modify mitmproxy's source, adding a custom addon with a next_layer event hook15that sets nextlayer.layer works just as well.16"""17 18from __future__ import annotations19 20import logging21import re22import sys23from collections.abc import Iterable24from collections.abc import Sequence25from typing import Any26from typing import cast27 28from mitmproxy import ctx29from mitmproxy.connection import Address30from mitmproxy.net.tls import starts_like_dtls_record31from mitmproxy.net.tls import starts_like_tls_record32from mitmproxy.proxy import layer33from mitmproxy.proxy import layers34from mitmproxy.proxy import mode_specs35from mitmproxy.proxy import tunnel36from mitmproxy.proxy.context import Context37from mitmproxy.proxy.layer import Layer38from mitmproxy.proxy.layers import ClientQuicLayer39from mitmproxy.proxy.layers import ClientTLSLayer40from mitmproxy.proxy.layers import DNSLayer41from mitmproxy.proxy.layers import HttpLayer42from mitmproxy.proxy.layers import modes43from mitmproxy.proxy.layers import RawQuicLayer44from mitmproxy.proxy.layers import ServerQuicLayer45from mitmproxy.proxy.layers import ServerTLSLayer46from mitmproxy.proxy.layers import TCPLayer47from mitmproxy.proxy.layers import UDPLayer48from mitmproxy.proxy.layers.http import HTTPMode49from mitmproxy.proxy.layers.quic import quic_parse_client_hello_from_datagrams50from mitmproxy.proxy.layers.tls import dtls_parse_client_hello51from mitmproxy.proxy.layers.tls import HTTP_ALPNS52from mitmproxy.proxy.layers.tls import parse_client_hello53from mitmproxy.tls import ClientHello54 55if sys.version_info < (3, 11):56    from typing_extensions import assert_never57else:58    from typing import assert_never59 60logger = logging.getLogger(__name__)61 62 63def stack_match(64    context: Context, layers: Sequence[type[Layer] | tuple[type[Layer], ...]]65) -> bool:66    if len(context.layers) != len(layers):67        return False68    return all(69        expected is Any or isinstance(actual, expected)70        for actual, expected in zip(context.layers, layers)71    )72 73 74class NeedsMoreData(Exception):75    """Signal that the decision on which layer to put next needs to be deferred within the NextLayer addon."""76 77 78class NextLayer:79    ignore_hosts: Sequence[re.Pattern] = ()80    allow_hosts: Sequence[re.Pattern] = ()81    tcp_hosts: Sequence[re.Pattern] = ()82    udp_hosts: Sequence[re.Pattern] = ()83 84    def configure(self, updated):85        if "tcp_hosts" in updated:86            self.tcp_hosts = [87                re.compile(x, re.IGNORECASE) for x in ctx.options.tcp_hosts88            ]89        if "udp_hosts" in updated:90            self.udp_hosts = [91                re.compile(x, re.IGNORECASE) for x in ctx.options.udp_hosts92            ]93        if "allow_hosts" in updated or "ignore_hosts" in updated:94            self.ignore_hosts = [95                re.compile(x, re.IGNORECASE) for x in ctx.options.ignore_hosts96            ]97            self.allow_hosts = [98                re.compile(x, re.IGNORECASE) for x in ctx.options.allow_hosts99            ]100 101    def next_layer(self, nextlayer: layer.NextLayer):102        if nextlayer.layer:103            return  # do not override something another addon has set.104        try:105            nextlayer.layer = self._next_layer(106                nextlayer.context,107                nextlayer.data_client(),108                nextlayer.data_server(),109            )110        except NeedsMoreData:111            logger.debug(112                f"Deferring layer decision, not enough data: {nextlayer.data_client().hex()!r}"113            )114 115    def _next_layer(116        self, context: Context, data_client: bytes, data_server: bytes117    ) -> Layer | None:118        assert context.layers119 120        def s(*layers):121            return stack_match(context, layers)122 123        tcp_based = context.client.transport_protocol == "tcp"124        udp_based = context.client.transport_protocol == "udp"125 126        # 1)  check for --ignore/--allow127        if self._ignore_connection(context, data_client, data_server):128            return (129                layers.TCPLayer(context, ignore=not ctx.options.show_ignored_hosts)130                if tcp_based131                else layers.UDPLayer(context, ignore=not ctx.options.show_ignored_hosts)132            )133 134        # 2)  Handle proxy modes with well-defined next protocol135        # 2a) Reverse proxy: derive from spec136        if s(modes.ReverseProxy):137            return self._setup_reverse_proxy(context, data_client)138        # 2b) Explicit HTTP proxies139        if s((modes.HttpProxy, modes.HttpUpstreamProxy)):140            return self._setup_explicit_http_proxy(context, data_client)141 142        # 3)  Handle security protocols143        # 3a) TLS/DTLS144        is_tls_or_dtls = (145            tcp_based146            and starts_like_tls_record(data_client)147            or udp_based148            and starts_like_dtls_record(data_client)149        )150        if is_tls_or_dtls:151            server_tls = ServerTLSLayer(context)152            server_tls.child_layer = ClientTLSLayer(context)153            return server_tls154        # 3b) QUIC155        if udp_based and _starts_like_quic(data_client, context.server.address):156            server_quic = ServerQuicLayer(context)157            server_quic.child_layer = ClientQuicLayer(context)158            return server_quic159 160        # 4)  Check for --tcp/--udp161        if tcp_based and self._is_destination_in_hosts(context, self.tcp_hosts):162            return layers.TCPLayer(context)163        if udp_based and self._is_destination_in_hosts(context, self.udp_hosts):164            return layers.UDPLayer(context)165 166        # 5)  Handle application protocol167        # 5a) Do we have a known ALPN negotiation?168        if context.client.alpn:169            if context.client.alpn in HTTP_ALPNS:170                return layers.HttpLayer(context, HTTPMode.transparent)171            elif context.client.tls_version == "QUICv1":172                # TODO: Once we support more QUIC-based protocols, relax force_raw here.173                return layers.RawQuicLayer(context, force_raw=True)174        # 5b) Is it DNS?175        if context.server.address and context.server.address[1] in (53, 5353):176            return layers.DNSLayer(context)177        # 5c) We have no other specialized layers for UDP, so we fall back to raw forwarding.178        if udp_based:179            return layers.UDPLayer(context)180        # 5d) Check for raw tcp mode.181        probably_no_http = (182            # the first three bytes should be the HTTP verb, so A-Za-z is expected.183            len(data_client) < 3184            # HTTP would require whitespace...185            or b" " not in data_client186            # ...and that whitespace needs to be in the first line.187            or (data_client.find(b" ") > data_client.find(b"\n"))188            or not data_client[:3].isalpha()189            # a server greeting would be uncharacteristic.190            or data_server191            or data_client.startswith(b"SSH")192        )193        if ctx.options.rawtcp and probably_no_http:194            return layers.TCPLayer(context)195        # 5c) Assume HTTP by default.196        return layers.HttpLayer(context, HTTPMode.transparent)197 198    def _ignore_connection(199        self,200        context: Context,201        data_client: bytes,202        data_server: bytes,203    ) -> bool | None:204        """205        Returns:206            True, if the connection should be ignored.207            False, if it should not be ignored.208 209        Raises:210            NeedsMoreData, if we need to wait for more input data.211        """212        if not ctx.options.ignore_hosts and not ctx.options.allow_hosts:213            return False214        # Special handling for wireguard mode: if the hostname is "10.0.0.53", do not ignore the connection215        if isinstance(216            context.client.proxy_mode, mode_specs.WireGuardMode217        ) and context.server.address == ("10.0.0.53", 53):218            return False219        hostnames: list[str] = []220        if context.server.peername:221            host, port, *_ = context.server.peername222            hostnames.append(f"{host}:{port}")223        if context.server.address:224            host, port, *_ = context.server.address225            hostnames.append(f"{host}:{port}")226 227            # We also want to check for TLS SNI and HTTP host headers, but in order to ignore connections based on that228            # they must have a destination address. If they don't, we don't know how to establish an upstream connection229            # if we ignore.230            if host_header := self._get_host_header(context, data_client, data_server):231                if not re.search(r":\d+$", host_header):232                    host_header = f"{host_header}:{port}"233                hostnames.append(host_header)234            if (235                client_hello := self._get_client_hello(context, data_client)236            ) and client_hello.sni:237                hostnames.append(f"{client_hello.sni}:{port}")238            if context.client.sni:239                # Hostname may be allowed, TLS is already established, and we have another next layer decision.240                hostnames.append(f"{context.client.sni}:{port}")241 242        if not hostnames:243            return False244 245        if ctx.options.allow_hosts:246            not_allowed = not any(247                re.search(rex, host, re.IGNORECASE)248                for host in hostnames249                for rex in ctx.options.allow_hosts250            )251            if not_allowed:252                return True253 254        if ctx.options.ignore_hosts:255            ignored = any(256                re.search(rex, host, re.IGNORECASE)257                for host in hostnames258                for rex in ctx.options.ignore_hosts259            )260            if ignored:261                return True262 263        return False264 265    @staticmethod266    def _get_host_header(267        context: Context,268        data_client: bytes,269        data_server: bytes,270    ) -> str | None:271        """272        Try to read a host header from data_client.273 274        Returns:275            The host header value, or None, if no host header was found.276 277        Raises:278            NeedsMoreData, if the HTTP request is incomplete.279        """280        if context.client.transport_protocol != "tcp" or data_server:281            return None282 283        host_header_expected = re.match(284            rb"[A-Z]{3,}.+HTTP/", data_client, re.IGNORECASE285        )286        if host_header_expected:287            if m := re.search(288                rb"\r\n(?:Host:\s+(.+?)\s*)?\r\n", data_client, re.IGNORECASE289            ):290                if host := m.group(1):291                    return host.decode("utf-8", "surrogateescape")292                else:293                    return None  # \r\n\r\n - header end came first.294            else:295                raise NeedsMoreData296        else:297            return None298 299    @staticmethod300    def _get_client_hello(context: Context, data_client: bytes) -> ClientHello | None:301        """302        Try to read a TLS/DTLS/QUIC ClientHello from data_client.303 304        Returns:305            A complete ClientHello, or None, if no ClientHello was found.306 307        Raises:308            NeedsMoreData, if the ClientHello is incomplete.309        """310        match context.client.transport_protocol:311            case "tcp":312                if starts_like_tls_record(data_client):313                    try:314                        ch = parse_client_hello(data_client)315                    except ValueError:316                        pass317                    else:318                        if ch is None:319                            raise NeedsMoreData320                        return ch321                return None322            case "udp":323                try:324                    return quic_parse_client_hello_from_datagrams([data_client])325                except ValueError:326                    pass327 328                try:329                    ch = dtls_parse_client_hello(data_client)330                except ValueError:331                    pass332                else:333                    if ch is None:334                        raise NeedsMoreData335                    return ch336                return None337            case _:  # pragma: no cover338                assert_never(context.client.transport_protocol)339 340    @staticmethod341    def _setup_reverse_proxy(context: Context, data_client: bytes) -> Layer:342        spec = cast(mode_specs.ReverseMode, context.client.proxy_mode)343        stack = tunnel.LayerStack()344 345        match spec.scheme:346            case "http":347                if starts_like_tls_record(data_client):348                    stack /= ClientTLSLayer(context)349                stack /= HttpLayer(context, HTTPMode.transparent)350            case "https":351                if context.client.transport_protocol == "udp":352                    stack /= ServerQuicLayer(context)353                    stack /= ClientQuicLayer(context)354                    stack /= HttpLayer(context, HTTPMode.transparent)355                else:356                    stack /= ServerTLSLayer(context)357                    if starts_like_tls_record(data_client):358                        stack /= ClientTLSLayer(context)359                    stack /= HttpLayer(context, HTTPMode.transparent)360 361            case "tcp":362                if starts_like_tls_record(data_client):363                    stack /= ClientTLSLayer(context)364                stack /= TCPLayer(context)365            case "tls":366                stack /= ServerTLSLayer(context)367                if starts_like_tls_record(data_client):368                    stack /= ClientTLSLayer(context)369                stack /= TCPLayer(context)370 371            case "udp":372                if starts_like_dtls_record(data_client):373                    stack /= ClientTLSLayer(context)374                stack /= UDPLayer(context)375            case "dtls":376                stack /= ServerTLSLayer(context)377                if starts_like_dtls_record(data_client):378                    stack /= ClientTLSLayer(context)379                stack /= UDPLayer(context)380 381            case "dns":382                # TODO: DNS-over-TLS / DNS-over-DTLS383                # is_tls_or_dtls = (384                #     context.client.transport_protocol == "tcp" and starts_like_tls_record(data_client)385                #     or386                #     context.client.transport_protocol == "udp" and starts_like_dtls_record(data_client)387                # )388                # if is_tls_or_dtls:389                #     stack /= ClientTLSLayer(context)390                stack /= DNSLayer(context)391 392            case "http3":393                stack /= ServerQuicLayer(context)394                stack /= ClientQuicLayer(context)395                stack /= HttpLayer(context, HTTPMode.transparent)396            case "quic":397                stack /= ServerQuicLayer(context)398                stack /= ClientQuicLayer(context)399                stack /= RawQuicLayer(context, force_raw=True)400 401            case _:  # pragma: no cover402                assert_never(spec.scheme)403 404        return stack[0]405 406    @staticmethod407    def _setup_explicit_http_proxy(context: Context, data_client: bytes) -> Layer:408        stack = tunnel.LayerStack()409 410        if context.client.transport_protocol == "udp":411            stack /= layers.ClientQuicLayer(context)412        elif starts_like_tls_record(data_client):413            stack /= layers.ClientTLSLayer(context)414 415        if isinstance(context.layers[0], modes.HttpUpstreamProxy):416            stack /= layers.HttpLayer(context, HTTPMode.upstream)417        else:418            stack /= layers.HttpLayer(context, HTTPMode.regular)419 420        return stack[0]421 422    @staticmethod423    def _is_destination_in_hosts(context: Context, hosts: Iterable[re.Pattern]) -> bool:424        return any(425            (context.server.address and rex.search(context.server.address[0]))426            or (context.client.sni and rex.search(context.client.sni))427            for rex in hosts428        )429 430 431# https://www.iana.org/assignments/quic/quic.xhtml432KNOWN_QUIC_VERSIONS = {433    0x00000001,  # QUIC v1434    0x51303433,  # Google QUIC Q043435    0x51303436,  # Google QUIC Q046436    0x51303530,  # Google QUIC Q050437    0x6B3343CF,  # QUIC v2438    0x709A50C4,  # QUIC v2 draft codepoint439}440 441TYPICAL_QUIC_PORTS = {80, 443, 8443}442 443 444def _starts_like_quic(data_client: bytes, server_address: Address | None) -> bool:445    """446    Make an educated guess on whether this could be QUIC.447    This turns out to be quite hard in practice as 1-RTT packets are hardly distinguishable from noise.448 449    Returns:450        True, if the passed bytes could be the start of a QUIC packet.451        False, otherwise.452    """453    # Minimum size: 1 flag byte + 1+ packet number bytes + 16+ bytes encrypted payload454    if len(data_client) < 18:455        return False456    if starts_like_dtls_record(data_client):457        return False458    # TODO: Add more checks here to detect true negatives.459 460    # Long Header Packets461    if data_client[0] & 0x80:462        version = int.from_bytes(data_client[1:5], "big")463        if version in KNOWN_QUIC_VERSIONS:464            return True465        # https://www.rfc-editor.org/rfc/rfc9000.html#name-versions466        # Versions that follow the pattern 0x?a?a?a?a are reserved for use in forcing version negotiation467        if version & 0x0F0F0F0F == 0x0A0A0A0A:468            return True469    else:470        # ¯\_(ツ)_/¯471        # We can't even rely on the QUIC bit, see https://datatracker.ietf.org/doc/rfc9287/.472        pass473 474    return bool(server_address and server_address[1] in TYPICAL_QUIC_PORTS)475