Alexey Alter-Pesotskiy

2 minute read

Mitmproxy: tips, tricks and scripts

Precondition

Let’s put this code at the top of our mitmproxy script file:

from mitmproxy import http
import json
import time
import os

server_folder = "mitmproxy_data"

def parse_json(content: bytes):
  try:
    return json.loads(content.decode("utf-8"))
  except:
    return None

Tips, tricks, scripts

Filter by host and endpoints

def request(flow: http.HTTPFlow) -> None:
  content = flow.request.content
  if any(url in flow.request.url for url in target_list()):
    print(content)

def response(flow: http.HTTPFlow) -> None:
  content = flow.response.content
  if any(url in flow.request.url for url in target_list()):
    print(content)

def target_list():
  return ["foo.bar", "sample-website.com/test/endpoint"]

Catch all json requests and responses

def request(flow: http.HTTPFlow) -> None:
  content = parse_json(flow.request.content)
  if content is not None:
    print(content)

def response(flow: http.HTTPFlow) -> None:
  content = parse_json(flow.response.content)
  if content is not None:
    print(content)

Embed tracking beacon in jsons

def request(flow: http.HTTPFlow) -> None:
  content = parse_json(flow.request.content)
  if content is not None:
    content = add_tracker(flow, content)
    print(content)

def response(flow: http.HTTPFlow) -> None:
  content = parse_json(flow.response.content)
  if content is not None:
    content = add_tracker(flow, content)
    print(content)

def add_tracker(flow: http.HTTPFlow, content: json):
  if isinstance(content, list):
    content += [{"mitmproxy_tracker": flow.request.url}]
  else:
    content["mitmproxy_tracker"] = flow.request.url
  return content

Write json data to a file

def request(flow: http.HTTPFlow) -> None:
  content = parse_json(flow.request.content)
  if content is not None:
    file = open(file_path("request"), 'w')
    file.write(beautify(content))
    file.close()

def response(flow: http.HTTPFlow) -> None:
  content = parse_json(flow.response.content)
  if content is not None:
    file = open(file_path("response"), 'w')
    file.write(beautify(content))
    file.close()

def file_path(type: str):
  os.makedirs(server_folder, exist_ok=True)
  return server_folder + "/" + type + "_%s.json" % int(time.time())

def beautify(content: json):
  return json.dumps(content, indent=4, sort_keys=True)

Return all jsons combined into one

def request(flow: http.HTTPFlow) -> None:
  if "mitmproxy/collect" in flow.request.url:
    jsons = []
    for file in os.listdir(server_folder):
      file_path = os.path.join(server_folder, file)
      content = ""
      with open(file_path) as file:
        for line in file:
          content += line
      jsons.append(content)

    json = "[" + ",".join(jsons) + "]"
    flow.response = http.Response.make(
      200,
      json,
      {"Content-Type": "application/json"}
    )

Clear all saved files

  if "mitmproxy/clear" in flow.request.url:
    for file in os.listdir(server_folder):
      os.remove(os.path.join(server_folder, file))
    flow.response = http.Response.make(200)

Kill request

def request(flow: http.HTTPFlow) -> None:
  block_hosts = ["appldnld.apple.com", "mesu.apple.com"]
  if any(host in flow.request.host for host in block_hosts):
    flow.kill

Update headers and queries in requests and responses

def request(flow: http.HTTPFlow) -> None:
  flow.request.query["mitm"] = "proxy"
  flow.request.headers["newheader"] = "foo"

def response(flow: http.HTTPFlow) -> None:
  flow.response.query["proxy"] = "mitm"
  flow.response.headers["newheader"] = "bar"

Change response status code

def response(flow: http.HTTPFlow) -> None:
  flow.response.status_code = 503

Emulate slow connection

def response(flow: http.HTTPFlow) -> None:
  time.sleep(5.0)

Redirect requests

def request(flow: http.HTTPFlow) -> None:
  flow.request.host = "127.0.0.1"
  flow.request.scheme = "http"
  flow.request.port = 4567

Anything else?

Stay in touch (:

Updated: