How to use python mitm to capture requests and replay on request via flask

Question

The idea I have is to run a flask webapp, encapsulating mitmproxy, so if I use the proxy and perform a get request to specific URL, capture it (data + session), and replay it when initiated by 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'))

So here my test browser browsing dummy.com as per normal, but the moment it does get_details.html page, I like to capture the request for repeat replay (can be done with mitmdump easily). the response should be saved in a variable of the class.

So now if I like to replay the earlier request, I can call http://127.0.0.1:2888/get_again to replay the same request.

Any ideas? i can do this manually with mitmdump, but I trying to automate it

Answer 1

Storing replay content must be in response method. In request method, flow returns with 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'))

I have tested with http://wttr.in/Innsbruck?0

Output:

$ 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