Programmiererforum

Code: Select all

import asyncio
import base64

from environs import Env
from mitmproxy import options
from mitmproxy.http import HTTPFlow
from mitmproxy.net import server_spec
from mitmproxy.tools.console.master import ConsoleMaster
from proxy_server.constants import PROXY_SERVER_HOST, PROXY_SERVER_PORT
from proxy_server.logger import logger

env = Env()
env.read_env()

class OxylabsProxyAddon:
def __init__(self, protocol: str = "https"):
self.username = env("OXYLAB_USERNAME")
self.password = env("OXYLAB_PASSWORD")
self.host = "pr.oxylabs.io"
self.port = 7777
self.protocol = protocol

self.auth = f"customer-{self.username}:{self.password}"
self.credentials = base64.b64encode(self.auth.encode()).decode("utf8")
self.spec = server_spec.parse(
f"{self.protocol}://{self.host}:{self.port}", self.protocol
)

async def http_connect(self, flow: HTTPFlow):
if "whatismyip" not in flow.request.url.lower():
return

flow.request.headers["Proxy-Authorization"] = f"Basic {self.credentials}"
flow.metadata["upstream_proxy"] = self.spec

logger.info(f"Upstream proxy set for {flow.request.url}")

async def run_proxy():
opts = options.Options(
listen_host=PROXY_SERVER_HOST, listen_port=PROXY_SERVER_PORT, mode=["regular"]
)
master = ConsoleMaster(opts)
master.addons.add(OxylabsProxyAddon())
await master.run()

if __name__ == "__main__":
asyncio.run(run_proxy())