相关文章推荐
失恋的稀饭  ·  如何使用Jacoco从覆盖率分析中排除测试类 ...·  11 月前    · 
近视的镜子  ·  Python搞定表格可视化！-CSDN博客·  11 月前    · 
傲视众生的葡萄  ·  conda的安装与使用（最新版）-阿里云开发者社区·  1 年前    · 
仗义的青蛙  ·  李飞飞等人提出Auto-DeepLab：自动 ...·  1 年前    · 
聪明伶俐的长颈鹿  ·  [Unity]代码设置color与color ...·  1 年前    · 
Code  ›  Python 实现代理拦截http请求：mitmproxy_mitmproxy 透明代理
python函数 代理模式 http代理 python继承
https://blog.csdn.net/freeking101/article/details/83901842
文雅的数据线
2 月前


mitmproxy 官网： https://mitmproxy.org
mitmproxy 官网文档： https://docs.mitmproxy.org/stable
​github地址： https://github.com/mitmproxy/mitmproxy
github示例： https://github.com/mitmproxy/mitmproxy/tree/main/examples

http 抓包在 web 渗透上占据着非常重要的地位，这方面的工具也非常多，像 burp suite， Fiddler，Charles 每个都是搞 web 的必备神器。mitmproxy 是基于 python 编写定制脚本的代理拦截工具。anyproxy 是基于 JavaScript 编写定制脚本，功能与 mitmproxy 基本一致

1、相关概念

​MITM (Man-in-the-middle attack) 中间人攻击，也可译为 中间人代理 ，可以用来拦截、监听和篡改 HTTP/HTTPS 请求。 中间人代理 一般用在客户端和服务器之间的网络中。

5 种代理模式

5种代理模式，官网介绍： https://docs.mitmproxy.org/stable/concepts-modes/

最常用的是 mitmproxy 的正向代理，通过调整配置 mitmproxy 还可以作为透明代理、反向代理、上游代理、SOCKS 代理等

  • 1、 正向代理（regular proxy）启动时默认的模式。 是一个位于客户端和原始服务器(origin server)之间的服务器，为了从原始服务器取得内容，客户端向mitmproxy代理发送一个请求并指定目标(原始服务器)，然后代理向原始服务器转交请求并将获得的内容返回给客户端。客户端必须要进行一些特别的设置才能使用正向代理。

正向代理：所谓正向代理就是顺着请求的方向进行的代理。比如我们要去访问国外的网站，直接访问不通，那么我们就可以找一个代理服务器为我们服务，我们通过代理服务器请求到国外的网站。对于国外的网站而言，他只知道有一个服务器访问了自己，并不知道这件事你是通过代理服务器访问自己。

一个通俗的例子：你需要钱，C正好有钱，但是C不直接借给你。你和B关系比较好，B可以找C借到钱。你和B沟通后，由B来找C借到钱后在给你。

正向代理类似一个跳板机，代理访问外部资源

正向代理的用途：
（1） 通过间接方式，访问原来无法直接访问的资源。
（2） 可以做缓存，加速访问资源
（3） 对客户端访问授权，上网进行认证
（4） 代理可以记录用户访问记录（上网行为管理），对外隐藏用户信息

  • 正向代理 是为了隐藏真实客户端，代理的对象是客户端，

    反向代理 是为了隐藏真实服务端，代理的对象是服务端，

  • 反向代理 和正向代理正好相反，对于客户端而言它就像是原始服务器，并且客户端不需要进行任何特别的设置。客户端向 mitmproxy代理服务器发送普通请求，mitmproxy 转发请求到指定的服务器，并将获得的内容返回给客户端。

    看图理解：

    正向代理中，proxy 和 client 同属一个 LAN，对 server 透明；
    反向代理中，proxy 和 server 同属一个 LAN，对 client 透明。

    实际上 proxy 在两种代理中做的事都是代为收发请求和响应，不过从结构上来看正好左右互换了下，所以把后出现的那种代理方式叫成了反向代理。虽然 整体的请求返回路线都是一样的都是Client 到 Proxy 到 Server 。

反向代理的作用：

  • 保证内网的安全，阻止 web 攻击，大型网站，通常将反向代理作为公网访问地址，Web 服务器是内网
  • 负载均衡，通过反向代理服务器来优化网站的负载

上行代理 ( 多级代理 )

上行代理（upstream proxy）启动参数 -U host。 mitmproxy 接受代理请求，并将所有请求无条件转发到指定的上游代理服务器。

透明代理：客户端不需要任何配置，只需要在网关或者路由器上，执行数据包重定向到代理服务器即可。因为客户端感觉不到代理的存在，所以说是透明代理。

设置透明代理，需要两步。

  • 第一，需要一个代理服务器，这里可以使用 mitmproxy 启用透明代理：mitmproxy -T
  • 第二，在网关或者路由器上，执行数据包重定向到代理服务器上。
    iptable 设置：
    iptables -t nat -A PREROUTING -i eth0 -p tcp --dport 80 -j REDIRECT --to-port 8080
    iptables -t nat -A PREROUTING -i eth0 -p tcp --dport 443 -j REDIRECT --to-port 8080

mitmproxy 透明代理（transparent proxy）启动参数 -T。

SOCKS5 代理

在此模式下，mitmproxy 充当 SOCKS5 代理。 这类似于常规代理模式，但使用 SOCKS5 而不是 HTTP 建立连接 与代理。

mitmproxy 支持的代理模式

mitmproxy 简介

mitmproxy 是一组工具，用于为 HTTP/1、HTTP/2 和 WebSocket 提供交互式、支持 SSL/TLS 的拦截代理。

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

不同于 fiddler 或 charles 等抓包工具，mitmproxy 不仅可以拦截请求，更可以通过自定义脚本进行二次开发。 mitmproxy 是基于 Python 开发的开源工具，它提供了 Python API，这样就可以通过载入自定义 python 脚本轻松实现使用 Python 代码来控制请求和响应。 由于 mitmproxy 工作在 HTTP 层，所以要让 mitmproxy 能够正常工作，必须要让客户端（APP 或浏览器）主动信任 mitmproxy 的SSL证书，或忽略证书异常。

mitmproxy 目前比较广泛的应用是做爬虫，利用手机模拟器、无头浏览器来爬取 APP 或网站的数据，mitmpproxy 作为代理可以拦截、存储爬虫获取到的数据，或修改数据调整爬虫的行为。

mitmproxy 实现原理：

  1. 客户端发起一个到 mitmproxy 的连接，并且发出 HTTP CONNECT 请求，
  2. mitmproxy 作出响应 (200)，模拟已经建立了 CONNECT 通信管道，
  3. 客户端确信它正在和远端服务器会话，然后启动 SSL 连接。在 SSL 连接中指明了它正在连接的主机名 (SNI)，
  4. mitmproxy 连接服务器，然后使用客户端发出的 SNI 指示的主机名建立 SSL 连接，
  5. 服务器以匹配的 SSL 证书作出响应，这个 SSL 证书里包含生成的拦截证书所必须的通用名 (CN) 和服务器备用名 (SAN)，
  6. mitmproxy 生成拦截证书，然后继续进行与第 3 步暂停的客户端 SSL 握手，
  7. 客户端通过已经建立的 SSL 连接发送请求，
  8. mitmproxy 通过第 4 步建立的 SSL 连接传递这个请求给服务器。

mitmproxy 工作步骤：

  1. 设置系统、浏览器、终端等的代理地址和端口为同一局域网中 mitmproxy 所在电脑的 IP 地址，比如我的 PC 开启 mitmproxy 之后，设置 8080 端口，本地 IP 为 192.168.1.130，那么设置 Android HTTP 代理为 192.168.1.130:8080
  2. 浏览器 或 移动端 访问 mitm.it 来安装 mitmproxy 提供的证书
  3. 在 mitmproxy 提供的命令行下，或者 mitmweb 提供的浏览器界面中就能看到 Android 端发出的请求。

2、安装、运行 mitmproxy

安装 mitmproxy

安装最新版本的 Python3.10+，安装： pip install mitmproxy
完成后，命令行/终端启动。默认监听： http://localhost:8080

  • 执行命令： mitmproxy 启动交互式的控制台版本。
  • 执行命令： mitmweb 启动基于浏览器web界面版本。
  • 执行命令： mitmdump 启动 mitmproxy的命令行版本。

用 mitmdump 测试一下安装是否成功

mitmproxy 使用最多的还是 正向代理模式。下面主要以正向代理说明。结合 Python 强大的功能，可以衍生出很多应用途径。除此之外，mitmproxy 还提供了强大的API，在这些 API 的基础上，完全可以自己定制一个实现了特殊功能的专属代理服务器。

安装 证书

安装证书： https://docs.mitmproxy.org/stable/concepts-certificates
如果想要截获 HTTPS 请求，就得安装 mitmproxy的CA证书。
首先启动 mitmproxy ，然后浏览器访问 http://mitm.it 选择对应平台的图标下载并安装证书。

将证书设置为 " 受信任的根证书颁发机构 "。

把 charles，Fiddler 证书安装到安卓根目录，解决安卓微信 7.0 版本以后安装证书也无法抓包问题，需要 root： https://blog.csdn.net/freeking101/article/details/118914275

启动 mitmproxy 命令

执行 mitmproxy 命令启动后，会提供一个命令行界面，用户可以实时看到发生的请求，并通过命令过滤请求，查看请求数据。形如：

执行 mitmproxy 命令后在终端显示数据包，通过 Shift + ? 可以查看查看当前页面可用的命令。

q    退出（相当于返回键，可一级一级返回）
d    删除当前（黄色箭头）指向的链接
D    恢复刚才删除的请求
G    跳到最新一个请求
g    跳到第一个请求
C    清空控制台（C是大写）
i    可输入需要拦截的文件或者域名（逗号需要用\来做转译，栗子：feezu.cn）
a    放行请求
A    放行所有请求
?    查看界面帮助信息
^ v    上下箭头移动光标
enter    查看光标所在列的内容
tab    分别查看 Request 和 Response 的详细信息
/    搜索body里的内容
esc    退出编辑
e    进入编辑模式

mitmproxy 帮助

常用的几个命令参数：

-p PORT, --port PORT    设置 mitmproxy 的代理端口
-T, --transparent       设置透明代理
--socks                 设置 SOCKS5 代理
-s "script.py --bar", --script "script.py --bar"   可以执行脚本，通过双引号来添加参数
-t FILTER               过滤参数

启动 mitmweb 命令

mitmweb 命令启动后，会提供一个 web 界面，用户可以实时看到发生的请求，并通过 GUI 交互来过滤请求，查看请求数据

启动 mitmdump 命令

mitmdump 命令启动后，控制台只能查看抓到的包，无法提供过滤请求、查看数据的功能。可以结合自己编写的Python脚本来使用。

命令行启动 Chrome

命令行启动 Chrome 并设置 mitmproxy 为代理，同时忽略证书错误。

接下来关闭所有 Chrome 窗口，否则命令行启动时的附加参数将失效。打开 cmd，执行： "C:\Program Files (x86)\Google\Chrome\Application\chrome.exe" --proxy-server=127.0.0.1:8080 -ignore-certificate-errors

cd C:\Program Files (x86)\Google\Chrome\Application
chrome.exe --proxy-server=127.0.0.1:8080 -ignore-certificate-errors

前面那一长串是 Chrome 的的安装路径，应当根据系统实际情况修改，后面两参数设置了代理地址并强制忽略掉证书错误。用 Chrome 打开一个网站，可以看到：

mitmproxy 所有功能

3、开发 addon(插件)

完成上述工作，已经具备操作 mitmproxy 的基本能力 了。接下来开始开发自定义脚本，这才是 mitmproxy 真正强大的地方。使用 -s 加载 Python 脚本：mitmproxy -s script.py

完整的 HTTP flow 会依次触发：requestheaders、request、responseheaders  response

开发 addon

addon 翻译成中文是 "插件"。Mitmproxy 的 addon 机制是 mitmproxy 中非常强大的部分。事实上 mitmproxy 的大部分功能在 一套内置插件 中定义，实现了 mitmproxy 几乎所有的功能。

开发插件时，许多方法都以 flow(流) 为参数。

addon 通过 event 与 mitmproxy 交互，并修改 mitmproxy 的行为。event 可以通过 "Options选项" 进行配置，Options 可以在脚本中设置，或者在命令行上传递，最后，它们可以公开命令，这允许用户直接调用它们的操作，或者将它们绑定到交互式工具中的键。

定义一个 类，类里面的方法实现了某些 mitmproxy 提供的事件。

编写一个 py 文件供 mitmproxy 加载，文件定义了变量 addons，addons 是个数组，每个元素是一个类实例，这些类有若干方法，这些方法实现了某些 mitmproxy 提供的事件，mitmproxy 会在某个事件发生时调用对应的方法。这些类，称为一个个 addon

示例：一个简单的插件，用于跟踪 HTTP 请求流(flow)

Basic skeleton of a mitmproxy addon. Run as follows: mitmproxy -s anatomy.py import logging class Counter: def __init__(self): self.num = 0 def request(self, flow): self.num = self.num + 1 logging.info("We've seen %d flows" % self.num) addons = [Counter()]

执行命令： mitmdump -s ./anatomy.py

  • Mitmproxy 插件机制：找到 全局列表 addons，并把列表的内容作为插件进行加载。
  • 插件是一个对象，这里是 类Counter的实例化 Counter()
  • 该 request 方法是 事件 的一个示例。插件只是实现这一个方法。每个事件及其签名都记录在案 在 API 文档 中。 更多的脚本可以 参考 
#!/usr/bin/env python
# -*- coding: UTF-8 -*-
def request(flow):
    if flow.request.pretty_host == 'api.github.com':
        flow.request.host = '127.0.0.1'
        flow.request.port = 8085

缩写 脚本

有时，只是想快速编写一个脚本，但是又不想创建类。 addon 机制有一个简写，允许将模块作为一个整体被视为 addon 对象。其实， 就是定义若干函数，这些函数实现了某些 mitmproxy 提供的事件 。例如，这是一个完整的脚本，它为每个请求添加一个标头： 

"""An addon using the abbreviated scripting syntax."""
def request(flow):
    flow.request.headers["myheader"] = "value"

Event Hooks (事件钩子)

只要 addon 方法实现了相关函数，addon 就可以通过 Event Hooks 钩入 mitmproxy 的内部机制。许多事件将 Flow 对象作为参数接收，通过修改这些对象，插件可以动态更改流量。例如，这里有一个插件，它添加了一个响应标头，其中包含查看的响应数量： 

"""Add an HTTP header to each response."""
class AddHeader:
    def __init__(self):
        self.num = 0
    def response(self, flow):
        self.num = self.num + 1
        flow.response.headers["count"] = str(self.num)
addons = [AddHeader()]

可用 钩子

所有可用的事件钩子 ： https://docs.mitmproxy.org/stable/api/events.html

事件针对不同生命周期分为 5 类。“生命周期”这里指在哪一个层面看待事件，举例来说，同样是一次 web 请求，可以理解为“HTTP 请求 -> HTTP 响应”的过程，也可以理解为“TCP 连接 -> TCP 通信 -> TCP 断开”的过程。那么，如果我想拒绝来个某个 IP 的客户端请求，应当注册函数到针对 TCP 生命周期 的tcp_start事件，又或者，我想阻断对某个特定域名的请求时，则应当注册函数到针对 HTTP 声明周期的http_connect事件。其他情况同理。

Lifecycle Events 生命周期事件

  • def load(loader mitmproxy.addonmanager.Loader):
  • def running():
  • def configure(updated: set[str]):
  • def done():

Connection Events 连接事件

  • def client_connected(client: mitmproxy.connection.Client):
  • def client_disconnected(client: mitmproxy.connection.Client):
  • def server_connect(data: mitmproxy.proxy.server_hooks.ServerConnectionHookData):
  • def server_connected(data: mitmproxy.proxy.server_hooks.ServerConnectionHookData):
  • def server_disconnected(data: mitmproxy.proxy.server_hooks.ServerConnectionHookData):

HTTP Events

  • def requestheaders(flow: mitmproxy.http.HTTPFlow):
  • def request(flow: mitmproxy.http.HTTPFlow):   当发送请求时被调用。
  • def responseheaders(flow: mitmproxy.http.HTTPFlow):
  • def response(flow: mitmproxy.http.HTTPFlow):  当接收到回复时被调用。
  • def error(flow: mitmproxy.http.HTTPFlow):
  • def http_connect(flow: mitmproxy.http.HTTPFlow):
  • def http_connect_upstream(flow: mitmproxy.http.HTTPFlow):

DNS Events

  • def dns_request(flow: mitmproxy.dns.DNSFlow):
  • def dns_response(flow: mitmproxy.dns.DNSFlow):
  • def dns_error(flow: mitmproxy.dns.DNSFlow):

TCP Events

  • def tcp_start(flow: mitmproxy.tcp.TCPFlow):
  • def tcp_message(flow: mitmproxy.tcp.TCPFlow):
  • def tcp_end(flow: mitmproxy.tcp.TCPFlow):
  • def tcp_error(flow: mitmproxy.tcp.TCPFlow):

UDP Events

  • def udp_start(flow: mitmproxy.udp.UDPFlow):
  • def udp_message(flow: mitmproxy.udp.UDPFlow):
  • def udp_end(flow: mitmproxy.udp.UDPFlow):
  • def udp_error(flow: mitmproxy.udp.UDPFlow):

QUIC Events

  • def quic_start_client(data: mitmproxy.proxy.layers.quic.QuicTlsData):
  • def quic_start_server(data: mitmproxy.proxy.layers.quic.QuicTlsData):

TLS Events

  • def tls_clienthello(data: mitmproxy.tls.ClientHelloData):
  • def tls_start_client(data: mitmproxy.tls.TlsData):
  • def tls_start_server(data: mitmproxy.tls.TlsData):
  • def tls_established_client(data: mitmproxy.tls.TlsData):
  • def tls_established_server(data: mitmproxy.tls.TlsData):
  • def tls_failed_client(data: mitmproxy.tls.TlsData):
  • def tls_failed_server(data: mitmproxy.tls.TlsData):

WebSocket Events

  • def websocket_start(flow: mitmproxy.http.HTTPFlow):
  • def websocket_message(flow: mitmproxy.http.HTTPFlow):
  • def websocket_end(flow: mitmproxy.http.HTTPFlow):

SOCKSv5 Events

  • def socks5_auth(data: mitmproxy.proxy.layers.modes.Socks5AuthData):

Advanced Lifecycle Events

  • def next_layer(data: mitmproxy.proxy.layer.NextLayer):
  • def update(flows: collections.abc.Sequence[mitmproxy.flow.Flow]):
  • def add_log(entry: mitmproxy.log.LogEntry):

相关 API

addon 官网示例

mitmproxy 的 内置插件

执行脚本的3种方法

mitmproxy -s xxx.py

示例：mitmproxy -s sample.py 

import mitmproxy.http
from mitmproxy import ctx
class Counter:
    def __init__(self):
        self.num = 0
    def request(self, flow: mitmproxy.http.HTTPFlow):
        self.num = self.num + 1
        ctx.log.info("We've seen %d flows" % self.num)
addons = [
    Counter()

mitmproxy 7+ 最新版本
 

mitmproxy 7+ 之后的版本，使用 asgncio 异步方式启动。
 

import asyncio
import mitmproxy.http
from mitmproxy.tools import main
from mitmproxy.tools import dump
from mitmproxy import options
from loguru import logger
class InterceptRequest(object):
    def __init__(self):
        self.logger = logger
    def __del__(self):
    def request(self, flow: mitmproxy.http.HTTPFlow):
        # print(flow.request)
        self.logger.info(flow.request)
    def response(self, flow: mitmproxy.http.HTTPFlow):
async def start_proxy(host, port):
    # opts = main.options.Options(listen_host=host, listen_port=port)
    opts = options.Options(listen_host=host, listen_port=port)
    master = dump.DumpMaster(
        opts,
        with_termlog=False,
        with_dumper=False,
    master.addons.add(InterceptRequest())
    await master.run()
    return master
def func_main():
    asyncio.run(start_proxy('127.0.0.1', 8080))
if __name__ == '__main__':
    func_main()
 

import time
import asyncio
import threading
from loguru import logger
from mitmproxy import addons
import mitmproxy.http
from mitmproxy.options import Options
class MyAddon1(object):
    def __init__(self):
        self.logger = logger
        self.logger.info('初始化 MyAddon1 成功')
    def request(self, flow: mitmproxy.http.HTTPFlow):
        self.logger.info(f'request ---> {flow.request.url}')
class MyAddon2(object):
    def __init__(self):
        self.logger = logger
        self.logger.info('初始化 MyAddon2 成功')
    def response(self, flow: mitmproxy.http.HTTPFlow):
        self.logger.info(f'response ---> status_code:{flow.response.status_code}')
def main_1():
    # DumpMaster 是继承 mitmproxy.master.Master
    from mitmproxy.tools.dump import DumpMaster
    async def func_temp():
        opts = Options(listen_host='0.0.0.0', listen_port=8080)
        dm = DumpMaster(opts, with_termlog=False, with_dumper=False)
        # dm.addons.add(MyAddon1())
        dm.addons.add(*[MyAddon1(), MyAddon2()])
            await dm.run()
        except BaseException as be:
            dm.shutdown()
    asyncio.run(func_temp())
def main_2():
    from mitmproxy.master import Master
    async def func_temp():
        opts = Options(listen_host='0.0.0.0', listen_port=8080)
        m = Master(opts, with_termlog=False)
        addon_list = [
            # default_addons 这个好像必须加上，不加上拦截不住
            *addons.default_addons(),
            MyAddon1(), MyAddon2()
        m.addons.add(*addon_list)
            await m.run()
        except BaseException as be:
            m.shutdown()
    asyncio.run(func_temp())
def main_3():
    # 创建一个事件循环thread_loop
    thread_loop = asyncio.new_event_loop()
    def func_new_thread(new_loop: asyncio.AbstractEventLoop = None):
        from mitmproxy.master import Master
        # 为子线程设置自己的事件循环
        asyncio.set_event_loop(new_loop)
        opts = Options(listen_host='127.0.0.1', listen_port=8080)
        m = Master(opts, new_loop)
        addon_list = [
            *addons.default_addons(),
            MyAddon1(), MyAddon2()
        m.addons.add(*addon_list)
        new_loop.run_until_complete(m.run())
    thd = threading.Thread(target=func_new_thread, args=(thread_loop, ))
    thd.daemon = True
    thd.start()
    thd.join()
def main_4():
    from mitmproxy.master import Master
    from mitmproxy.addons import dumper, errorcheck, keepserving, readfile, termlog
    # 创建一个事件循环thread_loop
    new_loop = asyncio.new_event_loop()
    async def func_temp():
        opts = Options(listen_host='0.0.0.0', listen_port=8080)
        m = Master(opts, new_loop)
        addon_list = [
            *addons.default_addons(),
            MyAddon1(), MyAddon2()
        m.addons.add(*addon_list)
        await m.run()
    coro = func_temp()
    future = asyncio.run_coroutine_threadsafe(coro, new_loop)
    ####################################################################
    # 在一个线程中专门运行时间循环。
    def start_event_loop(loop):
        asyncio.set_event_loop(loop)
        loop.run_forever()
    t = threading.Thread(target=start_event_loop, args=(new_loop,))
    t.start()
    ####################################################################
    future.add_done_callback(lambda f: new_loop.call_soon_threadsafe(new_loop.stop))
    future.result()
    t.join()
if __name__ == "__main__":
    main_1()
    # main_2()
    # main_3()
    # main_4()

推荐使用 main_1，即导入包 from mitmproxy.tools.dump import DumpMaster
 

mitmproxy7 之前版本
 

mitmproxy7 之前的版本，都是同步执行，没有使用 asyncio。
 

from mitmproxy import proxy, options
from mitmproxy.tools.dump import DumpMaster
from mitmproxy.addons import core
class AddHeader(object):
    def __init__(self):
        self.num = 0
    def response(self, flow):
        self.num = self.num + 1
        print(self.num)
        flow.response.headers["count"] = str(self.num)
addons = [
    AddHeader()
opts = options.Options(listen_host='127.0.0.1', listen_port=9999)
m = DumpMaster(options=opts)
m.addons.add(*addons)
print(m.addons)
# m.addons.add(core.Core())
    m.run()
except KeyboardInterrupt:
    m.shutdown()
 

示例 2：
 

# -*- coding: utf-8 -*-
import json
import datetime
from mitmproxy.options import Options
from mitmproxy.tools.dump import DumpMaster
from mitmproxy import ctx, http
class TTAddon(object):
    def __init__(self):
        self._temp = None
    def request(self, flow: http.HTTPFlow):
        self._temp = None
        # do something in response
        r_url = flow.request.url
    def response(self, flow: http.HTTPFlow):
        self._temp = None
        # do something in response
        r_url = flow.request.url
        if 'api/news/feed/v88/' in r_url and 'category=news_car' in r_url:
            print(r_url)
            resp = flow.response.text
            resp_dict = json.loads(resp)
            print(json.dumps(resp_dict, ensure_ascii=False, indent=4))
class ProxyMaster(DumpMaster):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
    def start_run(self):
            DumpMaster.run(self)
        except KeyboardInterrupt:
            self.shutdown()
if __name__ == "__main__":
    opts = Options(listen_host='0.0.0.0', listen_port=8080, http2=True)
    # master = ProxyMaster(options=opts, with_termlog=False, with_dumper=False)
    master = ProxyMaster(options=opts, with_termlog=True, with_dumper=False)
    master.addons.add((TTAddon(), ))
    master.start_run()

mitmdump
 
 
mitmdump 是对 mitmproxy 的简单封装，实现以编程的方式运行 mitmproxy 服务, 方便开箱即用，而不再需要记住复杂的命令参数，当然原来的命令行方式运行也是支持的，另外对于某些特殊的应用场景，还可以借助 Pycharm 对编写的 mitmproxy 脚本进行断点调试。
 
 
pypi 地址（ 示例代码 ）：https://pypi.org/project/mitmdump/ (已经好久没更新)
 

示例 1：
 

# sample1.py
from mitmproxy.http import HTTPFlow
from mitmdump import DumpMaster, Options
class AddHeader:
    def __init__(self):
        self.num = 0
    def response(self, flow: HTTPFlow):
        self.num = self.num + 1
        flow.response.headers["count"] = str(self.num)
addons = [
    AddHeader()
if __name__ == '__main__':
    opts = Options(listen_host='0.0.0.0', listen_port=8888, scripts=__file__)
    m = DumpMaster(opts)
    m.run()

示例 2：
 

# sample2.py
from mitmproxy import flowfilter, ctx, addonmanager
from mitmproxy.http import HTTPFlow
from mitmdump import DumpMaster, Options
class FilterFlow:
    def __init__(self):
        self.filter = None
    def load(self, loader: addonmanager.Loader):
        self.filter = flowfilter.parse(ctx.options.dumper_filter)
    def request(self, flow: HTTPFlow):
        if flowfilter.match(self.filter, flow):
            print(flow.request.url)
    def response(self, flow: HTTPFlow):
        if flowfilter.match(self.filter, flow):
            print(flow.response.headers)
addons = [
    FilterFlow()
if __name__ == '__main__':
    opts = Options(
        listen_host='0.0.0.0', listen_port=8888, scripts=None, dumper_filter='~m POST',
        flow_detail=1, termlog_verbosity='info', show_clientconnect_log=False
    m = DumpMaster(opts)
    # It's necessary if scripts parameter is None
    # 如果你的 scripts 参数为 None，则下方加载插件的语句是必须要有的
    m.addons.add(*addons)
    m.run()

4、HTTPFlow、Request、Response
 

HTTPFlow、Request、Response 都是在 mitmproxy 中 http.py 中
 

HTTPFlow
 

from mitmproxy import addons, http
from mitmproxy.options import Options 
class MyAddon1(object):
    def __init__(self):
        self.logger = logger
        self.logger.info('初始化 MyAddon1 成功')
    def request(self, flow: http.HTTPFlow):
        req = flow.request
        self.logger.info(f'request ---> {req.url}')
    def response(self, flow: http.HTTPFlow):
        resp = flow.response
        self.logger.info(f'response ---> status_code:{resp.status_code}')
 

Request 相关属性
 

Response 相关属性
 

示例：常用的操作 http 请求
 

：https://ranjuan.cn/mitmproxy-python-http-methods-use/
 

import mitmproxy.http
from mitmproxy import ctx, http
import time
class Action1:
    def request(self, flow: mitmproxy.http.HTTPFlow):
        #flow.request.host='www.google.com'
        #t=1/0
        #print("请求将被kill，后续将不会有response")
        #flow.kill()
        if 'action' in flow.request.url:
            print("触发了拦截,会阻塞后面的请求，即使请求不进入本逻辑")
            flow.intercept()
            a = input("输入1 放行: ")
            if a=='1':
                #time.sleep(20)
                flow.resume()#继续流动 - 在一个intercept（）之后调用
            else:
                print("输入不是1，此请求不再放行")
        else:
            print("不拦截")
        #pretty_host #类似于host，但使用主机头作为附加的首选数据源。这在透明模式下很有用，host只有IP地址，但可能不会反映实际的目的地，因为主机头可能被欺骗。
        flow.request.method #请求方式。POST、GET等
        flow.request.scheme #什么请求 ，应为“http”或“https”
        #flow.response = http.Response.make(200,"111",)
        flow.request.query #返回MultiDictView类型的数据，url直接带的键值参数
        flow.request.query.keys()#取得所有请求参数(不包含参数对应的值)
        print(list(flow.request.query.keys()))
        print(" ")
        act1 =flow.request.query.get('action')#取得请求参数wd的值
        print(f"act1={act1}")
        #flow.request.query.set_all(key,[value])#修改请求参数
        flow.request.query.set_all('action',['edit-action'])
        act2 =flow.request.query.get('action')
        print(f"act2={act2}")
        flow.request.cookies["log_id"] = "007"#修改cookie
        flow.request.get_content()#bytes,结果如flow.request.get_text()
        flow.request.raw_content #bytes,结果如flow.request.get_content()
        flow.request.urlencoded_form #MultiDictView，content-type：application/x-www-form-urlencoded时的请求参数，不包含url直接带的键值参数
        flow.request.urlencoded_form["code"] = "123456"#修改或赋值
        flow.request.urlencoded_form = [("code", "123456"),("name","lucy")]
        flow.request.multipart_form #MultiDictView，content-type：multipart/form-data
        return
    def response(self, flow):
        #t =1/0
        #flow.response = http.Response.make(200,"111+111",)
        #flow.response = flow.response.make(404)#返回404
        #print(flow.response.headers)
        for (k,v) in flow.response.headers.items():
            print(f"{k}:{v}")
        print(" ")
        #print(flow.response.get_text())
        flow.response.status_code #状态码
        flow.response.text#返回内容，已解码
        #print(flow.response.text)#返回内容，已解码
        flow.response.content #返回内容，二进制
        #flow.response.setText()#修改返回内容，不需要转码
        flow.response.set_text(flow.response.get_text().replace('<title>', '<title>返回title——'))
        flow.response.headers["isMitmproxy"]='yes'#给返回添加返回头
        #读取文件，在当前文件路径下执行脚本，否则需要写文件的绝对路径；不然会找不到该json文件
        with open('1.json','rb') as f:
            #从json文件中读取数据成python对象
            res = json.load(f)
        #将读取的python对象转成json字符串发送给客户端
        flow.response.set_text(json.dumps(res))
        return
 

如果是接口测试用，有时候需要利用postman或其他apifox、apipost之类的工具导入接口，手动创建接口参数这些填写很麻烦，即使curl导入也不太方便，这时候我们可以把经过mitmproxy代理的相关参数进行整理并打印出来，然后复制到软件中。
 

    def request(self, flow):
        ctx.log.error(f"【1】请求url：{flow.request.url}")
        ctx.log.error(f"【1】请求url(某些情况下地址是ip这时需要取pretty_url)：{flow.request.pretty_url}")
        ctx.log.error(f"【1】请求host：{flow.request.host}")
        ctx.log.error(f"【1】请求host(某些情况下地址是ip这时需要取pretty_host)：{flow.request.pretty_host}")
        ctx.log.error(f"【1】请求路径：{flow.request.path}")
        ctx.log.error(f"【1】请求content：{flow.request.content}")
        ctx.log.error("请求参数Params：")
        for k in flow.request.query:
            ctx.log.error(f"{k}:{flow.request.query[k]}")
        ctx.log.error(" ")
        ctx.log.error(f"【2】请求Cookie：")
        for k in flow.request.cookies:
            ctx.log.error(f"{k}:{flow.request.cookies[k]}")
        ctx.log.error(" ")
        ctx.log.error(f"【3】请求头：")
        for k in flow.request.headers:
            ctx.log.error(f"{k}:{flow.request.headers[k]}")
        ctx.log.error(" ")
        if 'Content-Type' in flow.request.headers:
            ctx.log.error(f"【4】请求类型：{flow.request.headers['Content-Type']}")
        else:
            ctx.log.error(f"【4】请求类型：无")
        #body 需要根据Content-Type去生成
        ctx.log.error("")
        ctx.log.error(f"【5】请求体文本：{flow.request.get_text()}")
        ctx.log.error("")
        if 'application/x-www-form-urlencoded' in str(flow.request.headers):
            #s = urllib.parse.unquote(flow.request.get_text().replace("=",":").replace("&","rn"))
            #ctx.log.error("【6】请求体x-www-form-urlencoded：n"+s)
            ctx.log.error("【6】请求体x-www-form-urlencoded：")
            for k in flow.request.urlencoded_form:
                ctx.log.error(f"{k}:{flow.request.urlencoded_form[k]}")
            ctx.log.error("")
        elif 'application/json' in str(flow.request.headers):
            ctx.log.error("【6】请求体json：n"+flow.request.get_text())
        else:
            ctx.log.error("【6】获取请求体失败,或请求体格式暂不支持")
        ctx.log.error("")

处理请求时常用的代码片段
 

将url参数转为字典
 

from urllib.parse import parse_qs, urlparse
url = 'https://www.test.com?action=add&id=007&user=&wd=&action=add2'
query = urlparse(url).query
print("原始url："+url)
params = parse_qs(query)
print(params)
 

将字典转为url参数
 

from urllib.parse import urlencode
params = {'action': 'add', 'id': 2}
result = urlencode(params)
print(result)
 

def allowed_urls(ori_str):#判断提供的字符串/url 是否符合匹配条件
	check_dic=["ranjuan","ntest"]#需要包含的关键字
	check_pass=["act","99"]#需要排除的关键字， 匹配条件是既需要包含关键字，又不能出现要排除的关键字
	#return all(any(i in j for j in bvalue) for i in avalue)
	if any(e in ori_str for e in check_dic):
		if any(e in ori_str for e in check_pass):
			return False
		else:
			return True
	return False
print(allowed_urls('http://ranjuan.cn?ction=1')) #True
print(allowed_urls('http://ranjuan.cn?action=1'))#False
print(allowed_urls('http://ranjuan.cn?cction=199'))#False
print(allowed_urls('http://ranjuantesst.cn?ction=1'))#True
print(allowed_urls('http://rantestjuan.cn?ction=1'))#True
 

md5函数
 

import hashlib
stringA='123456'
str_md5 = hashlib.md5(stringA.encode(encoding='utf-8')).hexdigest()
print(str_md5)

AES加密函数
 

aes加密使用会有一些坑，第一个就是python模块安装crypto后需要重命令相关的2个文件夹名字（首字母改成大写），然后再安装pycryptodome模块。另外就是密钥key建议一定要用16位长度，否则跨语言进行AES加解密时会自己python上写的可以加解密，但是对方那边就是解密不了（对方在自己的开发语言里面加解密也是正常的）。如果长度确实不够，那么填充key的方式两边需要确保一致才行！
 

def pads(text,length=16):
	#填充函数，使被加密数据的字节码长度是block_size的整数倍
	print(f"pads明文"+str(text))
	count = len(text.encode('utf-8'))
	add = length - (count % length)
	entext = text + (chr(add) * add)
	print(f"pads结果"+str(entext.encode("utf8")))
	return entext.encode("utf8")
from Crypto.Cipher import AES
import base64
#1、pip install Crypto
#2、要去安装目录下把两个crypto的文件夹（我这是crypto、crypto-1.4.1.dist-info）首字母大写后再安装pycryptodome。
#3、pip install pycryptodome
text = 'hello world'
password = 'ABCdefG890123456' #秘钥，b就是表示为bytes类型;限制为16
aes = AES.new(password.encode('utf-8'),AES.MODE_ECB) #密码正好长度为16时可不用填充，一定要16位很关键
en_text = aes.encrypt(pads(text)) #加密明文
encrypted_text = str(base64.encodebytes(en_text), encoding='utf-8')
print("密文：",encrypted_text) 
 

5、Python 通过代码设置 windows 代理
 

Windows 中一般程序都会默认走 IE 的代理设置。而 IE 的代理设置是记录在注册表中。
 具体位置在：HKEY_CURRENT_USER\Software\Microsoft\Windows\CurrentVersion\Internet Settings 该目录下有三个键值：
 

  • ProxyEnable  代理开关
  • ProxyOverride  例外（比如控制本地不走代理）
  • ProxyServer  代理地址
 

找到地方后，就可以很方便的使用标准库 winreg 进行修改了。注意，修改注册表涉及到系统权限，建议谨慎操作，并确保在修改前备份相关设置。
 

示例 1：
 

import winreg as reg
def set_proxy(proxy_enable, proxy_server=None):
    设置Windows代理
    :param proxy_enable: 1为开启代理，0为关闭代理
    :param proxy_server: 代理服务器地址和端口，例如 '127.0.0.1:8080'
    path = r'Software\Microsoft\Windows\CurrentVersion\Internet Settings'
        key = reg.OpenKey(reg.HKEY_CURRENT_USER, path, 0, reg.KEY_WRITE)
        reg.SetValueEx(key, "ProxyEnable", 0, reg.REG_DWORD, proxy_enable)
        if proxy_server:
            reg.SetValueEx(key, "ProxyServer", 0, reg.REG_SZ, proxy_server)
        reg.CloseKey(key)
        print("代理设置成功")
    except Exception as e:
        print(f"代理设置失败: {e}")
# 开启代理
set_proxy(1, '127.0.0.1:8080')
# 关闭代理
# set_proxy(0)
 
顾名思义，mitmproxy 就是用于 MITM 的 proxy，MITM 即中间人攻击（Man-in-the-middle attack）。用于中间人攻击的代理首先会向正常的代理一样转发请求，保障服务端与客户端的通信，其次，会适时的查、记录其截获的数据，或篡改数据，引发服务端或客户端特定的行为。
不同于 fiddler 或 wireshark 等抓包工具，mitmproxy 不仅可以截获请求帮助开发者查看、分析，更可以通过自定义脚本进行二次开发。举例来说，利用 f.
                                    使用WebView组件造一工具，工具可输入网页地址并显示网页内容及播放视频。把工具的代理设置指向mitmproxy的端口服务。配合使用mitmproxy的MITM技术，监听和拦截用户在短视频平台上发起的网络请求。MITM技术允许应用程序在数据包从客户端到服务器的传输过程中进行干预，从而捕获和分析请求和响应数据，从响应数据中提取视频的真实下载链接。
在微信小程序等环境中，并不能直接在WebView中配置代理。小程序的WebView组件直接使用的是微信小程序、支付宝小程序等平台提供的WebView实现，这些平
                                    今天给大家分享一个非常实用抓包工具，不需要安装软件，直接可以在浏览器上进行抓包，mitmproxy其实是基于python的第三方库，并且配合python脚本可以篡改请求和响应数据，使用起来相对于简单，易于上手，推荐给大家。
                                    举例来说，利用 Fiddler 可以过滤出浏览器对某个特定 url 的请求，并查看、分析其数据，但实现不了高度定制化的需求，类似于：“截获对浏览器对该 url 的请求，将返回内容置空，并将真实的返回内容存到某个数据库，出现异常时发出邮件通知”。：最好是配置python3.8及以上的python环境，如果是以下的版本，则最好安装5.2.0版本的mitmproxy，因为其他的版本有编码问题。3、手机浏览器访问http://mitm.it/ ，下载对应版本的证书并安装。前置条件：安装并配置好python环境。
                                    1、MitmProxy介绍
mitmproxy 就是用于 MITM 的 代理(proxy)， 而MITM，也就是(Man-in-the-middle attack)中间人攻击，指的就是通过各种技术手段将入侵者的一台计算机(服务器)虚拟放置在网络连接的两台计算机之间。(SMB会话劫持、DNS欺骗等)
......
                                    mitmproxy是一款用Python编写的支持HTTP(S)的中间人代理工具。它可以拦截、查看、修改、重放和保存HTTP/HTTPS流量 ，支持命令行界面和图形界面，可用于安全测试、网络调试、API开发和反向工程等场景。mitmproxy具有很高的灵活性和扩展性，可以通过插件机制进行定制化开发和功能扩展。本文注意介绍命令行模式。
                                    一、mitmproxy介绍
mitmproxy 就是用于 MITM 的 proxy，MITM 即中间人攻击（Man-in-the-middle attack）。用于中间人攻击的代理首先会向正常的代理一样转发请求，保障服务端与客户端的通信，其次，会适时的查、记录其截获的数据，或篡改数据，引发服务端或客户端特定的行为。
mitmproxy可以拦截http和https请求，如果需要拦截https请求需要安装证书
二、mitmproxy安装(windows下)
1. 运行下面命令进行安装
pip install m
 
推荐文章
失恋的稀饭  ·  如何使用Jacoco从覆盖率分析中排除测试类-腾讯云开发者社区-腾讯云
11 月前
近视的镜子  ·  Python搞定表格可视化！-CSDN博客
11 月前
傲视众生的葡萄  ·  conda的安装与使用（最新版）-阿里云开发者社区
1 年前
仗义的青蛙  ·  李飞飞等人提出Auto-DeepLab：自动搜索图像语义分割架构 - 知乎
1 年前
聪明伶俐的长颈鹿  ·  [Unity]代码设置color与color32的值_unity new color_BuladeMian的博客-CSDN博客
1 年前
今天看啥   ·   Py中国   ·   codingpro   ·   藏经阁   ·   小百科   ·   link之家   ·   卧龙AI搜索
删除内容请联系邮箱 2879853325@qq.com
Code - 代码工具平台
© 2024 ~ 沪ICP备11025650号