forked from xiaohei/taiji-AI-PAD
148 lines
4.7 KiB
Python
148 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
HTTPS 代理服务器
|
|
用于测试环境，将 HTTPS 请求转发到本地 HTTP 服务
|
|
|
|
使用方法:
|
|
python3 https_proxy.py --port 8989 --target http://localhost:8000 \
|
|
--cert ssl_certs/cert.pem --key ssl_certs/key.pem
|
|
"""
|
|
|
|
import argparse
|
|
import ssl
|
|
import http.server
|
|
import urllib.request
|
|
import urllib.error
|
|
import socketserver
|
|
import json
|
|
from urllib.parse import urlparse
|
|
|
|
|
|
class ProxyHandler(http.server.BaseHTTPRequestHandler):
|
|
"""HTTPS 到 HTTP 代理处理器"""
|
|
|
|
target_url = "http://localhost:8000"
|
|
|
|
def do_request(self, method: str):
|
|
"""处理所有 HTTP 方法"""
|
|
# 构建目标 URL
|
|
target = f"{self.target_url}{self.path}"
|
|
|
|
# 读取请求体
|
|
content_length = int(self.headers.get('Content-Length', 0))
|
|
body = self.rfile.read(content_length) if content_length > 0 else None
|
|
|
|
# 构建请求头（转发原始头）
|
|
headers = {}
|
|
for key, value in self.headers.items():
|
|
# 跳过 hop-by-hop 头
|
|
if key.lower() not in ('host', 'connection', 'keep-alive',
|
|
'transfer-encoding', 'te', 'trailer',
|
|
'proxy-authorization', 'proxy-authenticate',
|
|
'upgrade'):
|
|
headers[key] = value
|
|
|
|
# 添加 X-Forwarded 头
|
|
headers['X-Forwarded-For'] = self.client_address[0]
|
|
headers['X-Forwarded-Proto'] = 'https'
|
|
|
|
try:
|
|
# 创建请求
|
|
req = urllib.request.Request(
|
|
target,
|
|
data=body,
|
|
headers=headers,
|
|
method=method
|
|
)
|
|
|
|
# 发送请求
|
|
with urllib.request.urlopen(req, timeout=30) as response:
|
|
# 发送响应状态
|
|
self.send_response(response.status)
|
|
|
|
# 转发响应头
|
|
for key, value in response.headers.items():
|
|
if key.lower() not in ('transfer-encoding', 'connection'):
|
|
self.send_header(key, value)
|
|
self.end_headers()
|
|
|
|
# 转发响应体
|
|
self.wfile.write(response.read())
|
|
|
|
except urllib.error.HTTPError as e:
|
|
self.send_response(e.code)
|
|
for key, value in e.headers.items():
|
|
if key.lower() not in ('transfer-encoding', 'connection'):
|
|
self.send_header(key, value)
|
|
self.end_headers()
|
|
self.wfile.write(e.read())
|
|
|
|
except urllib.error.URLError as e:
|
|
self.send_error(502, f"Bad Gateway: {e.reason}")
|
|
|
|
except Exception as e:
|
|
self.send_error(500, f"Internal Server Error: {str(e)}")
|
|
|
|
def do_GET(self):
|
|
self.do_request('GET')
|
|
|
|
def do_POST(self):
|
|
self.do_request('POST')
|
|
|
|
def do_PUT(self):
|
|
self.do_request('PUT')
|
|
|
|
def do_DELETE(self):
|
|
self.do_request('DELETE')
|
|
|
|
def do_PATCH(self):
|
|
self.do_request('PATCH')
|
|
|
|
def do_OPTIONS(self):
|
|
self.do_request('OPTIONS')
|
|
|
|
def log_message(self, format, *args):
|
|
"""自定义日志格式"""
|
|
print(f"[HTTPS Proxy] {self.client_address[0]} - {format % args}")
|
|
|
|
|
|
class ThreadedHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
|
|
"""支持多线程的 HTTP 服务器"""
|
|
daemon_threads = True
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description='HTTPS to HTTP Proxy Server')
|
|
parser.add_argument('--port', type=int, default=8989, help='HTTPS port to listen on')
|
|
parser.add_argument('--target', default='http://localhost:8000', help='Target HTTP URL')
|
|
parser.add_argument('--cert', required=True, help='SSL certificate file')
|
|
parser.add_argument('--key', required=True, help='SSL private key file')
|
|
|
|
args = parser.parse_args()
|
|
|
|
# 设置目标 URL
|
|
ProxyHandler.target_url = args.target.rstrip('/')
|
|
|
|
# 创建 SSL 上下文
|
|
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
|
ssl_context.load_cert_chain(args.cert, args.key)
|
|
|
|
# 创建服务器
|
|
server = ThreadedHTTPServer(('0.0.0.0', args.port), ProxyHandler)
|
|
server.socket = ssl_context.wrap_socket(server.socket, server_side=True)
|
|
|
|
print(f"HTTPS Proxy Server started on port {args.port}")
|
|
print(f"Forwarding to: {args.target}")
|
|
print(f"Press Ctrl+C to stop")
|
|
print()
|
|
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
print("\nShutting down...")
|
|
server.shutdown()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|