Python实现带认证转发的本地透明代理：解决应用代理账密兼容问题

Hey there! Let's tackle this problem step by step. You need a local proxy B that handles authentication for your target proxy, so app A (which doesn't support authenticated proxies) can send requests through it. Here's how to implement the request flow and build it in Python:

First, let's map out the end-to-end path to make sure we cover every step:

• App A: Sends unauthenticated HTTP/HTTPS requests directly to your local proxy B (no proxy auth required here)
• Proxy B: Receives the request, adds the target proxy's username/password authentication, then forwards the request to the authenticated target proxy
• Target Proxy: Validates the credentials, forwards the request to the final target service, and sends the response back to proxy B
• Proxy B: Takes the response from the target proxy and relays it straight back to app A

The tricky part is handling both plain HTTP and encrypted HTTPS requests (since HTTPS requires a CONNECT tunnel to establish the secure connection before forwarding traffic).

We'll use built-in modules for the proxy core and requests for simplified forwarding, plus Flask for the REST management API.

1. Basic HTTP Request Handling #

For plain HTTP requests, we can directly modify the request to include proxy authentication and forward it to the target proxy.

from http.server import HTTPServer, BaseHTTPRequestHandler
import requests
import os

# Load config from environment variables (never hardcode credentials!)
TARGET_PROXY = os.getenv("TARGET_PROXY", "http://your-target-proxy:port")
PROXY_USER = os.getenv("PROXY_USER", "your-username")
PROXY_PASS = os.getenv("PROXY_PASS", "your-password")

class LocalProxyHandler(BaseHTTPRequestHandler):
    # Handle all standard HTTP methods
    def do_GET(self):
        self._forward_request("GET")
    
    def do_POST(self):
        self._forward_request("POST")
    
    def do_PUT(self):
        self._forward_request("PUT")
    
    def do_DELETE(self):
        self._forward_request("DELETE")
    
    def _forward_request(self, method):
        # Read request body from app A
        content_length = int(self.headers.get("Content-Length", 0))
        request_body = self.rfile.read(content_length) if content_length > 0 else None

        # Prepare headers to forward (exclude conflicting headers)
        forward_headers = {k: v for k, v in self.headers.items() if k not in ["Host", "Connection"]}

        # Configure target proxy with authentication
        proxies = {
            "http": TARGET_PROXY,
            "https": TARGET_PROXY
        }
        auth = (PROXY_USER, PROXY_PASS)

        try:
            # Forward request to target proxy
            target_url = f"http://{self.headers['Host']}{self.path}"
            response = requests.request(
                method=method,
                url=target_url,
                headers=forward_headers,
                data=request_body,
                proxies=proxies,
                auth=auth,
                allow_redirects=False
            )

            # Send response back to app A
            self.send_response(response.status_code)
            for header_name, header_value in response.headers.items():
                self.send_header(header_name, header_value)
            self.end_headers()
            self.wfile.write(response.content)
        except Exception as e:
            self.send_error(500, f"Failed to forward request: {str(e)}")

if __name__ == "__main__":
    # Start local proxy on port 8080
    server = HTTPServer(("0.0.0.0", 8080), LocalProxyHandler)
    print(f"Local proxy B running on http://0.0.0.0:8080")
    server.serve_forever()

2. HTTPS Request Handling (CONNECT Tunnel) #

HTTPS requires establishing a CONNECT tunnel first. We'll add support for this in our handler:

import socket
import base64
import select
import ssl

# Add this method to the LocalProxyHandler class above
def do_CONNECT(self):
    # Parse target host and port from the CONNECT request
    host, port = self.path.split(":")
    port = int(port) if port else 443

    # Connect to the target proxy
    proxy_host, proxy_port = TARGET_PROXY.replace("http://", "").split(":")
    proxy_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        proxy_socket.connect((proxy_host, int(proxy_port)))

        # Send proxy authentication to the target proxy
        auth_encoded = base64.b64encode(f"{PROXY_USER}:{PROXY_PASS}".encode()).decode()
        connect_request = (
            f"CONNECT {self.path} HTTP/1.1\r\n"
            f"Host: {self.path}\r\n"
            f"Proxy-Authorization: Basic {auth_encoded}\r\n\r\n"
        )
        proxy_socket.sendall(connect_request.encode())

        # Check if target proxy accepted the connection
        proxy_response = b""
        while b"\r\n\r\n" not in proxy_response:
            proxy_response += proxy_socket.recv(4096)
        
        if b"200 Connection established" not in proxy_response:
            self.send_error(502, "Target proxy refused connection")
            return

        # Notify app A the tunnel is ready
        self.send_response(200)
        self.end_headers()

        # Forward encrypted traffic between app A and target proxy (bidirectional)
        client_socket = self.connection
        client_socket.setblocking(False)
        proxy_socket.setblocking(False)

        while True:
            readable, _, _ = select.select([client_socket, proxy_socket], [], [], 5)
            if not readable:
                break
            
            for sock in readable:
                data = sock.recv(4096)
                if not data:
                    return
                if sock == client_socket:
                    proxy_socket.sendall(data)
                else:
                    client_socket.sendall(data)
    except Exception as e:
        self.send_error(500, f"Failed to establish HTTPS tunnel: {str(e)}")
    finally:
        proxy_socket.close()

3. Add REST Management API #

We'll use Flask to build a simple API to manage the proxy (update config, start/stop):

from flask import Flask, request, jsonify
import threading

app = Flask(__name__)
proxy_server = None

@app.route("/api/proxy/config", methods=["GET", "PUT"])
def manage_config():
    global TARGET_PROXY, PROXY_USER, PROXY_PASS
    if request.method == "GET":
        # Return config (omit password for security)
        return jsonify({
            "target_proxy": TARGET_PROXY,
            "username": PROXY_USER
        })
    elif request.method == "PUT":
        new_config = request.json
        TARGET_PROXY = new_config.get("target_proxy", TARGET_PROXY)
        PROXY_USER = new_config.get("username", PROXY_USER)
        PROXY_PASS = new_config.get("password", PROXY_PASS)
        return jsonify({"message": "Proxy config updated successfully"})

@app.route("/api/proxy/start", methods=["POST"])
def start_proxy():
    global proxy_server
    if proxy_server and proxy_server._BaseServer__is_running:
        return jsonify({"message": "Proxy is already running"})
    
    def run_proxy():
        proxy_server = HTTPServer(("0.0.0.0", 8080), LocalProxyHandler)
        proxy_server.serve_forever()
    
    threading.Thread(target=run_proxy, daemon=True).start()
    return jsonify({"message": "Proxy started on port 8080"})

@app.route("/api/proxy/stop", methods=["POST"])
def stop_proxy():
    global proxy_server
    if not proxy_server or not proxy_server._BaseServer__is_running:
        return jsonify({"message": "Proxy is not running"})
    
    proxy_server.shutdown()
    return jsonify({"message": "Proxy stopped successfully"})

if __name__ == "__main__":
    # Start proxy in background thread
    threading.Thread(target=lambda: HTTPServer(("0.0.0.0", 8080), LocalProxyHandler).serve_forever(), daemon=True).start()
    # Start Flask API on port 5000
    app.run(host="0.0.0.0", port=5000)

• Security: Always use environment variables or encrypted config files to store credentials—never hardcode them in your code.
• Performance: The above example is single-threaded. For high-concurrency use cases, switch to an async framework like aiohttp instead of http.server.
• Header Handling: Be careful with headers like Connection and Host—they can cause conflicts if not filtered properly when forwarding requests.
• Error Handling: Expand the exception handling to cover cases like target proxy downtime, authentication failures, and timeouts.

内容的提问来源于stack exchange，提问作者Sonique