mitmproxystartstop.py

import logging

from mitmproxy.tools.dump import DumpMaster
from mitmproxy import options
from mitmproxy import http
import asyncio

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# sudo lsof -i :8080
# sudo kill -9 PID

class ProxyManager:
    def __init__(self, port=8080):
        self.port = port
        self.master = None

    def start_proxy(self):
        """Start MITMproxy"""
        opts = options.Options(
            listen_host='0.0.0.0',  # Listen on all interfaces
            listen_port=self.port,
            mode='regular',  # Explicit proxy mode
            ssl_insecure=True  # Allow insecure connections

        )
        logger.info(f"Starting proxy on 0.0.0.0:{self.port}")

        async def run_proxy():
            self.master = DumpMaster(opts)
            await self.master.run()

        asyncio.run(run_proxy())


    def view_request(self, url, flow:http.HTTPFlow) -> None:
        """View a specific request based on url."""

        if flow.request.pretty_url == url:
            logger.info(f"Found the URL: {flow.request.pretty_url}")
            logger.info(f"Request Method: {flow.request.method}")
            logger.info(f"Request Headers: {flow.request.headers}")
            logger.info(f"Request Body: {flow.request.content}")


    def capture_url_request(self, url):
        """
        This should return a request that is specific to a url.


        :parameter: url - Url to watch for.
        :return: The request
        """

        pass

    def alter_request_header(self, url, header:dict):
        """Alter a header of the request, send it and return the response."""

        pass

    def alter_request_content(self, url, content: dict):
        """Alter a part of the request, send it and return the response."""

        pass