python mitmproxy例子

1、添加头部： from mitmproxy import http

def response(flow: http.HTTPFlow) -> None: flow.response.headers["newheader"] = "foo"

2、添加头部，类的方式： from mitmproxy import http

class AddHeader: def response(self, flow: http.HTTPFlow) -> None: flow.response.headers["newheader"] = "foo"

addons = [AddHeader()] 3、contentviews: from mitmproxy import contentviews

class ViewSwapCase(contentviews.View): name = "swapcase" content_types = ["text/plain"]

def __call__(self, data, **metadata) -> contentviews.TViewResult:
    return "case-swapped text", contentviews.format_text(data.swapcase())

view = ViewSwapCase()

def load(l): contentviews.add(view)

def done(): contentviews.remove(view)

4、custom_option from mitmproxy import ctx

def load(l): ctx.log.info("Registering option 'custom'") l.add_option("custom", bool, False, "A custom option")

def configure(updated): if "custom" in updated: ctx.log.info("custom option value: %s" % ctx.options.custom)

5、flowfilter from mitmproxy import flowfilter from mitmproxy import ctx, http

class Filter: def init(self): self.filter: flowfilter.TFilter = None

def configure(self, updated):
    self.filter = flowfilter.parse(ctx.options.flowfilter)

def load(self, l):
    l.add_option(
        "flowfilter", str, "", "Check that flow matches filter."
    )

def response(self, flow: http.HTTPFlow) -> None:
    if flowfilter.match(self.filter, flow):
        ctx.log.info("Flow matches filter:")
        ctx.log.info(flow)

addons = [Filter()]

6、替换内容： from mitmproxy import http

def response(flow: http.HTTPFlow) -> None: reflector = b"<style>body {transform: scaleX(-1);}</style></head>" flow.response.content = flow.response.content.replace(b"</head>", reflector)

7、呵呵 from mitmproxy import http

def response(flow: http.HTTPFlow) -> None: flow.response.content = b"hehe"

8、io?? from mitmproxy import io from mitmproxy.exceptions import FlowReadException import pprint import sys

with open(sys.argv[1], "rb") as logfile: freader = io.FlowReader(logfile) pp = pprint.PrettyPrinter(indent=4) try: for f in freader.stream(): print(f) print(f.request.host) pp.pprint(f.get_state()) print("") except FlowReadException as e: print("Flow file corrupted: {}".format(e))

-- import random import sys from mitmproxy import io, http import typing # noqa

class Writer: def init(self, path: str) -> None: self.f: typing.IO[bytes] = open(path, "wb") self.w = io.FlowWriter(self.f)

def response(self, flow: http.HTTPFlow) -> None:
    if random.choice([True, False]):
        self.w.add(flow)

def done(self):
    self.f.close()

addons = [Writer(sys.argv[1])]

9、修改： import re from urllib.parse import urljoin

def response(flow):

if "Content-Type" in flow.response.headers and flow.response.headers["Content-Type"].find("text/html") != -1:
    pageUrl = flow.request.url
    pageText = flow.response.text
    pattern = (r"<a\s+(?:[^>]*?\s+)?href=(?P<delimiter>[\"'])"
               r"(?P<link>(?!https?:\/\/|ftps?:\/\/|\/\/|#|javascript:|mailto:).*?)(?P=delimiter)")
    rel_matcher = re.compile(pattern, flags=re.IGNORECASE)
    rel_matches = rel_matcher.finditer(pageText)
    map_dict = {}
    for match_num, match in enumerate(rel_matches):
        (delimiter, rel_link) = match.group("delimiter", "link")
        abs_link = urljoin(pageUrl, rel_link)
        map_dict["{0}{1}{0}".format(delimiter, rel_link)] = "{0}{1}{0}".format(delimiter, abs_link)
    for map in map_dict.items():
        pageText = pageText.replace(*map)
        # Uncomment the following to print the expansion mapping
        # print("{0} -> {1}".format(*map))
    flow.response.text = pageText

10、载入： from mitmproxy import ctx

def load(l): ctx.log.info("This is some informative text.") ctx.log.warn("This is a warning.") ctx.log.error("This is an error.") 11、修改query:

from mitmproxy import http

def request(flow: http.HTTPFlow) -> None: flow.request.query["mitmproxy"] = "rocks" 12、修改form: from mitmproxy import http

def request(flow: http.HTTPFlow) -> None: if flow.request.urlencoded_form: # If there's already a form, one can just add items to the dict: flow.request.urlencoded_form["mitmproxy"] = "rocks" else: # One can also just pass new form data. # This sets the proper content type and overrides the body. flow.request.urlencoded_form = [ ("foo", "bar") ] 13\重定向： """ This example shows two ways to redirect flows to another server. """ from mitmproxy import http

def request(flow: http.HTTPFlow) -> None: # pretty_host takes the "Host" header of the request into account, # which is useful in transparent mode where we usually only have the IP # otherwise. if flow.request.pretty_host == "example.org": flow.request.host = "mitmproxy.org"

-- from mitmproxy import http

def request(flow: http.HTTPFlow) -> None: # pretty_url takes the "Host" header of the request into account, which # is useful in transparent mode where we usually only have the IP otherwise.

if flow.request.pretty_url == "http://example.com/path":
    flow.response = http.HTTPResponse.make(
        200,  # (optional) status code
        b"Hello World",  # (optional) content
        {"Content-Type": "text/html"}  # (optional) headers
    )