1
[project.scripts]
2
proxierinite = "proxierinite.cli:cli"

1
@cli.command()
2
@click.option("--port", default=8001, help="代理服务端口")
3
@click.option("--web-port", default=8002, help="Web 界面端口")
4
@click.option("--config", default=None, help="配置文件路径")
5
@click.option("--save", "save_path", default=None, help="保存请求数据到文件（jsonl）")
6
@click.option("--silent", "-s", is_flag=True, help="静默模式，不输出任何信息")
7
@click.option("--daemon", "-d", is_flag=True, help="后台模式启动")
8
def start(port: int, web_port: int, config: str, save_path: str | None, silent: bool, daemon: bool):
9
    """启动代理服务器"""
10
    ...

1
class ProxyServer:
2
    """代理服务器主类"""
3

4
    def __init__(
5
        self,
6
        config_path: str | None = None,
7
        save_path: str | None = None,
8
        silent: bool = False,
9
    ):
10
        self.config_path = config_path or default_config_path()
11
        self.save_path = save_path
12
        self.silent = silent
13
        self.rules_engine = RulesEngine(self.config_path, silent=self.silent)
14
        self.web_interface = WebInterface()
15
        self.addon = ProxyAddon(
16
            self.rules_engine,
17
            self.web_interface,
18
            save_path=save_path,
19
            silent=self.silent,
20
            config_path=self.config_path,
21
        )
22
        self.web_interface.register_clear_traffic_handler(
23
            self.addon.clear_traffic_state
24
        )

1
opts = options.Options(
2
    listen_host=host,
3
    listen_port=port,
4
    http2=False,
5
)
6

7
self.web_interface.start(web_port, host=web_host, silent=self.silent)
8

9
async def _run_master():
10
    self.master = DumpMaster(opts)
11
    self.master.addons.add(self.addon)
12
    await self.master.run()
13

14
asyncio.run(_run_master())

1
def requestheaders(self, flow: http.HTTPFlow) -> None:
2
    """
3
    在仅收到请求头时就创建最小记录。
4

5
    mitmproxy 的 request() 要等到整个请求体读完才会触发；
6
    对带 body 的普通 POST，如果中途被客户端取消/断开，request() 可能根本不会触发。
7
    """
8
    request_info = self._ensure_request_tracking_row(flow, log_capture_skip=True)
9
    if not request_info:
10
        return
11
    self._emit_pending_jsonl_once(request_info)
12
    self.web_interface.update_traffic_data(
13
        self.traffic_data, changed=request_info
14
    )

1
def request(self, flow: http.HTTPFlow) -> None:
2
    request_info = self._ensure_request_tracking_row(flow)
3
    if request_info is None:
4
        return
5

6
    request_body = self._safe_message_content(flow.request)
7
    content_type = (get_header_value(flow.request.headers, "content-type") or "").lower()
8
    transfer_encoding = (get_header_value(flow.request.headers, "transfer-encoding") or "").lower()
9
    content_length = get_header_value(flow.request.headers, "content-length") or ""
10

11
    is_multipart = content_type.startswith("multipart/form-data")
12
    is_binary = any(
13
        t in content_type
14
        for t in ["video/", "audio/", "image/", "application/octet-stream"]
15
    )
16

17
    request_info.update(
18
        {
19
            "headers": dict(flow.request.headers),
20
            "content": content_info,
21
            "content_size": len(request_body) if request_body else 0,
22
            "is_multipart": is_multipart,
23
            "is_binary": is_binary,
24
            "request_fully_received": True,
25
        }
26
    )
27

28
    modified_request, applied_rule_names = (
29
        self.rules_engine.apply_request_rules(flow.request)
30
    )

1
sc_resp = getattr(flow.request, "short_circuit_response", None)
2
if sc_resp is not None:
3
    flow.response = sc_resp
4
    flow.response.headers["X-Short-Circuit"] = "true"
5
    request_info.update(
6
        {
7
            "status": "completed",
8
            "response_status": flow.response.status_code,
9
            "response_headers": dict(flow.response.headers),
10
            "response_content": flow.response.get_text(strict=False),
11
            "response_time": 0,
12
        }
13
    )

1
def response(self, flow: http.HTTPFlow) -> None:
2
    if not flow.metadata.get("proxierinite_capture"):
3
        return
4

5
    end_time = time.time()
6
    request_info = self._find_traffic_row_for_flow(flow)
7

8
    orig_resp_headers = dict(flow.response.headers)
9
    _orig = self._analyze_response_content(
10
        orig_resp_headers, flow.response.content
11
    )
12

13
    setattr(flow.response, "request", flow.request)
14
    modified_response = self.rules_engine.apply_response_rules(flow.response)
15
    if modified_response:
16
        flow.response = modified_response
17
        request_info["response_modified"] = True
18
        request_info["response_original_headers"] = orig_resp_headers
19
        request_info["response_original_content"] = _orig["preview"]

1
request_info.update(
2
    {
3
        "status": "completed",
4
        "response_status": flow.response.status_code,
5
        "response_headers": dict(flow.response.headers),
6
        "response_content": response_content_info,
7
        "response_content_size": (
8
            len(flow.response.content) if flow.response.content else 0
9
        ),
10
        "response_time": (end_time - flow.request.timestamp_start),
11
        "response_timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
12
    }
13
)
14

15
self.web_interface.update_traffic_data(
16
    self.traffic_data, changed=request_info
17
)
18
self._maybe_persist(request_info)

1
def http_connect(self, flow: http.HTTPFlow) -> None:
2
    """HTTP-only 模式：拒绝隧道请求。"""
3
    flow.response = http.Response.make(
4
        405,
5
        b"Tunnel requests are not supported by this HTTP-only proxy.\n",
6
        {"Content-Type": "text/plain; charset=utf-8"},
7
    )

1
def load_rules(self) -> None:
2
    config_path = Path(self.config_path)
3
    if not config_path.exists():
4
        self.create_default_config()
5

6
    with open(config_path, encoding="utf-8") as f:
7
        config = yaml.safe_load(f)
8

9
    if "extends" in config:
10
        config = self._resolve_extends(config, config_path.parent)
11

12
    self.rules = []
13
    config_dir = str(config_path.parent)
14
    for idx, rule_config in enumerate(config.get("rules", [])):
15
        self._validate_rule_config(rule_config, idx)
16
        rule = Rule(rule_config, config_dir)
17
        self.rules.append(rule)
18

19
    self.rules.sort(key=lambda x: x.priority, reverse=True)

1
def apply_request_rules(
2
    self, request: http.Request
3
) -> tuple[http.Request | None, list[str]]:
4
    candidates: list[Rule] = []
5
    host_l = request.pretty_host.lower() if hasattr(request, "pretty_host") else ""
6
    path = request.path if hasattr(request, "path") else ""
7

8
    if host_l in self._host_index:
9
        candidates.extend(self._host_index[host_l])
10

11
    for prefix, rules in self._path_index.items():
12
        if path.startswith(prefix):
13
            candidates.extend(rules)
14

15
    candidates.extend(self._generic_rules)
16

17
    for rule in ordered:
18
        if rule.match(request):
19
            modified_request, actions = rule.apply_request_actions(request)
20
            if modified_request is not None:
21
                result_request = modified_request
22
                applied_rule_names.append(rule.name)
23
                if rule.stop_after_match:
24
                    break

1
class ActionProcessor(ABC):
2
    @property
3
    @abstractmethod
4
    def action_name(self) -> str:
5
        pass
6

7
    @abstractmethod
8
    def process_request(self, request: http.Request, params: dict[str, Any]) -> bool:
9
        pass
10

11
    @abstractmethod
12
    def process_response(self, response: http.Response, params: dict[str, Any]) -> bool:
13
        pass

1
class SetHeaderProcessor(ActionProcessor):
2
    @property
3
    def action_name(self) -> str:
4
        return "set_header"
5

6
    def process_request(self, request: http.Request, params: dict[str, Any]) -> bool:
7
        modified = False
8
        for k, v in (params or {}).items():
9
            request.headers[k] = v
10
            modified = True
11
        return modified
12

13
    def process_response(self, response: http.Response, params: dict[str, Any]) -> bool:
14
        modified = False
15
        for k, v in (params or {}).items():
16
            response.headers[k] = v
17
            modified = True
18
        return modified

1
def process_request_action(
2
    self, action: str, request: http.Request, params: dict[str, Any]
3
) -> bool:
4
    processor = self.get_processor(action)
5
    if processor:
6
        return processor.process_request(request, params)
7
    return False
8

9
def process_response_action(
10
    self, action: str, response: http.Response, params: dict[str, Any]
11
) -> bool:
12
    processor = self.get_processor(action)
13
    if processor:
14
        return processor.process_response(response, params)
15
    return False

1
class ReplaceBodyJsonProcessor(ActionProcessor):
2
    def process_response(self, response: http.Response, params: dict[str, Any]) -> bool:
3
        if not response.content:
4
            response.headers["X-ReplaceBodyJson-Error"] = "No content"
5
            return False
6

7
        processed_params = process_template_dict(params or {})
8
        content_str = response.content.decode("utf-8", errors="ignore") or "null"
9
        obj = json.loads(content_str)
10

11
        if "path" in processed_params and "value" in processed_params:
12
            _set_deep(obj, processed_params["path"], processed_params["value"])
13
        elif "values" in processed_params:
14
            for path, value in processed_params["values"].items():
15
                _set_deep(obj, path, value)
16

17
        response.content = json.dumps(obj, ensure_ascii=False).encode("utf-8")
18
        return True

1
self._traffic_queue: Queue[TrafficQueueItem] = Queue(maxsize=0)
2
self._traffic_worker = threading.Thread(
3
    target=self._traffic_worker_loop,
4
    daemon=True,
5
    name="proxierinite-traffic-ui",
6
)
7
self._traffic_worker.start()

1
def update_traffic_data(
2
    self,
3
    traffic_data: list[dict[str, Any]],
4
    changed: dict[str, Any] | None = None,
5
):
6
    """将流量更新投递到后台线程，避免阻塞代理主路径。"""
7
    with self._traffic_lock:
8
        gen = self._traffic_clear_gen
9
    if changed is not None:
10
        self._traffic_queue.put(("upsert", copy.deepcopy(changed), gen))
11
    else:
12
        rows = [dict(r) for r in (traffic_data or [])]
13
        self._traffic_queue.put(("replace_all", rows, gen))

1
def _traffic_worker_loop(self) -> None:
2
    while True:
3
        item = self._traffic_queue.get()
4
        if item is None:
5
            break
6
        op, payload, gen = item
7
        with self._traffic_lock:
8
            if gen != self._traffic_clear_gen:
9
                continue
10
            if op == "upsert":
11
                row_for_sse = self._worker_apply_upsert_locked(payload)
12
            elif op == "replace_all":
13
                row_for_sse = self._worker_apply_replace_all_locked(payload)
14
        self._publish_traffic_sse_after_mutation(op, row_for_sse)
15
        self._publish_stats_sse()

1
@self.app.route("/api/traffic")
2
def get_traffic():
3
    limit = request.args.get("limit", 100, type=int)
4
    with self._traffic_lock:
5
        if limit > 0:
6
            limit = min(max(limit, 1), 500)
7
            filtered_data = list(self.traffic_data[-limit:])
8
        else:
9
            filtered_data = list(self.traffic_data)
10
        total = len(self.traffic_data)
11
    return jsonify(
12
        {
13
            "data": [record_for_json_output(dict(r)) for r in filtered_data],
14
            "total": total,
15
            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
16
        }
17
    )

1
@self.app.route("/api/export")
2
def export_traffic():
3
    fmt = (request.args.get("format") or "json").lower()
4
    limit = request.args.get("limit", type=int)
5
    ...
6
    if fmt == "jsonl":
7
        ...
8
    elif fmt == "csv":
9
        ...
10
    else:
11
        out_rows = [record_for_json_output(r) for r in data]
12
        return Response(
13
            json.dumps(out_rows, ensure_ascii=False),
14
            mimetype="application/json",
15
            headers={"Content-Disposition": 'attachment; filename="traffic.json"'},
16
        )

1
def validate_config(
2
    self, config: dict[str, Any], config_path: str | None = None
3
) -> dict[str, Any]:
4
    self.validation_errors = []
5
    self.validation_warnings = []
6

7
    result = {"valid": True, "errors": [], "warnings": [], "suggestions": []}
8

9
    self._validate_basic_structure(config)
10

11
    if "capture" in config:
12
        self._validate_capture_config(config["capture"])
13

14
    if "rules" in config:
15
        self._validate_rules_config(config["rules"], config_path)
16

17
    if "extends" in config:
18
        self._validate_extends_config(config["extends"], config_path)
19

20
    result["errors"] = self.validation_errors
21
    result["warnings"] = self.validation_warnings
22
    result["valid"] = len(self.validation_errors) == 0
23

24
    return result

1
Usage: python -m proxierinite [OPTIONS] COMMAND [ARGS]...
2

3
  代理服务器命令行工具
4

5
Options:
6
  -v, --verbose  详细输出
7
  --version      Show the version and exit.
8
  --help         Show this message and exit.
9

10
Commands:
11
  examples  管理规则示例
12
  info      显示版本信息
13
  init      初始化代理服务器配置
14
  start     启动代理服务器
15
  status    查看服务器状态
16
  stop      停止后台运行的服务器
17
  validate  验证和分析配置文件

1
配置验证:
2
----------------------------------------
3
❌ 配置无效
4
错误 (1 个):
5
  ❌ 缺少必需的配置字段: capture
6

7
规则分析:
8
----------------------------------------
9
总规则数: 2
10
启用规则数: 2
11

12
动作类型统计:
13
  set_header: 2 次

1
配置验证:
2
----------------------------------------
3
❌ 配置无效
4
错误 (3 个):
5
  ❌ 规则 8.response_pipeline[0].params.file 引用的文件不存在
6
  ❌ 规则 11.response_pipeline[0] 缺少 params 字段
7
  ❌ 规则 13.response_pipeline[0] 缺少 params 字段
8

9
规则分析:
10
----------------------------------------
11
总规则数: 16
12
启用规则数: 9
13

14
动作类型统计:
15
  set_header: 5 次
16
  remove_header: 1 次
17
  rewrite_url: 1 次
18
  set_query_param: 1 次
19
  set_body_param: 1 次
20
  replace_body_json: 1 次
21
  mock_response: 3 次
22
  set_status: 1 次
23
  delay: 1 次
24
  conditional: 2 次
25
  short_circuit: 1 次

1
capture:
2
  include:
3
    hosts:
4
      - "^.*$"
5
  enable_streaming: false
6
  enable_large_files: false
7
  large_file_threshold: 1048576
8
  save_binary_content: false
9

10
rules:
11
  - name: "测试 - 设置响应头"
12
    enabled: true
13
    priority: 10
14
    match:
15
      host: ".*"
16
    response_pipeline:
17
      - action: set_header
18
        params:
19
          X-Proxierinite-Test: "ok"