来至菜鸟的祝福

国庆节到了，首先，祝大家国庆节快乐....（此处省略 500 字）该去哪玩呢，这是个难搞的问题，但是想了一下，作为菜鸟狗的我，还有脸玩耍吗，再加上，去哪都是看人头，天气又热，再回想自己的菜，干脆在家学习吧，正所谓弯道快，才是真的快，那个直线不会加速啊，对不起兄弟们，国庆节我决定弯道超车，废话不多说，上车

理论知识储备

参考文档：

• 官方网站：https://mitmproxy.org/

• 官方文档：https://docs.mitmproxy.org/stable/

什么是 mitmproxy？

• mitm 为 Man-In-The-Middle attack；

• mitmproxy 即为 中间人攻击代理。

为什么要用 mitmproxy？相比 Fiddler 和 Charles 它有什么优势？

• mitmproxy 不仅可以截获请求帮助开发者查看、分析，更可以通过自定义脚本进行二次开发。举例来说，利用 Fiddler 可以过滤出浏览器对某个特定 url 的请求，并查看、分析其数据，但实现不了高度定制化的需求，类似于：“截获对浏览器对该 url 的请求，将返回内容置空，并将真实的返回内容存到某个数据库，出现异常时发出邮件通知”。而对于 mitmproxy，这样的需求可以通过载入自定义 python 脚本轻松实现。

特征

• 拦截 HTTP 和 HTTPS 请求和响应并即时修改它们
• 保存完整的 HTTP 对话以供以后重播和分析
• 重播 HTTP 对话的客户端
• 重播先前记录的服务器的 HTTP 响应
• 反向代理模式将流量转发到指定的服务器
• macOS 和 Linux 上的透明代理模式
• 使用 Python 对 HTTP 流量进行脚本化更改
• 实时生成用于拦截的 SSL / TLS 证书
• 还有更多……

环境准备

模块安装

安装：

pip install mitmproxy

查看安装成功与否：

• cmd窗口，查看版本

mitmdump --version

出现以下字眼，则是成功安装了。

Mitmproxy: 7.0.4
Python:    3.9.6
OpenSSL:   OpenSSL 1.1.1l  24 Aug 2021
Platform:  Windows-10-10.0.22000-SP0

证书安装

模块安装完成后，首次运行 mitmproxy 或 mitmdump，在当前用户下面会生成几个ca证书。

从Windows用户界面的 .mitmproxy中，点击进去，可以看到有多个证书，

证书	作用
mitmproxy-ca.pem	PEM 格式的证书和私钥。
mitmproxy-ca-cert.pem	PEM 格式的证书。使用它可以在大多数非 Windows 平台上分发。
mitmproxy-ca-cert.p12	PKCS12 格式的证书。适用于 Windows（安装这个
mitmproxy-ca-cert.cer	与.pem 相同的文件，但某些 Android 设备需要扩展名。

pc 端

• mitmproxy-ca-cert.p12

移动端

• 配置好 wifi 连接之后，访问 mitm.it
• 下载对应手机系统的证书，然后安装即可。

抓包示例： (这里就不细说，官网或者百度都有教程，拒绝伸手党)

• Windows端
  • 要使用代理 + 走指定的端口哦！！！
• 手机端
  • 配置Windows端的 ip + 指定代理！！

mitmproxy 三大组件

熟悉 mitmproxy 的大佬都知道” mitmproxy“，的核心组件通常指这三种工具中的任何一种 -- 它们只是同一核心代理的不同前端。

Tools	Description
mitmproxy	供交互式界面（Windows系统不可用
mitmdump	提供简单明了的终端输出
mitmweb	提供基于浏览器的图形界面

温馨提示

mitmproxy默认绑定的端口为 127.0.0.1:8080

注意一下：

• 如果端口被占用了，会提示报错哦！

---

基础了解

mitmproxy

Windows系统不可用，这里暂不展示（别问，问就是贫）

mitmdump-------命令行模式

查看所有命令：

mitmdump --help

查看版本：

mitmdump --version

常用命令：

-p 8888         # 指定端口
-s xxx.py       # 执行指定脚本
-w outfile      # 指定输出文件
-q quiet        # 仅匹配脚本过滤后的数据包
"~m post"       # 仅匹配Post请求

带有颜色的 log：

• log，带有输出不同颜色的功能（个人觉得没有什么用
  • info 白色
  • warn 黄色
  • error 红色

注意这里要使用 cmd，使用 PowerShell 显示出来的颜色效果不完整。

mitmDemoOne.py

class Demo:
    def request(self, flow: mitmproxy.http.HTTPFlow):
        """Print different colors"""
        url = flow.request.url
        if 'testerhome' in url:
            print(type(url))
            ctx.log.info('Color White：' + url)
            ctx.log.warn('Color Yellow：' + url)
            ctx.log.error('Color Red：' + url)


addons = [
    Demo()
]

---

mitmweb------web 界面模式

监听的端口是 127.0.0.1:8080，

同时提供一个 web 交互界面在 127.0.0.1:8081。

功能介绍
基础介绍：

• 拦截

  • 修改请求前数据
  • 修改请求后数据

• 筛选

• 高亮

• 重放请求

不信你动手试试

操作模式：https://docs.mitmproxy.org/stable/concepts-modes/

脚本编写：https://docs.mitmproxy.org/stable/addons-scripting/

如何工作：https://docs.mitmproxy.org/stable/concepts-howmitmproxyworks/

测试网站：

• http://www.httpbin.org/get（如果比较慢的话，自己高台服务器，或者虚拟机直接 docker 一条命令，安装下）

常用的两个函数简单演示：

这里介绍一下使用的比较多的两个函数，其他的可以通过官方文档去进行一个系统的学习。

def request(flow):
    pass

def response(flow):
    pass

• request

common	Description
request = flow.request
request.url	url
request.host	域名
request.headers	请求头
request.method	方式：POST、GET 等
request.scheme	类型：http、https
request.path	路径，URL 除域名之外的内容
request.query	返回MultiDictView类型的数据，URL 的键值参数
request.query.keys()	获取所有请求参数键值的键
request.query.values()	获取所有请求参数键值的值
request.query.get('wd')	获取请求参数中wd 键的值（前提是要有 wb 参数
request.query.set_all('wd', ['python'])	将wd 参数的值修改为 python

修改请求头：

mitmDemoTwo.py

flow.request.headers['User-Agent'] = 'Mozilla/5.0'

将百度搜索修改为 python：

mitmDemoThree.py

def request(flow):
    if 'https://www.baidu.com' in flow.request.url:
        # 取得请求参数wd的值
        print(flow.request.query.get('wd'))
        # 获取所有请求参数键值的键
        print(list(flow.request.query.keys()))
        # 获取所有请求参数键值的值
        print(list(flow.request.query.values()))
        # 修改请求参数
        flow.request.query.set_all('wd',['python'])
        # 打印修改过后的参数
        print(flow.request.query.get('wd'))

---

• response

common	Description
response = flow.response
response.status_code	响应码
response.text	文本 (同下)
response.content	Bytes 类型
response.headers	响应头
response.cookies	响应 cookie
response.set_text()	修改响应的文本
response.get_text()	文本 (同上)
flow.response= flow.response.make(404)	响应 404

修改文本

mitmDemoFour.py

flow.response.set_text(text)

拒绝响应

mitmDemoFour.py

# 同下
flow.response = mitmproxy.http.HTTPResponse.make(401)
# 同下
flow.response= flow.response.make(404)

拒绝响应：在百度搜索 小黄人

mitmDemoFive.py

if flow.request.query.get('wd') == '小黄人':
    flow.response = mitmproxy.http.HTTPResponse.make(
        404,                                    # (optional) status code
        b"You son of a bitch, Please leave.",   # (optional) content
        {"Content-Type": "text/html"}           # (optional) headers
    )

简单应用

需求：

1. 修改请求（如果是搜索测试，则修改为 听说mtsc大会门票有折扣
2. 修改响应（将页面所有 Python 字眼 替换为 testerhome
3. 如果存在大小黄人等信息(例：大黄人、小黄人)，则拒绝响应，重定向到指定的页面

代码：

mitmDemoSix.py

# -*- encoding: utf-8 -*-
"""
@File    : mit_01.py
@Time    : 2021/10/2 20:44
@Author  : YuYe
@Software: PyCharm
"""

from mitmproxy import http


class Dome(object):
    @staticmethod
    def request(flow: http.HTTPFlow):
        fire_key = ["小黄人", "大黄人"]
        if "https://www.baidu.com" in flow.request.url:
            keyword = flow.request.query.get("wd")
            if keyword == "测试":
                flow.request.query.set_all("wd", ["听说mtsc大会门票有折扣"])
            if keyword in fire_key:
                flow.response = http.Response.make(
                    status_code=400,
                    content="""  <title>国庆节弯道超车</title>
                                 <h1>MTsc大会面基不</h1>
                                 <h2>一起去深圳吃猪脚饭啊</h2>
                                 <a>点击</a>
                                 <a href="https://www.bagevent.com/event/7689076?bag_track=sqsybanner#website_moduleId_935251" target="_blank">买票不,主办方说可以打骨折 /a>
                                 """,
                    headers={"Content-Type": "text/html"}
                )

    @staticmethod
    def response(flow: http.HTTPFlow):
        responseS = flow.response
        if flow.request.host == 'www.baidu.com':
            replace_words = ['你好', 'python', 'Python']
            text = responseS.get_text()
            text = list(map(lambda x: text.replace(x, 'testerhome'), replace_words))[0]
            text = list(map(lambda x: text.replace(x, 'testerhome'), replace_words))[1]
            text = list(map(lambda x: text.replace(x, 'testerhome'), replace_words))[2]
            print(text)
            flow.response.set_text(text=text)


addons = [
    Dome()
]




### 日常报错解决

```python
502 Bad Gateway
Certificate verification error for xxx: unable to get local issuer certificate (errno: 20, depth: 0)

网关证书验证错误，解决方法有二：

1. 执行--ssl-insecure
2. 下载最新的cacert.pem替换 ( Python安装路径\Lib\site-packages\certifi ) 的目录证书

案例展示

mitmproxy + Selenium

电脑端自动化爬虫

案例说明：

• Selenium 自动翻页，
• mitmproxy 进行信息采集，
• 在指定网站，输入 指定关键词 以及 爬取的页码数量，即可。

注意点：

• 评论数量是另外一个文件，需要另外进行解析。
• 返回评论适量的链接有两个，要区别做判断。

selenium JD：

"""输入关键词 + 页码数量 Jd自动翻页程序"""

import time
from selenium import webdriver


class JdSpider:
    """OK"""

    def __init__(self, keyword=None, page=None):
        self.url = 'https://www.jd.com/'
        self.browser = None
        self.page = int(page)
        self.keyword = keyword

    def __del__(self):
        self.browser.close()

    def open_browser(self):
        """打开浏览器"""
        self.browser = webdriver.Chrome()
        # self.browser.maximize_window()
        self.browser.set_window_size(1350, 850)

    def search_keyword(self):
        '''搜索关键字'''
        self.browser.get(self.url)
        # 输入内容
        self.browser.find_element_by_xpath('//*[@id="key"]').send_keys(self.keyword)
        # 模拟点击
        self.browser.find_element_by_xpath('//*[@id="search"]/div/div[2]/button').click()

    def turn_page(self):
        '''翻页'''
        self.browser.execute_script('window.scrollTo(0,document.body.scrollHeight)')
        time.sleep(3)
        if self.browser.page_source.find('pn-next disabled') == -1:
            self.browser.find_element_by_class_name('pn-next').click()

    def main(self):
        '''函数启动接口'''
        self.open_browser()
        self.search_keyword()
        for count in range(self.page):
            self.turn_page()


if __name__ == '__main__':
    keyword = input("Enter the keywords to search：")
    page = input("Enter the Page to download：")
    spider = JdSpider(keyword=keyword, page=page)
    spider.main()

网页解析 及 保存：

import re
import os
import csv
import json
from lxml import etree


def format_common(_list: list):
    """格式化函数"""
    _str = ''.join(_list)
    _str = _str.replace('\n', '').replace('\t', '').replace('￥', '')
    return _str


def format_state(_list: list):
    """格式化函数"""
    _str = ' '.join(_list)
    _str = _str.replace('\n', '').replace('\t', '').replace('￥', '')
    return _str


class SaveData:
    """OK"""

    def __init__(self, data):
        self.comment_data = data[0]
        self.other_data = data[1]

    def judge_exists(self, path):
        """判断文件是否已存在"""
        if os.path.exists(path):
            return
        title = ["商铺名称", "说明", "价格", "评价人数", "商品名称"]
        with open(path, 'a+', encoding='utf-8', newline='') as f:
            writer = csv.writer(f)  # 创建写 对象
            writer.writerow(title)  # 写入单行

    def save_to_csv(self, data: list):
        """保存为csv"""
        path = r'./data/JdGoodsInfo.csv'
        self.judge_exists(path)
        with open(path, 'a+', encoding='utf-8', newline='') as f:
            writer = csv.writer(f)  # 创建写 对象
            writer.writerows(data)  # 写入多行

    def parse_data(self):
        """解析网页"""
        if not self.other_data or not self.comment_data:
            return

        comments_item = json.loads(re.findall("jQuery\d+\((.*?)\);", self.comment_data)[0])['CommentsCount']
        if len(comments_item) != 30:
            return

        _data = list()
        xpath_html = etree.HTML(self.other_data)
        xpath_items = xpath_html.xpath('//li[@class="gl-item"]')
        for xpath_item, comment_item in zip(xpath_items, comments_item):
            _parse = xpath_item.xpath
            shop = format_common(_parse('.//div[@class="p-shop"]//text()'))
            icons = format_state(_parse('.//div[@class="p-icons"]//text()'))
            price = format_common(_parse('.//div[@class="p-price"]//text()'))
            name = format_common(_parse('.//div[@class="p-name p-name-type-2"]//text()'))
            comment = comment_item['CommentCountStr']
            _data.append((shop, icons, price, comment, name))
        self.save_to_csv(data=_data)

    def main(self):
        """开始干活"""
        self.parse_data()

mitm 代码：

# -*- coding:utf-8 -*-
# author   : SunriseCai
# datetime : 2020/11/21 10:47
# software : PyCharm

import json
import mitmproxy.http
from mitmSaveData import SaveData


class Demo:
    def __init__(self):
        self.other_data = None

    def response(self, flow: mitmproxy.http.HTTPFlow):
        url = flow.request.url
        if 'jd.com' not in url:
            return
        # 商品信息链接
        if 'https://search.jd.com/s_new.php?keyword=' in url:
            self.other_data = flow.response.text or None
            # 评论链接
        if 'https://club.jd.com/comment' in url:
            comment_data = flow.response.text
            SaveData([comment_data, self.other_data]).main()
            comment_data, self.other_data = None, None


addons = [
    Demo(),
]

遗留问题：

• 搜索的首页不是 XHR 形式加载出来的，这个不想做适配了。

当然大家要是感兴趣的话可以继续试试 mitmproxy + Appium 手机端自动数据采集

小结一下

很多人可能会说 mitmproxy，没有 charles，fiddler 在日常工作中使用方便，但是试想一下，要是你需要成千上万的数据，或者你在跑自动化的时候，需要替换替换数据的时候，难道还是使用手动吗，那是不可能的，累都得累死，或者你在多环境测试的时候，如何快速完成呢，举个例子，假如你本次项目版本只有服务端做了改动，客户端没有做了改动，你回归的时候，就可以使用 mitmproxy 把本地的生产环境直接批量性的重定向到你的测试或者预发环境，还有本事 mitmproxy 才可以采集数据，那么换个思路，可不可以把生产的所有请求采集，然后到测试环境在跑一遍呢 ，这样，是不是就可以，在更加多的时间，干个更多事情了，也相对减少了加班的节奏，我记得社区有个大佬和我说过，“你要想学习进步，首先第一步就是让自己不加班，这样才有机会进步”，换个想法，下班，去路边买杯奶茶，蹲在路边，看过往的小姐姐不香吗。