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:  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".   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   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: 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())