硕士生
 
- 金币
- 3032
- 好评
- 14
- 信誉
- 101
|
本帖最后由 则铭 于 2026-8-24 19:23 编辑
配合cloudflare内网穿透，或者其他有ssl的穿透服务使用。服务器应该也能用吧
嗯，新账号有一天会员。弄不了
- #!/usr/bin/env python3
- # -*- coding: utf-8 -*-
- """
- 反代 + 篡改脚本（生产可用版）
- 总并发能力：4 × 1000 = 4000 个连接
- 部署方式
- 1. 安装依赖
- pip install gunicorn gevent flask requests
- 2. 启动（推荐 gevent 异步模型）
- gunicorn -w 4 -k gevent \
- --worker-connections 1000 \
- --bind 127.0.0.1:8080 \
- --timeout 120 \
- --keep-alive 30 \
- --max-requests 10000 --max-requests-jitter 1000 \
- --access-logfile - --error-logfile - \
- app:app
- 参数解释：
- -w 4 4 个 worker 进程
- -k gevent 用 gevent 协程，单 worker 内可挂 1000 并发
- --worker-connections 1000 每 worker 1000 并发连接
- --timeout 120 慢请求/大文件上传下载给够时间
- --keep-alive 30 长连接保活 30 秒，降低 TLS 握手开销
- --max-requests 10000 防 worker 内存泄漏，每处理 1 万请求重启 worker
- Windows 用户请使用下面的命令启动：
- # 安装Waitress（Windows 友好的生产级 WSGI 服务器）
- pip install waitress flask requests
- # 启动
- waitress-serve --host 127.0.0.1 --port 8080 --threads 50 "流光助手替换服务:app"
- 流光助手替换服务:app(流光助手替换服务.py) 你的脚本文件名，比如 app.py 就是 "app:app"
- 流光助手需要把libapp.so文件里的 https://pan-api.yodlx.com (只有三处)改为 你的cloudflared内网穿透的url,长度必须一致，否则会出错。
- 比如我的是 https://pan-apiy.luac.top (68 74 74 70 73 3A 2F 2F 70 61 6E 2D 61 70 69 79 2E 6C 75 61 63 2E 74 6F 70)
- 运行脚本，浏览器测试访问 https://yourUrl/api/v1/auth/me
- 如 https://pan-apiy.luac.top/api/v1/auth/me
- 返回{"error":{"code":"missing_token","message":"需要 Bearer 登录凭证"}} 说明你的脚本就能用了
- cloudflared tunnel 仍指向 http://127.0.0.1:8080 即可。
- """
- from flask import Flask, request, Response, stream_with_context
- import requests
- from requests.adapters import HTTPAdapter
- from urllib3.util.retry import Retry
- import json
- import logging
- import threading
- # ---------- 日志 ----------
- logging.basicConfig(
- level=logging.INFO,
- format='%(asctime)s [%(process)d] %(levelname)s - %(message)s'
- )
- logger = logging.getLogger("proxy")
- app = Flask(__name__)
- # ---------- 配置 ----------
- REAL_SERVER = "https://pan-api.yodlx.com"
- TAMPER_PATHS = {"/api/v1/auth/me", "/api/v1/auth/login"}
- FAKE_EXPIRES = "2099-12-31 23:59:59"
- # 必须由 Flask/上游重新生成的响应头
- REMOVE_HEADERS = {
- 'transfer-encoding', 'content-encoding', 'content-length',
- 'connection', 'keep-alive', 'proxy-connection', 'upgrade',
- 'content-type', # 响应 Content-Type 由 Flask 根据 content_type 参数设置
- }
- # 转发请求时需要移除的头（避免与 requests 自动生成冲突）
- FORWARD_EXCLUDE_HEADERS = {
- 'host',
- 'content-length',
- 'transfer-encoding',
- 'connection',
- }
- # 单进程内连接池配置（gunicorn 多 worker 时每个 worker 各自有池）
- POOL_CONNECTIONS = 50 # 不同 host 的连接缓存
- POOL_MAXSIZE = 200 # 同一 host 的最大并发连接
- CONNECT_TIMEOUT = 5 # TCP/TLS 建立超时
- READ_TIMEOUT = 120 # 上游响应读超时（大文件也要传完）
- CHUNK_SIZE = 64 * 1024
- # ---------- 全局 Session（线程安全，连接复用） ----------
- _session = None
- _session_lock = threading.Lock()
- def get_session() -> requests.Session:
- global _session
- if _session is None:
- with _session_lock:
- if _session is None:
- s = requests.Session()
- retry = Retry(
- total=2,
- backoff_factor=0.2,
- status_forcelist=(502, 503, 504),
- allowed_methods=("GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"),
- raise_on_status=False,
- )
- adapter = HTTPAdapter(
- pool_connections=POOL_CONNECTIONS,
- pool_maxsize=POOL_MAXSIZE,
- max_retries=retry,
- pool_block=False,
- )
- s.mount('http://', adapter)
- s.mount('https://', adapter)
- _session = s
- logger.info("HTTP session/pool initialized")
- return _session
- # ---------- 工具函数 ----------
- def _filter_headers(raw_headers):
- """过滤掉会引起冲突的响应头，保留其他原始头（含大小写）。"""
- return [(k, v) for k, v in raw_headers.items() if k.lower() not in REMOVE_HEADERS]
- def _build_response(content, status_code, headers_list, content_type):
- """构造响应，正确处理重复的 Set-Cookie 等头。"""
- resp = Response(content, status=status_code, content_type=content_type)
- for k, v in headers_list:
- if k.lower() == 'set-cookie':
- resp.headers.add(k, v) # 允许多个 Set-Cookie
- else:
- resp.headers[k] = v
- return resp
- def _tamper_body(content: bytes, current_path: str) -> bytes:
- """篡改响应体；失败则原样返回。"""
- try:
- j = json.loads(content)
- except json.JSONDecodeError as e:
- logger.warning(f"[{current_path}] json decode failed: {e}")
- return content
- data = j.get('data') if isinstance(j, dict) else None
- if isinstance(data, dict) and 'user' in data:
- data['user']['is_pro'] = 1
- data['user']['pro_expires_at'] = FAKE_EXPIRES
- logger.info(f"✅ Tampered {current_path} -> is_pro=1")
- else:
- logger.warning(f"[{current_path}] unexpected json structure")
- return json.dumps(j, ensure_ascii=False).encode('utf-8')
- # ---------- 路由 ----------
- @app.route('/', defaults={'path': ''},
- methods=['GET', 'POST', 'PUT', 'DELETE', 'PATCH', 'HEAD', 'OPTIONS'])
- @app.route('/<path:path>',
- methods=['GET', 'POST', 'PUT', 'DELETE', 'PATCH', 'HEAD', 'OPTIONS'])
- def proxy(path):
- full_url = REAL_SERVER + request.full_path
- logger.info(f"-> {request.method} {request.path}")
- # 请求头：去掉 host 和会冲突的长度/连接头
- headers = {
- k: v for k, v in request.headers.items()
- if k.lower() not in FORWARD_EXCLUDE_HEADERS
- }
- # 请求体：完整读取，避免流式 body 与 Content-Length 冲突
- data = request.get_data() if request.method in ('POST', 'PUT', 'PATCH') else None
- current_path = request.path
- needs_tamper = current_path in TAMPER_PATHS
- try:
- resp = get_session().request(
- method=request.method,
- url=full_url,
- headers=headers,
- data=data, # 完整 bytes 请求体
- allow_redirects=False,
- verify=True,
- timeout=(CONNECT_TIMEOUT, READ_TIMEOUT),
- stream=True, # 流式下载
- )
- except requests.exceptions.RequestException as e:
- logger.error(f"forward failed [{current_path}]: {e}")
- return Response(f"Proxy error: {e}", status=502)
- # 清理响应头
- response_headers = _filter_headers(resp.headers)
- status_code = resp.status_code
- content_type = resp.headers.get('content-type', '')
- # ---------- 篡改路径：必须读全文 ----------
- if needs_tamper and status_code == 200 and 'application/json' in content_type:
- try:
- content = resp.content
- except Exception as e:
- logger.error(f"read tamper body failed: {e}")
- resp.close()
- return Response("Proxy error: read body", status=502)
- finally:
- resp.close()
- content = _tamper_body(content, current_path)
- # 因为可能改了长度，去掉原始 Content-Length，让 Flask 自动算
- response_headers = [(k, v) for k, v in response_headers
- if k.lower() != 'content-length']
- return _build_response(
- content,
- status_code,
- response_headers,
- content_type
- )
- # ---------- 非篡改路径：纯流式转发 ----------
- def generate():
- try:
- for chunk in resp.iter_content(chunk_size=CHUNK_SIZE):
- if chunk:
- yield chunk
- except Exception as e:
- logger.error(f"stream error [{current_path}]: {e}")
- finally:
- resp.close()
- # HEAD 请求没有 body，Flask 会自动处理
- return _build_response(
- stream_with_context(generate()),
- status_code,
- response_headers,
- content_type or None
- )
- # ---------- 健康检查（cloudflared / 负载均衡用） ----------
- @app.route('/__health', methods=['GET'])
- def health():
- return Response('ok', status=200, content_type='text/plain')
- if __name__ == '__main__':
- # 仅本地调试；生产请用 gunicorn
- app.run(host='127.0.0.1', port=8080, threaded=True)
|
本帖子中包含更多资源
您需要 登录 才可以下载或查看，没有账号？立即注册
x
|
苏公网安备 32010602011309号
|
苏ICP备19064627号
