如何使用 python mitm 捕获请求并通过 flask 按请求重放
How to use python mitm to capture requests and replay on request via flask
我的想法是 运行 一个 flask webapp,封装 mitmproxy,所以如果我使用代理并对特定 URL 执行获取请求,捕获它(数据 + 会话),并在 flask
启动时重播
from flask import Flask
from mitmproxy import http
from mitmproxy.options import Options
from mitmproxy.proxy.config import ProxyConfig
from mitmproxy.proxy.server import ProxyServer
from mitmproxy.tools.dump import DumpMaster
import requests
import threading
import asyncio
proxies = {
'http': 'http://127.0.0.1:8080',
'https': 'http://127.0.0.1:8080',
}
class repeat:
def __init__(self):
self.response=''
def request(self, flow):
if 'http://dummy.com/get_details.html' in flow.request.pretty_url:
'''
STEP 1
Capture authenticated requests for replay later
'''
elif 'repeat_again' in flow.request.pretty_url:
'''
STEP 2
Repeat earlier get_details request
then save data
'''
def response(self, flow: http.HTTPFlow):
'''
return ddynamic details
save the get_details data into class variable self.response
'''
addons=[repeat()]
app = Flask("proxapp")
@app.route('/get_again')
def get_again():
requests.get('repeat_again',proxies=proxies)
return repeat.response
'''
cause a repeat request
'''
def loop_in_thread(loop, m):
asyncio.set_event_loop(loop)
m.run_loop(loop.run_forever)
if __name__ == "__main__":
options = Options(listen_host='0.0.0.0', listen_port=8080, http2=True)
m = DumpMaster(options, with_termlog=True, with_dumper=True)
config = ProxyConfig(options)
m.server = ProxyServer(config)
m.addons.add(addons)
# run mitmproxy in backgroud, especially integrated with other server
loop = asyncio.get_event_loop()
t = threading.Thread( target=loop_in_thread, args=(loop,m) )
t.start()
app.run(debug=True, use_reloader=False, host='0.0.0.0', port=int('28888'))
所以这里我的测试浏览器按正常方式浏览 dummy.com,但是当它显示 get_details.html 页面时,我喜欢捕获重复重播的请求(可以使用 mitmdump 轻松完成)。响应应保存在 class.
的变量中
所以现在如果我想重播之前的请求,我可以调用 http://127.0.0.1:2888/get_again 来重播相同的请求。
有什么想法吗?我可以使用 mitmdump 手动执行此操作,但我试图将其自动化
存储回放内容必须在response
方法中。在 request
方法中,流 returns 与 replay_response
.
from flask import Flask
from mitmproxy.options import Options
from mitmproxy.proxy.config import ProxyConfig
from mitmproxy.proxy.server import ProxyServer
from mitmproxy.tools.dump import DumpMaster
import requests
import threading
import asyncio
proxies = {
'http': 'http://127.0.0.1:8080',
'https': 'http://127.0.0.1:8080',
}
# Change replay_url -> http://dummy.com/get_details.html
replay_url = 'http://wttr.in/Innsbruck?0'
class Repeat:
def __init__(self):
self.replay_response = None
def request(self, flow):
if 'repeat_again' in flow.request.pretty_url:
flow.response = self.replay_response
def response(self, flow):
if replay_url in flow.request.pretty_url:
self.replay_response = flow.response
app = Flask("proxapp")
repeat = Repeat()
@app.route('/get_again')
def get_again():
resp = requests.get('http://repeat_again', proxies=proxies)
return (resp.text, resp.status_code, resp.headers.items())
def loop_in_thread(loop, m):
asyncio.set_event_loop(loop)
m.run_loop(loop.run_forever)
if __name__ == "__main__":
options = Options(listen_host='0.0.0.0', listen_port=8080, http2=True)
m = DumpMaster(options, with_termlog=True, with_dumper=True)
m.addons.add(repeat)
config = ProxyConfig(options)
m.server = ProxyServer(config)
# run mitmproxy in background, especially integrated with other server
loop = asyncio.get_event_loop()
t = threading.Thread(target=loop_in_thread, args=(loop, m))
t.start()
app.run(debug=True, use_reloader=False, host='0.0.0.0', port=int('28888'))
我用 http://wttr.in/Innsbruck?0
测试过
输出:
$ curl --proxy http://localhost:8080 "http://wttr.in/Innsbruck?0"
Weather report: Innsbruck
\ / Partly cloudy
_ /"".-. 2..3 °C
\_( ). → 7 km/h
/(___(__) 10 km
0.0 mm
$ curl --proxy http://localhost:8080 "http://repeat_again"
Weather report: Innsbruck
\ / Partly cloudy
_ /"".-. 2..3 °C
\_( ). → 7 km/h
/(___(__) 10 km
0.0 mm
$ curl http://localhost:28888/get_again
Weather report: Innsbruck
\ / Partly cloudy
_ /"".-. 2..3 °C
\_( ). → 7 km/h
/(___(__) 10 km
0.0 mm
我的想法是 运行 一个 flask webapp,封装 mitmproxy,所以如果我使用代理并对特定 URL 执行获取请求,捕获它(数据 + 会话),并在 flask
启动时重播from flask import Flask
from mitmproxy import http
from mitmproxy.options import Options
from mitmproxy.proxy.config import ProxyConfig
from mitmproxy.proxy.server import ProxyServer
from mitmproxy.tools.dump import DumpMaster
import requests
import threading
import asyncio
proxies = {
'http': 'http://127.0.0.1:8080',
'https': 'http://127.0.0.1:8080',
}
class repeat:
def __init__(self):
self.response=''
def request(self, flow):
if 'http://dummy.com/get_details.html' in flow.request.pretty_url:
'''
STEP 1
Capture authenticated requests for replay later
'''
elif 'repeat_again' in flow.request.pretty_url:
'''
STEP 2
Repeat earlier get_details request
then save data
'''
def response(self, flow: http.HTTPFlow):
'''
return ddynamic details
save the get_details data into class variable self.response
'''
addons=[repeat()]
app = Flask("proxapp")
@app.route('/get_again')
def get_again():
requests.get('repeat_again',proxies=proxies)
return repeat.response
'''
cause a repeat request
'''
def loop_in_thread(loop, m):
asyncio.set_event_loop(loop)
m.run_loop(loop.run_forever)
if __name__ == "__main__":
options = Options(listen_host='0.0.0.0', listen_port=8080, http2=True)
m = DumpMaster(options, with_termlog=True, with_dumper=True)
config = ProxyConfig(options)
m.server = ProxyServer(config)
m.addons.add(addons)
# run mitmproxy in backgroud, especially integrated with other server
loop = asyncio.get_event_loop()
t = threading.Thread( target=loop_in_thread, args=(loop,m) )
t.start()
app.run(debug=True, use_reloader=False, host='0.0.0.0', port=int('28888'))
所以这里我的测试浏览器按正常方式浏览 dummy.com,但是当它显示 get_details.html 页面时,我喜欢捕获重复重播的请求(可以使用 mitmdump 轻松完成)。响应应保存在 class.
的变量中所以现在如果我想重播之前的请求,我可以调用 http://127.0.0.1:2888/get_again 来重播相同的请求。
有什么想法吗?我可以使用 mitmdump 手动执行此操作,但我试图将其自动化
存储回放内容必须在response
方法中。在 request
方法中,流 returns 与 replay_response
.
from flask import Flask
from mitmproxy.options import Options
from mitmproxy.proxy.config import ProxyConfig
from mitmproxy.proxy.server import ProxyServer
from mitmproxy.tools.dump import DumpMaster
import requests
import threading
import asyncio
proxies = {
'http': 'http://127.0.0.1:8080',
'https': 'http://127.0.0.1:8080',
}
# Change replay_url -> http://dummy.com/get_details.html
replay_url = 'http://wttr.in/Innsbruck?0'
class Repeat:
def __init__(self):
self.replay_response = None
def request(self, flow):
if 'repeat_again' in flow.request.pretty_url:
flow.response = self.replay_response
def response(self, flow):
if replay_url in flow.request.pretty_url:
self.replay_response = flow.response
app = Flask("proxapp")
repeat = Repeat()
@app.route('/get_again')
def get_again():
resp = requests.get('http://repeat_again', proxies=proxies)
return (resp.text, resp.status_code, resp.headers.items())
def loop_in_thread(loop, m):
asyncio.set_event_loop(loop)
m.run_loop(loop.run_forever)
if __name__ == "__main__":
options = Options(listen_host='0.0.0.0', listen_port=8080, http2=True)
m = DumpMaster(options, with_termlog=True, with_dumper=True)
m.addons.add(repeat)
config = ProxyConfig(options)
m.server = ProxyServer(config)
# run mitmproxy in background, especially integrated with other server
loop = asyncio.get_event_loop()
t = threading.Thread(target=loop_in_thread, args=(loop, m))
t.start()
app.run(debug=True, use_reloader=False, host='0.0.0.0', port=int('28888'))
我用 http://wttr.in/Innsbruck?0
输出:
$ curl --proxy http://localhost:8080 "http://wttr.in/Innsbruck?0"
Weather report: Innsbruck
\ / Partly cloudy
_ /"".-. 2..3 °C
\_( ). → 7 km/h
/(___(__) 10 km
0.0 mm
$ curl --proxy http://localhost:8080 "http://repeat_again"
Weather report: Innsbruck
\ / Partly cloudy
_ /"".-. 2..3 °C
\_( ). → 7 km/h
/(___(__) 10 km
0.0 mm
$ curl http://localhost:28888/get_again
Weather report: Innsbruck
\ / Partly cloudy
_ /"".-. 2..3 °C
\_( ). → 7 km/h
/(___(__) 10 km
0.0 mm