Class NextLayer

mitmproxy/addons/next_layer.py:78–428

class NextLayer:
    ignore_hosts: Sequence[re.Pattern] = ()
    allow_hosts: Sequence[re.Pattern] = ()
    tcp_hosts: Sequence[re.Pattern] = ()
    udp_hosts: Sequence[re.Pattern] = ()

    def configure(self, updated):
        if "tcp_hosts" in updated:
            self.tcp_hosts = [
                re.compile(x, re.IGNORECASE) for x in ctx.options.tcp_hosts
            ]
        if "udp_hosts" in updated:
            self.udp_hosts = [
                re.compile(x, re.IGNORECASE) for x in ctx.options.udp_hosts
            ]
        if "allow_hosts" in updated or "ignore_hosts" in updated:
            self.ignore_hosts = [
                re.compile(x, re.IGNORECASE) for x in ctx.options.ignore_hosts
            ]
            self.allow_hosts = [
                re.compile(x, re.IGNORECASE) for x in ctx.options.allow_hosts
            ]

    def next_layer(self, nextlayer: layer.NextLayer):
        if nextlayer.layer:
            return  # do not override something another addon has set.
        try:
            nextlayer.layer = self._next_layer(
                nextlayer.context,
                nextlayer.data_client(),
                nextlayer.data_server(),
            )
        except NeedsMoreData:
            logger.debug(
                f"Deferring layer decision, not enough data: {nextlayer.data_client().hex()!r}"
            )

    def _next_layer(
        self, context: Context, data_client: bytes, data_server: bytes
    ) -> Layer | None:
        assert context.layers

        def s(*layers):
            return stack_match(context, layers)

        tcp_based = context.client.transport_protocol == "tcp"
        udp_based = context.client.transport_protocol == "udp"

        # 1)  check for --ignore/--allow
        if self._ignore_connection(context, data_client, data_server):
            return (
                layers.TCPLayer(context, ignore=not ctx.options.show_ignored_hosts)
                if tcp_based
                else layers.UDPLayer(context, ignore=not ctx.options.show_ignored_hosts)
            )

        # 2)  Handle proxy modes with well-defined next protocol
        # 2a) Reverse proxy: derive from spec