proxypin

mirror of https://github.com/wanghongenpin/proxypin.git synced 2026-01-25 18:32:32 +08:00

Table of Contents

上报服务使用说明
概述
请求 & 响应字段详解（示例）

请求字段示例（request）
响应字段示例（response）

Python (Flask) 服务端示例
curl 测试示例（单条 entry）

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

上报服务使用说明

概述

本说明描述如何接受并处理来自 ProxyPin 的单条请求上报。上报通常以 JSON 格式 POST 到你的 HTTP 服务。上报格式使用单个 HAR Entry 方式。即每次上报仅包含一个 HAR entry 对象，结构类似于：

{
  "startedDateTime": "2025-10-25T12:34:56.789Z",
  "time": 123,
  "request": { /* 请求部分 */ },
  "response": { /* 响应部分 */ },
  "timings": { /* timing 信息 */ }
}

服务端只需接收该 entry（可以选择直接保存 entry 或将其包装为标准 HAR 的 log 对象再保存）。下面示例中服务器会将收到的 entry 包装为带单条 entry 的 HAR log 并保存成 JSON 文件，方便用 HAR 工具（DevTools / Charles / mitmproxy）打开或导入。

说明：客户端会使用 Content-Type: application/json格式请求；如果客户端启用了压缩（gzip），请求会有 Content-Encoding: gzip，服务端需先解压再解析 JSON。 ProxyPin除了会向服务器发送请求和响应数据，还会在请求头中携带下面这些数据：

X-Report-Name：命中的上报的规则名称

请求 & 响应字段详解（示例）

下面给出更完整的 request 与 response 字段示例（放在单条 entry 内），以及每个字段的说明。你可以直接把这些结构当作后端解析与保存的参考。

请求字段示例（request）

"request": {
  "method": "POST", 
  "url": "https://api.example.com/v1/upload?user=1", //完整请求 URL（包含 scheme、主机、路径与查询字符串）。
  "httpVersion": "HTTP/1.1",
  "cookies": [],
  "headers": [  //一个数组，每项包含 name/value（HAR 标准）
    { "name": "Host", "value": "api.example.com" },
    { "name": "Content-Type", "value": "application/json; charset=utf-8" },
    { "name": "Authorization", "value": "Bearer abcdef" }
  ],
  "queryString": [ //查询参数数组（name/value）。
    { "name": "user", "value": "1" }
  ],
  "headersSize": -1,
  "bodySize": 1234, // 原始 body 大小（字节），若未知可用 -1。
  "postData": { //若有请求体，包含 mimeType 与 text，
    "mimeType": "application/json",
    "text": "{\"name\":\"Alice\",\"age\":30}",
    "params": []
  }
}

响应字段示例（response）

"response": {
  "status": 200, 
  "statusText": "OK",
  "httpVersion": "HTTP/1.1",
  "cookies": [],
  "headers": [
    { "name": "Content-Type", "value": "application/json; charset=utf-8" },
    { "name": "Content-Encoding", "value": "gzip" }
  ],
  "content": {
    "size": 2048, //响应正文的原始大小（字节）。
    "mimeType": "application/json",
    "text": "{\"result\":\"ok\",\"id\":123}",
  },
  "redirectURL": "",
  "headersSize": -1,
  "bodySize": 2048
}

Python (Flask) 服务端示例

该最小示例：解压（如需）、解析 JSON、验证为单个 HAR entry，然后把它包装为一个标准 HAR log 并保存到 received_har/。

环境准备：

python3 -m venv venv
source venv/bin/activate
pip install flask

保存为 report_server.py：

# report_server.py
import os
import gzip
import json
from datetime import datetime
from flask import Flask, request, jsonify

app = Flask(__name__)
OUT_DIR = 'received_har'
os.makedirs(OUT_DIR, exist_ok=True)


def try_decompress(body: bytes, encoding: str):
    if not encoding:
        return body
    enc = encoding.lower()
    if 'gzip' in enc:
        try:
            return gzip.decompress(body)
        except Exception as e:
            raise RuntimeError(f'gzip decompress error: {e}')
    return body


@app.route('/report', methods=['POST'])
def report():
    try:
        raw = request.get_data()
        content_encoding = request.headers.get('Content-Encoding', '') or ''

        # 解压（若有）
        try:
            body_bytes = try_decompress(raw, content_encoding)
        except Exception as e:
            return jsonify({'ok': False, 'error': f'decompress failed: {e}'}), 400

        # 解析 JSON
        try:
            payload = json.loads(body_bytes.decode('utf-8', errors='replace'))
        except Exception as e:
            return jsonify({'ok': False, 'error': f'json decode failed: {e}'}), 400

        # 验证为单个 HAR entry（必须包含 request 或 response）
        if not isinstance(payload, dict) or ('request' not in payload and 'response' not in payload):
            return jsonify({'ok': False, 'error': 'expected a single HAR entry with request or response field'}), 400

        # 将 entry 包装为标准 HAR log，便于后续导出或兼容工具
        har = {
            'log': {
                'version': '1.2',
                'creator': {'name': 'ProxyPin-report-server', 'version': '1.0'},
                'entries': [payload]
            }
        }

        # persist to file
        ts = datetime.utcnow().strftime('%Y%m%dT%H%M%S%f')[:-3]
        filename = os.path.join(OUT_DIR, f'har_{ts}.har')
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(har, f, ensure_ascii=False, indent=2)

        return jsonify({'ok': True, 'saved': filename}), 200

    except Exception as e:
        return jsonify({'ok': False, 'error': str(e)}), 500


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

启动服务：

python report_server.py

curl 测试示例（单条 entry）

假设你有 entry.json 文件，内容为单个 HAR entry（如上所示）：

未压缩上传：

curl -X POST http://localhost:8080/report \
  -H "Content-Type: application/json" \
  --data-binary @entry.json

GZIP 压缩上传：

gzip -c entry.json > entry.json.gz

curl -X POST http://localhost:8080/report \
  -H "Content-Type: application/json" \
  -H "Content-Encoding: gzip" \
  --data-binary @entry.json.gz

快速 inline 测试（最小 entry）：

echo '{"startedDateTime":"2025-10-25T12:34:56.789Z","time":1,"request":{},"response":{},"timings":{}}' \
  | curl -X POST http://localhost:8080/report -H "Content-Type: application/json" -d @-