How to Dynamically Route and Authenticate Upstream Proxies in mitmproxy Based on URL?

Question

Hello Stack Overflow community,

I am working on a project using mitmproxy and I'm facing a challenge where I need to dynamically route requests to different upstream proxies based on the URL, along with handling authentication for these proxies. I would appreciate any guidance or suggestions on how to implement this.

Requirements:

1. Dynamic Proxy Routing:

• If the incoming request URL is https://example.com/123, mitmproxy should forward it through "Proxy A".
• If the URL is https://example.com/456, it should use "Proxy B".

2. Authentication for Each Proxy:

• Both "Proxy A" and "Proxy B" require authentication. The solution needs to handle this, ensuring the correct credentials are used based on which proxy is selected

3. Implementation in an Addon

• I am looking to implement this as an addon in mitmproxy, without using specific command-line arguments like --mode upstream:http://example.com:8081.

My Attempts/Research: I've looked into the documentation but haven't found a clear way to change the upstream proxy dynamically based on the request URL, especially when it comes to incorporating authentication for different proxies.

Questions:

• How can I programmatically route requests to different upstream proxies based on the URL in mitmproxy?
• What is the most efficient method to authenticate with these proxies, keeping in mind that each proxy has different credentials?
• Are there particular functions or modules within mitmproxy that I should look into for achieving this?

Any code examples, documentation references, or insights into how to approach this in mitmproxy would be extremely helpful.

Thank you in advance for your help!

below is the code I tried but not satisfied

import base64
from mitmproxy import http

class DynamicUpstreamProxy:
    def __init__(self):
        self.proxy_A = ("upstream-proxy-A.com", 8081)
        self.proxy_B = ("upstream-proxy-B.com", 8082)
        self.proxy_A_auth = self.encode_credentials("usernameA", "passwordA")
        self.proxy_B_auth = self.encode_credentials("usernameB", "passwordB")

    def encode_credentials(self, username, password):
        credentials = f"{username}:{password}"
        encoded_credentials = base64.b64encode(credentials.encode()).decode()
        return f"Basic {encoded_credentials}"

    def request(self, flow: http.HTTPFlow):
        url = flow.request.pretty_url

        if url.startswith("https://example.com/123"):
            # Upstream Proxy A
            flow.live.change_upstream_proxy_server(self.proxy_A)
            flow.request.headers["Proxy-Authorization"] = self.proxy_A_auth

        elif url.startswith("https://example.com/456"):
            #  Upstream Proxy B
            flow.live.change_upstream_proxy_server(self.proxy_B)
            flow.request.headers["Proxy-Authorization"] = self.proxy_B_auth

addons = [
    DynamicUpstreamProxy()
]

then run addon

mitmproxy -s my_upstream_addon.py

Answer 1

How about something like the below? This routes each request to an upstream proxy based on the value of a custom header called "X-Upstream-Proxy" or no upstream if the header does not exist (tested with mitmproxy v10.1.3).

Regarding authentication with the upstream proxy server, I haven't tested this but I assume an upstream proxy value of "http://user:pass@proxy-hostname:8080" or similar should work.

This code can be easily modified to run as a command line add-on to mitmproxy, take a look at a relevant example here: https://github.com/mitmproxy/mitmproxy/blob/main/examples/contrib/change_upstream_proxy.py

import asyncio

from urllib.parse import urlparse

from mitmproxy.addons.proxyserver import Proxyserver
from mitmproxy.options import Options
from mitmproxy.tools.dump import DumpMaster
from mitmproxy.http import HTTPFlow
from mitmproxy.connection import Server
from mitmproxy.net.server_spec import ServerSpec

UPSTREAM_PROXY_HEADER = 'X-Upstream-Proxy'


def get_upstream_proxy(flow: HTTPFlow) -> tuple[str, tuple[str, int]] | None:
    upstream_proxy = flow.request.headers.get(UPSTREAM_PROXY_HEADER)
    if upstream_proxy is not None:
        parsed_upstream_proxy = urlparse(upstream_proxy)

        if parsed_upstream_proxy.scheme not in ('http', 'https'):
            return None

        del flow.request.headers[UPSTREAM_PROXY_HEADER]
        return parsed_upstream_proxy.scheme, (parsed_upstream_proxy.hostname, parsed_upstream_proxy.port)

    return None


class DynamicUpstreamProxy:

    def request(self, flow: HTTPFlow) -> None:
        upstream_proxy = get_upstream_proxy(flow)

        print(flow.request)

        if upstream_proxy is not None:
            has_proxy_changed = upstream_proxy != flow.server_conn.via
            server_connection_already_open = flow.server_conn.timestamp_start is not None

            if has_proxy_changed and server_connection_already_open:
                # server_conn already refers to an existing connection (which cannot be modified),
                # so we need to replace it with a new server connection object.
                flow.server_conn = Server(address=flow.server_conn.address)

            flow.server_conn.via = ServerSpec(upstream_proxy)
        else:
            flow.server_conn.via = None
        
if __name__ == '__main__':
    options = Options(listen_host='127.0.0.1', listen_port=8080, http2=True, mode=['upstream:http://dummy:8888/'])
    m = DumpMaster(options, with_termlog=True, with_dumper=False, loop=asyncio.get_event_loop())

    m.server = Proxyserver()
    m.addons.add(DynamicUpstreamProxy())
    asyncio.run(m.run())