如何让Python HTTP代理拦截转发HTTPS流量并拦截HTTPS网站？

问题背景 #

现有Python HTTP代理可正常处理HTTP请求，但不支持HTTPS协议。尝试添加ssl库并使用自签名证书修改后，仍无法加载HTTPS网站页面，需要实现HTTPS流量的拦截、转发功能，同时支持HTTPS网站拦截（如youtube.com）。

原始HTTP代理代码 #

import socket
import signal
import threading
import logging
from datetime import datetime

class ProxyServer:

    def __init__(self, config):
        # Shutdown on Ctrl+C
        signal.signal(signal.SIGINT, self.shutdown) 

        # Create a TCP socket
        self.serverSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

        # Re-use the socket
        self.serverSocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

        # Bind the socket to a public host, and a port   
        self.serverSocket.bind((config['HOST_NAME'], config['BIND_PORT']))
        
        self.serverSocket.listen(10) # Become a server socket
        self.__clients = {}

        self.config = config

        # Setup logging
        logging.basicConfig(filename=config['LOG_FILE'], level=logging.INFO, 
                            format='%(asctime)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S')

        # Blocklist
        self.blocklist = config.get('BLOCKLIST',[])

    def shutdown(self, signum, frame):
        print("Shutting down server...")
        self.serverSocket.close()
        exit(0)

    def _getClientName(self, client_address):
        return f"{client_address[0]}:{client_address[1]}"

    def proxy_thread(self, clientSocket, client_address):
        # Get the request from the browser
        request = clientSocket.recv(self.config['MAX_REQUEST_LEN'])

        # Log the request
        self.log_request(request, client_address)

        # Parse the first line
        first_line = request.split(b'\n')[0]

        # Get URL
        url = first_line.split(b' ')[1].decode('utf-8')
        http_pos = url.find("://") # Find pos of ://
        if http_pos == -1:
            temp = url
        else:
            temp = url[(http_pos+3):] # Get the rest of URL

        port_pos = temp.find(":") # Find the port pos (if any)

        # Find end of web server
        webserver_pos = temp.find("/")
        if webserver_pos == -1:
            webserver_pos = len(temp)

        webserver = ""
        port = -1
        if port_pos == -1 or webserver_pos < port_pos: 
            # Default port 
            port = 80 
            webserver = temp[:webserver_pos] 
        else: # Specific port 
            port = int((temp[(port_pos+1):])[:webserver_pos-port_pos-1])
            webserver = temp[:port_pos]

        # Check if the domain is in the blocklist
        if self.is_blocked(webserver):
            self.deny_request(clientSocket, webserver)
            return

        try:
            # Create a socket to connect to the web server
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 
            s.settimeout(self.config['CONNECTION_TIMEOUT'])
            s.connect((webserver, port))
            s.sendall(request)

            while True:
                data = s.recv(self.config['MAX_REQUEST_LEN'])
                if len(data) > 0:
                    clientSocket.send(data)
                else:
                    break
        except socket.error as e:
            print(f"Socket error: {e}")
        finally:
            clientSocket.close()
            s.close()

    def is_blocked(self, domain):
        return any(blocked_domain in domain for blocked_domain in self.blocklist)

    def deny_request(self, clientSocket, domain):
        response = (
        "HTTP/1.1 403 Forbidden\r\n"
        "Content-Type: text/html\r\n"
        "Connection: close\r\n"
        "\r\n"
        "<html>"
        "<head>"
        "<style>"
        "body {"
        "  font-family: Arial, sans-serif;"
        "  background-color: #FF0000;"
        "  display: flex;"
        "  justify-content: center;"
        "  align-items: center;"
        "  height: 100vh;"
        "  margin: 0;"
        "}"
        "h1 {"
        "  color: #333;"
        "  font-size: 36px;"
        "  margin-bottom: 20px;"
        "}"
        "p {"
        "  color: #666;"
        "  font-size: 18px;"
        "}"
        ".container {"
        "  background-color: #fff;"
        "  padding: 40px;"
        "  border-radius: 8px;"
        "  box-shadow: 0 2px 8px rgba(0, 0, 0, 0.1);"
        "  text-align: center;"
        "}"
        "</style>"
        "</head>"
        "<body>"
        "<div class='container'>"
        "<h1>403 Forbidden</h1>"
        f"<p>Access to {domain} is blocked.</p>"
        "</div>"
        "</body>"
        "</html>"
        )
        clientSocket.sendall(response.encode('utf-8'))
        clientSocket.close()
        logging.info(f"Blocked access to {domain}")

    def log_request(self, request, client_address):
        request_lines = request.decode('utf-8', errors='replace').split('\n')
        log_message = f"Source IP: {client_address[0]}\n"

        for line in request_lines:
            log_message += line + "\n"

        logging.info(log_message)

    def start(self):
        while True:
            try:
                # Establish the connection
                clientSocket, client_address = self.serverSocket.accept() 
                
                d = threading.Thread(name=self._getClientName(client_address), 
                                     target=self.proxy_thread, 
                                     args=(clientSocket, client_address))
                d.setDaemon(True)
                d.start()
            except Exception as e:
                print(f"Error:{e}")
if __name__ == "__main__":
    blocklist_file=open("blocklist.txt","r")
    blocklist=blocklist_file.read().splitlines()
    blocklist_file.close()
    config = {
        'HOST_NAME': '127.0.0.1',
        'BIND_PORT': 8888,
        'MAX_REQUEST_LEN': 8192,
        'CONNECTION_TIMEOUT': 5,
        'LOG_FILE': 'proxy_server.log',
        'BLOCKLIST': blocklist
    }
    
    server = ProxyServer(config)
    server.start()

尝试修改后的HTTPS处理代码片段 #

def proxy_thread(self, clientSocket, client_address):
        request = clientSocket.recv(self.config['MAX_REQUEST_LEN'])
        self.log_request(request, client_address)
        first_line = request.split(b'\n')[0]
        url = first_line.split(b' ')[1].decode('utf-8')
        http_pos = url.find("://")
        if http_pos == -1:
            temp = url
        else:
            temp = url[(http_pos+3):]
        port_pos = temp.find(":")
        webserver_pos = temp.find("/")
        if webserver_pos == -1:
            webserver_pos = len(temp)
        webserver = ""
        port = -1
        if port_pos == -1 or webserver_pos < port_pos:
            port = 80
            webserver = temp[:webserver_pos]
        else:
            port = int((temp[(port_pos+1):])[:webserver_pos-port_pos-1])
            webserver = temp[:port_pos]
        if self.is_blocked(webserver):
            self.deny_request(clientSocket, webserver)
            return

        try:
            if port == 443:  # HTTPS
                context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
                context.load_cert_chain(certfile='server.crt', keyfile='server.key')
                s = context.wrap_socket(socket.socket(socket.AF_INET), server_hostname=webserver)
            else:  # HTTP
                s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.settimeout(self.config['CONNECTION_TIMEOUT'])
            s.connect((webserver, port))
            s.sendall(request)
            while True:
                data = s.recv(self.config['MAX_REQUEST_LEN'])
                if len(data) > 0:
                    clientSocket.send(data)
                else:
                    break
        except socket.error as e:
            print(f"Socket error: {e}")
        except ssl.SSLError as e:
            print(f"SSL error: {e}")
        finally:
            clientSocket.close()
            if 's' in locals():
                s.close()

核心解决方案 #

1. 处理HTTPS的CONNECT请求 #

浏览器访问HTTPS网站时会先发送CONNECT请求建立隧道，原代码未处理该逻辑，导致SSL连接无法建立。需要先解析CONNECT请求，返回200响应后再建立双向SSL通道。

2. 实现SSL中间人（MITM）模式 #

要拦截HTTPS内容，代理需作为中间人：

• 接收客户端CONNECT请求，解析目标域名和端口
• 返回HTTP/1.1 200 Connection Established响应，告知客户端隧道已建立
• 用自签名证书与客户端建立SSL连接（代理作为服务器）
• 代理本身与目标服务器建立SSL连接（代理作为客户端）
• 双向转发客户端与服务器之间的加密流量

3. 修正SSL上下文配置 #

• 与客户端建立连接时，代理是SSL服务器，需使用ssl.Purpose.CLIENT_AUTH上下文
• 与目标服务器建立连接时，代理是SSL客户端，使用ssl.Purpose.SERVER_AUTH上下文
• 确保自签名证书已导入浏览器信任列表

完整修改后的proxy_thread方法 #

import ssl

def proxy_thread(self, clientSocket, client_address):
    request = clientSocket.recv(self.config['MAX_REQUEST_LEN'])
    self.log_request(request, client_address)
    first_line = request.split(b'\n')[0]
    request_method = first_line.split(b' ')[0].decode('utf-8')

    # 处理CONNECT请求（HTTPS）
    if request_method == 'CONNECT':
        # 解析CONNECT目标：格式为 "hostname:port"
        target = first_line.split(b' ')[1].decode('utf-8')
        webserver, port = target.split(':')
        port = int(port) if port else 443

        # 检查是否在黑名单
        if self.is_blocked(webserver):
            self.deny_request(clientSocket, webserver)
            return

        try:
            # 先返回200响应给客户端，告知隧道建立
            clientSocket.send(b'HTTP/1.1 200 Connection Established\r\n\r\n')

            # 与客户端建立SSL连接（代理作为服务器）
            client_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
            client_context.load_cert_chain(certfile='server.crt', keyfile='server.key')
            ssl_client = client_context.wrap_socket(clientSocket, server_side=True)

            # 与目标服务器建立SSL连接（代理作为客户端）
            server_context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
            ssl_server = server_context.wrap_socket(socket.socket(socket.AF_INET), server_hostname=webserver)
            ssl_server.connect((webserver, port))

            # 双向转发数据
            def forward(src, dst):
                while True:
                    data = src.recv(self.config['MAX_REQUEST_LEN'])
                    if not data:
                        break
                    dst.sendall(data)

            # 启动两个线程分别转发客户端<->服务器的流量
            threading.Thread(target=forward, args=(ssl_client, ssl_server), daemon=True).start()
            forward(ssl_server, ssl_client)

        except socket.error as e:
            print(f"Socket error: {e}")
        except ssl.SSLError as e:
            print(f"SSL error: {e}")
        finally:
            if 'ssl_client' in locals():
                ssl_client.close()
            if 'ssl_server' in locals():
                ssl_server.close()
        return

    # 处理普通HTTP请求（原有逻辑保留）
    url = first_line.split(b' ')[1].decode('utf-8')
    http_pos = url.find("://")
    if http_pos == -1:
        temp = url
    else:
        temp = url[(http_pos+3):]
    port_pos = temp.find(":")
    webserver_pos = temp.find("/")
    if webserver_pos == -1:
        webserver_pos = len(temp)
    webserver = ""
    port = -1
    if port_pos == -1 or webserver_pos < port_pos:
        port = 80
        webserver = temp[:webserver_pos]
    else:
        port = int((temp[(port_pos+1):])[:webserver_pos-port_pos-1])
        webserver = temp[:port_pos]

    if self.is_blocked(webserver):
        self.deny_request(clientSocket, webserver)
        return

    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(self.config['CONNECTION_TIMEOUT'])
        s.connect((webserver, port))
        s.sendall(request)
        while True:
            data = s.recv(self.config['MAX_REQUEST_LEN'])
            if len(data) > 0:
                clientSocket.send(data)
            else:
                break
    except socket.error as e:
        print(f"Socket error: {e}")
    finally:
        clientSocket.close()
        if 's' in locals():
            s.close()

4. 自签名证书生成命令 #

确保生成包含正确域名的证书：

openssl req -x509 -newkey rsa:4096 -keyout server.key -out server.crt -days 365 -nodes -subj "/CN=localhost"

• 如需拦截任意域名，可使用通配符/CN=*.com或添加多个SAN域名
• 将生成的server.crt导入浏览器的信任根证书列表

5. 关键注意事项 #

• 双向转发需启动两个线程，避免单向阻塞
• 代理作为SSL服务器时，必须设置server_side=True
• 与目标服务器建立SSL连接时，需指定server_hostname以支持SNI

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