我来帮你了解如何用Python编写一个简单的HTTP代理服务器,这里提供两种实现方法：
使用socket基础实现
import socket
import threading
class SimpleProxy:
def __init__(self, host='127.0.0.1', port=8888):
self.host = host
self.port = port
self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
def start(self):
self.server.bind((self.host, self.port))
self.server.listen(100)
print(f"代理服务器启动在 {self.host}:{self.port}")
while True:
client_socket, client_addr = self.server.accept()
print(f"收到连接: {client_addr}")
# 为每个客户端创建新线程处理
thread = threading.Thread(target=self.handle_client, args=(client_socket,))
thread.daemon = True
thread.start()
def handle_client(self, client_socket):
try:
# 接收客户端请求
request = client_socket.recv(4096)
if not request:
return
# 解析请求行
request_line = request.split(b'\r\n')[0].decode()
print(f"请求: {request_line}")
# 提取目标主机和端口
method, url, _ = request_line.split()
if method == 'CONNECT':
# 处理HTTPS连接
self.handle_connect(client_socket, url)
else:
# 处理HTTP请求
self.handle_http(client_socket, request, url)
except Exception as e:
print(f"处理请求错误: {e}")
finally:
client_socket.close()
def handle_http(self, client_socket, request, url):
# 解析目标主机
if url.startswith('http://'):
url = url[7:]
host_port = url.split('/')[0]
if ':' in host_port:
host, port = host_port.split(':')
port = int(port)
else:
host = host_port
port = 80
# 连接目标服务器
remote_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
remote_socket.connect((host, port))
# 转发请求
remote_socket.send(request)
# 接收响应并转发给客户端
while True:
data = remote_socket.recv(4096)
if not data:
break
client_socket.send(data)
finally:
remote_socket.close()
def handle_connect(self, client_socket, url):
# 解析目标主机和端口
host, port = url.split(':')
port = int(port)
try:
# 连接目标服务器
remote_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
remote_socket.connect((host, port))
# 发送200连接成功响应
client_socket.send(b'HTTP/1.1 200 Connection Established\r\n\r\n')
# 双向数据传输
self.relay_data(client_socket, remote_socket)
except Exception as e:
print(f"CONNECT错误: {e}")
client_socket.send(b'HTTP/1.1 502 Bad Gateway\r\n\r\n')
finally:
remote_socket.close()
def relay_data(self, client_socket, remote_socket):
# 创建两个线程进行双向数据传输
def forward(src, dst):
try:
while True:
data = src.recv(4096)
if not data:
break
dst.send(data)
except:
pass
t1 = threading.Thread(target=forward, args=(client_socket, remote_socket))
t2 = threading.Thread(target=forward, args=(remote_socket, client_socket))
t1.daemon = True
t2.daemon = True
t1.start()
t2.start()
t1.join()
t2.join()
if __name__ == '__main__':
proxy = SimpleProxy()
proxy.start()
使用asyncio实现（更现代的方式）
import asyncio
import aiohttp
from aiohttp import web
class AsyncProxy:
def __init__(self, host='127.0.0.1', port=8888):
self.host = host
self.port = port
async def handle_request(self, request):
# 获取目标URL
url = str(request.url)
print(f"代理请求: {url}")
try:
async with aiohttp.ClientSession() as session:
# 转发请求
async with session.request(
method=request.method,
url=url,
headers=request.headers,
data=request.content
) as response:
# 构建响应
resp_headers = dict(response.headers)
# 移除某些头部
resp_headers.pop('Transfer-Encoding', None)
return web.Response(
body=await response.read(),
status=response.status,
headers=resp_headers
)
except Exception as e:
return web.Response(
text=f"代理错误: {str(e)}",
status=502
)
async def handle_connect(self, request):
# 处理CONNECT方法（HTTPS）
host, port = request.path.split(':')
print(f"CONNECT代理: {host}:{port}")
response = web.StreamResponse()
response.status = 200
response.headers['Connection'] = 'keep-alive'
await response.prepare(request)
# 这里需要处理隧道连接
# 由于aiohttp的限制，这个实现较复杂，完整实现需要底层socket
await response.write(b'HTTP/1.1 200 Connection Established\r\n\r\n')
return response
async def start(self):
app = web.Application()
app.router.add_route('*', '/{path:.*}', self.handle_request)
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, self.host, self.port)
await site.start()
print(f"代理服务器启动在 {self.host}:{self.port}")
# 保持运行
await asyncio.Event().wait()
if __name__ == '__main__':
proxy = AsyncProxy()
asyncio.run(proxy.start())
使用说明
-
安装依赖（方法二需要）：
pip install aiohttp
-
运行代理：
# 方法一 python proxy_server.py
方法二
python async_proxy.py
3. **配置浏览器使用代理**：
- HTTP代理: `127.0.0.1:8888`
- 浏览器设置中配置代理服务器
## 注意事项
1. **功能限制**：
- 基础版本只支持HTTP，HTTPS支持有限
- 没有缓存功能
- 缺少认证机制
2. **安全考虑**：
- 仅用于开发和测试环境
- 不要在生产环境使用
- 建议只在本地网络使用
3. **性能优化**：
- 可以使用线程池代替每请求创建线程
- 添加连接池复用
- 实现请求缓存
这个实现可以帮助你理解HTTP代理的基本工作原理，适合学习和测试使用。标签： HTTP代理