Skip to content

Instantly share code, notes, and snippets.

Last active December 12, 2023 07:51
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save linw1995/4630163c5f3fb2b575bb6fd50e89aa80 to your computer and use it in GitHub Desktop.
Save linw1995/4630163c5f3fb2b575bb6fd50e89aa80 to your computer and use it in GitHub Desktop.
A simple example shows how to hook response with mitmproxy in script.
# Standard Library
import asyncio
import contextlib
from pathlib import Path
from typing import Callable, Optional
# Third Party Library
import httpx
from mitmproxy.addons import core
from mitmproxy.http import HTTPFlow
from mitmproxy.master import Master
from mitmproxy.options import Options
from mitmproxy.proxy import ProxyConfig, ProxyServer
ResponseHook = Callable[[HTTPFlow], None]
CAFILE_PATH = Path("~/.mitmproxy/mitmproxy-ca.pem").expanduser().absolute()
class Interceptor:
def __init__(self):
self.response_hook: Optional[ResponseHook] = None
def hook_response(self, hook_function: ResponseHook):
self.response_hook = hook_function
self.response_hook = None
def response(self, flow: HTTPFlow):
if self.response_hook is not None:
def control(interceptor: Interceptor, proxy_port: int):
def wait_resposne(flow: HTTPFlow):
# run in main thread
print("in sync control")
print("received response url:", flow.request.url)
with httpx.Client(
proxies=f"http://localhost:{proxy_port}", verify=CAFILE_PATH
) as client:
# run in another thread
with interceptor.hook_response(wait_resposne):
resp = client.get("")
resp = client.get("")
async def async_control(interceptor: Interceptor, proxy_port: int):
def wait_resposne(flow: HTTPFlow):
print("in async control")
print("received response url:", flow.request.url)
async with httpx.AsyncClient(
) as client:
with interceptor.hook_response(wait_resposne):
resp = await client.get("")
resp = await client.get("")
async def intercept(listen_port: int = 8080) -> None:
options = Options(listen_host="", listen_port=listen_port)
master = Master(options)
master.server = server = ProxyServer(ProxyConfig(options))
# TODO: Keeps this line until the major version 7 is released
await master.running()
interceptor = Interceptor()
yield interceptor
# Don't use `Master.shutdown()`.
# It will make the process being unable to exit.
# And I don't know what is causing this.
async def main():
port = 8888
async with intercept(listen_port=port) as interceptor:
# run in other thread to avoid blocking main thread
await asyncio.to_thread(control, interceptor=interceptor, proxy_port=port)
# async function is invoked normally via await statement
await async_control(interceptor, port)
if __name__ == "__main__":
in sync control
received response url:
in async control
received response url:
name = ""
version = ""
description = ""
authors = [
{name = "林玮 (Jade Lin)", email = ""},
dependencies = [
requires-python = ">=3.9"
dynamic = ["classifiers"]
license = {text = "MIT"}
homepage = ""
requires = ["pdm-pep517"]
build-backend = "pdm.pep517.api"
Copy link

milahu commented Dec 11, 2023

fails with mitmproxy-9.0.1

ImportError: cannot import name 'ProxyConfig' from 'mitmproxy.proxy'

code search: from mitmproxy.master import Master

Copy link

@milahu This script has only been tested on the version "mitmproxy~=6.0."

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment