server_side_tcp.py

import typing

from mitmproxy.ctx import options
from mitmproxy.proxy import config, RootContext, protocol

force_tcp = config.HostMatcher()


def load(l):
    l.add_option(
        "force_tcp", typing.Sequence[str], [],
        "Force plain (non-TLS) TCP with server-side greetings",
    )


    # Monkeypatch RootContext.next_layer
    RootContext.actual_next_layer = RootContext.next_layer

    def next_layer(self, top_layer):
        if force_tcp(top_layer.server_conn.address):
            return protocol.RawTCPLayer(top_layer)
        return RootContext.actual_next_layer(self, top_layer)

    RootContext.next_layer = next_layer


def configure(updated):
    if "force_tcp" in updated:
        global force_tcp
        force_tcp = config.HostMatcher(options.force_tcp)


def done():
    RootContext.next_layer = RootContext.actual_next_layer
    del RootContext.actual_next_layer