怎么用脚本实现简易代理

wen 实用脚本 2026-08-01 2

本文目录导读：

怎么用脚本实现简易代理

Python + Flask 实现HTTP代理
Python socket实现TCP代理
Bash + socat 快速代理
Node.js 简易代理
Python实现支持HTTP/HTTPS的代理
最简单的Python HTTP代理（不带认证）
使用建议

我来介绍几种用脚本实现简易代理的方法,从简单到复杂：

Python + Flask 实现HTTP代理

最简单的HTTP转发代理：

# simple_proxy.py
from flask import Flask, request, Response
import requests
app = Flask(__name__)
@app.route('/<path:path>', methods=['GET', 'POST', 'PUT', 'DELETE'])
def proxy(path):
    # 目标服务器地址
    target_url = f'http://target-server.com/{path}'
    # 转发请求
    resp = requests.request(
        method=request.method,
        url=target_url,
        headers=request.headers,
        data=request.get_data(),
        params=request.args,
        stream=True
    )
    # 返回响应
    return Response(
        resp.content,
        status=resp.status_code,
        headers=dict(resp.headers)
    )
if __name__ == '__main__':
    app.run(port=8080)

Python socket实现TCP代理

# tcp_proxy.py
import socket
import threading
class TCPProxy:
    def __init__(self, listen_host='127.0.0.1', listen_port=8080, 
                 target_host='example.com', target_port=80):
        self.listen_host = listen_host
        self.listen_port = listen_port
        self.target_host = target_host
        self.target_port = target_port
    def start(self):
        server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server.bind((self.listen_host, self.listen_port))
        server.listen(5)
        print(f'代理服务器启动: {self.listen_host}:{self.listen_port}')
        while True:
            client_sock, client_addr = server.accept()
            thread = threading.Thread(target=self.handle_client, 
                                    args=(client_sock,))
            thread.start()
    def handle_client(self, client_sock):
        try:
            # 连接目标服务器
            target_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            target_sock.connect((self.target_host, self.target_port))
            # 双向转发数据
            t1 = threading.Thread(target=self.forward, 
                                args=(client_sock, target_sock))
            t2 = threading.Thread(target=self.forward, 
                                args=(target_sock, client_sock))
            t1.start()
            t2.start()
            t1.join()
            t2.join()
        except Exception as e:
            print(f'错误: {e}')
        finally:
            client_sock.close()
            if 'target_sock' in locals():
                target_sock.close()
    @staticmethod
    def forward(src, dst):
        try:
            while True:
                data = src.recv(4096)
                if not data:
                    break
                dst.sendall(data)
        except:
            pass
# 使用示例
if __name__ == '__main__':
    proxy = TCPProxy(target_host='www.example.com', target_port=80)
    proxy.start()

Bash + socat 快速代理

# 简单的TCP端口转发
socat TCP4-LISTEN:8080 TCP4:target-server:443
# 反向代理
socat TCP4-LISTEN:8080,fork,reuseaddr TCP4:target-server:80

Node.js 简易代理

// proxy.js
const http = require('http');
const httpProxy = require('http-proxy');
const proxy = httpProxy.createProxyServer({});
const server = http.createServer((req, res) => {
    // 转发到目标服务器
    proxy.web(req, res, {
        target: 'http://target-server.com',
        changeOrigin: true
    });
});
server.listen(8080, () => {
    console.log('代理服务器运行在端口 8080');
});

Python实现支持HTTP/HTTPS的代理

# http_https_proxy.py
import socket
import threading
import ssl
import base64
class HTTPProxy:
    def __init__(self, host='127.0.0.1', port=8080):
        self.host = host
        self.port = port
    def start(self):
        server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server.bind((self.host, self.port))
        server.listen(100)
        print(f'HTTP代理启动: {self.host}:{self.port}')
        while True:
            client_sock, addr = server.accept()
            threading.Thread(target=self.handle_client, 
                           args=(client_sock,)).start()
    def handle_client(self, client_sock):
        try:
            # 读取请求头
            request_data = client_sock.recv(8192).decode('utf-8')
            lines = request_data.split('\r\n')
            # 解析请求
            method, url, protocol = lines[0].split(' ')
            # 解析目标地址
            if url.startswith('http://'):
                # HTTP请求
                target_host = url.split('://')[1].split('/')[0]
                target_port = 80
                path = '/' + '/'.join(url.split('/')[3:])
            else:
                # HTTPS连接
                target_host, target_port = self.parse_host_port(url)
                path = '/'
            # 连接目标服务器
            target_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            target_sock.connect((target_host, target_port))
            # 如果是HTTPS请求
            if method == 'CONNECT':
                client_sock.send(b'HTTP/1.1 200 Connection Established\r\n\r\n')
            else:
                # 修改请求头并转发
                modified_request = self.modify_request(request_data, path)
                target_sock.send(modified_request.encode('utf-8'))
            # 双向转发
            t1 = threading.Thread(target=self.forward, 
                                args=(client_sock, target_sock))
            t2 = threading.Thread(target=self.forward, 
                                args=(target_sock, client_sock))
            t1.start()
            t2.start()
            t1.join()
            t2.join()
        except Exception as e:
            print(f'错误: {e}')
        finally:
            client_sock.close()
    def parse_host_port(self, url):
        # 解析 host:port
        if ':' in url:
            host, port = url.split(':')
            return host, int(port)
        return url, 443
    def modify_request(self, request_data, new_path):
        # 修改请求路径
        lines = request_data.split('\r\n')
        request_line = lines[0].split(' ')
        request_line[1] = new_path
        lines[0] = ' '.join(request_line)
        return '\r\n'.join(lines)
    @staticmethod
    def forward(src, dst):
        try:
            while True:
                data = src.recv(8192)
                if not data:
                    break
                dst.sendall(data)
        except:
            pass
if __name__ == '__main__':
    proxy = HTTPProxy()
    proxy.start()

最简单的Python HTTP代理（不带认证）

import http.client
import http.server
import urllib.parse
class SimpleProxy(http.server.BaseHTTPRequestHandler):
    def do_GET(self):
        self._proxy()
    def do_POST(self):
        self._proxy()
    def _proxy(self):
        # 解析URL
        url = urllib.parse.urlparse(self.path)
        # 连接到目标服务器
        if url.scheme == 'https':
            conn = http.client.HTTPSConnection(url.hostname, url.port or 443)
        else:
            conn = http.client.HTTPConnection(url.hostname, url.port or 80)
        # 获取请求体
        content_length = int(self.headers.get('Content-Length', 0))
        body = self.rfile.read(content_length) if content_length else None
        # 发送请求
        conn.request(self.command, url.path + ('?' + url.query if url.query else ''), 
                    body=body, headers=dict(self.headers))
        # 获取响应
        response = conn.getresponse()
        # 返回响应
        self.send_response(response.status)
        for key, value in response.getheaders():
            self.send_header(key, value)
        self.end_headers()
        self.wfile.write(response.read())
        conn.close()
if __name__ == '__main__':
    server = http.server.HTTPServer(('127.0.0.1', 8080), SimpleProxy)
    print('代理服务器启动在 127.0.0.1:8080')
    server.serve_forever()

使用建议

简单测试：使用 socat 或 Python 版本
生产环境：建议使用成熟的工具如 nginx、haproxy
调试用：Flask版本最容易理解和修改
性能优先：考虑使用 Go 或 C++ 实现

运行示例：

# 运行Python代理
python simple_proxy.py
# 测试代理
curl -x http://127.0.0.1:8080 http://example.com

这些脚本可以帮助你理解代理的基本原理，但在实际生产环境中，建议使用成熟、经过测试的代理软件。