硕士生
 
- 金币
- 3027
- 好评
- 14
- 信誉
- 101
|
发表于
前天 13:36
来自手机
|
显示全部楼层
来自 重庆
感谢大佬的教程 
- #!/usr/bin/env python3
- # -*- coding: utf-8 -*-
- from flask import Flask, request, Response
- import requests
- import json
- import logging
- logging.basicConfig(
- level=logging.INFO,
- format='%(asctime)s - %(levelname)s - %(message)s'
- )
- logger = logging.getLogger(__name__)
- 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', # 让 Flask 自动计算
- 'connection',
- 'keep-alive',
- 'proxy-connection',
- 'upgrade',
- }
- @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"Proxying {request.method} {full_url}")
- # 准备请求头
- headers = {k: v for k, v in request.headers.items() if k.lower() != 'host'}
-
- # 获取请求体
- data = request.get_data()
- try:
- resp = requests.request(
- method=request.method,
- url=full_url,
- headers=headers,
- data=data,
- params=request.args,
- allow_redirects=False,
- verify=True,
- timeout=30,
- stream=True # 流式获取,避免内存问题
- )
- except requests.exceptions.RequestException as e:
- logger.error(f"Forward failed: {e}")
- return Response(f"Proxy error: {e}", status=502)
- # 读取完整响应体(但注意大文件可能消耗内存)
- content = resp.content
- status_code = resp.status_code
- # 清理响应头,移除可能导致冲突的字段
- response_headers = {}
- for key, value in resp.headers.items():
- key_lower = key.lower()
- if key_lower not in REMOVE_HEADERS:
- response_headers[key] = value
- # 判断是否需要篡改
- current_path = request.path
- if current_path in TAMPER_PATHS and status_code == 200:
- content_type = resp.headers.get('content-type', '')
- if 'application/json' in content_type and content:
- try:
- j = json.loads(content)
- if 'data' in j and isinstance(j['data'], dict) and 'user' in j['data']:
- j['data']['user']['is_pro'] = 1
- j['data']['user']['pro_expires_at'] = FAKE_EXPIRES
- content = json.dumps(j, ensure_ascii=False).encode('utf-8')
- logger.info(f"✅ Tampered {current_path} -> is_pro=1")
- else:
- logger.warning(f"Unexpected JSON structure for {current_path}")
- except json.JSONDecodeError as e:
- logger.warning(f"JSON decode failed: {e}")
- # 构造响应(Flask 会自动处理 Content-Length 和 Transfer-Encoding)
- return Response(
- content,
- status=status_code,
- headers=response_headers,
- content_type=resp.headers.get('content-type') # 保持原 Content-Type
- )
- if __name__ == '__main__':
- app.run(host='127.0.0.1', port=8080, threaded=True)
复制代码
直接白嫖的cloudflare内网穿透。 |
本帖子中包含更多资源
您需要 登录 才可以下载或查看,没有账号?立即注册
x
|