mitmproxy
基础
- github地址：https://github.com/mitmproxy/mitmprox
- mitmproxy 官网：https://mitmproxy.org
- mitmproxy 中文网址：http://www.hpaizhao.cn
- mitmproxy 官网文档：https://docs.mitmproxy.org/stable
- mitmproxy 官方示例 及 API：（推荐从 simple 开始）：https://github.com/mitmproxy/mitmproxy/tree/master/examples
使命行模式
https://www.jianshu.com/p/f6af1d57186e
常用的几个命令参数：
-
-p PORT, --port PORT设置 mitmproxy 的代理端口 -
-T, --transparent设置透明代理 -
--socks设置 SOCKS5 代理 -
-s "script.py --bar", --script "script.py --bar"来执行脚本，通过双引号来添加参数 -
-t FILTER过滤参数
命令模式下，在终端显示请求流，可以通过Shift + ?来开启帮助查看当前页面可用的命令：
基本快捷键
b 保存请求 / 返回头 C 将请求内容导出到粘贴板，按 C 之后会有选择导出哪一部分 d 删除 flow 请求 D 恢复刚才删除的请求 E 将 flow 导出到文件 w 保存所有 flow 或者该 flow W 保存该 flow L 加载保存的 Flow m 添加 / 取消 Mark 标记，会在请求列表该请求前添加红色圆圈 z 清空 flow list 和 eventlog / 在详情界面，可以使用 / 来搜索，大小写敏感 i 开启 interception pattern 拦截请求 G 跳到最新一个请求 g 跳到第一个请求 C 清空控制台（C是大写） i 可输入需要拦截的文件或者域名（逗号需要用\来做转译，栗子：feezu.cn） a 放行请求 A 放行所有请求 ? 查看界面帮助信息 ^ v 上下箭头移动光标 enter 查看光标所在列的内容 tab 分别查看 Request 和 Response 的详细信息 / 搜索body里的内容 esc 退出编辑 e 进入编辑模式
移动
j, k 上下 h, l 左右 g, G go to beginning, end space 下一页 pg up/down 上一页 / 下一页 ctrl+b/ctrl+f 上一页 / 下一页 arrows 箭头 上下左右
全局快捷键
q 退出，或者后退 Q 不提示直接退出
代理模式
有五种代理模式：
1、正向代理（regular proxy）启动时默认选择的模式。
是一个位于客户端和原始服务器(origin server)之间的服务器，为了从原始服务器取得内容， 客户端向mitmproxy代理发送一个请求并指定目标(原始服务器)， 然后代理向原始服务器转交请求并将获得的内容返回给客户端。 客户端必须要进行一些特别的设置才能使用正向代理。
2、反向代理（reverse proxy）启动参数-R host。跟正向代理正好相反，
对于客户端而言它就像是原始服务器，并且客户端不需要进行任何特别的设置。
客户端向mitmproxy代理服务器发送普通请求，mitmproxy转发请求到指定的服务器，
并将获得的内容返回给客户端，就像这些内容 原本就是它自己的一样。
3、上行代理（upstream proxy）启动参数-U host。mitmproxy接受代理请求，
并将所有请求无条件转发到指定的上游代理服务器。这与反向代理相反，
其中mitmproxy将普通HTTP请求转发给上游服务器。
4、透明代理（transparent proxy）启动参数-T。当使用透明代理时，
流量将被重定向到网络层的代理，而不需要任何客户端配置。
这使得透明代理非常适合那些无法更改客户端行为的情况 -
代理的Android应用程序是一个常见的例子。要设置透明代理，我们需要两个新的组件。
第一个是重定向机制，可以将目的地为Internet上的服务器的TCP连接透明地重新路由
到侦听代理服务器。这通常采用与代理服务器相同的主机上的防火墙形式。
比如Linux下的iptables, 或者OSX中的pf,一旦客户端初始化了连接，
它将作出一个普通的HTTP请求(注意，这种请求就是客户端不知道代理存在)
请求头中没有scheme(比如http://或者https://),
也没有主机名(比如example.com)我们如何知道上游的主机是哪个呢？
路由机制执行了重定向，但保持了原始的目的地址。
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 -T
5、socks5 proxy 启动参数--socks。采用socks协议的代理服务器
web界面模式
拦截8888端口，web界面8080
mitmweb --no-web-open-browser -p 8888 --web-port 8080
--no-web-open-browser是阻止每次启动都弹一个新的浏览器窗口出来。
在浏览器中以localhost:8080打开web界面。
如果mitmproxy之前还有一层代理（比如shadowshocket），则通过--mode upstream:...
指定：
mitmweb --no-web-open-browser --mode upstream:https://127.0.0.1:9999 -p 8888 --web-port 8080
Chrome浏览器在没有插件辅助的情况下要用启动参数指定代理：
chrome.exe --proxy-server=127.0.0.1:8888 --ignore-certificate-errors
安装证书
当我们初次运行 mitmproxy 或 mitmdump 时，会在当前目录下生成~/.mitmproxy文件夹
，其中该文件下包含4个文件，这就是我们要的证书了：
-
mitmproxy-ca.pemPEM格式的证书私钥 -
mitmproxy-ca-cert.pemPEM格式证书，适用于大多数非Windows平台 -
mitmproxy-ca-cert.cer与mitmproxy-ca-cert.pem相同（只是后缀名不同），适用于大部分Android平台 -
mitmproxy-ca-cert.p12PKCS12格式的证书，适用于大多数Windows平台 -
mitmproxy-dhparam.pemPEM格式的秘钥文件，用于增强SSL安全性。
把代理设置为mitmproxy代理的IP与端口，然后访问 http://mitm.it 可以直接选择证书安装。
macOS
- 打开系统配置（System Preferences.app）- 网络（Network）- 高级（Advanced）- 代理（Proxies）- Web Proxy(HTTP)和Secure Web Proxy(HTTPS)
- 填上代理服务器IP和端口
- 打开Keychain Access.app
- 选择login(Keychains)和Certificates(Category)中找到mitmproxy
- 点击mitmproxy，在Trust中选择Always Trust
chrome
把代理设置为mitmproxy代理的IP与端口，然后访问 http://mitm.it ，
选择others下载.pem格式的证书文件。
chrome中setting > Manage certificates > import导入证书。
linux
把代理设置为mitmproxy代理的IP与端口，然后访问 http://mitm.it ，
选择others下载.pem格式的证书文件，然后转为.crt模式的证书：
openssl x509 -in mitmproxy-ca-cert.pem -inform PEM -out mitmproxy-ca-cert.crt
然后导入：
sudo mkdir -p /usr/share/ca-certificates/extra/ sudo cp mitmproxy.crt /usr/share/ca-certificates/extra/mitmproxy.crt sudo cp ./mitmproxy-ca-cert.crt /usr/share/ca-certificates/extra/mitmproxy.crt sudo dpkg-reconfigure ca-certificates
基本概念
过滤表达式
匹配表达式
-
~a：Match asset in response: CSS, Javascript, Flash, images. -
~b regex：Body -
~bq regex：Request body -
~bs regex：Response body -
~c int：HTTP response code -
~d regex：Domain -
~dst regex：Match destination address -
~e：Match error -
~h regex：Header -
~hq regex：Request header -
~hs regex：Response header -
~http：Match HTTP flows -
~m regex：Method -
~marked：Match marked flows -
~q：Match request with no response -
~s：Match response -
~src regex：Match source address -
~t regex：Content-type header -
~tcp：Match TCP flows -
~tq regex：Request Content-Type header -
~ts regex：Response Content-Type header -
~u regex：URL -
~websocket：Match WebSocket flows -
!：unary not -
&：and -
|：or -
(...)：grouping
- Regexes are Python-style
- Regexes can be specified as quoted strings
- Header matching (~h, ~hq, ~hs) is against a string of the form “name: value”.
- Strings with no operators are matched against the request URL.
- The default binary operator is &.
流选择器
-
@all：All flows -
@focus：The currently focused flow -
@shown：All flows currently shown -
@hidden：All flows currently hidden -
@marked：All marked flows -
@unmarked：All unmarked flows
例子
URL containing google.com:
google\.com
Requests whose body contains the string test:
~q ~b test
Anything but requests with a text/html content type:
!(~q & ~t "text/html")
数据结构
三个比较重要的数据结构
-
mitmproxy.models.http.HTTPRequest -
mitmproxy.models.http.HTTPResponse -
mitmproxy.models.http.HTTPFlow
生命周期
import typing
import mitmproxy.addonmanager
import mitmproxy.connections
import mitmproxy.http
import mitmproxy.log
import mitmproxy.tcp
import mitmproxy.websocket
import mitmproxy.proxy.protocol
def requestheaders(self, flow: mitmproxy.http.HTTPFlow):
"""
HTTP request headers were successfully read. At this point, the body
is empty.
"""
def request(self, flow: mitmproxy.http.HTTPFlow):
"""
The full HTTP request has been read.
"""
def responseheaders(self, flow: mitmproxy.http.HTTPFlow):
"""
HTTP response headers were successfully read. At this point, the body
is empty.
"""
def response(self, flow: mitmproxy.http.HTTPFlow):
"""
The full HTTP response has been read.
"""
def error(self, flow: mitmproxy.http.HTTPFlow):
"""
An HTTP error has occurred, e.g. invalid server responses, or
interrupted connections. This is distinct from a valid server HTTP
error response, which is simply a response with an HTTP error code.
"""
# TCP lifecycle
def tcp_start(self, flow: mitmproxy.tcp.TCPFlow):
"""
A TCP connection has started.
"""
def tcp_message(self, flow: mitmproxy.tcp.TCPFlow):
"""
A TCP connection has received a message. The most recent message
will be flow.messages[-1]. The message is user-modifiable.
"""
def tcp_error(self, flow: mitmproxy.tcp.TCPFlow):
"""
A TCP error has occurred.
"""
def tcp_end(self, flow: mitmproxy.tcp.TCPFlow):
"""
A TCP connection has ended.
"""
# Websocket lifecycle
def websocket_handshake(self, flow: mitmproxy.http.HTTPFlow):
"""
Called when a client wants to establish a WebSocket connection. The
WebSocket-specific headers can be manipulated to alter the
handshake. The flow object is guaranteed to have a non-None request
attribute.
"""
def websocket_start(self, flow: mitmproxy.websocket.WebSocketFlow):
"""
A websocket connection has commenced.
"""
def websocket_message(self, flow: mitmproxy.websocket.WebSocketFlow):
"""
Called when a WebSocket message is received from the client or
server. The most recent message will be flow.messages[-1]. The
message is user-modifiable. Currently there are two types of
messages, corresponding to the BINARY and TEXT frame types.
"""
def websocket_error(self, flow: mitmproxy.websocket.WebSocketFlow):
"""
A websocket connection has had an error.
"""
def websocket_end(self, flow: mitmproxy.websocket.WebSocketFlow):
"""
A websocket connection has ended.
"""
# Network lifecycle
def clientconnect(self, layer: mitmproxy.proxy.protocol.Layer):
"""
A client has connected to mitmproxy. Note that a connection can
correspond to multiple HTTP requests.
"""
def clientdisconnect(self, layer: mitmproxy.proxy.protocol.Layer):
"""
A client has disconnected from mitmproxy.
"""
def serverconnect(self, conn: mitmproxy.connections.ServerConnection):
"""
Mitmproxy has connected to a server. Note that a connection can
correspond to multiple requests.
"""
def serverdisconnect(self, conn: mitmproxy.connections.ServerConnection):
"""
Mitmproxy has disconnected from a server.
"""
def next_layer(self, layer: mitmproxy.proxy.protocol.Layer):
"""
Network layers are being switched. You may change which layer will
be used by returning a new layer object from this event.
"""
# General lifecycle
def configure(self, updated: typing.Set[str]):
"""
Called when configuration changes. The updated argument is a
set-like object containing the keys of all changed options. This
event is called during startup with all options in the updated set.
"""
def done(self):
"""
Called when the addon shuts down, either by being removed from
the mitmproxy instance, or when mitmproxy itself shuts down. On
shutdown, this event is called after the event loop is
terminated, guaranteeing that it will be the final event an addon
sees. Note that log handlers are shut down at this point, so
calls to log functions will produce no output.
"""
def load(self, entry: mitmproxy.addonmanager.Loader):
"""
Called when an addon is first loaded. This event receives a Loader
object, which contains methods for adding options and commands. This
method is where the addon configures itself.
"""
def log(self, entry: mitmproxy.log.LogEntry):
"""
Called whenever a new log entry is created through the mitmproxy
context. Be careful not to log from this event, which will cause an
infinite loop!
"""
def running(self):
"""
Called when the proxy is completely up and running. At this point,
you can expect the proxy to be bound to a port, and all addons to be
loaded.
"""
def update(self, flows: typing.Sequence[mitmproxy.flow.Flow]):
"""
Update is called when one or more flow objects have been modified,
usually from a different addon.
"""
1. 针对 HTTP 生命周期
def http_connect(self, flow: mitmproxy.http.HTTPFlow):
(Called when) 收到了来自客户端的 HTTP CONNECT 请求。在 flow 上设置非 2xx 响应将返回该响应并断开连接。CONNECT 不是常用的 HTTP 请求方法，目的是与服务器建立代理连接，仅是 client 与 proxy 的之间的交流，所以 CONNECT 请求不会触发 request、response 等其他常规的 HTTP 事件。
def requestheaders(self, flow: mitmproxy.http.HTTPFlow):
(Called when) 来自客户端的 HTTP 请求的头部被成功读取。此时 flow 中的 request 的 body 是空的。
def request(self, flow: mitmproxy.http.HTTPFlow):
(Called when) 来自客户端的 HTTP 请求被成功完整读取。
def responseheaders(self, flow: mitmproxy.http.HTTPFlow):
(Called when) 来自服务端的 HTTP 响应的头部被成功读取。此时 flow 中的 response 的 body 是空的。
def response(self, flow: mitmproxy.http.HTTPFlow):
(Called when) 来自服务端端的 HTTP 响应被成功完整读取。
def error(self, flow: mitmproxy.http.HTTPFlow):
(Called when) 发生了一个 HTTP 错误。比如无效的服务端响应、连接断开等。注意与“有效的 HTTP 错误返回”不是一回事，后者是一个正确的服务端响应，只是 HTTP code 表示错误而已。
2. 针对 TCP 生命周期
def tcp_start(self, flow: mitmproxy.tcp.TCPFlow):
(Called when) 建立了一个 TCP 连接。
def tcp_message(self, flow: mitmproxy.tcp.TCPFlow):
(Called when) TCP 连接收到了一条消息，最近一条消息存于 flow.messages[-1]。消息是可修改的。
def tcp_error(self, flow: mitmproxy.tcp.TCPFlow):
(Called when) 发生了 TCP 错误。
def tcp_end(self, flow: mitmproxy.tcp.TCPFlow):
(Called when) TCP 连接关闭。
3. 针对 Websocket 生命周期
def websocket_handshake(self, flow: mitmproxy.http.HTTPFlow):
(Called when) 客户端试图建立一个 websocket 连接。可以通过控制 HTTP 头部中针对 websocket 的条目来改变握手行为。flow 的 request 属性保证是非空的的。
def websocket_start(self, flow: mitmproxy.websocket.WebSocketFlow):
(Called when) 建立了一个 websocket 连接。
def websocket_message(self, flow: mitmproxy.websocket.WebSocketFlow):
(Called when) 收到一条来自客户端或服务端的 websocket 消息。最近一条消息存于 flow.messages[-1]。消息是可修改的。目前有两种消息类型，对应 BINARY 类型的 frame 或 TEXT 类型的 frame。
def websocket_error(self, flow: mitmproxy.websocket.WebSocketFlow):
(Called when) 发生了 websocket 错误。
def websocket_end(self, flow: mitmproxy.websocket.WebSocketFlow):
(Called when) websocket 连接关闭。
4. 针对网络连接生命周期
def clientconnect(self, layer: mitmproxy.proxy.protocol.Layer):
(Called when) 客户端连接到了 mitmproxy。注意一条连接可能对应多个 HTTP 请求。
def clientdisconnect(self, layer: mitmproxy.proxy.protocol.Layer):
(Called when) 客户端断开了和 mitmproxy 的连接。
def serverconnect(self, conn: mitmproxy.connections.ServerConnection):
(Called when) mitmproxy 连接到了服务端。注意一条连接可能对应多个 HTTP 请求。
def serverdisconnect(self, conn: mitmproxy.connections.ServerConnection):
(Called when) mitmproxy 断开了和服务端的连接。
def next_layer(self, layer: mitmproxy.proxy.protocol.Layer):
(Called when) 网络 layer 发生切换。你可以通过返回一个新的 layer 对象来改变将被使用的 layer。详见 layer 的定义。
5. 通用生命周期
def configure(self, updated: typing.Set[str]):
(Called when) 配置发生变化。updated 参数是一个类似集合的对象，包含了所有变化了的选项。在 mitmproxy 启动时，该事件也会触发，且 updated 包含所有选项。
def done(self):
(Called when) addon 关闭或被移除，又或者 mitmproxy 本身关闭。由于会先等事件循环终止后再触发该事件，所以这是一个 addon 可以看见的最后一个事件。由于此时 log 也已经关闭，所以此时调用 log 函数没有任何输出。
def load(self, entry: mitmproxy.addonmanager.Loader):
(Called when) addon 第一次加载时。entry 参数是一个 Loader 对象，包含有添加选项、命令的方法。这里是 addon 配置它自己的地方。
def log(self, entry: mitmproxy.log.LogEntry):
(Called when) 通过 mitmproxy.ctx.log 产生了一条新日志。小心不要在这个事件内打日志，否则会造成死循环。
def running(self):
(Called when) mitmproxy 完全启动并开始运行。此时，mitmproxy 已经绑定了端口，所有的 addon 都被加载了。
def update(self, flows: typing.Sequence[mitmproxy.flow.Flow]):
(Called when) 一个或多个 flow 对象被修改了，通常是来自一个不同的 addon。
常用方法
Request的一些方法：
-
get_query()：得到请求的url的参数，被存放成了字典。 -
set_query(odict)：设置请求的url参数，参数是字典。 -
get_url()：请求的url。 -
set_url(url)：设置url的域。 -
get_cookies()：得到请求的cookie。 -
headers： 请求的header的字典。 -
content：
# http.HTTPFlow 实例 flow flow.request.headers # 获取所有头信息，包含Host、User-Agent、Content-type等字段 flow.request.url # 完整的请求地址，包含域名及请求参数，但是不包含放在body里面的请求参数 flow.request.pretty_url # 同flow.request.url目前没看出什么差别 flow.request.host # 域名 flow.request.method # 请求方式。POST、GET等 flow.request.scheme # 什么请求 ，如 https flow.request.path # 请求的路径，url除域名之外的内容 flow.request.get_text() # 请求中body内容，有一些http会把请求参数放在body里面，那么可通过此方法获取，返回字典类型 flow.request.query # 返回MultiDictView类型的数据，url直接带的键值参数 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.multipart_form # MultiDictView，content-type：multipart/form-data 时的请求参数，不包含url直接带的键值参数
Response的一些方法如下：
-
Headers：返回的header的字典。 -
Code：返回数据包的状态，比如200,301之类的状态。 -
Httpversion：http版本。
flow.response.status_code # 状态码
flow.response.text # 返回内容，已解码
flow.response.content # 返回内容，二进制
flow.response.setText() # 修改返回内容，不需要转码。
#请求的内容，如果请求时post，那么content就是指代post的参数。
例，修改响应的内容：
def response(flow:http.HTTPFlow)-> None:
#特定接口需要返回1001结果
interface_list=["page/**"] #由于涉及公司隐私问题，隐藏实际的接口
url_path=flow.request.path
if url_path.split("?")[0] in interface_list:
ctx.log.info("#"*50)
ctx.log.info("待修改路径的内容："+url_path)
ctx.log.info("修改成：1001错误返回")
ctx.log.info("修改前：\n")
ctx.log.info(flow.response.text)
flow.response.set_text(json.dumps({"result":"1001","message":"服务异常"}))#修改，使用set_text不用转码
ctx.log.info("修改后：\n")
ctx.log.info(flow.response.text)
ctx.log.info("#"*50)
elif flow.request.host in host_list:#host_list 域名列表，作为全局变量，公司有多个域名，也隐藏
ctx.log.info("response= "+flow.response.text)
自定义插件
脚本的方式：不推荐
按mitmproxy事件的实现对应的函数：
import mitmproxy.http
from mitmproxy import ctx
num = 0
def request(flow: mitmproxy.http.HTTPFlow):
global num
num = num + 1
ctx.log.info("We've seen %d flows" % num)
使用插件的方式：推荐
把事件对应的方法封闭在对象中，然后把对象放到addons列表中。
这样更加容易扩展和管理：
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() ]
指定加载脚本：
mitmweb --listen-port 8888 --web-port 8080 -s ./addons.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)
定义另一个插件：
import mitmproxy.http
from mitmproxy import ctx, http
class Joker:
# 拦截的baidu关键字搜索请求，把查询关键字替换为“请使用360”
def request(self, flow: mitmproxy.http.HTTPFlow):
if flow.request.host != "www.baidu.com" or not flow.request.path.startswith("/s"):
return
if "wd" not in flow.request.query.keys():
ctx.log.warn("can not get search word from %s" % flow.request.pretty_url)
return
ctx.log.info("catch search word: %s" % flow.request.query.get("wd"))
flow.request.query.set_all("wd", ["360搜索"])
# 拦截360的页面，把页面上所有“搜索”替换成“请使用谷歌”
def response(self, flow: mitmproxy.http.HTTPFlow):
if flow.request.host != "www.so.com":
return
text = flow.response.get_text()
text = text.replace("搜索", "请使用谷歌")
flow.response.set_text(text)
# 拦截全部google的访问并返回404
def http_connect(self, flow: mitmproxy.http.HTTPFlow):
if flow.request.host == "www.google.com":
flow.response = http.HTTPResponse.make(404)
同时引入两个插件：
import counter
import joker
addons = [
counter.Counter(),
joker.Joker(),
]
指定加载脚本：
mitmweb --listen-port 8888 --web-port 8080 -s ./addons2.py
全局配置：Options
mitmproxy和加载的插件的行为都是由全局的option store控制的。option有多个来源： 配置文件、命令行、用户在交互中动态修改。
配置项要注明类型（限于mitmproxy支持的几种类型），这样才能被序列化与反序列化， 或是在程序中被处理。
增加一条配置
from mitmproxy import ctx
class AddHeader:
def __init__(self):
self.num = 0
def load(self, loader):
loader.add_option( # 添加了一条配置，名称是`addheader`，类型是`bool`
name = "addheader",
typespec = bool,
default = False,
help = "Add a count header to responses",
)
def response(self, flow):
if ctx.options.addheader:
self.num = self.num + 1
flow.response.headers["count"] = str(self.num)
addons = [ AddHeader() ]
load()方法在插件第一次被加载时调用，loader的类型是mitmproxy.addonmanager.Loader
。通过loader可能声明option和commands。
$ mitmproxy -s ./examples/addons/options-simple.py # 加载脚本 .... $ env http_proxy=http://localhost:8080 curl -I http://google.com # 使用代理 HTTP/1.1 301 Moved Permanently Location: http://www.google.com/ Content-Length: 219 count: 1 # 响应中多了的头部
更新配置
当configure事件被触发时，有变化的多个配置项以key-value的形式的参数updates
传递给configure()函数：
https://docs.mitmproxy.org/stable/addons-options/
例子
基础
mitmproxy环境上下文
-
mitmproxy.ctc上下文环境 -
flow请求
from mitmproxy import ctx
class Counter:
def __init__(self):
self.num = 0
def request(self, flow):
self.num = self.num + 1
ctx.log.info("We've seen %d flows" % self.num)
addons = [ Counter() ]
添加头部
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() ]
配置插件
from mitmproxy import ctx
class AddHeader:
def __init__(self):
self.num = 0
def load(self, loader):
"""
这个自定义的插件第一次被调用时，用来初始化自己的
"""
loader.add_option( # 把自己的信息添加到上下文环境中
name = "addheader",
typespec = bool,
default = False,
help = "Add a count header to responses",
)
def response(self, flow):
if ctx.options.addheader: # 在上下文环境中找自己第一次被加载时的配置
self.num = self.num + 1
flow.response.headers["count"] = str(self.num)
addons = [ AddHeader() ]
配置发生变化
配置发生变化。updated 参数是一个类似集合的对象，包含了所有变化了的选项。在 mitmproxy 启动时，该事件也会触发，且 updated 包含所有选项。
import typing
from mitmproxy import ctx
from mitmproxy import exceptions
class AddHeader:
def load(self, loader):
loader.add_option(
name = "addheader",
typespec = typing.Optional[int],
default = None,
help = "Add a header to responses",
)
def configure(self, updates):
if "addheader" in updates:
if ctx.options.addheader is not None and ctx.options.addheader > 100:
raise exceptions.OptionsError("addheader must be <= 100")
def response(self, flow):
if ctx.options.addheader is not None:
flow.response.headers["addheader"] = str(ctx.options.addheader)
addons = [
AddHeader()
]
应用
替换URL地址
比如将指定 url 的请求指向新的地址
用于调试 Android 或者 iOS 客户端，打包比较复杂的时候， 强行将客户端请求从线上地址指向本地调试地址。可以使用 mitmproxy scripting API mitmproxy 提供的事件驱动接口。
加上将线上地址，指向本地 8085 端口，
def request(flow):
if flow.request.pretty_host == 'api.github.com':
flow.request.host = '127.0.0.1'
flow.request.port = 8085
替换响应
import sys
import json
from mitmproxy import flowfilter
from pymongo import MongoClient
reload(sys)
sys.setdefaultencoding('utf-8')
'''
头脑王者即时显示答案脚本
'''
class TNWZ:
'''
初始化
'''
def __init__(self):
#添加一个过滤器，只针对地址中包含`findQuiz`
self.filter = flowfilter.parse('~u findQuiz')
#连接答案数据库
self.conn = MongoClient('localhost', 27017)
self.db = self.conn.tnwz
self.answer_set = self.db.quizzes
def request(self, flow):
'''
演示request事件效果, 请求的时候输出提示
:param flow:
:return:
'''
if flowfilter.match(self.filter, flow):
print(u'准备请求答案')
def responseheaders(self, flow):
'''
演示responseheaders事件效果, 添加头信息
:param flow:
:return:
'''
if flowfilter.match(self.filter, flow):
flow.response.headers['Cache-Control'] = 'no-cache'
flow.response.headers['Pragma'] = 'no-cache'
def response(self, flow):
'''
HTTPEvent 下面所有事件参数都是 flow 类型 HTTPFlow
可以在API下面查到 HTTPFlow, 下面有一个属性response 类型 TTPResponse
HTTPResponse 有个属性为 content 就是response在内容,更多属性可以查看 文档
:param flow:
:return:
'''
if flowfilter.match(self.filter, flow):
#匹配上后证明抓到的是问题了, 查答案
data = flow.response.content
quiz = json.loads(data)
#获取问题
question = quiz['quiz']
print(question)
#获取答案
answer = self.answer_set.find_one({"quiz":question})
if answer is None:
print('no answer')
else:
answerIndex = int(answer['answer'])-1
options = answer['options']
print(options[answerIndex])
#这里简单演示下start事件
def start():
return TNWZ()
登录获取 Android cookies
利用appium和mitmproxy登录获取cookies
环境搭建
windows中Appium-desktop配合夜神模拟器的使用 :
https://www.cnblogs.com/c-x-a/p/9163221.html
appium
start_appium.py
# -*- coding: utf-8 -*-
# @Time : 2018/10/8 11:00
# @Author : cxa
# @File : test.py
# @Software: PyCharmctx
from appium import webdriver
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By
import time
import base64
def start_appium():
desired_caps = {}
desired_caps['platformName'] = 'Android' # 设备系统
desired_caps['deviceName'] = '127.0.0.1:62001' # 设备名称
desired_caps['appPackage'] = 'com.xxxx.xxxx' # 测试app包名，如何获取包名方式看上面的环境搭建。
desired_caps['appActivity'] = 'com.xxxx.xxxx.xxx.xxxx' # 测试appActivity，如何获取包名方式看上面的环境搭建。
desired_caps['platformVersion'] = '4.4.2' # 设备系统的安卓版本，版本不要太高，设计安全策略得外部因素。
desired_caps['noReset'] = True # 启动后结束后不清空应用数据
desired_caps['unicodeKeyboard'] = True # 此两行是为了解决字符输入不正确的问题
desired_caps['resetKeyboard'] = True # 运行完成后重置软键盘的状态
driver = webdriver.Remote('http://localhost:4723/wd/hub', desired_caps) # 启动app，启动前记得打开appium服务。
wait = WebDriverWait(driver, 60)#设置等待事件
try:
btn_xpath = '//android.widget.Button[@resource-id="com.alicom.smartdail:id/m_nonum_confirm_btn"]'
btn_node = wait.until(EC.presence_of_element_located((By.XPATH, btn_xpath)))#等元素出现再继续，最长等待时间上面设置的60s。
# btn_node=driver.find_element_by_xpath(btn_xpath)
btn_node.click()
except:
driver.back()
btn_xpath = '//android.widget.Button[@resource-id="com.alicom.smartdail:id/m_nonum_confirm_btn"]'
btn_node = wait.until(EC.presence_of_element_located((By.XPATH, btn_xpath)))
# btn_node = driver.find_element_by_xpath(btn_xpath)
btn_node.click()
# sleep 30s
# 点击
def login_in(driver):
id_xpath = '//android.widget.EditText[@content-desc="账户名输入框"]'
id_node = driver.find_element_by_xpath(id_xpath)
id_node.clear()
id_node.send_keys("test")
pwd = str(base64.b64decode("MTIzNHF3ZXI="), 'u8')
pwd_xpath = '//android.widget.EditText[@content-desc="密码输入框"]'
pwd_node = driver.find_element_by_xpath(pwd_xpath)
pwd_node.clear()
pwd_node.send_keys(pwd)
submit = "//android.widget.Button[@text='登录']"
submit_node = driver.find_element_by_xpath(submit)
submit_node.click()
time.sleep(10)
if __name__ == '__main__':
start_appium()
mitmproxy
代码 mitm_proxy_script.py
# -*- coding: utf-8 -*-
# @Time : 2018/10/8 11:00
# @Author : cxa
# @File : mitm_proxy_script.py
# @Software: PyCharm
import sys
sitename = 'ali'
def response(flow):
request = flow.request
if '.png' in request.url or 'xxx.x.xxx.com' not in request.url:
return #如果不在观察的url内则返回
if 'xxx.x.xxx.com' in request .url:
print(request .url)
cookies = dict(request.cookies) #转换cookies格式为dict
if cookies:
save_cookies(repr(cookies))#如果不为空保存cookies
def save_cookies(cookies):
sys.path.append("../")
from database import getcookies
getcookies.insert_data(sitename, cookies) #保存cookies
拦截猫眼跳转到美团的验证码
参考：selenium + chromedriver 被反爬的解决方法 ：https://blog.csdn.net/weixin_39847926/article/details/82262048
如何突破网站对selenium的屏蔽 : https://blog.csdn.net/qq_26877377/article/details/83307208
使用 selenium 驱动 Chrome 浏览器时，猫眼的 js 代码 中有检测，所以需要使用 mitmproxy 做代理拦截请求，然后把 js 文件中对 selenium 检测的关键字全部替换了。
如果不替换，selenium 驱动浏览器出现美团验证码的时候，手动输入验证码点击确定会报错，
拦截替换后，手动输入验证码可以通过。
拦截脚本 （mitmproxy_script.py）：
#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author :
# @File : mitmproxy_script.py
# @Software : PyCharm
# @description : XXX
from mitmproxy import http
import winreg
import requests
import ctypes
import socks
import socket
# ############################# 改系统代理 ############################################
proxy = "117.60.167.134"
port = 4532
INTERNET_SETTINGS = winreg.OpenKey(
winreg.HKEY_CURRENT_USER,
r'Software\Microsoft\Windows\CurrentVersion\Internet Settings',
0,
winreg.KEY_ALL_ACCESS
)
INTERNET_OPTION_REFRESH = 37
INTERNET_OPTION_SETTINGS_CHANGED = 39
internet_set_option = ctypes.windll.Wininet.InternetSetOptionW
socks.setdefaultproxy(socks.PROXY_TYPE_SOCKS5, proxy, port)
socks.setdefaultproxy(socks.PROXY_TYPE_SOCKS4, proxy, port)
socks.setdefaultproxy(socks.PROXY_TYPE_HTTP, proxy, port)
socket.socket = socks.socksocket
def set_key(name, value):
"""
# 修改键值
:param name:
:param value:
:return:
"""
_, reg_type = winreg.QueryValueEx(INTERNET_SETTINGS, name)
winreg.SetValueEx(INTERNET_SETTINGS, name, 0, reg_type, value)
##########################################################################
def chang_ip():
"""
启用代理
:return:
"""
set_key('ProxyEnable', 1) # 启用
set_key('ProxyOverride', u'*.local;<local>') # 绕过本地
set_key('ProxyServer', u'{}:{}'.format(proxy,port)) # 代理IP及端口
internet_set_option(0, INTERNET_OPTION_REFRESH, 0, 0)
internet_set_option(0, INTERNET_OPTION_SETTINGS_CHANGED, 0, 0)
# 停用代理
set_key('ProxyEnable', 0)
internet_set_option(0, INTERNET_OPTION_REFRESH, 0, 0)
internet_set_option(0, INTERNET_OPTION_SETTINGS_CHANGED, 0, 0)
def request(flow):
print(flow.request.url)
# if 'bs/yoda-static/file' in flow.request.url:
# print('*' * 100)
# print(flow.request.url)
# flow.response.text = flow.response.text.replace("webdriver", "fuck_that")
# flow.response.text = flow.response.text.replace("Webdriver", "fuck_that")
# flow.response.text = flow.response.text.replace("WEBDRIVER", "fuck_that")
# chang_ip()
pass
def response(flow):
# print(type(flow.response.text))
if 'webdriver' in flow.response.text:
print('*' * 100)
print('find web_driver key')
flow.response.text = flow.response.text.replace("webdriver", "fuck_that_1")
if 'Webdriver' in flow.response.text:
print('*' * 100)
print('find web_driver key')
flow.response.text = flow.response.text.replace("Webdriver", "fuck_that_2")
if 'WEBDRIVER' in flow.response.text:
print('*' * 100)
print('find web_driver key')
flow.response.text = flow.response.text.replace("WEBDRIVER", "fuck_that_3")
猫眼 对 selenium 驱动 的关键字有三个 ：webdriver、Webdriver、WEBDRIVER，把这三个全替换了。
然后执行：mitmweb -s mitmproxy_script.py
使用 selenium 驱动 Chrome 访问猫眼时，如果再出现验证码，手动输入就可以通过了。