start mitmproxy in background programmatically

start mitmproxy in background programmatically

from mitmproxy.options import Options
from mitmproxy.proxy.config import ProxyConfig
from mitmproxy.proxy.server import ProxyServer
from mitmproxy.tools.dump import DumpMaster

import threading
import asyncio
import time

class Addon(object):
    def __init__(self):
        self.num = 1

    def request(self, flow):
        flow.request.headers["count"] = str(self.num)

    def response(self, flow):
        self.num = self.num + 1
        flow.response.headers["count"] = str(self.num)
        print(self.num)


# see source mitmproxy/master.py for details
def loop_in_thread(loop, m):
    asyncio.set_event_loop(loop)  # This is the key.
    m.run_loop(loop.run_forever)


if __name__ == "__main__":
    options = Options(listen_host='0.0.0.0', listen_port=8080, http2=True)
    m = DumpMaster(options, with_termlog=False, with_dumper=False)
    config = ProxyConfig(options)
    m.server = ProxyServer(config)
    m.addons.add(Addon())

    # run mitmproxy in backgroud, especially integrated with other server
    loop = asyncio.get_event_loop()
    t = threading.Thread( target=loop_in_thread, args=(loop,m) )
    t.start()

    # Other servers, such as a web server, might be started then.
    time.sleep(20)
    print('going to shutdown mitmproxy')
    m.shutdown()

Your demo is great, but I have some doubts:
I want to be able to interrupt or restart mitmproxy in the thread.
I used m.shutdown() to interrupt mitmproxy, and then I got an error when restarting the thread:
RuntimeError: Event loop is closed
This means that when I use asyncio.get_event_loop(), I get the same loop, so when I execute shutdown() to close the loop, I can’t start it again.
I tried to use asyncio.new_event_loop() to get a new loop, but this seems to conflict with the loop inside mitmproxy.

what version of mitmproxy are you using?

what version of mitmproxy are you using?

mitmproxy 5.3.0

I can't find mitmproxy.proxy.config: Import "mitmproxy.proxy.config" could not be resolved. Is it possible to update this script with the latest version @BigSully ?

Mitmproxy works with asyncio now... Here's a working solution

import asyncio
import threading
from typing import Any, Callable, Self

from mitmproxy import http
from mitmproxy.addons import default_addons, script
from mitmproxy.master import Master
from mitmproxy.options import Options


class Addon:
    def __init__(self) -> None:
        self.n_reponse = 0

    def response(self, flow: http.HTTPFlow) -> None:
        if flow.response:
            self.n_reponse += 1
            print(f"reponse {self.n_reponse}")


class ThreadedMitmProxy(threading.Thread):
    def __init__(self, user_addon: Callable, **options: Any) -> None:
        self.loop = asyncio.get_event_loop()
        self.master = Master(Options(), event_loop=self.loop)
        # replace the ScriptLoader with the user addon
        self.master.addons.add(
            *(
                user_addon() if isinstance(addon, script.ScriptLoader) else addon
                for addon in default_addons()
            )
        )
        # set the options after the addons since some options depend on addons
        self.master.options.update(**options)
        super().__init__()

    def run(self) -> None:
        self.loop.run_until_complete(self.master.run())

    def __enter__(self) -> Self:
        self.start()
        return self

    def __exit__(self, *_) -> None:
        self.master.shutdown()
        self.join()


if __name__ == "__main__":
    with ThreadedMitmProxy(Addon, listen_host="127.0.0.1", listen_port=8080):
        # Other stuff might be started then
        input("hit <Enter> to quit")
        print("shutdown mitmproxy")