proxy.py

#!/bin/env python
import asyncio
from mitmproxy import options
from mitmproxy.tools import dump
from mitmproxy.http import Headers

from base64 import b64decode, b64encode
import json


class MITMparams:
    def __init__(self):
        self.config = {}
        self.clients = {}
    
    def set_config(self, config):
        self.config = config
    
    def set_clients(self, clients):
        self.clients = clients

    def request(self, flow):
        port = "" if flow.request.port == 443 else ':' + str(flow.request.port)
        url = flow.request.scheme + "://" + flow.request.host + port + flow.request.path
        try:
            if (url not in self.config["proxy"]["do_not_proxify"]["urls"] and flow.request.host + port not in self.config["proxy"]["do_not_proxify"]["domains"] and flow.request.headers["Sec-Fetch-Dest"] and flow.request.headers["Sec-Fetch-Dest"] not in self.config["proxy"]["do_not_proxify"]["sec_fetch_dest"]) or url[-12:] == "&proxyforced":
                if url[-12:] == "&proxyforced":
                    url = url[:-12]
                _req_cookies_str = "full_url=" + url
                if "Referer" in flow.request.headers and flow.request.headers["Referer"].split('/')[2] + port not in self.config["proxy"]["do_not_proxify"]["domains"]:
                    splitted = flow.request.headers["Referer"].split('/')
                    flow.request.path = self.clients[splitted[0] + "//" + splitted[2]]
                elif "Origin" in flow.request.headers:
                    flow.request.path = self.clients[flow.request.headers["Origin"]]
                else:
                    flow.request.path = self.clients[flow.request.scheme + "://" + flow.request.host + port]
                if flow.request.text:
                    flow.request.text = b64encode(bytes.fromhex(flow.request.text.encode('latin-1').hex())).decode('utf-8')

                flow.request.host = self.config["web_server"]["ip_address"]
                flow.request.port = self.config["web_server"]["port"]

                flow.request.headers["cookie"] = _req_cookies_str
        except Exception as e:
            print(f"Exception in request : {e}", flush=True)

    
    def response(self, flow):
        try:
            if "x-content-type-options" in flow.response.headers:
                del flow.response.headers["x-content-type-options"]
            if "Content-Type" not in flow.response.headers:
                flow.response.headers["Content-Type"] = "text/html"
            if "mitmproxy_override" in flow.response.headers and flow.response.headers["mitmproxy_override"]:
                headers = json.loads(flow.response.headers["mitmproxy_override"])
                new_headers = Headers()
                for i in headers:
                    if i not in ["content-encoding"]:
                        new_headers[i] = headers[i]
                flow.response.headers = new_headers
                flow.response.content = b64decode(flow.response.content)
            elif 300 < flow.response.status_code < 500 and self.config["proxy"]["do_not_proxify"]["force_proxify_on_errors"]:
                port = "" if flow.request.port == 443 else ':' + str(flow.request.port)
                path = flow.request.path
                if path[0] != "/":
                    path = '/' + path
                url = flow.request.scheme + "://" + flow.request.host + port + path
                flow.response.status_code = 301
                flow.response.headers["Location"] = url + "&proxyforced"


        except Exception as e:
            print(f"Exception in response : {e}",flush=True)


async def start_proxy(params, config):
    opts = options.Options(listen_host=config["proxy"]["interface"], listen_port=config["proxy"]["port"], ssl_insecure=True, confdir="./certificates", ssl_verify_upstream_trusted_confdir="./certificates", ssl_verify_upstream_trusted_ca="./certificates/mitmproxy-ca.pem")

    master = dump.DumpMaster(
        opts,
        with_termlog=False,
        with_dumper=False,
    )
    master.addons.add(params)
    
    await master.run()
    return master