之前一篇文章 主要是使用 mitmproxy 进行抓包,但是其实 mitmproxy 自带的 feature 远远不止于抓包,使用 mitmdump 可以自定义脚本来修改 response 返回,或者将请求结果 dump 到本地以便于之后的分析。

之前的那篇文章在 mitmdump 的时候只是简单的介绍了一下功能,并没有展开,所以有了这篇文章。mitmdump 可以理解为 mitmproxy 的命令行版本,他提供了 tcpdump 类似的功能来查看,记录,甚至编程改写 HTTP 流量。

保存流量

开启代理模式,并将所有的请求写入文件

mitmdump -w outfile

过滤保存的流量

-n 参数表示不开启代理, -r 表示读入 infile,然后将将所有 match ~m post POST 流量写入 outfile 文件中。

mitmdump -nr infile -w outfile "~m post"

关于过滤的规则,可以具体参考这里

客户端重放请求

使用 -n 参数不开启代理,然后 -c filename 参数进行重放。

mitmdump -nc outfile

甚至是可以重放请求,然后将结果保存到另外的文件中

mitmdump -nc srcfile -w dstfile

添加脚本

可以在启动 mitmdump 时添加自定义的脚本用来改写请求。

mitmdump -s examples/add_header.py

如果脚本文件带有参数,则需要在 -s 参数后面增加双引号,比如 mitmdump -s "add_header.py custom_header"

组合使用

将这些参数组合一起使用

mitmdump -ns examples/add_header.py -r srcfile -w dstfile

从 srcfile 文件中加载流量,然后使用特定的脚本改写,然后将结果写入 dstfile 文件中。

实例

将请求结果保存到本地文件

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
import os
import sys

from mitmproxy import flowfilter, http, ctx


# events run ordr: start, request, responseheaders, response, error, done

class Filter:
    MOVIE_TOP250 = '/api/v2/subject_collection/movie_top250/items'

    def __init__(self, path):
        self.folder_path = path
        # 构造一个 HTTP response code
        self.http_code_ok = flowfilter.parse('~c 200')
        if not os.path.exists(self.folder_path):
            os.makedirs(self.folder_path)
        # 构造一个 URL 过滤器
        self.douban_path = flowfilter.parse(
            '~u frodo.douban.com/api/v2/subject_collection/movie_top250/items')

    # @concurrent  # Remove this and see what happens
    def request(self, flow: http.HTTPFlow):
        if flowfilter.match(self.douban_path, flow):
            if flow.request.host:
                ctx.log(
                    "handle request: %s%s" % (
                        flow.request.host, flow.request.path))

    def response(self, flow: http.HTTPFlow):
        if flowfilter.match(self.http_code_ok, flow):
            """只有 200 状态进入"""
            ctx.log('code %s' % flow.response.status_code)
            if flowfilter.match(self.MOVIE_TOP250, flow):
                if flow.response.content:
                    pretty_path = str(flow.request.path.rstrip())
                    pretty_path = pretty_path.replace('/', '_') \
                        .replace(':', '_') \
                        .replace('&', '_')
                    pretty_path = pretty_path[:250] + '.json'
                    res_content = flow.response.content.decode('utf-8')
                    path = os.path.join(self.folder_path, pretty_path)
                    with open(path, 'w+') as f:
                        f.write(res_content)


def start():
    if len(sys.argv) != 2:
        raise ValueError('Usage: -s "save_response.py path"')
    # 保存结果的 folder 路径
    return Filter(sys.argv[1])

将上面的脚本执行

/usr/local/bin/mitmdump -s "save_response.py /tmp/response_result/"

然后在结果路径中就能得到请求的豆瓣 Top250 电影结果,然后再对电影结果进行解析即可。

或者可以将请求的 webp 或者 jpg 的图全都保存到另外的文件夹中

pretty_url = flow.request.pretty_url
if pretty_url.endswith(".webp") or pretty_url.endswith('.jpg'):
    # ctx.log('pretty url %s' % flow.request.pretty_url)
    filename = os.path.join(self.folder_path,
                            os.path.basename(pretty_url))
    with open(filename, 'wb') as f:
        f.write(flow.response.content)

然后只要浏览过的图片就全都保存在本地的文件夹中了。

按规则过滤请求

mitm 的过滤都是依靠 flowfilter.py 来实现的,可以匹配的规则有如下

    The following operators are understood:

        ~q          Request
        ~s          Response

    Headers:

        Patterns are matched against "name: value" strings. Field names are
        all-lowercase.

        ~a          Asset content-type in response. Asset content types are:
                        text/javascript
                        application/x-javascript
                        application/javascript
                        text/css
                        image/*
                        application/x-shockwave-flash
        ~h rex      Header line in either request or response
        ~hq rex     Header in request
        ~hs rex     Header in response

        ~b rex      Expression in the body of either request or response
        ~bq rex     Expression in the body of request
        ~bs rex     Expression in the body of response
        ~t rex      Shortcut for content-type header.

        ~d rex      Request domain
        ~m rex      Method
        ~u rex      URL
        ~c CODE     Response code.
        rex         Equivalent to ~u rex

从这些匹配规则就能看出来过滤规则可以非常精细,比如过滤结果为 500 的请求,比如过滤 header 中 content-type 为某种类型的请求,比如按照正则去匹配 URL 等等。

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
import sys

from mitmproxy import flowfilter, http, ctx


# events run ordr: start, request, responseheaders, response, error, done

class Filter:
    MOVIE_TOP250 = '/api/v2/subject_collection/movie_top250/items'

    def __init__(self):
        # 构造一个 URL 过滤器
        self.douban_path = flowfilter.parse(
            '~u frodo.douban.com/api/v2/elendil/home_timeline')
        # 构造一个 HTTP response code
        self.http_code_ok = flowfilter.parse('~c 200')
        # Domain
        self.my_domain = flowfilter.parse('~d douban.com')
        # Method
        self.filter_mathod = flowfilter.parse('~m POST')
        # content-type header
        self.filter_content_type = flowfilter.parse('~t json')

    # @concurrent  # Remove this and see what happens
    def request(self, flow: http.HTTPFlow):
        if flowfilter.match(self.douban_path, flow):
            if flow.request.host:
                ctx.log(
                    "handle request: %s%s" % (
                        flow.request.host, flow.request.path))

    def response(self, flow: http.HTTPFlow):
        if flowfilter.match(self.http_code_ok, flow):
            """只有 200 状态进入"""
            ctx.log('code %s' % (flow.response.status_code))
        if flowfilter.match(self.my_domain, flow):
            """只有匹配域名"""
            ctx.log('domain %s' % flow.response.text)
        if flowfilter.match(self.douban_path, flow):
            """只有 特定 url 可以进入"""
            ctx.log('douban text' + flow.response.text)
            ctx.log('douban reason ' + flow.response.reason)
            ctx.log('douban http version ' + flow.response.http_version)
        pretty_url = flow.request.pretty_url
        if flowfilter.match(self.MOVIE_TOP250, flow):
            if flow.response.content:
                res_content = flow.response.content.decode('utf-8')
                ctx.log("content: " + res_content)


def start():
    if len(sys.argv) != 2:
        raise ValueError('Usage: -s "dump.py"')
    return Filter()

reference