python-Mitmproxy抓包

一、使用

安装pip install mitmproxy
mitmproxy 是具有控制台界面的交互式，支持SSL的拦截代理
mitmdump是mitmproxy的命令行版本。想想tcpdump为HTTP
mitmweb 是一个基于web的界面，适用于mitmproxy
mitmproxy(mac)、mitmdump、mitmweb(win) 这三个命令中的任意一个即可

mitmweb -s mitm.py 命令行启动默认端口8080
mitmweb -p 8888 -s mitm.py 指定端口8888
ctrl+c退出

启动后设置电脑或手机代理(电脑ip，端口8888)，安装证书
打开 cmd，执行 "C:Program Files (x86)GoogleChromeApplicationchrome.exe" --proxy-server=10.12.2.28:8888 --ignore-certificate-errors

二、过滤、修改

"""
flow.request.scheme  请求协议
flow.request.host    请求host
flow.request.url     请求URL链接
flow.request.method  请求方法
flow.request.query   请求URL查询参数
flow.request.path    请求URL https://www.baidu.com/ 
flow.request.path_components  #请求URL不包含域名的元祖 ('project', 'classify', 'list')
flow.request.urlencoded_form  请求POST数据
flow.response.status_code  HTTP响应状态码
flow.response.headers    HTTP响应头信息
flow.response.get_text   HTTP响应内容

"""

class Counter:
    def __init__(self):
        self.result = {}  # 存接口请求和返回信息
        # url filter  去掉
        self.url_filter = ['baidu.com','qq.com','360']
        # url screen  仅访问
        self.url_race = ['10.162.16.39:8091']
        # http static extension
        self.static_ext = ['js', 'css', 'ico', 'jpg', 'png', 'gif', 'jpeg', 'bmp','xml']
        # http Content-Type
        self.static_files = ['text/css','image/jpeg', 'image/gif','image/png','text/html','application/octet-stream','application/x-protobuf']
        # http Content-Type media resource files type
        self.media_types = ['image', 'video', 'audio']

    def parser_data(self,query,data = {}):
        for key, value in query.items():
            data[key] = value
        return data

    def get_extension(self, url_tup):
        if not url_tup:
            return ''
        else:
            end_path = url_tup[-1]
            split_ext = end_path.split('.')       #1148e88a9d97.jpg  #list
            return '' if not split_ext or len(split_ext) == 1 else split_ext[-1]

    # 拒绝连接
    def http_connect(self, flow: mitmproxy.http.HTTPFlow):
        for i in self.url_filter:   #过滤url
            if i in flow.request.host:
                flow.response = http.HTTPResponse.make(404)

    #存在筛选就返回true拦截，Flase通过
    def capture_pass(self,request,response):
        if self.url_race:
            if request.host not in self.url_race: #筛选url
                return True
        url_tup = request.path_components #获取url的tup
        extension = self.get_extension(url_tup)
        if extension in self.static_ext:  #判断后缀
            return True
        try:
            content_type = response.headers['Content-Type'].split(';')[0]
            if not content_type:
                return False
            elif content_type in self.static_files:   #判断Content-Type
                return True
            else:
                http_mime_type = content_type.split('/')[0]
                if http_mime_type in self.media_types:  #判断Content-Type的files type
                    return True
                else:
                    return False
        except Exception:
            return False



    def request(self, flow: mitmproxy.http.HTTPFlow):
        request = flow.request
        # 修改请求头
        # request.headers["shuzf"] = "shuzf"
        # # 修改get参数
        # if "shuzf" in flow.request.query.keys():
        #     request.query.set_all("shuzf", ["舒志福"])
        # # 修改post参数
        # if "shuzf" in flow.request.urlencoded_form.keys():
        #     request.urlencoded_form.set_all('shuzf', '舒志福')
        scheme = request.scheme
        domain = request.host
        self.result['url'] = parse.unquote(request.url)  # url解码
        self.result['method'] = request.method
        self.result['request_headers'] = {}
        for item in request.headers:
            self.result['request_headers'][item] = request.headers[item]
        self.result['get_data'] = self.parser_data(request.query)  # 将表单转字典
        self.result['post_data'] = self.parser_data(request.urlencoded_form)  # 将表单转字典

    def response(self, flow: mitmproxy.http.HTTPFlow):
        request = flow.request
        response = flow.response
        # # 修改返回头
        # response.headers["shuzf"] = "shuzf"
        # # 修改返回体
        # text = response.text
        # text = text.replace("shuzf", "舒志福")
        # flow.response.set_text(text)
        if not self.capture_pass(request,response):
            print(request.url)
            self.result['status_code '] = response.status_code
            self.result['response_headers'] = {}
            for item in response.headers:
                self.result['response_headers'][item] = response.headers[item]
            # HTTPResponse内部使用了iso-8859-1编码，先进行解码为Unicode再进行utf-8编码  response.text.encode("iso-8859-1").decode("utf-8")
            self.result['response_content'] = response.text 
            # 添加result至数据库
            new_url = Proxy(url=self.result['url'],res=self.result['response_content'], content=json.dumps(self.result))

            session.add(new_url)
            session.commit()

            # 关闭session:
            # session.close()

addons = [Counter()]  # 实例类