Merge branch 'fork/TizenryA/feat/warp-flaresolverr-runtime'

.env.example

@@ -46,3 +46,37 @@ STORAGE_BACKEND=json
 
 # Git 仓库中的文件路径（可选，默认: accounts.json）
 # GIT_FILE_PATH=accounts.json
+
+# ============================================
+# WARP / FlareSolverr 稳定代理配置（可选）
+# ============================================
+
+# docker-compose.warp.yml 暴露到宿主机的端口
+# WARP_SOCKS_PORT=40000
+# PRIVOXY_PORT=40080
+# FLARESOLVERR_PORT=8191
+
+# FlareSolverr 日志与时区
+# FLARESOLVERR_LOG_LEVEL=info
+# TZ=Asia/Shanghai
+
+# Cloudflare WARP+ License Key（可选）
+# WARP_LICENSE_KEY=
+
+# 初始化 proxy_runtime 默认值
+# CHATGPT2API_PROXY_RUNTIME_ENABLED=true
+# CHATGPT2API_PROXY_RUNTIME_EGRESS_MODE=single_proxy
+# CHATGPT2API_PROXY_RUNTIME_PROXY_URL=http://privoxy:8118
+# CHATGPT2API_PROXY_RUNTIME_RESOURCE_PROXY_URL=
+# CHATGPT2API_PROXY_RUNTIME_SKIP_SSL_VERIFY=false
+# CHATGPT2API_PROXY_RUNTIME_RESET_STATUS_CODES=403
+
+# Cloudflare clearance 刷新配置
+# CHATGPT2API_PROXY_RUNTIME_CLEARANCE_ENABLED=true
+# CHATGPT2API_PROXY_RUNTIME_CLEARANCE_MODE=flaresolverr
+# CHATGPT2API_FLARESOLVERR_URL=http://flaresolverr:8191
+# CHATGPT2API_PROXY_RUNTIME_CLEARANCE_TIMEOUT_SEC=60
+# CHATGPT2API_PROXY_RUNTIME_CLEARANCE_REFRESH_INTERVAL=3600
+# CHATGPT2API_PROXY_RUNTIME_WARM_UP_ON_START=false
+# CHATGPT2API_PROXY_RUNTIME_BROWSER=chrome
+# CHATGPT2API_PROXY_RUNTIME_USER_AGENT=

CHANGELOG.md

@@ -2,6 +2,14 @@
 
 ## Unreleased
 
+## 1.4.2 - 2026-06-11
+
++ [新增] 稳定代理运行时。
++ [新增] FlareSolverr 清障。
++ [新增] WARP Compose 部署。
++ [新增] 设置页代理面板。
++ [优化] 注册拦截后重试。
+
 ## 1.4.1 - 2026-06-03
 
 + [新增] 账号刷新改为异步模式，支持前端轮询刷新/重新登录进度。

README.md

@@ -15,6 +15,19 @@
 > - 使用本项目即视为你已充分理解并同意本免责声明全部内容；如因滥用、违规或违法使用造成任何后果，均由使用者自行承担。
 > - 本项目基于对 ChatGPT 官网相关能力的逆向研究实现，存在账号受限、临时封禁或永久封禁的风险。请勿使用你自己的重要账号、常用账号或高价值账号进行测试。
 
+## 本分支修改内容
+
+本分支在 1.4.2 基础上补充注册与生图链路稳定性能力，核心改动如下：
+
+- 新增默认关闭的稳定代理运行时 `proxy_runtime`。
+- 引入 WARP + Privoxy + FlareSolverr 组合部署。
+- 注册遇到 Cloudflare 拦截后刷新 clearance 并重试。
+- 增加 FlareSolverr clearance 缓存、测试接口与脱敏展示。
+- 后台设置页新增「稳定代理运行时」配置面板。
+- 新增 `docker-compose.warp.yml` 与幂等初始化脚本。
+- 代理优先级：账号代理 > 稳定代理运行时 > 显式代理 > 全局代理。
+- 补充代理运行时、注册重试和初始化脚本单元测试。
+
 ## 快速开始
 
 已发布镜像支持 `linux/amd64` 与 `linux/arm64`，在 x86 服务器和 Apple Silicon / ARM Linux 设备上都会自动拉取匹配架构的版本。
@@ -33,6 +46,27 @@ docker compose up -d
 - API 地址：`http://localhost:3000/v1`
 - 数据目录：`./data`
 
+### WARP / FlareSolverr 稳定代理部署
+
+如果注册或图片链路经常遇到 Cloudflare 拦截，可以启用附带的 WARP + Privoxy + FlareSolverr 方案：
+
+```bash
+cp .env.example .env
+docker compose -f docker-compose.warp.yml up -d --build
+```
+
+该 compose 会启动：
+
+- `warp-proxy`：提供 WARP SOCKS5 出口。
+- `privoxy`：把 WARP SOCKS5 转成 HTTP 代理。
+- `flaresolverr`：刷新 Cloudflare clearance。
+- `init-config`：幂等写入 `proxy_runtime` 默认配置。
+- `app`：启动 ChatGPT2API 主服务。
+
+默认只让上游 OpenAI / ChatGPT 请求走稳定代理，账号邮箱、CPA 等辅助链路不会被强制接管。账号自身配置的代理优先级最高，其次是稳定代理运行时，再其次是显式代理和旧版全局代理。
+
+可在 `.env` 中调整端口和代理运行时参数，也可在后台设置页的「稳定代理运行时」面板手动保存、测试代理和测试 clearance。
+
 ### 本地开发
 
 启动后端：
@@ -113,6 +147,7 @@ environment:
 - 定时检查限流账号并自动刷新
 - 支持密码重新登录恢复异常账号，刷新后可自动重登
 - 支持网页端配置全局 HTTP / HTTPS / SOCKS5 / SOCKS5H 代理
+- 支持 WARP / FlareSolverr 稳定代理运行时
 - 支持搜索、筛选、批量刷新、导出、手动编辑和清理账号
 - 支持四种导入方式：本地 CPA JSON 文件导入、远程 CPA 服务器导入、`sub2api` 服务器导入、`access_token` 导入
 - 支持在设置页配置 `sub2api` 服务器，筛选并批量导入其中的 OpenAI OAuth 账号

VERSION

@@ -1 +1 @@
-1.4.1
+1.4.2

api/system.py

@@ -24,7 +24,7 @@ from services.image_service import (
 from services.image_storage_service import ImageStorageError, image_storage_service
 from services.image_tags_service import delete_tag, get_all_tags, set_tags
 from services.log_service import log_service
-from services.proxy_service import test_proxy
+from services.proxy_service import proxy_settings, test_clearance, test_proxy
 
 
 class SettingsUpdateRequest(BaseModel):
@@ -35,6 +35,10 @@ class ProxyTestRequest(BaseModel):
     url: str = ""
 
 
+class ClearanceTestRequest(BaseModel):
+    target_url: str = "https://chatgpt.com"
+
+
 class ImageDeleteRequest(BaseModel):
     paths: list[str] = []
     start_date: str = ""
@@ -131,10 +135,32 @@ def create_router(app_version: str) -> APIRouter:
     @router.post("/api/proxy/test")
     async def test_proxy_endpoint(body: ProxyTestRequest, authorization: str | None = Header(default=None)):
         require_admin(authorization)
-        candidate = (body.url or "").strip() or config.get_proxy_settings()
-        if not candidate:
-            raise HTTPException(status_code=400, detail={"error": "proxy url is required"})
-        return {"result": await run_in_threadpool(test_proxy, candidate)}
+        return {"result": await run_in_threadpool(test_proxy, (body.url or "").strip())}
+
+    @router.get("/api/proxy/runtime")
+    async def get_proxy_runtime_endpoint(authorization: str | None = Header(default=None)):
+        require_admin(authorization)
+        return {
+            "runtime": config.get_public_proxy_runtime_settings(),
+            "status": proxy_settings.get_runtime_status(),
+        }
+
+    @router.post("/api/proxy/runtime")
+    async def save_proxy_runtime_endpoint(body: SettingsUpdateRequest, authorization: str | None = Header(default=None)):
+        require_admin(authorization)
+        try:
+            config.update({"proxy_runtime": body.model_dump(mode="python")})
+        except ValueError as exc:
+            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc
+        return {
+            "runtime": config.get_public_proxy_runtime_settings(),
+            "status": proxy_settings.get_runtime_status(),
+        }
+
+    @router.post("/api/proxy/clearance/test")
+    async def test_proxy_clearance_endpoint(body: ClearanceTestRequest, authorization: str | None = Header(default=None)):
+        require_admin(authorization)
+        return {"result": await run_in_threadpool(test_clearance, body.target_url)}
 
     @router.get("/api/storage/info")
     async def get_storage_info(authorization: str | None = Header(default=None)):
@@ -275,6 +301,7 @@ def create_router(app_version: str) -> APIRouter:
             "healthy": healthy,
             "version": app_version,
             "storage": {"backend": storage.get_backend_info(), "health": storage_health},
+            "proxy_runtime": proxy_settings.get_runtime_status(),
             "accounts": stats,
         }
         if format == "json":

config.json

@@ -12,6 +12,28 @@
     "warning"
   ],
   "proxy": "",
+  "proxy_runtime": {
+    "enabled": false,
+    "egress_mode": "direct",
+    "proxy_url": "",
+    "resource_proxy_url": "",
+    "skip_ssl_verify": false,
+    "reset_session_status_codes": [
+      403
+    ],
+    "clearance": {
+      "enabled": false,
+      "mode": "none",
+      "cf_cookies": "",
+      "cf_clearance": "",
+      "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/145.0.0.0 Safari/537.36",
+      "browser": "chrome",
+      "flaresolverr_url": "",
+      "timeout_sec": 60,
+      "refresh_interval": 3600,
+      "warm_up_on_start": false
+    }
+  },
   "base_url": "",
   "sensitive_words": [],
   "global_system_prompt": "",

docker-compose.warp.yml

@@ -0,0 +1,108 @@
+services:
+  init-config:
+    image: python:3.13-alpine
+    container_name: chatgpt2api-warp-init
+    working_dir: /app
+    volumes:
+      - ./config.json:/app/config.json
+      - ./scripts:/app/scripts:ro
+    environment:
+      CHATGPT2API_CONFIG_FILE: /app/config.json
+      CHATGPT2API_PROXY_RUNTIME_ENABLED: ${CHATGPT2API_PROXY_RUNTIME_ENABLED:-true}
+      CHATGPT2API_PROXY_RUNTIME_EGRESS_MODE: ${CHATGPT2API_PROXY_RUNTIME_EGRESS_MODE:-single_proxy}
+      CHATGPT2API_PROXY_RUNTIME_PROXY_URL: ${CHATGPT2API_PROXY_RUNTIME_PROXY_URL:-http://privoxy:8118}
+      CHATGPT2API_PROXY_RUNTIME_RESOURCE_PROXY_URL: ${CHATGPT2API_PROXY_RUNTIME_RESOURCE_PROXY_URL:-}
+      CHATGPT2API_PROXY_RUNTIME_SKIP_SSL_VERIFY: ${CHATGPT2API_PROXY_RUNTIME_SKIP_SSL_VERIFY:-false}
+      CHATGPT2API_PROXY_RUNTIME_RESET_STATUS_CODES: ${CHATGPT2API_PROXY_RUNTIME_RESET_STATUS_CODES:-403}
+      CHATGPT2API_PROXY_RUNTIME_CLEARANCE_ENABLED: ${CHATGPT2API_PROXY_RUNTIME_CLEARANCE_ENABLED:-true}
+      CHATGPT2API_PROXY_RUNTIME_CLEARANCE_MODE: ${CHATGPT2API_PROXY_RUNTIME_CLEARANCE_MODE:-flaresolverr}
+      CHATGPT2API_PROXY_RUNTIME_CLEARANCE_TIMEOUT_SEC: ${CHATGPT2API_PROXY_RUNTIME_CLEARANCE_TIMEOUT_SEC:-60}
+      CHATGPT2API_PROXY_RUNTIME_CLEARANCE_REFRESH_INTERVAL: ${CHATGPT2API_PROXY_RUNTIME_CLEARANCE_REFRESH_INTERVAL:-3600}
+      CHATGPT2API_PROXY_RUNTIME_WARM_UP_ON_START: ${CHATGPT2API_PROXY_RUNTIME_WARM_UP_ON_START:-false}
+      CHATGPT2API_PROXY_RUNTIME_BROWSER: ${CHATGPT2API_PROXY_RUNTIME_BROWSER:-chrome}
+      CHATGPT2API_PROXY_RUNTIME_USER_AGENT: ${CHATGPT2API_PROXY_RUNTIME_USER_AGENT:-}
+      CHATGPT2API_FLARESOLVERR_URL: ${CHATGPT2API_FLARESOLVERR_URL:-http://flaresolverr:8191}
+    command: ["python3", "/app/scripts/init_proxy_config.py"]
+    restart: "no"
+
+  warp-proxy:
+    image: caomingjun/warp:latest
+    container_name: chatgpt2api-warp-proxy
+    restart: unless-stopped
+    environment:
+      WARP_SLEEP: "2"
+      # 可选：填写 Cloudflare WARP+ License Key
+      # WARP_LICENSE_KEY: ${WARP_LICENSE_KEY:-}
+    cap_add:
+      - NET_ADMIN
+      - SYS_MODULE
+    sysctls:
+      net.ipv6.conf.all.disable_ipv6: 0
+      net.ipv4.conf.all.src_valid_mark: 1
+    ports:
+      - "127.0.0.1:${WARP_SOCKS_PORT:-40000}:1080"
+    healthcheck:
+      test: ["CMD-SHELL", "python3 -c \"import socket; s=socket.create_connection(('127.0.0.1', 1080), 5); s.close()\""]
+      interval: 30s
+      timeout: 5s
+      retries: 5
+      start_period: 30s
+
+  privoxy:
+    image: vimagick/privoxy:latest
+    container_name: chatgpt2api-privoxy
+    restart: unless-stopped
+    depends_on:
+      warp-proxy:
+        condition: service_started
+    volumes:
+      - ./scripts/privoxy-warp.conf:/etc/privoxy/config:ro
+    ports:
+      - "127.0.0.1:${PRIVOXY_PORT:-40080}:8118"
+    healthcheck:
+      test: ["CMD-SHELL", "nc -z 127.0.0.1 8118 || exit 1"]
+      interval: 30s
+      timeout: 5s
+      retries: 5
+      start_period: 10s
+
+  flaresolverr:
+    image: ghcr.io/flaresolverr/flaresolverr:latest
+    container_name: chatgpt2api-flaresolverr
+    restart: unless-stopped
+    environment:
+      LOG_LEVEL: ${FLARESOLVERR_LOG_LEVEL:-info}
+      TZ: ${TZ:-Asia/Shanghai}
+    ports:
+      - "127.0.0.1:${FLARESOLVERR_PORT:-8191}:8191"
+    healthcheck:
+      test: ["CMD-SHELL", "curl -fsS http://127.0.0.1:8191/ >/dev/null"]
+      interval: 30s
+      timeout: 10s
+      retries: 5
+      start_period: 30s
+
+  app:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    image: chatgpt2api:warp
+    container_name: chatgpt2api-warp
+    restart: unless-stopped
+    depends_on:
+      init-config:
+        condition: service_completed_successfully
+      privoxy:
+        condition: service_started
+      flaresolverr:
+        condition: service_started
+    ports:
+      - "${CHATGPT2API_PORT:-3000}:80"
+    volumes:
+      - ./data:/app/data
+      - ./config.json:/app/config.json
+    environment:
+      STORAGE_BACKEND: ${STORAGE_BACKEND:-json}
+      DATABASE_URL: ${DATABASE_URL:-}
+      CHATGPT2API_AUTH_KEY: ${CHATGPT2API_AUTH_KEY:-}
+      CHATGPT2API_BASE_URL: ${CHATGPT2API_BASE_URL:-}

scripts/init_proxy_config.py

@@ -0,0 +1,186 @@
+#!/usr/bin/env python3
+"""Idempotently merge WARP/FlareSolverr proxy_runtime defaults into config.json."""
+
+from __future__ import annotations
+
+import copy
+import json
+import os
+import re
+import sys
+from pathlib import Path
+from typing import Any
+
+DEFAULT_USER_AGENT = (
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+    "AppleWebKit/537.36 (KHTML, like Gecko) "
+    "Chrome/145.0.0.0 Safari/537.36"
+)
+
+DEFAULT_PROXY_RUNTIME: dict[str, Any] = {
+    "enabled": False,
+    "egress_mode": "direct",
+    "proxy_url": "",
+    "resource_proxy_url": "",
+    "skip_ssl_verify": False,
+    "reset_session_status_codes": [403],
+    "clearance": {
+        "enabled": False,
+        "mode": "none",
+        "cf_cookies": "",
+        "cf_clearance": "",
+        "user_agent": DEFAULT_USER_AGENT,
+        "browser": "chrome",
+        "flaresolverr_url": "",
+        "timeout_sec": 60,
+        "refresh_interval": 3600,
+        "warm_up_on_start": False,
+    },
+}
+
+
+def _env_bool(name: str, default: bool) -> bool:
+    value = os.getenv(name)
+    if value is None:
+        return default
+    lowered = value.strip().lower()
+    if lowered in {"1", "true", "yes", "on"}:
+        return True
+    if lowered in {"0", "false", "no", "off"}:
+        return False
+    return default
+
+
+def _env_int(name: str, default: int, minimum: int = 0) -> int:
+    try:
+        return max(minimum, int(os.getenv(name, str(default))))
+    except (TypeError, ValueError, OverflowError):
+        return default
+
+
+def _warp_runtime_defaults() -> dict[str, Any]:
+    clearance_enabled = _env_bool("CHATGPT2API_PROXY_RUNTIME_CLEARANCE_ENABLED", True)
+    clearance_mode = os.getenv("CHATGPT2API_PROXY_RUNTIME_CLEARANCE_MODE", "flaresolverr").strip().lower()
+    if clearance_mode not in {"none", "manual", "flaresolverr"}:
+        clearance_mode = "flaresolverr" if clearance_enabled else "none"
+    if not clearance_enabled:
+        clearance_mode = "none"
+
+    egress_mode = os.getenv("CHATGPT2API_PROXY_RUNTIME_EGRESS_MODE", "single_proxy").strip().lower()
+    if egress_mode not in {"direct", "single_proxy"}:
+        egress_mode = "single_proxy"
+
+    runtime = copy.deepcopy(DEFAULT_PROXY_RUNTIME)
+    runtime.update(
+        {
+            "enabled": _env_bool("CHATGPT2API_PROXY_RUNTIME_ENABLED", True),
+            "egress_mode": egress_mode,
+            "proxy_url": os.getenv("CHATGPT2API_PROXY_RUNTIME_PROXY_URL", "http://privoxy:8118").strip(),
+            "resource_proxy_url": os.getenv("CHATGPT2API_PROXY_RUNTIME_RESOURCE_PROXY_URL", "").strip(),
+            "skip_ssl_verify": _env_bool("CHATGPT2API_PROXY_RUNTIME_SKIP_SSL_VERIFY", False),
+            "reset_session_status_codes": [
+                int(part.strip())
+                for part in os.getenv("CHATGPT2API_PROXY_RUNTIME_RESET_STATUS_CODES", "403").split(",")
+                if part.strip().isdigit() and 100 <= int(part.strip()) <= 599
+            ] or [403],
+        }
+    )
+    runtime["clearance"].update(
+        {
+            "enabled": clearance_enabled,
+            "mode": clearance_mode,
+            "user_agent": os.getenv("CHATGPT2API_PROXY_RUNTIME_USER_AGENT", DEFAULT_USER_AGENT).strip() or DEFAULT_USER_AGENT,
+            "browser": os.getenv("CHATGPT2API_PROXY_RUNTIME_BROWSER", "chrome").strip() or "chrome",
+            "flaresolverr_url": os.getenv("CHATGPT2API_FLARESOLVERR_URL", "http://flaresolverr:8191").strip(),
+            "timeout_sec": _env_int("CHATGPT2API_PROXY_RUNTIME_CLEARANCE_TIMEOUT_SEC", 60, 1),
+            "refresh_interval": _env_int("CHATGPT2API_PROXY_RUNTIME_CLEARANCE_REFRESH_INTERVAL", 3600, 60),
+            "warm_up_on_start": _env_bool("CHATGPT2API_PROXY_RUNTIME_WARM_UP_ON_START", False),
+        }
+    )
+    return runtime
+
+
+def _deep_fill_missing(target: dict[str, Any], defaults: dict[str, Any]) -> bool:
+    changed = False
+    for key, value in defaults.items():
+        if key not in target:
+            target[key] = copy.deepcopy(value)
+            changed = True
+            continue
+        if isinstance(target[key], dict) and isinstance(value, dict):
+            changed = _deep_fill_missing(target[key], value) or changed
+    return changed
+
+
+def _looks_like_repository_default(runtime: Any) -> bool:
+    if not isinstance(runtime, dict):
+        return True
+    candidate = copy.deepcopy(runtime)
+    _deep_fill_missing(candidate, DEFAULT_PROXY_RUNTIME)
+    return candidate == DEFAULT_PROXY_RUNTIME
+
+
+def _mask_url(value: str) -> str:
+    return re.sub(r"(https?://)([^\s/@:]+):([^\s/@]+)@", r"\1[REDACTED]@", value or "", flags=re.I)
+
+
+def main() -> int:
+    config_path = Path(os.getenv("CHATGPT2API_CONFIG_FILE", "/app/config.json"))
+    if not config_path.exists():
+        print(f"Config file not found, creating {config_path}")
+        data: dict[str, Any] = {}
+    else:
+        try:
+            data = json.loads(config_path.read_text(encoding="utf-8"))
+        except json.JSONDecodeError as exc:
+            print(f"Invalid JSON in {config_path}: {exc}", file=sys.stderr)
+            return 1
+        if not isinstance(data, dict):
+            print(f"Config root must be an object: {config_path}", file=sys.stderr)
+            return 1
+
+    desired = _warp_runtime_defaults()
+    existing = data.get("proxy_runtime")
+    changed = False
+
+    if _looks_like_repository_default(existing) or _env_bool("CHATGPT2API_PROXY_RUNTIME_FORCE", False):
+        data["proxy_runtime"] = desired
+        changed = True
+        print("Created proxy_runtime defaults")
+    else:
+        runtime = existing if isinstance(existing, dict) else {}
+        changed = _deep_fill_missing(runtime, DEFAULT_PROXY_RUNTIME)
+        data["proxy_runtime"] = runtime
+        print("Proxy runtime already configured")
+
+    if changed:
+        config_path.parent.mkdir(parents=True, exist_ok=True)
+        payload = json.dumps(data, ensure_ascii=False, indent=2) + "\n"
+        tmp_path = config_path.with_suffix(config_path.suffix + ".tmp")
+        try:
+            tmp_path.write_text(payload, encoding="utf-8")
+            tmp_path.replace(config_path)
+        except OSError as exc:
+            # Docker bind-mounted single files can reject atomic rename with EBUSY.
+            # Fall back to in-place write so the init job works with both file and
+            # directory mounts.
+            if getattr(exc, "errno", None) != 16:
+                raise
+            config_path.write_text(payload, encoding="utf-8")
+            tmp_path.unlink(missing_ok=True)
+
+    runtime = data.get("proxy_runtime") if isinstance(data.get("proxy_runtime"), dict) else {}
+    clearance = runtime.get("clearance") if isinstance(runtime.get("clearance"), dict) else {}
+    print(
+        "Proxy runtime summary: "
+        f"enabled={bool(runtime.get('enabled'))}, "
+        f"egress_mode={runtime.get('egress_mode')}, "
+        f"proxy_url={_mask_url(str(runtime.get('proxy_url') or ''))}, "
+        f"clearance_mode={clearance.get('mode')}, "
+        f"flaresolverr_url={_mask_url(str(clearance.get('flaresolverr_url') or ''))}"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/privoxy-warp.conf

@@ -0,0 +1,13 @@
+listen-address  0.0.0.0:8118
+toggle  1
+enable-remote-toggle  0
+enable-remote-http-toggle  0
+enable-edit-actions  0
+enforce-blocks  0
+buffer-limit  4096
+keep-alive-timeout  300
+socket-timeout  300
+
+# Forward every HTTP/HTTPS request through the WARP SOCKS5 proxy.
+# The trailing dot is required by Privoxy syntax.
+forward-socks5t   /   warp-proxy:1080 .

services/config.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import copy
 from dataclasses import dataclass
 import json
 import os
@@ -48,6 +49,33 @@ DEFAULT_CHAT_COMPLETION_CACHE = {
     "drop_assistant_history": False,
 }
 
+DEFAULT_PROXY_RUNTIME_USER_AGENT = (
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+    "AppleWebKit/537.36 (KHTML, like Gecko) "
+    "Chrome/145.0.0.0 Safari/537.36"
+)
+
+DEFAULT_PROXY_RUNTIME = {
+    "enabled": False,
+    "egress_mode": "direct",
+    "proxy_url": "",
+    "resource_proxy_url": "",
+    "skip_ssl_verify": False,
+    "reset_session_status_codes": [403],
+    "clearance": {
+        "enabled": False,
+        "mode": "none",
+        "cf_cookies": "",
+        "cf_clearance": "",
+        "user_agent": DEFAULT_PROXY_RUNTIME_USER_AGENT,
+        "browser": "chrome",
+        "flaresolverr_url": "",
+        "timeout_sec": 60,
+        "refresh_interval": 3600,
+        "warm_up_on_start": False,
+    },
+}
+
 
 def _normalize_bool(value: object, default: bool = False) -> bool:
     if isinstance(value, str):
@@ -56,6 +84,7 @@ def _normalize_bool(value: object, default: bool = False) -> bool:
             return True
         if lowered in {"0", "false", "no", "off"}:
             return False
+        return default
     if value is None:
         return default
     return bool(value)
@@ -64,7 +93,7 @@ def _normalize_bool(value: object, default: bool = False) -> bool:
 def _normalize_positive_int(value: object, default: int, minimum: int = 0) -> int:
     try:
         normalized = int(value)
-    except (TypeError, ValueError):
+    except (OverflowError, TypeError, ValueError):
         normalized = default
     return max(minimum, normalized)
 
@@ -163,6 +192,84 @@ def _normalize_chat_completion_cache_settings(value: object) -> dict[str, object
     }
 
 
+def _normalize_status_codes(value: object) -> list[int]:
+    items = value if isinstance(value, list) else DEFAULT_PROXY_RUNTIME["reset_session_status_codes"]
+    normalized: list[int] = []
+    for item in items:
+        if isinstance(item, bool):
+            continue
+        try:
+            status = int(item)
+        except (OverflowError, TypeError, ValueError):
+            continue
+        if 100 <= status <= 599 and status not in normalized:
+            normalized.append(status)
+    if not normalized:
+        return list(DEFAULT_PROXY_RUNTIME["reset_session_status_codes"])
+    return normalized
+
+
+def _normalize_proxy_runtime_settings(value: object) -> dict[str, object]:
+    source = value if isinstance(value, dict) else {}
+    default_clearance = DEFAULT_PROXY_RUNTIME["clearance"]
+    clearance_source = source.get("clearance") if isinstance(source.get("clearance"), dict) else {}
+
+    egress_mode = str(source.get("egress_mode") or DEFAULT_PROXY_RUNTIME["egress_mode"]).strip().lower()
+    if egress_mode not in {"direct", "single_proxy"}:
+        egress_mode = str(DEFAULT_PROXY_RUNTIME["egress_mode"])
+
+    clearance_mode = str(clearance_source.get("mode") or default_clearance["mode"]).strip().lower()
+    if clearance_mode not in {"none", "manual", "flaresolverr"}:
+        clearance_mode = str(default_clearance["mode"])
+
+    user_agent = str(clearance_source.get("user_agent") or default_clearance["user_agent"]).strip()
+    browser = str(clearance_source.get("browser") or default_clearance["browser"]).strip()
+
+    existing_clearance_cookies = str(source.get("_existing_cf_cookies") or "").strip()
+    existing_cf_clearance = str(source.get("_existing_cf_clearance") or "").strip()
+    cf_cookies = str(clearance_source.get("cf_cookies") or "").strip()
+    cf_clearance = str(clearance_source.get("cf_clearance") or "").strip()
+    if not cf_cookies and _normalize_bool(clearance_source.get("has_cf_cookies"), False):
+        cf_cookies = existing_clearance_cookies
+    if not cf_clearance and _normalize_bool(clearance_source.get("has_cf_clearance"), False):
+        cf_clearance = existing_cf_clearance
+
+    return {
+        "enabled": _normalize_bool(source.get("enabled"), bool(DEFAULT_PROXY_RUNTIME["enabled"])),
+        "egress_mode": egress_mode,
+        "proxy_url": str(source.get("proxy_url") or "").strip(),
+        "resource_proxy_url": str(source.get("resource_proxy_url") or "").strip(),
+        "skip_ssl_verify": _normalize_bool(
+            source.get("skip_ssl_verify"),
+            bool(DEFAULT_PROXY_RUNTIME["skip_ssl_verify"]),
+        ),
+        "reset_session_status_codes": _normalize_status_codes(source.get("reset_session_status_codes")),
+        "clearance": {
+            "enabled": _normalize_bool(clearance_source.get("enabled"), bool(default_clearance["enabled"])),
+            "mode": clearance_mode,
+            "cf_cookies": cf_cookies,
+            "cf_clearance": cf_clearance,
+            "user_agent": user_agent or str(default_clearance["user_agent"]),
+            "browser": browser or str(default_clearance["browser"]),
+            "flaresolverr_url": str(clearance_source.get("flaresolverr_url") or "").strip(),
+            "timeout_sec": _normalize_positive_int(
+                clearance_source.get("timeout_sec"),
+                int(default_clearance["timeout_sec"]),
+                1,
+            ),
+            "refresh_interval": _normalize_positive_int(
+                clearance_source.get("refresh_interval"),
+                int(default_clearance["refresh_interval"]),
+                60,
+            ),
+            "warm_up_on_start": _normalize_bool(
+                clearance_source.get("warm_up_on_start"),
+                bool(default_clearance["warm_up_on_start"]),
+            ),
+        },
+    }
+
+
 def _validate_image_storage_settings(settings: dict[str, object]) -> None:
     if not _normalize_bool(settings.get("enabled"), False):
         return
@@ -433,12 +540,28 @@ class ConfigStore:
         data["backup"] = self.get_backup_settings()
         data["image_storage"] = self.get_image_storage_settings()
         data["chat_completion_cache"] = self.get_chat_completion_cache_settings()
+        data["proxy_runtime"] = self.get_public_proxy_runtime_settings()
         data.pop("auth-key", None)
         return data
 
     def get_proxy_settings(self) -> str:
         return str(self.data.get("proxy") or "").strip()
 
+    def get_proxy_runtime_settings(self) -> dict[str, object]:
+        return _normalize_proxy_runtime_settings(self.data.get("proxy_runtime"))
+
+    def get_public_proxy_runtime_settings(self) -> dict[str, object]:
+        runtime = copy.deepcopy(self.get_proxy_runtime_settings())
+        clearance = runtime.get("clearance") if isinstance(runtime.get("clearance"), dict) else {}
+        if isinstance(clearance, dict):
+            cf_cookies = str(clearance.get("cf_cookies") or "").strip()
+            cf_clearance = str(clearance.get("cf_clearance") or "").strip()
+            clearance["cf_cookies"] = ""
+            clearance["cf_clearance"] = ""
+            clearance["has_cf_cookies"] = bool(cf_cookies)
+            clearance["has_cf_clearance"] = bool(cf_clearance)
+        return runtime
+
     def update(self, data: dict[str, object]) -> dict[str, object]:
         next_data = dict(self.data)
         next_data.update(dict(data or {}))
@@ -451,6 +574,15 @@ class ConfigStore:
             next_data["chat_completion_cache"] = _normalize_chat_completion_cache_settings(
                 next_data.get("chat_completion_cache")
             )
+        if "proxy_runtime" in next_data:
+            incoming_runtime = next_data.get("proxy_runtime")
+            if isinstance(incoming_runtime, dict):
+                previous_clearance = self.get_proxy_runtime_settings().get("clearance")
+                if isinstance(previous_clearance, dict):
+                    incoming_runtime = dict(incoming_runtime)
+                    incoming_runtime["_existing_cf_cookies"] = previous_clearance.get("cf_cookies")
+                    incoming_runtime["_existing_cf_clearance"] = previous_clearance.get("cf_clearance")
+            next_data["proxy_runtime"] = _normalize_proxy_runtime_settings(incoming_runtime)
         next_data.pop("backup_state", None)
         self.data = next_data
         self._save()

services/proxy_service.py

@@ -1,8 +1,14 @@
-"""Global outbound proxy helpers for upstream ChatGPT and CPA requests."""
+"""Global outbound proxy and Cloudflare clearance helpers."""
 
 from __future__ import annotations
 
+from dataclasses import dataclass, field, replace
+import json
+import re
+import threading
 import time
+from typing import Callable, Mapping
+from urllib import request as urllib_request
 from urllib.parse import urlparse
 
 from curl_cffi.requests import Session
@@ -10,30 +16,543 @@ from curl_cffi.requests import Session
 from services.config import config
 
 
+FlareSolverrRequestMethod = Callable[[str, bytes, dict[str, str], float], bytes]
+
+
+def normalize_proxy_url(url: str) -> str:
+    """Normalize proxy URLs for curl_cffi.
+
+    SOCKS proxies should use remote-DNS resolution by default, so generic
+    ``socks://`` and ``socks5://`` inputs are upgraded to ``socks5h://``.
+    HTTP/HTTPS/socks5h inputs are otherwise left untouched except trimming.
+    """
+    candidate = str(url or "").strip()
+    lowered = candidate.lower()
+    if lowered.startswith("socks://"):
+        return "socks5h://" + candidate[len("socks://") :]
+    if lowered.startswith("socks5://"):
+        return "socks5h://" + candidate[len("socks5://") :]
+    return candidate
+
+
+@dataclass(frozen=True)
+class ProxyRuntimeProfile:
+    proxy_url: str = ""
+    proxy_source: str = "direct"
+    resource: bool = False
+    runtime_enabled: bool = False
+    egress_mode: str = "direct"
+    skip_ssl_verify: bool = False
+    reset_session_status_codes: tuple[int, ...] = field(default_factory=lambda: (403,))
+    clearance: dict[str, object] = field(default_factory=dict, repr=False)
+
+    @property
+    def clearance_enabled(self) -> bool:
+        return (
+            self.runtime_enabled
+            and bool(self.clearance.get("enabled"))
+            and self.clearance_mode in {"manual", "flaresolverr"}
+        )
+
+    @property
+    def clearance_mode(self) -> str:
+        return str(self.clearance.get("mode") or "none").strip().lower()
+
+    @property
+    def refresh_interval(self) -> int:
+        try:
+            return max(0, int(self.clearance.get("refresh_interval") or 0))
+        except (OverflowError, TypeError, ValueError):
+            return 0
+
+    @property
+    def timeout_sec(self) -> int:
+        try:
+            return max(1, int(self.clearance.get("timeout_sec") or 60))
+        except (OverflowError, TypeError, ValueError):
+            return 60
+
+
+@dataclass(frozen=True)
+class ClearanceBundle:
+    target_host: str
+    proxy_url: str = ""
+    cookies: dict[str, str] = field(default_factory=dict, repr=False)
+    user_agent: str = ""
+    created_at: float = field(default_factory=time.time)
+    expires_at: float | None = None
+
+    def is_valid_for(self, target_host: str, proxy_url: str, *, now: float | None = None) -> bool:
+        host = _normalize_host(target_host)
+        if self.target_host and host and _normalize_host(self.target_host) != host:
+            return False
+        if normalize_proxy_url(self.proxy_url) != normalize_proxy_url(proxy_url):
+            return False
+        if self.expires_at is not None and (time.time() if now is None else now) >= self.expires_at:
+            return False
+        return bool(self.cookies or self.user_agent)
+
+    def cookie_header(self) -> str:
+        return _cookies_to_header(self.cookies)
+
+
+class FlareSolverrClearanceProvider:
+    def __init__(self, flaresolverr_url: str, request_method: FlareSolverrRequestMethod | None = None) -> None:
+        self.flaresolverr_url = str(flaresolverr_url or "").strip().rstrip("/")
+        self._request_method = request_method or self._urllib_post
+
+    def get_clearance(self, target_url: str, proxy_url: str = "", timeout_sec: int = 60) -> ClearanceBundle | None:
+        if not self.flaresolverr_url:
+            return None
+
+        timeout = _coerce_timeout(timeout_sec)
+        payload: dict[str, object] = {
+            "cmd": "request.get",
+            "url": str(target_url or ""),
+            "maxTimeout": int(timeout * 1000),
+        }
+        proxy_url = normalize_proxy_url(proxy_url)
+        if proxy_url:
+            payload["proxy"] = {"url": proxy_url}
+
+        endpoint = f"{self.flaresolverr_url}/v1"
+        try:
+            body = json.dumps(payload).encode("utf-8")
+            raw_response = self._request_method(
+                endpoint,
+                body,
+                {"Content-Type": "application/json"},
+                timeout,
+            )
+            data = json.loads(raw_response.decode("utf-8") if isinstance(raw_response, bytes) else raw_response)
+        except Exception:
+            return None
+
+        if not isinstance(data, dict) or str(data.get("status") or "").lower() != "ok":
+            return None
+        solution = data.get("solution")
+        if not isinstance(solution, dict):
+            return None
+
+        target_host = _host_from_url(target_url)
+        cookies = _filter_flaresolverr_cookies(solution.get("cookies"), target_host)
+        user_agent = str(solution.get("userAgent") or "").strip()
+        if not cookies and not user_agent:
+            return None
+        return ClearanceBundle(
+            target_host=target_host,
+            proxy_url=proxy_url,
+            cookies=cookies,
+            user_agent=user_agent,
+        )
+
+    @staticmethod
+    def _urllib_post(endpoint: str, body: bytes, headers: dict[str, str], timeout: float) -> bytes:
+        req = urllib_request.Request(endpoint, data=body, headers=headers, method="POST")
+        with urllib_request.urlopen(req, timeout=timeout) as response:
+            return response.read()
+
+
 class ProxySettingsStore:
-    def build_session_kwargs(self, account: dict | None = None, proxy: str = "", **session_kwargs) -> dict[str, object]:
-        account_proxy = str((account or {}).get("proxy") or "").strip()
-        proxy = str(proxy or account_proxy or config.get_proxy_settings()).strip()
-        if proxy:
-            session_kwargs["proxy"] = proxy
+    def __init__(
+        self,
+        config_store=None,
+        clearance_provider_factory: Callable[[str], FlareSolverrClearanceProvider] | None = None,
+    ) -> None:
+        self._config = config_store or config
+        self._clearance_provider_factory = clearance_provider_factory or FlareSolverrClearanceProvider
+        self._clearance_cache: dict[tuple[str, str], ClearanceBundle] = {}
+        self._provider_cache: dict[str, FlareSolverrClearanceProvider] = {}
+        self._flight_locks: dict[tuple[str, str], threading.Lock] = {}
+        self._lock = threading.RLock()
+
+    def get_profile(
+        self,
+        account: dict | None = None,
+        proxy: str = "",
+        resource: bool = False,
+        upstream: bool = False,
+    ) -> ProxyRuntimeProfile:
+        runtime = self._get_runtime_settings()
+        clearance = dict(runtime.get("clearance") if isinstance(runtime.get("clearance"), dict) else {})
+        runtime_enabled = bool(runtime.get("enabled"))
+        egress_mode = str(runtime.get("egress_mode") or "direct").strip().lower()
+
+        account_proxy = _clean((account or {}).get("proxy") if isinstance(account, dict) else "")
+        explicit_proxy = _clean(proxy)
+        legacy_proxy = _clean(self._config.get_proxy_settings())
+
+        runtime_proxy = ""
+        runtime_proxy_source = "runtime"
+        if upstream and runtime_enabled and egress_mode == "single_proxy":
+            resource_proxy = _clean(runtime.get("resource_proxy_url")) if resource else ""
+            runtime_proxy = resource_proxy or _clean(runtime.get("proxy_url"))
+            runtime_proxy_source = "runtime_resource" if resource_proxy else "runtime"
+
+        selected_proxy = ""
+        source = "direct"
+        if account_proxy:
+            selected_proxy = account_proxy
+            source = "account"
+        elif runtime_proxy:
+            selected_proxy = runtime_proxy
+            source = runtime_proxy_source
+        elif explicit_proxy:
+            selected_proxy = explicit_proxy
+            source = "explicit"
+        elif legacy_proxy:
+            selected_proxy = legacy_proxy
+            source = "global"
+
+        return ProxyRuntimeProfile(
+            proxy_url=normalize_proxy_url(selected_proxy),
+            proxy_source=source,
+            resource=bool(resource),
+            runtime_enabled=runtime_enabled,
+            egress_mode=egress_mode,
+            skip_ssl_verify=bool(runtime.get("skip_ssl_verify")) if runtime_enabled else False,
+            reset_session_status_codes=_status_codes_tuple(runtime.get("reset_session_status_codes")),
+            clearance=clearance,
+        )
+
+    def build_session_kwargs(
+        self,
+        account: dict | None = None,
+        proxy: str = "",
+        resource: bool = False,
+        upstream: bool = False,
+        **session_kwargs,
+    ) -> dict[str, object]:
+        profile = self.get_profile(account=account, proxy=proxy, resource=resource, upstream=upstream)
+        if profile.proxy_url:
+            session_kwargs["proxy"] = profile.proxy_url
+        if profile.runtime_enabled and profile.skip_ssl_verify:
+            session_kwargs["verify"] = False
         return session_kwargs
 
+    def build_headers(
+        self,
+        headers: Mapping[str, object] | None = None,
+        target_url: str = "https://chatgpt.com",
+        account: dict | None = None,
+        proxy: str = "",
+        resource: bool = False,
+        upstream: bool = True,
+    ) -> dict[str, object]:
+        merged_headers: dict[str, object] = dict(headers or {})
+        profile = self.get_profile(account=account, proxy=proxy, resource=resource, upstream=upstream)
+        if not profile.clearance_enabled:
+            return merged_headers
+
+        target_host = _host_from_url(target_url)
+        bundle = self._bundle_for_headers(profile, target_host)
+        if bundle is None or not bundle.is_valid_for(target_host, profile.proxy_url):
+            return merged_headers
+
+        if bundle.user_agent and _find_header_key(merged_headers, "user-agent") is None:
+            merged_headers["User-Agent"] = bundle.user_agent
+
+        if bundle.cookies:
+            cookie_key = _find_header_key(merged_headers, "cookie") or "Cookie"
+            existing_cookie = str(merged_headers.get(cookie_key) or "")
+            cookie_header = _merge_cookie_header(existing_cookie, bundle.cookies)
+            if cookie_header:
+                merged_headers[cookie_key] = cookie_header
+        return merged_headers
+
+    def refresh_clearance(
+        self,
+        target_url: str = "https://chatgpt.com",
+        account: dict | None = None,
+        proxy: str = "",
+        resource: bool = False,
+        force: bool = False,
+        upstream: bool = True,
+    ) -> ClearanceBundle | None:
+        profile = self.get_profile(account=account, proxy=proxy, resource=resource, upstream=upstream)
+        if not profile.clearance_enabled:
+            return None
+
+        target_host = _host_from_url(target_url)
+        key = self._cache_key(profile.proxy_url, target_host)
+        if profile.clearance_mode == "manual":
+            bundle = self._build_manual_bundle(profile, target_host)
+            if bundle is not None:
+                self._set_cached_bundle(key, bundle)
+            return bundle
+        if profile.clearance_mode != "flaresolverr":
+            return None
+
+        cached_before = self._get_cached_bundle(key)
+        if cached_before is not None and not force and cached_before.is_valid_for(target_host, profile.proxy_url):
+            return cached_before
+
+        lock = self._get_flight_lock(key)
+        if not lock.acquire(blocking=False):
+            with lock:
+                pass
+            return self._get_cached_bundle(key) or cached_before
+
+        try:
+            cached_now = self._get_cached_bundle(key)
+            if cached_now is not None and not force and cached_now.is_valid_for(target_host, profile.proxy_url):
+                return cached_now
+
+            flaresolverr_url = str(profile.clearance.get("flaresolverr_url") or "").strip()
+            provider = self._get_provider(flaresolverr_url)
+            new_bundle = provider.get_clearance(target_url, proxy_url=profile.proxy_url, timeout_sec=profile.timeout_sec)
+            if new_bundle is not None:
+                expires_at = time.time() + profile.refresh_interval if profile.refresh_interval else None
+                if (
+                    not new_bundle.target_host
+                    or normalize_proxy_url(new_bundle.proxy_url) != normalize_proxy_url(profile.proxy_url)
+                    or new_bundle.expires_at != expires_at
+                ):
+                    new_bundle = replace(
+                        new_bundle,
+                        target_host=new_bundle.target_host or target_host,
+                        proxy_url=profile.proxy_url,
+                        expires_at=expires_at,
+                    )
+                self._set_cached_bundle(key, new_bundle)
+                return new_bundle
+            return cached_now or cached_before
+        finally:
+            lock.release()
+
+    def invalidate_clearance(
+        self,
+        target_url: str = "https://chatgpt.com",
+        account: dict | None = None,
+        proxy: str = "",
+        resource: bool = False,
+        upstream: bool = True,
+    ) -> None:
+        profile = self.get_profile(account=account, proxy=proxy, resource=resource, upstream=upstream)
+        target_host = _host_from_url(target_url)
+        key = self._cache_key(profile.proxy_url, target_host)
+        with self._lock:
+            self._clearance_cache.pop(key, None)
+
+    def get_runtime_status(self) -> dict[str, object]:
+        profile = self.get_profile(upstream=True)
+        with self._lock:
+            cached_hosts = [host for _proxy, host in self._clearance_cache]
+            cached_count = len(self._clearance_cache)
+        return {
+            "enabled": profile.runtime_enabled,
+            "egress_mode": profile.egress_mode,
+            "proxy_source": profile.proxy_source,
+            "has_proxy": bool(profile.proxy_url),
+            "clearance_enabled": profile.clearance_enabled,
+            "clearance_mode": profile.clearance_mode,
+            "has_clearance_bundle": cached_count > 0,
+            "cached_clearance_hosts": sorted(set(cached_hosts)),
+        }
+
+    def _get_runtime_settings(self) -> dict[str, object]:
+        try:
+            runtime = self._config.get_proxy_runtime_settings()
+        except AttributeError:
+            runtime = {}
+        return runtime if isinstance(runtime, dict) else {}
+
+    def _bundle_for_headers(self, profile: ProxyRuntimeProfile, target_host: str) -> ClearanceBundle | None:
+        key = self._cache_key(profile.proxy_url, target_host)
+        if profile.clearance_mode == "manual":
+            bundle = self._build_manual_bundle(profile, target_host)
+            if bundle is not None:
+                self._set_cached_bundle(key, bundle)
+            return bundle
+        if profile.clearance_mode == "flaresolverr":
+            return self._get_cached_bundle(key)
+        return None
+
+    def _build_manual_bundle(self, profile: ProxyRuntimeProfile, target_host: str) -> ClearanceBundle | None:
+        cookies = _parse_cookie_header(str(profile.clearance.get("cf_cookies") or ""))
+        cf_clearance = str(profile.clearance.get("cf_clearance") or "").strip()
+        if cf_clearance and "cf_clearance" not in cookies:
+            cookies["cf_clearance"] = cf_clearance
+        user_agent = str(profile.clearance.get("user_agent") or "").strip()
+        if not cookies and not user_agent:
+            return None
+
+        now = time.time()
+        expires_at = now + profile.refresh_interval if profile.refresh_interval else None
+        return ClearanceBundle(
+            target_host=target_host,
+            proxy_url=profile.proxy_url,
+            cookies=cookies,
+            user_agent=user_agent,
+            created_at=now,
+            expires_at=expires_at,
+        )
+
+    def _get_provider(self, flaresolverr_url: str) -> FlareSolverrClearanceProvider:
+        url = str(flaresolverr_url or "").strip().rstrip("/")
+        with self._lock:
+            provider = self._provider_cache.get(url)
+            if provider is None:
+                provider = self._clearance_provider_factory(url)
+                self._provider_cache[url] = provider
+            return provider
+
+    def _get_flight_lock(self, key: tuple[str, str]) -> threading.Lock:
+        with self._lock:
+            lock = self._flight_locks.get(key)
+            if lock is None:
+                lock = threading.Lock()
+                self._flight_locks[key] = lock
+            return lock
+
+    def _get_cached_bundle(self, key: tuple[str, str]) -> ClearanceBundle | None:
+        with self._lock:
+            return self._clearance_cache.get(key)
+
+    def _set_cached_bundle(self, key: tuple[str, str], bundle: ClearanceBundle) -> None:
+        with self._lock:
+            self._clearance_cache[key] = bundle
+
+    @staticmethod
+    def _cache_key(proxy_url: str, target_host: str) -> tuple[str, str]:
+        return (normalize_proxy_url(proxy_url), _normalize_host(target_host))
+
 
 def _clean(value: object) -> str:
     return str(value or "").strip()
 
 
+def _normalize_host(host: str) -> str:
+    return str(host or "").strip().strip(".").lower()
+
+
+def _host_from_url(url: str) -> str:
+    candidate = str(url or "").strip()
+    parsed = urlparse(candidate)
+    if not parsed.hostname and candidate and "://" not in candidate:
+        parsed = urlparse(f"https://{candidate}")
+    return _normalize_host(parsed.hostname or "")
+
+
+def _status_codes_tuple(value: object) -> tuple[int, ...]:
+    source = value if isinstance(value, list) else [403]
+    codes: list[int] = []
+    for item in source:
+        if isinstance(item, bool):
+            continue
+        try:
+            code = int(item)
+        except (OverflowError, TypeError, ValueError):
+            continue
+        if 100 <= code <= 599 and code not in codes:
+            codes.append(code)
+    return tuple(codes or [403])
+
+
+def _coerce_timeout(value: object) -> float:
+    try:
+        timeout = float(value)
+    except (OverflowError, TypeError, ValueError):
+        timeout = 60.0
+    return max(1.0, timeout)
+
+
 def _is_valid_proxy_url(url: str) -> bool:
-    parsed = urlparse(url)
+    parsed = urlparse(normalize_proxy_url(url))
     return parsed.scheme in {"http", "https", "socks5", "socks5h"} and bool(parsed.netloc)
 
 
-def test_proxy(url: str, *, timeout: float = 15.0) -> dict:
-    candidate = _clean(url)
+def _domain_matches(host: str, domain: str) -> bool:
+    normalized_host = _normalize_host(host)
+    normalized_domain = _normalize_host(domain.lstrip("."))
+    if not normalized_domain:
+        return True
+    return normalized_host == normalized_domain or normalized_host.endswith(f".{normalized_domain}")
+
+
+def _filter_flaresolverr_cookies(raw_cookies: object, target_host: str) -> dict[str, str]:
+    if not isinstance(raw_cookies, list):
+        return {}
+
+    filtered_cookies: dict[str, str] = {}
+    for item in raw_cookies:
+        if not isinstance(item, dict):
+            continue
+        name = str(item.get("name") or "").strip()
+        if not name:
+            continue
+        value = str(item.get("value") or "")
+        domain = str(item.get("domain") or "").strip()
+        if not domain or _domain_matches(target_host, domain):
+            filtered_cookies[name] = value
+    return filtered_cookies
+
+
+def _parse_cookie_header(header: str) -> dict[str, str]:
+    cookies: dict[str, str] = {}
+    for part in str(header or "").split(";"):
+        name, sep, value = part.strip().partition("=")
+        if sep and name:
+            cookies[name.strip()] = value.strip()
+    return cookies
+
+
+def _cookies_to_header(cookies: Mapping[str, str]) -> str:
+    return "; ".join(f"{name}={value}" for name, value in cookies.items() if name)
+
+
+def _merge_cookie_header(existing_header: str, cookies: Mapping[str, str]) -> str:
+    existing = str(existing_header or "").strip()
+    existing_names = set(_parse_cookie_header(existing).keys())
+    additions = [f"{name}={value}" for name, value in cookies.items() if name and name not in existing_names]
+    if existing and additions:
+        return existing.rstrip("; ") + "; " + "; ".join(additions)
+    if existing:
+        return existing
+    return "; ".join(additions)
+
+
+def _find_header_key(headers: Mapping[str, object], name: str) -> str | None:
+    target = name.lower()
+    for key in headers:
+        if str(key).lower() == target:
+            return str(key)
+    return None
+
+
+def _redact_url_credentials(text: str) -> str:
+    return re.sub(
+        r"(https?://)([^\s/@:]+):([^\s/@]+)@",
+        r"\1[REDACTED]@",
+        str(text or ""),
+        flags=re.IGNORECASE,
+    )
+
+
+def test_proxy(url: str = "", *, timeout: float = 15.0) -> dict:
+    candidate = normalize_proxy_url(_clean(url))
+    proxy_source = "input"
+    if not candidate:
+        profile = proxy_settings.get_profile(upstream=True)
+        candidate = profile.proxy_url
+        proxy_source = profile.proxy_source
+    result_base = {"proxy_source": proxy_source, "has_proxy": bool(candidate)}
     if not candidate:
-        return {"ok": False, "status": 0, "latency_ms": 0, "error": "proxy url is required"}
+        return {
+            "ok": False,
+            "status": 0,
+            "latency_ms": 0,
+            "error": "no active proxy configured",
+            **result_base,
+        }
     if not _is_valid_proxy_url(candidate):
-        return {"ok": False, "status": 0, "latency_ms": 0, "error": "invalid proxy url"}
+        return {
+            "ok": False,
+            "status": 0,
+            "latency_ms": 0,
+            "error": "invalid proxy url",
+            **result_base,
+        }
     session = Session(impersonate="edge101", verify=True, proxy=candidate)
     started = time.perf_counter()
     try:
@@ -48,6 +567,7 @@ def test_proxy(url: str, *, timeout: float = 15.0) -> dict:
             "status": int(response.status_code),
             "latency_ms": latency_ms,
             "error": None if response.status_code < 500 else f"HTTP {response.status_code}",
+            **result_base,
         }
     except Exception as exc:
         latency_ms = int((time.perf_counter() - started) * 1000)
@@ -55,10 +575,62 @@ def test_proxy(url: str, *, timeout: float = 15.0) -> dict:
             "ok": False,
             "status": 0,
             "latency_ms": latency_ms,
-            "error": str(exc) or exc.__class__.__name__,
+            "error": _redact_url_credentials(str(exc) or exc.__class__.__name__),
+            **result_base,
         }
     finally:
         session.close()
 
-proxy_settings = ProxySettingsStore()
 
+def test_clearance(target_url: str = "https://chatgpt.com") -> dict:
+    target_url = str(target_url or "https://chatgpt.com").strip() or "https://chatgpt.com"
+    started = time.perf_counter()
+    status = proxy_settings.get_runtime_status()
+    if not status.get("clearance_enabled"):
+        return {
+            "ok": False,
+            "status": "disabled",
+            "latency_ms": 0,
+            "has_cookies": False,
+            "user_agent": "",
+            "error": "clearance is disabled",
+            "runtime": status,
+        }
+    try:
+        bundle = proxy_settings.refresh_clearance(target_url=target_url, force=True, upstream=True)
+    except Exception as exc:
+        latency_ms = int((time.perf_counter() - started) * 1000)
+        return {
+            "ok": False,
+            "status": "error",
+            "latency_ms": latency_ms,
+            "has_cookies": False,
+            "user_agent": "",
+            "error": _redact_url_credentials(str(exc) or exc.__class__.__name__),
+            "runtime": proxy_settings.get_runtime_status(),
+        }
+
+    latency_ms = int((time.perf_counter() - started) * 1000)
+    runtime = proxy_settings.get_runtime_status()
+    if bundle is None:
+        return {
+            "ok": False,
+            "status": "failed",
+            "latency_ms": latency_ms,
+            "has_cookies": False,
+            "user_agent": "",
+            "error": "clearance refresh returned no bundle",
+            "runtime": runtime,
+        }
+    return {
+        "ok": True,
+        "status": "ok",
+        "latency_ms": latency_ms,
+        "has_cookies": bool(bundle.cookies),
+        "user_agent": bundle.user_agent or "",
+        "error": None,
+        "runtime": runtime,
+    }
+
+
+proxy_settings = ProxySettingsStore()

services/register/openai_register.py

@@ -17,6 +17,7 @@ from urllib.parse import parse_qs, urlencode, urlparse
 from curl_cffi import requests
 
 from services.account_service import account_service
+from services.proxy_service import ClearanceBundle, proxy_settings
 from services.register import mail_provider
 
 base_dir = Path(__file__).resolve().parent
@@ -196,11 +197,22 @@ def _is_cloudflare_challenge(resp) -> bool:
     text = str(getattr(resp, "text", "") or "").lower()
     headers = getattr(resp, "headers", {}) or {}
     server = str(headers.get("server") or "").lower()
-    return (
-        "cloudflare" in server
-        or "challenges.cloudflare.com" in text
+    try:
+        status_code = int(getattr(resp, "status_code", 0) or 0)
+    except (TypeError, ValueError):
+        status_code = 0
+
+    explicit_challenge_markers = (
+        "challenges.cloudflare.com" in text
         or "<title>just a moment" in text
+        or "/cdn-cgi/challenge-platform/" in text
+        or "cf-chl-" in text
+        or "__cf_chl_" in text
     )
+    if explicit_challenge_markers:
+        return True
+
+    return "cloudflare" in server and status_code in (403, 503)
 
 
 def _mail_config() -> dict:
@@ -246,12 +258,54 @@ def build_sentinel_token(session: requests.Session, device_id: str, flow: str) -
 
 
 def create_session(proxy: str = "") -> Any:
-    kwargs = {"impersonate": "chrome", "verify": False}
-    if proxy:
-        kwargs["proxy"] = proxy
+    kwargs = proxy_settings.build_session_kwargs(
+        proxy=proxy,
+        upstream=True,
+        impersonate="chrome",
+        verify=False,
+    )
     return requests.Session(**kwargs)
 
 
+def _apply_clearance_to_session(session: requests.Session, bundle: ClearanceBundle | None) -> None:
+    if bundle is None:
+        return
+    if bundle.user_agent:
+        session.headers["User-Agent"] = bundle.user_agent
+        session.headers["user-agent"] = bundle.user_agent
+    for name, value in bundle.cookies.items():
+        try:
+            session.cookies.set(name, value, domain=f".{bundle.target_host or 'openai.com'}")
+            session.cookies.set(name, value, domain=bundle.target_host or "auth.openai.com")
+        except Exception:
+            continue
+
+
+def _headers_with_clearance(
+    headers: dict[str, str],
+    target_url: str,
+    proxy: str = "",
+    user_agent_override: str = "",
+) -> dict[str, str]:
+    merged = proxy_settings.build_headers(
+        headers=headers,
+        target_url=target_url,
+        proxy=proxy,
+        upstream=True,
+    )
+    normalized = {str(key): str(value) for key, value in merged.items()}
+    if user_agent_override:
+        ua_key = next((key for key in normalized if key.lower() == "user-agent"), "user-agent")
+        normalized[ua_key] = user_agent_override
+    return normalized
+
+
+def _cloudflare_block_message(resp, prefix: str = "被 Cloudflare 拦截") -> str:
+    status = getattr(resp, "status_code", "unknown")
+    debug = _response_debug_detail(resp)
+    return f"{prefix}，clearance 刷新失败或重试后仍失败，请更换 IP/代理重试: status={status}, {debug}"
+
+
 def request_with_local_retry(session: requests.Session, method: str, url: str, retry_attempts: int = 3, **kwargs):
     last_error = ""
     for _ in range(max(1, retry_attempts)):
@@ -329,7 +383,9 @@ def request_platform_oauth_token(session: requests.Session, code: str, code_veri
 
 class PlatformRegistrar:
     def __init__(self, proxy: str = "") -> None:
-        self.session = create_session(proxy)
+        self.proxy = str(proxy or "").strip()
+        self.session = create_session(self.proxy)
+        self.clearance_user_agent = ""
         self.device_id = str(uuid.uuid4())
         self.code_verifier = ""
         self.platform_auth_code = ""
@@ -350,6 +406,20 @@ class PlatformRegistrar:
         headers.update(_make_trace_headers())
         return headers
 
+    def _refresh_cloudflare_clearance(self, target_url: str, index: int) -> ClearanceBundle | None:
+        step(index, "检测到 Cloudflare 拦截，尝试刷新 clearance", "yellow")
+        bundle = proxy_settings.refresh_clearance(
+            target_url=target_url,
+            proxy=self.proxy,
+            force=True,
+            upstream=True,
+        )
+        if bundle is not None:
+            _apply_clearance_to_session(self.session, bundle)
+            self.clearance_user_agent = bundle.user_agent or self.clearance_user_agent
+            step(index, "Cloudflare clearance 刷新完成，重试当前请求", "yellow")
+        return bundle
+
     def _platform_authorize(self, email: str, index: int) -> None:
         step(index, "开始 platform authorize")
         self.session.cookies.set("oai-did", self.device_id, domain=".auth.openai.com")
@@ -376,12 +446,21 @@ class PlatformRegistrar:
             "code_challenge_method": "S256",
             "auth0Client": platform_auth0_client,
         }
-        resp, error = request_with_local_retry(self.session, "get", f"{auth_base}/api/accounts/authorize?{urlencode(params)}", headers=self._navigate_headers(f"{platform_base}/"), allow_redirects=True, verify=False)
+        target_url = f"{auth_base}/api/accounts/authorize?{urlencode(params)}"
+        headers = self._navigate_headers(f"{platform_base}/")
+        headers = _headers_with_clearance(headers, target_url, self.proxy, self.clearance_user_agent)
+        resp, error = request_with_local_retry(self.session, "get", target_url, headers=headers, allow_redirects=True, verify=False)
+        if _is_cloudflare_challenge(resp):
+            bundle = self._refresh_cloudflare_clearance(auth_base, index)
+            if bundle is None:
+                raise RuntimeError(_cloudflare_block_message(resp))
+            retry_headers = _headers_with_clearance(self._navigate_headers(f"{platform_base}/"), target_url, self.proxy, self.clearance_user_agent)
+            resp, error = request_with_local_retry(self.session, "get", target_url, headers=retry_headers, allow_redirects=True, verify=False)
+            if _is_cloudflare_challenge(resp):
+                raise RuntimeError(_cloudflare_block_message(resp, "Cloudflare clearance 重试仍被拦截"))
         if resp is None or resp.status_code != 200:
             err = _response_json(resp).get("error", {}) if resp is not None else {}
             detail = f": {err.get('code', '')} - {err.get('message', '')}".strip(" -") if err else ""
-            if _is_cloudflare_challenge(resp):
-                raise RuntimeError("被 Cloudflare 拦截，请更换 IP 重试")
             debug = _response_debug_detail(resp)
             status = getattr(resp, "status_code", "unknown")
             raise RuntimeError(error or f"platform_authorize_http_{status}{detail}, {debug}")
@@ -392,9 +471,21 @@ class PlatformRegistrar:
 
     def _register_user(self, email: str, password: str, index: int) -> None:
         step(index, "开始提交注册密码")
+        url = f"{auth_base}/api/accounts/user/register"
         headers = self._json_headers(f"{auth_base}/create-account/password")
         headers["openai-sentinel-token"] = build_sentinel_token(self.session, self.device_id, "username_password_create")
-        resp, error = request_with_local_retry(self.session, "post", f"{auth_base}/api/accounts/user/register", json={"username": email, "password": password}, headers=headers, verify=False)
+        headers = _headers_with_clearance(headers, url, self.proxy, self.clearance_user_agent)
+        resp, error = request_with_local_retry(self.session, "post", url, json={"username": email, "password": password}, headers=headers, verify=False)
+        if _is_cloudflare_challenge(resp):
+            bundle = self._refresh_cloudflare_clearance(auth_base, index)
+            if bundle is None:
+                raise RuntimeError(_cloudflare_block_message(resp))
+            headers = self._json_headers(f"{auth_base}/create-account/password")
+            headers["openai-sentinel-token"] = build_sentinel_token(self.session, self.device_id, "username_password_create")
+            headers = _headers_with_clearance(headers, url, self.proxy, self.clearance_user_agent)
+            resp, error = request_with_local_retry(self.session, "post", url, json={"username": email, "password": password}, headers=headers, verify=False)
+            if _is_cloudflare_challenge(resp):
+                raise RuntimeError(_cloudflare_block_message(resp, "Cloudflare clearance 重试仍被拦截"))
         if resp is None or resp.status_code != 200:
             data = _response_json(resp) if resp is not None else {}
             if data.get("message") == "Failed to create account. Please try again.":
@@ -405,7 +496,17 @@ class PlatformRegistrar:
 
     def _send_otp(self, index: int) -> None:
         step(index, "开始发送验证码")
-        resp, error = request_with_local_retry(self.session, "get", f"{auth_base}/api/accounts/email-otp/send", headers=self._navigate_headers(f"{auth_base}/create-account/password"), allow_redirects=True, verify=False)
+        url = f"{auth_base}/api/accounts/email-otp/send"
+        headers = _headers_with_clearance(self._navigate_headers(f"{auth_base}/create-account/password"), url, self.proxy, self.clearance_user_agent)
+        resp, error = request_with_local_retry(self.session, "get", url, headers=headers, allow_redirects=True, verify=False)
+        if _is_cloudflare_challenge(resp):
+            bundle = self._refresh_cloudflare_clearance(auth_base, index)
+            if bundle is None:
+                raise RuntimeError(_cloudflare_block_message(resp))
+            headers = _headers_with_clearance(self._navigate_headers(f"{auth_base}/create-account/password"), url, self.proxy, self.clearance_user_agent)
+            resp, error = request_with_local_retry(self.session, "get", url, headers=headers, allow_redirects=True, verify=False)
+            if _is_cloudflare_challenge(resp):
+                raise RuntimeError(_cloudflare_block_message(resp, "Cloudflare clearance 重试仍被拦截"))
         if resp is None or resp.status_code not in (200, 302):
             raise RuntimeError(error or f"send_otp_http_{getattr(resp, 'status_code', 'unknown')}")
         step(index, "发送验证码完成")
@@ -424,9 +525,21 @@ class PlatformRegistrar:
 
     def _create_account(self, name: str, birthdate: str, index: int) -> None:
         step(index, "开始创建账号资料")
+        url = f"{auth_base}/api/accounts/create_account"
         headers = self._json_headers(f"{auth_base}/about-you")
         headers["openai-sentinel-token"] = build_sentinel_token(self.session, self.device_id, "oauth_create_account")
-        resp, error = request_with_local_retry(self.session, "post", f"{auth_base}/api/accounts/create_account", json={"name": name, "birthdate": birthdate}, headers=headers, verify=False)
+        headers = _headers_with_clearance(headers, url, self.proxy, self.clearance_user_agent)
+        resp, error = request_with_local_retry(self.session, "post", url, json={"name": name, "birthdate": birthdate}, headers=headers, verify=False)
+        if _is_cloudflare_challenge(resp):
+            bundle = self._refresh_cloudflare_clearance(auth_base, index)
+            if bundle is None:
+                raise RuntimeError(_cloudflare_block_message(resp))
+            headers = self._json_headers(f"{auth_base}/about-you")
+            headers["openai-sentinel-token"] = build_sentinel_token(self.session, self.device_id, "oauth_create_account")
+            headers = _headers_with_clearance(headers, url, self.proxy, self.clearance_user_agent)
+            resp, error = request_with_local_retry(self.session, "post", url, json={"name": name, "birthdate": birthdate}, headers=headers, verify=False)
+            if _is_cloudflare_challenge(resp):
+                raise RuntimeError(_cloudflare_block_message(resp, "Cloudflare clearance 重试仍被拦截"))
         if resp is None or resp.status_code not in (200, 302):
             data = _response_json(resp) if resp is not None else {}
             if data.get("message") == "Failed to create account. Please try again.":

test/test_init_proxy_config.py

@@ -0,0 +1,117 @@
+import errno
+import importlib.util
+import json
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+
+SCRIPT_PATH = Path(__file__).resolve().parents[1] / "scripts" / "init_proxy_config.py"
+
+
+def load_script_module():
+    spec = importlib.util.spec_from_file_location("init_proxy_config_under_test", SCRIPT_PATH)
+    module = importlib.util.module_from_spec(spec)
+    assert spec is not None and spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+class InitProxyConfigTests(unittest.TestCase):
+    def test_creates_warp_defaults_when_proxy_runtime_missing(self) -> None:
+        module = load_script_module()
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "config.json"
+            path.write_text(json.dumps({"auth-key": "secret", "proxy": ""}), encoding="utf-8")
+            with patch.dict(os.environ, {"CHATGPT2API_CONFIG_FILE": str(path)}, clear=False):
+                self.assertEqual(module.main(), 0)
+
+            data = json.loads(path.read_text(encoding="utf-8"))
+            self.assertEqual(data["auth-key"], "secret")
+            runtime = data["proxy_runtime"]
+            self.assertTrue(runtime["enabled"])
+            self.assertEqual(runtime["egress_mode"], "single_proxy")
+            self.assertEqual(runtime["proxy_url"], "http://privoxy:8118")
+            self.assertTrue(runtime["clearance"]["enabled"])
+            self.assertEqual(runtime["clearance"]["mode"], "flaresolverr")
+            self.assertEqual(runtime["clearance"]["flaresolverr_url"], "http://flaresolverr:8191")
+
+    def test_existing_custom_runtime_is_not_overwritten(self) -> None:
+        module = load_script_module()
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "config.json"
+            path.write_text(
+                json.dumps(
+                    {
+                        "auth-key": "secret",
+                        "proxy_runtime": {
+                            "enabled": False,
+                            "egress_mode": "single_proxy",
+                            "proxy_url": "http://custom.proxy:8080",
+                            "clearance": {
+                                "enabled": True,
+                                "mode": "manual",
+                                "cf_clearance": "manual-token",
+                            },
+                        },
+                    }
+                ),
+                encoding="utf-8",
+            )
+            with patch.dict(os.environ, {"CHATGPT2API_CONFIG_FILE": str(path)}, clear=False):
+                self.assertEqual(module.main(), 0)
+
+            runtime = json.loads(path.read_text(encoding="utf-8"))["proxy_runtime"]
+            self.assertFalse(runtime["enabled"])
+            self.assertEqual(runtime["proxy_url"], "http://custom.proxy:8080")
+            self.assertEqual(runtime["clearance"]["mode"], "manual")
+            self.assertEqual(runtime["clearance"]["cf_clearance"], "manual-token")
+            self.assertIn("timeout_sec", runtime["clearance"])
+            self.assertIn("reset_session_status_codes", runtime)
+
+    def test_env_can_disable_runtime_defaults_for_warp_compose(self) -> None:
+        module = load_script_module()
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "config.json"
+            path.write_text(json.dumps({"auth-key": "secret"}), encoding="utf-8")
+            with patch.dict(
+                os.environ,
+                {
+                    "CHATGPT2API_CONFIG_FILE": str(path),
+                    "CHATGPT2API_PROXY_RUNTIME_ENABLED": "false",
+                    "CHATGPT2API_PROXY_RUNTIME_CLEARANCE_ENABLED": "false",
+                },
+                clear=False,
+            ):
+                self.assertEqual(module.main(), 0)
+
+            runtime = json.loads(path.read_text(encoding="utf-8"))["proxy_runtime"]
+            self.assertFalse(runtime["enabled"])
+            self.assertFalse(runtime["clearance"]["enabled"])
+            self.assertEqual(runtime["clearance"]["mode"], "none")
+
+    def test_bind_mounted_config_file_falls_back_when_atomic_replace_is_busy(self) -> None:
+        module = load_script_module()
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "config.json"
+            path.write_text(json.dumps({"auth-key": "secret", "proxy": ""}), encoding="utf-8")
+            original_replace = Path.replace
+
+            def replace_with_ebusy(self: Path, target: Path) -> Path:
+                if self.name == "config.json.tmp":
+                    raise OSError(errno.EBUSY, "Device or resource busy")
+                return original_replace(self, target)
+
+            with patch.dict(os.environ, {"CHATGPT2API_CONFIG_FILE": str(path)}, clear=False):
+                with patch.object(Path, "replace", replace_with_ebusy):
+                    self.assertEqual(module.main(), 0)
+
+            data = json.loads(path.read_text(encoding="utf-8"))
+            self.assertTrue(data["proxy_runtime"]["enabled"])
+            self.assertFalse(path.with_suffix(".json.tmp").exists())
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_proxy_runtime_api.py

@@ -0,0 +1,181 @@
+from __future__ import annotations
+
+import copy
+import unittest
+from unittest import mock
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+import api.system as system_module
+from services.config import DEFAULT_PROXY_RUNTIME
+
+
+AUTH_HEADERS = {"Authorization": "Bearer chatgpt2api"}
+
+
+class FakeStorage:
+    def get_backend_info(self) -> dict[str, object]:
+        return {"type": "json"}
+
+    def health_check(self) -> dict[str, object]:
+        return {"ok": True}
+
+
+class FakeConfig:
+    def __init__(self) -> None:
+        self.data: dict[str, object] = {
+            "proxy": "",
+            "proxy_runtime": copy.deepcopy(DEFAULT_PROXY_RUNTIME),
+        }
+
+    def get(self) -> dict[str, object]:
+        return copy.deepcopy(self.data)
+
+    def update(self, updates: dict[str, object]) -> dict[str, object]:
+        self.data.update(copy.deepcopy(updates))
+        return self.get()
+
+    def get_proxy_settings(self) -> str:
+        return str(self.data.get("proxy") or "").strip()
+
+    def get_proxy_runtime_settings(self) -> dict[str, object]:
+        return copy.deepcopy(self.data["proxy_runtime"])  # type: ignore[index]
+
+    def get_public_proxy_runtime_settings(self) -> dict[str, object]:
+        return self.get_proxy_runtime_settings()
+
+    def get_storage_backend(self) -> FakeStorage:
+        return FakeStorage()
+
+
+class FakeProxySettings:
+    def get_runtime_status(self) -> dict[str, object]:
+        return {
+            "enabled": True,
+            "egress_mode": "single_proxy",
+            "proxy_source": "proxy_runtime",
+            "has_proxy": True,
+            "clearance_enabled": True,
+            "clearance_mode": "flaresolverr",
+            "has_clearance_bundle": False,
+            "cached_clearance_hosts": [],
+        }
+
+
+class FakeAccountService:
+    def get_stats(self) -> dict[str, object]:
+        return {
+            "total": 1,
+            "cumulative_total": 1,
+            "active": 1,
+            "unlimited_quota_count": 0,
+            "total_quota": 1,
+            "limited": 0,
+            "abnormal": 0,
+            "disabled": 0,
+            "total_success": 0,
+            "total_fail": 0,
+            "by_type": {"web": 1},
+        }
+
+
+class ProxyRuntimeApiTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.fake_config = FakeConfig()
+        self.fake_proxy_settings = FakeProxySettings()
+        self.test_proxy_calls: list[str] = []
+        self.test_clearance_calls: list[str] = []
+
+        def fake_test_proxy(url: str = "") -> dict[str, object]:
+            self.test_proxy_calls.append(url)
+            return {
+                "ok": True,
+                "status": 204,
+                "latency_ms": 12,
+                "error": None,
+                "proxy_source": "proxy_runtime" if not url else "input",
+                "has_proxy": True,
+            }
+
+        def fake_test_clearance(target_url: str = "https://chatgpt.com") -> dict[str, object]:
+            self.test_clearance_calls.append(target_url)
+            return {
+                "ok": True,
+                "status": "ok",
+                "latency_ms": 34,
+                "has_cookies": True,
+                "user_agent": "Flare UA",
+                "error": None,
+                "runtime": self.fake_proxy_settings.get_runtime_status(),
+            }
+
+        self.patchers = [
+            mock.patch.object(system_module, "config", self.fake_config),
+            mock.patch.object(system_module, "require_admin", lambda _authorization: {"role": "admin"}),
+            mock.patch.object(system_module, "test_proxy", fake_test_proxy),
+            mock.patch.object(system_module, "test_clearance", fake_test_clearance, create=True),
+            mock.patch.object(system_module, "proxy_settings", self.fake_proxy_settings, create=True),
+            mock.patch("services.account_service.account_service", FakeAccountService()),
+        ]
+        for patcher in self.patchers:
+            patcher.start()
+            self.addCleanup(patcher.stop)
+        app = FastAPI()
+        app.include_router(system_module.create_router("9.9.9-test"))
+        self.client = TestClient(app)
+
+    def test_proxy_test_can_use_active_runtime_when_url_is_empty(self) -> None:
+        response = self.client.post("/api/proxy/test", headers=AUTH_HEADERS, json={})
+
+        self.assertEqual(response.status_code, 200, response.text)
+        payload = response.json()["result"]
+        self.assertTrue(payload["ok"])
+        self.assertEqual(payload["proxy_source"], "proxy_runtime")
+        self.assertEqual(self.test_proxy_calls, [""])
+
+    def test_proxy_runtime_endpoint_reads_and_updates_runtime_config(self) -> None:
+        get_response = self.client.get("/api/proxy/runtime", headers=AUTH_HEADERS)
+        self.assertEqual(get_response.status_code, 200, get_response.text)
+        self.assertEqual(get_response.json()["runtime"]["enabled"], False)
+        self.assertEqual(get_response.json()["status"]["proxy_source"], "proxy_runtime")
+
+        runtime = copy.deepcopy(DEFAULT_PROXY_RUNTIME)
+        runtime["enabled"] = True
+        runtime["egress_mode"] = "single_proxy"
+        runtime["proxy_url"] = "http://privoxy:8118"
+        runtime["clearance"]["enabled"] = True  # type: ignore[index]
+        runtime["clearance"]["mode"] = "flaresolverr"  # type: ignore[index]
+        runtime["clearance"]["flaresolverr_url"] = "http://flaresolverr:8191"  # type: ignore[index]
+
+        post_response = self.client.post("/api/proxy/runtime", headers=AUTH_HEADERS, json=runtime)
+
+        self.assertEqual(post_response.status_code, 200, post_response.text)
+        self.assertTrue(post_response.json()["runtime"]["enabled"])
+        self.assertEqual(self.fake_config.data["proxy_runtime"], runtime)
+
+    def test_clearance_test_endpoint_runs_clearance_refresh_without_returning_cookie_values(self) -> None:
+        response = self.client.post(
+            "/api/proxy/clearance/test",
+            headers=AUTH_HEADERS,
+            json={"target_url": "https://chatgpt.com/backend-api/models"},
+        )
+
+        self.assertEqual(response.status_code, 200, response.text)
+        payload = response.json()["result"]
+        self.assertTrue(payload["ok"])
+        self.assertTrue(payload["has_cookies"])
+        self.assertNotIn("cf_clearance", response.text)
+        self.assertEqual(self.test_clearance_calls, ["https://chatgpt.com/backend-api/models"])
+
+    def test_health_json_includes_proxy_runtime_status(self) -> None:
+        response = self.client.get("/health?format=json")
+
+        self.assertEqual(response.status_code, 200, response.text)
+        payload = response.json()
+        self.assertEqual(payload["version"], "9.9.9-test")
+        self.assertEqual(payload["proxy_runtime"]["clearance_mode"], "flaresolverr")
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_proxy_runtime_config.py

@@ -0,0 +1,220 @@
+import copy
+import json
+import math
+import tempfile
+import unittest
+from pathlib import Path
+
+from services.config import (
+    DEFAULT_PROXY_RUNTIME,
+    ConfigStore,
+    _normalize_proxy_runtime_settings,
+)
+
+
+class ProxyRuntimeConfigTests(unittest.TestCase):
+    def _make_store(self, initial: dict[str, object] | None = None) -> tuple[tempfile.TemporaryDirectory[str], ConfigStore]:
+        tmp_dir = tempfile.TemporaryDirectory()
+        path = Path(tmp_dir.name) / "config.json"
+        data = {"auth-key": "test-auth"}
+        if initial:
+            data.update(initial)
+        path.write_text(json.dumps(data), encoding="utf-8")
+        return tmp_dir, ConfigStore(path)
+
+    def test_defaults_are_safe_and_included_in_public_config(self) -> None:
+        tmp_dir, store = self._make_store({"proxy": " http://legacy.example:8080 "})
+        with tmp_dir:
+            expected_default = copy.deepcopy(DEFAULT_PROXY_RUNTIME)
+            runtime = store.get_proxy_runtime_settings()
+            self.assertEqual(runtime, expected_default)
+            self.assertEqual(store.get_proxy_settings(), "http://legacy.example:8080")
+
+            expected_public = copy.deepcopy(expected_default)
+            expected_public["clearance"]["has_cf_cookies"] = False
+            expected_public["clearance"]["has_cf_clearance"] = False
+            public_config = store.get()
+            self.assertEqual(public_config["proxy"], " http://legacy.example:8080 ")
+            self.assertEqual(public_config["proxy_runtime"], expected_public)
+            self.assertNotIn("auth-key", public_config)
+
+            runtime["enabled"] = True
+            runtime["clearance"]["enabled"] = True
+            self.assertEqual(DEFAULT_PROXY_RUNTIME, expected_default)
+            self.assertEqual(store.get_proxy_runtime_settings(), expected_default)
+
+    def test_normalize_proxy_runtime_sanitizes_invalid_values(self) -> None:
+        normalized = _normalize_proxy_runtime_settings(
+            {
+                "enabled": "yes",
+                "egress_mode": "tor",
+                "proxy_url": "  http://proxy.example:8080  ",
+                "resource_proxy_url": "  socks5://resource.example:1080  ",
+                "skip_ssl_verify": "on",
+                "reset_session_status_codes": ["403", 429, 99, 600, "bad", None, True],
+                "clearance": {
+                    "enabled": "0",
+                    "mode": "manual",
+                    "cf_cookies": "  a=b; c=d  ",
+                    "cf_clearance": "  token  ",
+                    "user_agent": "  Custom UA  ",
+                    "browser": "  firefox  ",
+                    "flaresolverr_url": "  http://flare.example/  ",
+                    "timeout_sec": 0,
+                    "refresh_interval": 59,
+                    "warm_up_on_start": "true",
+                },
+            }
+        )
+
+        self.assertTrue(normalized["enabled"])
+        self.assertEqual(normalized["egress_mode"], "direct")
+        self.assertEqual(normalized["proxy_url"], "http://proxy.example:8080")
+        self.assertEqual(normalized["resource_proxy_url"], "socks5://resource.example:1080")
+        self.assertTrue(normalized["skip_ssl_verify"])
+        self.assertEqual(normalized["reset_session_status_codes"], [403, 429])
+
+        clearance = normalized["clearance"]
+        self.assertFalse(clearance["enabled"])
+        self.assertEqual(clearance["mode"], "manual")
+        self.assertEqual(clearance["cf_cookies"], "a=b; c=d")
+        self.assertEqual(clearance["cf_clearance"], "token")
+        self.assertEqual(clearance["user_agent"], "Custom UA")
+        self.assertEqual(clearance["browser"], "firefox")
+        self.assertEqual(clearance["flaresolverr_url"], "http://flare.example/")
+        self.assertEqual(clearance["timeout_sec"], 1)
+        self.assertEqual(clearance["refresh_interval"], 60)
+        self.assertTrue(clearance["warm_up_on_start"])
+
+    def test_normalize_proxy_runtime_uses_defaults_for_missing_or_empty_values(self) -> None:
+        self.assertEqual(_normalize_proxy_runtime_settings(None), DEFAULT_PROXY_RUNTIME)
+        self.assertEqual(
+            _normalize_proxy_runtime_settings(
+                {
+                    "egress_mode": "single_proxy",
+                    "reset_session_status_codes": ["bad", 99, 600],
+                    "clearance": {
+                        "enabled": True,
+                        "mode": "invalid",
+                        "user_agent": "",
+                        "browser": "",
+                        "timeout_sec": "bad",
+                        "refresh_interval": "bad",
+                    },
+                }
+            ),
+            {
+                **DEFAULT_PROXY_RUNTIME,
+                "egress_mode": "single_proxy",
+                "reset_session_status_codes": [403],
+                "clearance": {
+                    **DEFAULT_PROXY_RUNTIME["clearance"],
+                    "enabled": True,
+                    "mode": "none",
+                },
+            },
+        )
+
+    def test_malformed_proxy_runtime_values_fall_back_safely(self) -> None:
+        normalized = _normalize_proxy_runtime_settings(
+            {
+                "enabled": "maybe",
+                "skip_ssl_verify": "maybe",
+                "reset_session_status_codes": [math.inf, "NaN", "403"],
+                "clearance": {
+                    "enabled": "maybe",
+                    "timeout_sec": math.inf,
+                    "refresh_interval": -math.inf,
+                    "warm_up_on_start": "maybe",
+                },
+            }
+        )
+
+        self.assertFalse(normalized["enabled"])
+        self.assertFalse(normalized["skip_ssl_verify"])
+        self.assertEqual(normalized["reset_session_status_codes"], [403])
+        clearance = normalized["clearance"]
+        self.assertFalse(clearance["enabled"])
+        self.assertEqual(clearance["timeout_sec"], DEFAULT_PROXY_RUNTIME["clearance"]["timeout_sec"])
+        self.assertEqual(clearance["refresh_interval"], DEFAULT_PROXY_RUNTIME["clearance"]["refresh_interval"])
+        self.assertFalse(clearance["warm_up_on_start"])
+
+    def test_update_normalizes_and_persists_proxy_runtime(self) -> None:
+        tmp_dir, store = self._make_store()
+        with tmp_dir:
+            public_config = store.update(
+                {
+                    "proxy_runtime": {
+                        "enabled": "true",
+                        "egress_mode": "single_proxy",
+                        "proxy_url": "  http://proxy.example  ",
+                        "reset_session_status_codes": [401, "403", "nope"],
+                        "clearance": {
+                            "enabled": "yes",
+                            "mode": "flaresolverr",
+                            "flaresolverr_url": " http://localhost:8191 ",
+                            "timeout_sec": "30",
+                            "refresh_interval": "120",
+                        },
+                    }
+                }
+            )
+
+            expected = {
+                **DEFAULT_PROXY_RUNTIME,
+                "enabled": True,
+                "egress_mode": "single_proxy",
+                "proxy_url": "http://proxy.example",
+                "reset_session_status_codes": [401, 403],
+                "clearance": {
+                    **DEFAULT_PROXY_RUNTIME["clearance"],
+                    "enabled": True,
+                    "mode": "flaresolverr",
+                    "flaresolverr_url": "http://localhost:8191",
+                    "timeout_sec": 30,
+                    "refresh_interval": 120,
+                },
+            }
+            expected_public = copy.deepcopy(expected)
+            expected_public["clearance"]["cf_cookies"] = ""
+            expected_public["clearance"]["cf_clearance"] = ""
+            expected_public["clearance"]["has_cf_cookies"] = False
+            expected_public["clearance"]["has_cf_clearance"] = False
+            self.assertEqual(public_config["proxy_runtime"], expected_public)
+
+            raw_saved = json.loads(store.path.read_text(encoding="utf-8"))
+            self.assertEqual(raw_saved["proxy_runtime"], expected)
+            reloaded = ConfigStore(store.path)
+            self.assertEqual(reloaded.get_proxy_runtime_settings(), expected)
+
+    def test_public_proxy_runtime_redacts_and_preserves_existing_clearance_values(self) -> None:
+        existing = copy.deepcopy(DEFAULT_PROXY_RUNTIME)
+        existing["enabled"] = True
+        existing["clearance"]["enabled"] = True
+        existing["clearance"]["mode"] = "manual"
+        existing["clearance"]["cf_cookies"] = "session=secret-cookie"
+        existing["clearance"]["cf_clearance"] = "secret-clearance"
+        tmp_dir, store = self._make_store({"proxy_runtime": existing})
+        with tmp_dir:
+            public_config = store.get()
+            public_runtime = public_config["proxy_runtime"]
+            public_clearance = public_runtime["clearance"]
+            self.assertEqual(public_clearance["cf_cookies"], "")
+            self.assertEqual(public_clearance["cf_clearance"], "")
+            self.assertTrue(public_clearance["has_cf_cookies"])
+            self.assertTrue(public_clearance["has_cf_clearance"])
+            self.assertNotIn("secret-cookie", json.dumps(public_config))
+            self.assertNotIn("secret-clearance", json.dumps(public_config))
+
+            public_clearance["user_agent"] = "Updated UA"
+            updated_public = store.update({"proxy_runtime": public_runtime})
+            updated_raw = json.loads(store.path.read_text(encoding="utf-8"))["proxy_runtime"]
+            self.assertEqual(updated_raw["clearance"]["cf_cookies"], "session=secret-cookie")
+            self.assertEqual(updated_raw["clearance"]["cf_clearance"], "secret-clearance")
+            self.assertEqual(updated_raw["clearance"]["user_agent"], "Updated UA")
+            self.assertEqual(updated_public["proxy_runtime"]["clearance"]["cf_cookies"], "")
+            self.assertEqual(updated_public["proxy_runtime"]["clearance"]["cf_clearance"], "")
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_proxy_service.py

@@ -0,0 +1,415 @@
+import copy
+import json
+import threading
+import time
+import unittest
+from concurrent.futures import ThreadPoolExecutor
+from unittest.mock import patch
+from urllib.parse import urlparse
+
+from services.config import DEFAULT_PROXY_RUNTIME
+from services.proxy_service import (
+    ClearanceBundle,
+    FlareSolverrClearanceProvider,
+    ProxySettingsStore,
+    normalize_proxy_url,
+)
+
+
+class FakeConfig:
+    def __init__(self, legacy_proxy: str = "", runtime: dict[str, object] | None = None) -> None:
+        self.legacy_proxy = legacy_proxy
+        self.runtime = runtime if runtime is not None else copy.deepcopy(DEFAULT_PROXY_RUNTIME)
+
+    def get_proxy_settings(self) -> str:
+        return self.legacy_proxy
+
+    def get_proxy_runtime_settings(self) -> dict[str, object]:
+        return copy.deepcopy(self.runtime)
+
+
+def make_runtime(**overrides: object) -> dict[str, object]:
+    runtime = copy.deepcopy(DEFAULT_PROXY_RUNTIME)
+    clearance = overrides.pop("clearance", None)
+    runtime.update(overrides)
+    if clearance is not None:
+        runtime["clearance"].update(clearance)  # type: ignore[index,union-attr]
+    return runtime
+
+
+class ProxyServiceTests(unittest.TestCase):
+    def test_normalize_proxy_url_strips_and_converts_socks_schemes(self) -> None:
+        self.assertEqual(normalize_proxy_url("  http://proxy.example:8080  "), "http://proxy.example:8080")
+        self.assertEqual(normalize_proxy_url("\thttps://proxy.example:8443\n"), "https://proxy.example:8443")
+        self.assertEqual(normalize_proxy_url(" socks://proxy.example:1080 "), "socks5h://proxy.example:1080")
+        self.assertEqual(normalize_proxy_url("socks5://proxy.example:1080"), "socks5h://proxy.example:1080")
+        self.assertEqual(normalize_proxy_url(" socks5h://proxy.example:1080 "), "socks5h://proxy.example:1080")
+        self.assertEqual(normalize_proxy_url("   "), "")
+
+    def test_build_session_kwargs_keeps_legacy_global_proxy_when_runtime_disabled(self) -> None:
+        store = ProxySettingsStore(FakeConfig(legacy_proxy="  http://legacy.example:8080  "))
+
+        kwargs = store.build_session_kwargs(impersonate="chrome")
+
+        self.assertEqual(kwargs["impersonate"], "chrome")
+        self.assertEqual(kwargs["proxy"], "http://legacy.example:8080")
+
+    def test_runtime_proxy_is_limited_to_upstream_scope_by_default(self) -> None:
+        runtime = make_runtime(enabled=True, egress_mode="single_proxy", proxy_url="http://runtime.example:8080")
+        store = ProxySettingsStore(FakeConfig(legacy_proxy="http://legacy.example:8080", runtime=runtime))
+
+        self.assertEqual(
+            store.build_session_kwargs()["proxy"],
+            "http://legacy.example:8080",
+        )
+        self.assertEqual(
+            store.build_session_kwargs(upstream=True)["proxy"],
+            "http://runtime.example:8080",
+        )
+
+    def test_account_proxy_wins_over_runtime_and_global_proxy(self) -> None:
+        runtime = make_runtime(enabled=True, egress_mode="single_proxy", proxy_url="http://runtime.example:8080")
+        store = ProxySettingsStore(FakeConfig(legacy_proxy="http://legacy.example:8080", runtime=runtime))
+
+        kwargs = store.build_session_kwargs(account={"proxy": " socks://account.example:1080 "}, upstream=True)
+
+        self.assertEqual(kwargs["proxy"], "socks5h://account.example:1080")
+
+    def test_proxy_runtime_single_proxy_wins_over_explicit_and_legacy_proxy_when_enabled(self) -> None:
+        runtime = make_runtime(enabled=True, egress_mode="single_proxy", proxy_url=" socks5://runtime.example:1080 ")
+        store = ProxySettingsStore(FakeConfig(legacy_proxy="http://legacy.example:8080", runtime=runtime))
+
+        kwargs = store.build_session_kwargs(proxy="http://explicit.example:8080", upstream=True)
+
+        self.assertEqual(kwargs["proxy"], "socks5h://runtime.example:1080")
+
+    def test_explicit_proxy_wins_over_legacy_global_proxy_when_runtime_disabled(self) -> None:
+        store = ProxySettingsStore(FakeConfig(legacy_proxy="http://legacy.example:8080"))
+
+        kwargs = store.build_session_kwargs(proxy=" socks5://explicit.example:1080 ")
+
+        self.assertEqual(kwargs["proxy"], "socks5h://explicit.example:1080")
+
+    def test_resource_requests_use_resource_proxy_url_when_configured(self) -> None:
+        runtime = make_runtime(
+            enabled=True,
+            egress_mode="single_proxy",
+            proxy_url="http://runtime.example:8080",
+            resource_proxy_url=" socks5://resource.example:1080 ",
+        )
+        store = ProxySettingsStore(FakeConfig(legacy_proxy="http://legacy.example:8080", runtime=runtime))
+
+        kwargs = store.build_session_kwargs(resource=True, upstream=True)
+
+        self.assertEqual(kwargs["proxy"], "socks5h://resource.example:1080")
+
+    def test_manual_clearance_merges_cookies_and_preserves_explicit_user_agent(self) -> None:
+        runtime = make_runtime(
+            enabled=True,
+            egress_mode="single_proxy",
+            proxy_url="http://runtime.example:8080",
+            clearance={
+                "enabled": True,
+                "mode": "manual",
+                "cf_cookies": "foo=bar; session=ok",
+                "cf_clearance": "manual-token",
+                "user_agent": "Manual UA",
+            },
+        )
+        store = ProxySettingsStore(FakeConfig(runtime=runtime))
+
+        headers = store.build_headers(headers={"Cookie": "existing=1"}, target_url="https://chatgpt.com/backend-api")
+
+        self.assertEqual(headers["User-Agent"], "Manual UA")
+        self.assertEqual(headers["Cookie"], "existing=1; foo=bar; session=ok; cf_clearance=manual-token")
+
+        headers_with_ua = store.build_headers(
+            headers={"User-Agent": "Caller UA", "Cookie": "cf_clearance=caller-token"},
+            target_url="https://chatgpt.com/backend-api",
+        )
+
+        self.assertEqual(headers_with_ua["User-Agent"], "Caller UA")
+        self.assertEqual(headers_with_ua["Cookie"], "cf_clearance=caller-token; foo=bar; session=ok")
+        self.assertNotIn("cf_clearance=manual-token", headers_with_ua["Cookie"])
+
+    def test_flaresolverr_provider_parses_solution_and_filters_cookies_by_host(self) -> None:
+        calls: list[tuple[str, dict[str, object], dict[str, str], float]] = []
+
+        def fake_request(endpoint: str, body: bytes, headers: dict[str, str], timeout: float) -> bytes:
+            payload = json.loads(body.decode("utf-8"))
+            calls.append((endpoint, payload, headers, timeout))
+            return json.dumps(
+                {
+                    "status": "ok",
+                    "solution": {
+                        "userAgent": "Flare UA",
+                        "cookies": [
+                            {"name": "cf_clearance", "value": "host-token", "domain": ".chatgpt.com"},
+                            {"name": "no_domain", "value": "kept"},
+                            {"name": "wrong_host", "value": "dropped", "domain": "example.net"},
+                        ],
+                    },
+                }
+            ).encode("utf-8")
+
+        provider = FlareSolverrClearanceProvider("http://flare.local/", request_method=fake_request)
+
+        bundle = provider.get_clearance(
+            "https://chatgpt.com/backend-api/conversation",
+            proxy_url="socks5h://proxy.example:1080",
+            timeout_sec=12,
+        )
+
+        self.assertIsNotNone(bundle)
+        assert bundle is not None
+        endpoint, payload, headers, timeout = calls[0]
+        self.assertEqual(endpoint, "http://flare.local/v1")
+        self.assertEqual(payload["cmd"], "request.get")
+        self.assertEqual(payload["url"], "https://chatgpt.com/backend-api/conversation")
+        self.assertEqual(payload["maxTimeout"], 12000)
+        self.assertEqual(payload["proxy"], {"url": "socks5h://proxy.example:1080"})
+        self.assertEqual(headers["Content-Type"], "application/json")
+        self.assertEqual(timeout, 12)
+        self.assertEqual(bundle.user_agent, "Flare UA")
+        self.assertEqual(bundle.cookies, {"cf_clearance": "host-token", "no_domain": "kept"})
+
+    def test_flaresolverr_provider_keeps_only_matching_or_no_domain_cookies(self) -> None:
+        def fake_request(endpoint: str, body: bytes, headers: dict[str, str], timeout: float) -> bytes:
+            return json.dumps(
+                {
+                    "status": "ok",
+                    "solution": {
+                        "userAgent": "Filtered UA",
+                        "cookies": [
+                            {"name": "wrong_host", "value": "dropped", "domain": "example.net"},
+                            {"name": "also_wrong", "value": "dropped-too", "domain": ".example.org"},
+                            {"name": "no_domain", "value": "kept"},
+                        ],
+                    },
+                }
+            ).encode("utf-8")
+
+        provider = FlareSolverrClearanceProvider("http://flare.local", request_method=fake_request)
+
+        bundle = provider.get_clearance("https://chatgpt.com", timeout_sec=5)
+
+        self.assertIsNotNone(bundle)
+        assert bundle is not None
+        self.assertEqual(bundle.cookies, {"no_domain": "kept"})
+        self.assertEqual(bundle.user_agent, "Filtered UA")
+
+    def test_flaresolverr_provider_drops_all_wrong_domain_cookies(self) -> None:
+        def fake_request(endpoint: str, body: bytes, headers: dict[str, str], timeout: float) -> bytes:
+            return json.dumps(
+                {
+                    "status": "ok",
+                    "solution": {
+                        "cookies": [
+                            {"name": "wrong_host", "value": "dropped", "domain": "example.net"},
+                        ],
+                    },
+                }
+            ).encode("utf-8")
+
+        provider = FlareSolverrClearanceProvider("http://flare.local", request_method=fake_request)
+
+        self.assertIsNone(provider.get_clearance("https://chatgpt.com", timeout_sec=5))
+
+    def test_cached_flaresolverr_bundle_is_merged_by_build_headers_and_can_be_invalidated(self) -> None:
+        runtime = make_runtime(
+            enabled=True,
+            egress_mode="single_proxy",
+            proxy_url="http://runtime.example:8080",
+            clearance={
+                "enabled": True,
+                "mode": "flaresolverr",
+                "flaresolverr_url": "http://flare.local",
+                "timeout_sec": 5,
+            },
+        )
+        bundle = ClearanceBundle(
+            target_host="chatgpt.com",
+            proxy_url="http://runtime.example:8080",
+            cookies={"cf_clearance": "flare-token"},
+            user_agent="Flare UA",
+        )
+
+        class FakeProvider:
+            def get_clearance(self, target_url: str, proxy_url: str = "", timeout_sec: int = 60) -> ClearanceBundle | None:
+                return bundle
+
+        store = ProxySettingsStore(FakeConfig(runtime=runtime), clearance_provider_factory=lambda _url: FakeProvider())
+        store.refresh_clearance(target_url="https://chatgpt.com", force=True)
+
+        headers = store.build_headers(headers={"Cookie": "existing=1"}, target_url="https://chatgpt.com/backend-api")
+
+        self.assertEqual(headers["User-Agent"], "Flare UA")
+        self.assertEqual(headers["Cookie"], "existing=1; cf_clearance=flare-token")
+
+        store.invalidate_clearance(target_url="https://chatgpt.com")
+        self.assertEqual(
+            store.build_headers(headers={"Cookie": "existing=1"}, target_url="https://chatgpt.com/backend-api"),
+            {"Cookie": "existing=1"},
+        )
+
+    def test_flaresolverr_refresh_failure_keeps_old_cached_bundle(self) -> None:
+        runtime = make_runtime(
+            enabled=True,
+            egress_mode="single_proxy",
+            proxy_url="http://runtime.example:8080",
+            clearance={
+                "enabled": True,
+                "mode": "flaresolverr",
+                "flaresolverr_url": "http://flare.local",
+                "timeout_sec": 5,
+            },
+        )
+        first_bundle = ClearanceBundle(
+            target_host="chatgpt.com",
+            proxy_url="http://runtime.example:8080",
+            cookies={"cf_clearance": "old-token"},
+            user_agent="Old UA",
+        )
+
+        class FakeProvider:
+            def __init__(self) -> None:
+                self.calls = 0
+
+            def get_clearance(self, target_url: str, proxy_url: str = "", timeout_sec: int = 60) -> ClearanceBundle | None:
+                self.calls += 1
+                return first_bundle if self.calls == 1 else None
+
+        provider = FakeProvider()
+        store = ProxySettingsStore(FakeConfig(runtime=runtime), clearance_provider_factory=lambda _url: provider)
+
+        refreshed = store.refresh_clearance(target_url="https://chatgpt.com", force=True)
+        fallback = store.refresh_clearance(target_url="https://chatgpt.com", force=True)
+
+        self.assertIsNotNone(refreshed)
+        assert refreshed is not None
+        self.assertEqual(refreshed.cookies, first_bundle.cookies)
+        self.assertIsNotNone(refreshed.expires_at)
+        assert refreshed.expires_at is not None
+        self.assertAlmostEqual(refreshed.expires_at - refreshed.created_at, 3600, places=2)
+        self.assertIs(fallback, refreshed)
+        self.assertEqual(provider.calls, 2)
+
+    def test_flaresolverr_bundle_expires_after_refresh_interval(self) -> None:
+        runtime = make_runtime(
+            enabled=True,
+            egress_mode="single_proxy",
+            proxy_url="http://runtime.example:8080",
+            clearance={
+                "enabled": True,
+                "mode": "flaresolverr",
+                "flaresolverr_url": "http://flare.local",
+                "timeout_sec": 5,
+                "refresh_interval": 60,
+            },
+        )
+
+        class FakeProvider:
+            def get_clearance(self, target_url: str, proxy_url: str = "", timeout_sec: int = 60) -> ClearanceBundle | None:
+                return ClearanceBundle(
+                    target_host="chatgpt.com",
+                    proxy_url=proxy_url,
+                    cookies={"cf_clearance": "expiring-token"},
+                    user_agent="Expiring UA",
+                    created_at=1000,
+                )
+
+        store = ProxySettingsStore(FakeConfig(runtime=runtime), clearance_provider_factory=lambda _url: FakeProvider())
+        with patch("services.proxy_service.time.time", return_value=1000):
+            bundle = store.refresh_clearance(target_url="https://chatgpt.com", force=True)
+
+        self.assertIsNotNone(bundle)
+        assert bundle is not None
+        self.assertEqual(bundle.expires_at, 1060)
+        self.assertTrue(bundle.is_valid_for("chatgpt.com", "http://runtime.example:8080", now=1059))
+        self.assertFalse(bundle.is_valid_for("chatgpt.com", "http://runtime.example:8080", now=1060))
+
+    def test_profile_repr_does_not_expose_clearance_secrets(self) -> None:
+        runtime = make_runtime(
+            enabled=True,
+            clearance={
+                "enabled": True,
+                "mode": "manual",
+                "cf_cookies": "foo=secret-cookie",
+                "cf_clearance": "secret-clearance",
+            },
+        )
+        profile = ProxySettingsStore(FakeConfig(runtime=runtime)).get_profile()
+
+        text = repr(profile)
+
+        self.assertNotIn("secret-cookie", text)
+        self.assertNotIn("secret-clearance", text)
+
+    def test_proxy_test_error_redacts_proxy_credentials(self) -> None:
+        class FailingSession:
+            def __init__(self, **kwargs: object) -> None:
+                pass
+
+            def get(self, *args: object, **kwargs: object) -> object:
+                raise RuntimeError("proxy failed for http://user:pass@proxy.example:8080")
+
+            def close(self) -> None:
+                pass
+
+        with patch("services.proxy_service.Session", FailingSession):
+            result = __import__("services.proxy_service", fromlist=["test_proxy"]).test_proxy(
+                "http://user:pass@proxy.example:8080"
+            )
+
+        self.assertFalse(result["ok"])
+        self.assertIn("[REDACTED]", result["error"])
+        self.assertNotIn("user:pass", result["error"])
+
+    def test_concurrent_flaresolverr_refresh_uses_single_flight_per_proxy_and_host(self) -> None:
+        runtime = make_runtime(
+            enabled=True,
+            egress_mode="single_proxy",
+            proxy_url="http://runtime.example:8080",
+            clearance={
+                "enabled": True,
+                "mode": "flaresolverr",
+                "flaresolverr_url": "http://flare.local",
+                "timeout_sec": 5,
+            },
+        )
+
+        class SlowProvider:
+            def __init__(self) -> None:
+                self.calls = 0
+                self.lock = threading.Lock()
+
+            def get_clearance(self, target_url: str, proxy_url: str = "", timeout_sec: int = 60) -> ClearanceBundle | None:
+                with self.lock:
+                    self.calls += 1
+                time.sleep(0.15)
+                return ClearanceBundle(
+                    target_host=urlparse(target_url).hostname or "",
+                    proxy_url=proxy_url,
+                    cookies={"cf_clearance": "thread-token"},
+                    user_agent="Thread UA",
+                )
+
+        provider = SlowProvider()
+        store = ProxySettingsStore(FakeConfig(runtime=runtime), clearance_provider_factory=lambda _url: provider)
+        workers = 6
+        start = threading.Barrier(workers)
+
+        def refresh() -> ClearanceBundle | None:
+            start.wait(timeout=5)
+            return store.refresh_clearance(target_url="https://chatgpt.com", force=True)
+
+        with ThreadPoolExecutor(max_workers=workers) as executor:
+            results = list(executor.map(lambda _index: refresh(), range(workers)))
+
+        self.assertEqual(provider.calls, 1)
+        self.assertTrue(all(result is results[0] for result in results))
+        self.assertIsNotNone(results[0])
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_register_proxy_runtime.py

@@ -0,0 +1,194 @@
+import unittest
+from unittest.mock import patch
+
+from services.proxy_service import ClearanceBundle
+from services.register import openai_register
+
+
+class FakeResponse:
+    def __init__(self, status_code=200, text="", headers=None, url="https://auth.openai.com/test"):
+        self.status_code = status_code
+        self.text = text
+        self.headers = headers or {}
+        self.url = url
+
+    def json(self):
+        return {}
+
+
+class FakeCookieJar:
+    def __init__(self):
+        self.items = []
+
+    def set(self, name, value, domain=None):
+        self.items.append({"name": name, "value": value, "domain": domain})
+
+
+class FakeSession:
+    def __init__(self, **kwargs):
+        self.kwargs = kwargs
+        self.headers = {}
+        self.cookies = FakeCookieJar()
+        self.closed = False
+
+    def close(self):
+        self.closed = True
+
+
+class FakeProxySettings:
+    def __init__(self, bundle=None):
+        self.bundle = bundle
+        self.refreshed = False
+        self.session_kwargs_calls = []
+        self.build_headers_calls = []
+        self.refresh_calls = []
+
+    def build_session_kwargs(self, **kwargs):
+        self.session_kwargs_calls.append(kwargs)
+        return dict(kwargs, proxy="http://runtime.example:8118")
+
+    def build_headers(self, headers=None, target_url="", proxy="", upstream=True, **kwargs):
+        self.build_headers_calls.append({"target_url": target_url, "proxy": proxy, "upstream": upstream})
+        merged = dict(headers or {})
+        if self.refreshed and self.bundle and self.bundle.cookies:
+            merged["Cookie"] = "; ".join(f"{key}={value}" for key, value in self.bundle.cookies.items())
+        return merged
+
+    def refresh_clearance(self, target_url="", proxy="", force=False, upstream=True, **kwargs):
+        self.refresh_calls.append({"target_url": target_url, "proxy": proxy, "force": force, "upstream": upstream})
+        self.refreshed = self.bundle is not None
+        return self.bundle
+
+
+class RegisterProxyRuntimeTests(unittest.TestCase):
+    def test_create_session_uses_proxy_settings_without_breaking_existing_proxy_argument(self):
+        fake_proxy = FakeProxySettings()
+        created = []
+
+        def fake_session_factory(**kwargs):
+            session = FakeSession(**kwargs)
+            created.append(session)
+            return session
+
+        with patch.object(openai_register, "proxy_settings", fake_proxy), patch.object(
+            openai_register.requests,
+            "Session",
+            side_effect=fake_session_factory,
+        ):
+            session = openai_register.create_session("http://legacy-register.example:8080")
+
+        self.assertIs(session, created[0])
+        self.assertEqual(fake_proxy.session_kwargs_calls[0]["proxy"], "http://legacy-register.example:8080")
+        self.assertTrue(fake_proxy.session_kwargs_calls[0]["upstream"])
+        self.assertEqual(fake_proxy.session_kwargs_calls[0]["impersonate"], "chrome")
+        self.assertFalse(fake_proxy.session_kwargs_calls[0]["verify"])
+        self.assertEqual(session.kwargs["proxy"], "http://runtime.example:8118")
+
+    def test_cloudflare_without_clearance_keeps_clear_register_error(self):
+        fake_proxy = FakeProxySettings(bundle=None)
+        cf_response = FakeResponse(
+            status_code=403,
+            text="<html><title>Just a moment...</title></html>",
+            headers={"server": "cloudflare", "content-type": "text/html"},
+            url="https://auth.openai.com/api/accounts/authorize",
+        )
+
+        with patch.object(openai_register, "proxy_settings", fake_proxy), patch.object(
+            openai_register,
+            "create_session",
+            return_value=FakeSession(),
+        ), patch.object(openai_register, "request_with_local_retry", return_value=(cf_response, "")):
+            registrar = openai_register.PlatformRegistrar(proxy="http://legacy-register.example:8080")
+            with self.assertRaisesRegex(RuntimeError, "Cloudflare") as ctx:
+                registrar._platform_authorize("user@example.com", 1)
+
+        self.assertEqual(len(fake_proxy.refresh_calls), 1)
+        self.assertIn("status=403", str(ctx.exception))
+        self.assertIn("Just a moment", str(ctx.exception))
+
+    def test_openai_html_behind_cloudflare_is_not_treated_as_challenge(self):
+        response = FakeResponse(
+            status_code=200,
+            text="""
+            <!DOCTYPE html><html lang=\"en-US\"><head>
+            <title>Create a password - OpenAI</title>
+            </head><body>OpenAI account page</body></html>
+            """,
+            headers={"server": "cloudflare", "content-type": "text/html; charset=utf-8"},
+            url="https://auth.openai.com/create-account/password",
+        )
+
+        self.assertFalse(openai_register._is_cloudflare_challenge(response))
+
+    def test_cloudflare_challenge_refreshes_clearance_and_retries_once_with_matching_headers(self):
+        bundle = ClearanceBundle(
+            target_host="auth.openai.com",
+            proxy_url="http://runtime.example:8118",
+            cookies={"cf_clearance": "flare-token"},
+            user_agent="Flare UA",
+        )
+        fake_proxy = FakeProxySettings(bundle=bundle)
+        responses = [
+            FakeResponse(
+                status_code=403,
+                text="<html><title>Just a moment...</title></html>",
+                headers={"server": "cloudflare", "content-type": "text/html"},
+                url="https://auth.openai.com/api/accounts/authorize",
+            ),
+            FakeResponse(status_code=200, text="{}", headers={"content-type": "application/json"}),
+        ]
+        request_calls = []
+
+        def fake_request(session, method, url, retry_attempts=3, **kwargs):
+            request_calls.append({"method": method, "url": url, "headers": dict(kwargs.get("headers") or {})})
+            return responses.pop(0), ""
+
+        with patch.object(openai_register, "proxy_settings", fake_proxy), patch.object(
+            openai_register,
+            "create_session",
+            return_value=FakeSession(),
+        ), patch.object(openai_register, "request_with_local_retry", side_effect=fake_request):
+            registrar = openai_register.PlatformRegistrar(proxy="http://legacy-register.example:8080")
+            registrar._platform_authorize("user@example.com", 1)
+
+        self.assertEqual(len(request_calls), 2)
+        self.assertEqual(len(fake_proxy.refresh_calls), 1)
+        retry_headers = {key.lower(): value for key, value in request_calls[1]["headers"].items()}
+        self.assertEqual(retry_headers["user-agent"], "Flare UA")
+        self.assertEqual(retry_headers["cookie"], "cf_clearance=flare-token")
+        self.assertEqual(fake_proxy.refresh_calls[0]["target_url"], openai_register.auth_base)
+        self.assertEqual(fake_proxy.refresh_calls[0]["proxy"], "http://legacy-register.example:8080")
+        self.assertTrue(fake_proxy.refresh_calls[0]["force"])
+
+    def test_refresh_failure_reports_cloudflare_detail_without_infinite_retry(self):
+        fake_proxy = FakeProxySettings(bundle=None)
+        cf_response = FakeResponse(
+            status_code=403,
+            text="<html><title>Just a moment...</title><body>challenge body</body></html>",
+            headers={"server": "cloudflare", "content-type": "text/html"},
+            url="https://auth.openai.com/api/accounts/authorize",
+        )
+        request_calls = []
+
+        def fake_request(session, method, url, retry_attempts=3, **kwargs):
+            request_calls.append({"method": method, "url": url})
+            return cf_response, ""
+
+        with patch.object(openai_register, "proxy_settings", fake_proxy), patch.object(
+            openai_register,
+            "create_session",
+            return_value=FakeSession(),
+        ), patch.object(openai_register, "request_with_local_retry", side_effect=fake_request):
+            registrar = openai_register.PlatformRegistrar(proxy="")
+            with self.assertRaisesRegex(RuntimeError, "Cloudflare") as ctx:
+                registrar._platform_authorize("user@example.com", 1)
+
+        self.assertEqual(len(request_calls), 1)
+        self.assertEqual(len(fake_proxy.refresh_calls), 1)
+        message = str(ctx.exception)
+        self.assertIn("status=403", message)
+        self.assertIn("challenge body", message)
+
+
+if __name__ == "__main__":
+    unittest.main()

web/src/app/settings/components/proxy-runtime-card.tsx

@@ -0,0 +1,359 @@
+"use client";
+
+import { Cookie, LoaderCircle, PlugZap, ShieldCheck } from "lucide-react";
+import { useState } from "react";
+import { toast } from "sonner";
+
+import { Button } from "@/components/ui/button";
+import { Card, CardContent } from "@/components/ui/card";
+import { Checkbox } from "@/components/ui/checkbox";
+import { Input } from "@/components/ui/input";
+import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select";
+import { Textarea } from "@/components/ui/textarea";
+import {
+  testProxy,
+  testProxyClearance,
+  type ClearanceTestResult,
+  type ProxyRuntimeClearanceMode,
+  type ProxyRuntimeEgressMode,
+  type ProxyTestResult,
+} from "@/lib/api";
+
+import { useSettingsStore } from "../store";
+
+export function ProxyRuntimeCard() {
+  const [isTestingProxy, setIsTestingProxy] = useState(false);
+  const [isTestingClearance, setIsTestingClearance] = useState(false);
+  const [proxyResult, setProxyResult] = useState<ProxyTestResult | null>(null);
+  const [clearanceResult, setClearanceResult] = useState<ClearanceTestResult | null>(null);
+  const [targetUrl, setTargetUrl] = useState("https://chatgpt.com");
+  const config = useSettingsStore((state) => state.config);
+  const isLoadingConfig = useSettingsStore((state) => state.isLoadingConfig);
+  const saveConfig = useSettingsStore((state) => state.saveConfig);
+  const setProxyRuntimeField = useSettingsStore((state) => state.setProxyRuntimeField);
+  const setProxyRuntimeClearanceField = useSettingsStore((state) => state.setProxyRuntimeClearanceField);
+  const setProxyRuntimeStatusCodesText = useSettingsStore((state) => state.setProxyRuntimeStatusCodesText);
+
+  if (isLoadingConfig || !config?.proxy_runtime) {
+    return (
+      <Card className="rounded-2xl border-white/80 bg-white/90 shadow-sm">
+        <CardContent className="flex items-center justify-center p-10">
+          <LoaderCircle className="size-5 animate-spin text-stone-400" />
+        </CardContent>
+      </Card>
+    );
+  }
+
+  const runtime = config.proxy_runtime;
+  const clearance = runtime.clearance;
+  const runtimeEnabled = Boolean(runtime.enabled);
+  const clearanceMode = clearance.mode;
+  const hasStoredClearance = Boolean(clearance.has_cf_cookies || clearance.has_cf_clearance);
+
+  const handleTestRuntimeProxy = async () => {
+    setIsTestingProxy(true);
+    setProxyResult(null);
+    try {
+      const saved = await saveConfig();
+      if (!saved) {
+        return;
+      }
+      const data = await testProxy();
+      setProxyResult(data.result);
+      if (data.result.ok) {
+        toast.success(`稳定代理可用（${data.result.latency_ms} ms，HTTP ${data.result.status}）`);
+      } else {
+        toast.error(`稳定代理不可用：${data.result.error ?? "未知错误"}`);
+      }
+    } catch (error) {
+      toast.error(error instanceof Error ? error.message : "测试稳定代理失败");
+    } finally {
+      setIsTestingProxy(false);
+    }
+  };
+
+  const handleTestClearance = async () => {
+    setIsTestingClearance(true);
+    setClearanceResult(null);
+    try {
+      const saved = await saveConfig();
+      if (!saved) {
+        return;
+      }
+      const data = await testProxyClearance(targetUrl.trim() || "https://chatgpt.com");
+      setClearanceResult(data.result);
+      if (data.result.ok) {
+        toast.success(`Clearance 获取成功（${data.result.latency_ms} ms）`);
+      } else {
+        toast.error(`Clearance 获取失败：${data.result.error ?? data.result.status}`);
+      }
+    } catch (error) {
+      toast.error(error instanceof Error ? error.message : "测试 Clearance 失败");
+    } finally {
+      setIsTestingClearance(false);
+    }
+  };
+
+  return (
+    <Card className="rounded-2xl border-white/80 bg-white/90 shadow-sm">
+      <CardContent className="space-y-5 p-6">
+        <div className="flex flex-col gap-3 sm:flex-row sm:items-start sm:justify-between">
+          <div>
+            <div className="flex items-center gap-2 text-base font-semibold text-stone-900">
+              <PlugZap className="size-5 text-stone-500" />
+              稳定代理运行时
+            </div>
+            <p className="mt-1 text-xs leading-6 text-stone-500">
+              默认关闭。启用后仅接管上游 OpenAI/ChatGPT 请求，可配合 WARP / Privoxy / FlareSolverr 提升注册和生图链路稳定性。
+            </p>
+          </div>
+          <span className={`rounded-full px-3 py-1 text-xs ${runtimeEnabled ? "bg-emerald-50 text-emerald-700" : "bg-stone-100 text-stone-500"}`}>
+            {runtimeEnabled ? "已启用" : "未启用"}
+          </span>
+        </div>
+
+        <div className="rounded-xl border border-stone-200 bg-stone-50 px-4 py-3 text-xs leading-6 text-stone-600">
+          代理优先级：账号代理 &gt; 稳定代理运行时 &gt; 显式代理 &gt; 全局代理。Cookie / cf_clearance 不会在接口响应中明文返回。
+        </div>
+
+        <div className="grid gap-4 md:grid-cols-2">
+          <label className="flex items-center gap-3 rounded-xl border border-stone-200 bg-white px-4 py-3 text-sm text-stone-700 md:col-span-2">
+            <Checkbox
+              checked={runtimeEnabled}
+              onCheckedChange={(checked) => setProxyRuntimeField("enabled", Boolean(checked))}
+            />
+            启用稳定代理运行时
+          </label>
+
+          <div className="space-y-2">
+            <label className="text-sm text-stone-700">出站模式</label>
+            <Select
+              value={runtime.egress_mode}
+              onValueChange={(value) => setProxyRuntimeField("egress_mode", value as ProxyRuntimeEgressMode)}
+              disabled={!runtimeEnabled}
+            >
+              <SelectTrigger className="h-10 rounded-xl border-stone-200 bg-white shadow-none">
+                <SelectValue />
+              </SelectTrigger>
+              <SelectContent>
+                <SelectItem value="direct">直连</SelectItem>
+                <SelectItem value="single_proxy">单代理/WARP</SelectItem>
+              </SelectContent>
+            </Select>
+            <p className="text-xs text-stone-500">WARP compose 默认使用 single_proxy。</p>
+          </div>
+
+          <div className="space-y-2">
+            <label className="text-sm text-stone-700">运行时代理 URL</label>
+            <Input
+              value={runtime.proxy_url}
+              onChange={(event) => setProxyRuntimeField("proxy_url", event.target.value)}
+              placeholder="http://privoxy:8118"
+              className="h-10 rounded-xl border-stone-200 bg-white"
+              disabled={!runtimeEnabled || runtime.egress_mode !== "single_proxy"}
+            />
+            <p className="text-xs text-stone-500">支持 http/https/socks，socks5 会转为 socks5h。</p>
+          </div>
+
+          <div className="space-y-2">
+            <label className="text-sm text-stone-700">资源代理 URL</label>
+            <Input
+              value={runtime.resource_proxy_url}
+              onChange={(event) => setProxyRuntimeField("resource_proxy_url", event.target.value)}
+              placeholder="留空则复用运行时代理"
+              className="h-10 rounded-xl border-stone-200 bg-white"
+              disabled={!runtimeEnabled || runtime.egress_mode !== "single_proxy"}
+            />
+          </div>
+
+          <div className="space-y-2">
+            <label className="text-sm text-stone-700">重置会话状态码</label>
+            <Input
+              value={runtime.reset_session_status_codes.join(",")}
+              onChange={(event) => setProxyRuntimeStatusCodesText(event.target.value)}
+              placeholder="403"
+              className="h-10 rounded-xl border-stone-200 bg-white"
+              disabled={!runtimeEnabled}
+            />
+            <p className="text-xs text-stone-500">默认 403，只对 Cloudflare/挑战类错误触发。</p>
+          </div>
+
+          <label className="flex items-center gap-3 rounded-xl border border-stone-200 bg-white px-4 py-3 text-sm text-stone-700">
+            <Checkbox
+              checked={Boolean(runtime.skip_ssl_verify)}
+              onCheckedChange={(checked) => setProxyRuntimeField("skip_ssl_verify", Boolean(checked))}
+              disabled={!runtimeEnabled}
+            />
+            跳过 SSL 校验
+          </label>
+
+          <div className="flex items-end justify-end">
+            <Button
+              type="button"
+              variant="outline"
+              className="h-9 rounded-xl border-stone-200 bg-white px-4 text-stone-700"
+              onClick={() => void handleTestRuntimeProxy()}
+              disabled={isTestingProxy || !runtimeEnabled}
+            >
+              {isTestingProxy ? <LoaderCircle className="size-4 animate-spin" /> : <PlugZap className="size-4" />}
+              测试当前运行时代理
+            </Button>
+          </div>
+
+          {proxyResult ? (
+            <div className={`rounded-xl border px-3 py-2 text-xs leading-6 md:col-span-2 ${proxyResult.ok ? "border-emerald-200 bg-emerald-50 text-emerald-800" : "border-rose-200 bg-rose-50 text-rose-800"}`}>
+              {proxyResult.ok
+                ? `代理可用：HTTP ${proxyResult.status}，用时 ${proxyResult.latency_ms} ms，来源 ${proxyResult.proxy_source ?? "unknown"}`
+                : `代理不可用：${proxyResult.error ?? "未知错误"}（用时 ${proxyResult.latency_ms} ms）`}
+            </div>
+          ) : null}
+        </div>
+
+        <div className="space-y-4 rounded-xl border border-stone-200 bg-white px-4 py-3">
+          <div className="flex flex-col gap-3 sm:flex-row sm:items-center sm:justify-between">
+            <div className="flex items-center gap-2 text-sm font-medium text-stone-800">
+              <Cookie className="size-4 text-stone-500" />
+              Cloudflare Clearance
+            </div>
+            <span className={`rounded-full px-3 py-1 text-xs ${clearance.enabled ? "bg-emerald-50 text-emerald-700" : "bg-stone-100 text-stone-500"}`}>
+              {clearance.enabled ? clearanceMode : "disabled"}
+            </span>
+          </div>
+
+          <div className="grid gap-4 md:grid-cols-2">
+            <div className="space-y-2">
+              <label className="text-sm text-stone-700">Clearance 模式</label>
+              <Select
+                value={clearanceMode}
+                onValueChange={(value) => {
+                  const mode = value as ProxyRuntimeClearanceMode;
+                  setProxyRuntimeClearanceField("mode", mode);
+                  setProxyRuntimeClearanceField("enabled", mode !== "none");
+                }}
+                disabled={!runtimeEnabled}
+              >
+                <SelectTrigger className="h-10 rounded-xl border-stone-200 bg-white shadow-none">
+                  <SelectValue />
+                </SelectTrigger>
+                <SelectContent>
+                  <SelectItem value="none">不启用</SelectItem>
+                  <SelectItem value="manual">手动 Cookie</SelectItem>
+                  <SelectItem value="flaresolverr">FlareSolverr</SelectItem>
+                </SelectContent>
+              </Select>
+            </div>
+
+            <div className="space-y-2">
+              <label className="text-sm text-stone-700">FlareSolverr URL</label>
+              <Input
+                value={clearance.flaresolverr_url}
+                onChange={(event) => setProxyRuntimeClearanceField("flaresolverr_url", event.target.value)}
+                placeholder="http://flaresolverr:8191"
+                className="h-10 rounded-xl border-stone-200 bg-white"
+                disabled={!runtimeEnabled || clearanceMode !== "flaresolverr"}
+              />
+            </div>
+
+            <div className="space-y-2 md:col-span-2">
+              <label className="text-sm text-stone-700">User-Agent</label>
+              <Input
+                value={clearance.user_agent}
+                onChange={(event) => setProxyRuntimeClearanceField("user_agent", event.target.value)}
+                className="h-10 rounded-xl border-stone-200 bg-white font-mono text-xs"
+                disabled={!runtimeEnabled || clearanceMode === "none"}
+              />
+            </div>
+
+            <div className="space-y-2">
+              <label className="text-sm text-stone-700">超时秒数</label>
+              <Input
+                value={String(clearance.timeout_sec)}
+                onChange={(event) => setProxyRuntimeClearanceField("timeout_sec", event.target.value)}
+                placeholder="60"
+                className="h-10 rounded-xl border-stone-200 bg-white"
+                disabled={!runtimeEnabled || clearanceMode === "none"}
+              />
+            </div>
+
+            <div className="space-y-2">
+              <label className="text-sm text-stone-700">刷新间隔秒数</label>
+              <Input
+                value={String(clearance.refresh_interval)}
+                onChange={(event) => setProxyRuntimeClearanceField("refresh_interval", event.target.value)}
+                placeholder="3600"
+                className="h-10 rounded-xl border-stone-200 bg-white"
+                disabled={!runtimeEnabled || clearanceMode === "none"}
+              />
+            </div>
+
+            <div className="space-y-2 md:col-span-2">
+              <label className="text-sm text-stone-700">手动 Cookie</label>
+              <Textarea
+                value={clearance.cf_cookies}
+                onChange={(event) => setProxyRuntimeClearanceField("cf_cookies", event.target.value)}
+                placeholder="可选：foo=bar; cf_clearance=..."
+                className="min-h-24 rounded-xl border-stone-200 bg-white font-mono text-xs shadow-none"
+                disabled={!runtimeEnabled || clearanceMode !== "manual"}
+              />
+              <p className="text-xs text-stone-500">
+                {hasStoredClearance ? "服务端已保存过 Cookie/clearance；留空保存不会清空已有值。" : "留空表示不使用手动 Cookie。"}
+              </p>
+            </div>
+
+            <div className="space-y-2 md:col-span-2">
+              <label className="text-sm text-stone-700">单独 cf_clearance</label>
+              <Input
+                value={clearance.cf_clearance}
+                onChange={(event) => setProxyRuntimeClearanceField("cf_clearance", event.target.value)}
+                placeholder="可选：只填写 cf_clearance 值"
+                className="h-10 rounded-xl border-stone-200 bg-white font-mono text-xs"
+                disabled={!runtimeEnabled || clearanceMode !== "manual"}
+              />
+            </div>
+
+            <label className="flex items-center gap-3 rounded-xl border border-stone-200 bg-stone-50 px-4 py-3 text-sm text-stone-700">
+              <Checkbox
+                checked={Boolean(clearance.warm_up_on_start)}
+                onCheckedChange={(checked) => setProxyRuntimeClearanceField("warm_up_on_start", Boolean(checked))}
+                disabled={!runtimeEnabled || clearanceMode === "none"}
+              />
+              启动时预热 Clearance
+            </label>
+
+            <div className="space-y-2">
+              <label className="text-sm text-stone-700">测试目标 URL</label>
+              <Input
+                value={targetUrl}
+                onChange={(event) => setTargetUrl(event.target.value)}
+                placeholder="https://chatgpt.com"
+                className="h-10 rounded-xl border-stone-200 bg-white"
+                disabled={!runtimeEnabled || clearanceMode === "none"}
+              />
+            </div>
+
+            <div className="flex justify-end md:col-span-2">
+              <Button
+                type="button"
+                variant="outline"
+                className="h-9 rounded-xl border-stone-200 bg-white px-4 text-stone-700"
+                onClick={() => void handleTestClearance()}
+                disabled={isTestingClearance || !runtimeEnabled || clearanceMode === "none"}
+              >
+                {isTestingClearance ? <LoaderCircle className="size-4 animate-spin" /> : <ShieldCheck className="size-4" />}
+                测试 Clearance
+              </Button>
+            </div>
+
+            {clearanceResult ? (
+              <div className={`rounded-xl border px-3 py-2 text-xs leading-6 md:col-span-2 ${clearanceResult.ok ? "border-emerald-200 bg-emerald-50 text-emerald-800" : "border-rose-200 bg-rose-50 text-rose-800"}`}>
+                {clearanceResult.ok
+                  ? `Clearance 可用：${clearanceResult.has_cookies ? "已获取 Cookie" : "无 Cookie"}，用时 ${clearanceResult.latency_ms} ms`
+                  : `Clearance 不可用：${clearanceResult.error ?? clearanceResult.status}（用时 ${clearanceResult.latency_ms} ms）`}
+              </div>
+            ) : null}
+          </div>
+        </div>
+      </CardContent>
+    </Card>
+  );
+}

web/src/app/settings/page.tsx

@@ -10,6 +10,7 @@ import { ConfigCard } from "./components/config-card";
 import { CPAPoolDialog } from "./components/cpa-pool-dialog";
 import { CPAPoolsCard } from "./components/cpa-pools-card";
 import { ImportBrowserDialog } from "./components/import-browser-dialog";
+import { ProxyRuntimeCard } from "./components/proxy-runtime-card";
 import { SettingsHeader } from "./components/settings-header";
 import { Sub2APIConnections } from "./components/sub2api-connections";
 import { UserKeysCard } from "./components/user-keys-card";
@@ -66,6 +67,7 @@ function SettingsPageContent() {
       <SettingsHeader />
       <section className="space-y-6">
         <ConfigCard />
+        <ProxyRuntimeCard />
         <BackupSettingsCard />
         <UserKeysCard />
         <CPAPoolsCard />

web/src/app/settings/store.ts

@@ -31,6 +31,9 @@ import {
   type CPARemoteFile,
   type ImageStorageMode,
   type ImageStorageSettings,
+  type ProxyRuntimeClearanceMode,
+  type ProxyRuntimeEgressMode,
+  type ProxyRuntimeSettings,
   type RegisterConfig,
   type SettingsConfig,
 } from "@/lib/api";
@@ -39,6 +42,71 @@ export const PAGE_SIZE_OPTIONS = ["50", "100", "200"] as const;
 
 export type PageSizeOption = (typeof PAGE_SIZE_OPTIONS)[number];
 
+const DEFAULT_PROXY_RUNTIME: ProxyRuntimeSettings = {
+  enabled: false,
+  egress_mode: "direct",
+  proxy_url: "",
+  resource_proxy_url: "",
+  skip_ssl_verify: false,
+  reset_session_status_codes: [403],
+  clearance: {
+    enabled: false,
+    mode: "none",
+    cf_cookies: "",
+    cf_clearance: "",
+    user_agent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/145.0.0.0 Safari/537.36",
+    browser: "chrome",
+    flaresolverr_url: "",
+    timeout_sec: 60,
+    refresh_interval: 3600,
+    warm_up_on_start: false,
+    has_cf_cookies: false,
+    has_cf_clearance: false,
+  },
+};
+
+function normalizeProxyRuntime(value: unknown): ProxyRuntimeSettings {
+  const source = typeof value === "object" && value !== null ? value as Partial<ProxyRuntimeSettings> : {};
+  const clearanceSource = typeof source.clearance === "object" && source.clearance !== null
+    ? source.clearance as Partial<ProxyRuntimeSettings["clearance"]>
+    : {};
+  const egressMode = source.egress_mode === "single_proxy" ? "single_proxy" : "direct";
+  const clearanceMode: ProxyRuntimeClearanceMode = clearanceSource.mode === "manual" || clearanceSource.mode === "flaresolverr"
+    ? clearanceSource.mode
+    : "none";
+  const statusCodes = Array.isArray(source.reset_session_status_codes)
+    ? source.reset_session_status_codes
+      .map((item) => Number(item))
+      .filter((item) => Number.isInteger(item) && item >= 100 && item <= 599)
+    : [];
+  return {
+    ...DEFAULT_PROXY_RUNTIME,
+    ...source,
+    enabled: Boolean(source.enabled),
+    egress_mode: egressMode as ProxyRuntimeEgressMode,
+    proxy_url: String(source.proxy_url || ""),
+    resource_proxy_url: String(source.resource_proxy_url || ""),
+    skip_ssl_verify: Boolean(source.skip_ssl_verify),
+    reset_session_status_codes: statusCodes.length > 0 ? statusCodes : [403],
+    clearance: {
+      ...DEFAULT_PROXY_RUNTIME.clearance,
+      ...clearanceSource,
+      enabled: Boolean(clearanceSource.enabled),
+      mode: clearanceMode,
+      cf_cookies: String(clearanceSource.cf_cookies || ""),
+      cf_clearance: String(clearanceSource.cf_clearance || ""),
+      user_agent: String(clearanceSource.user_agent || DEFAULT_PROXY_RUNTIME.clearance.user_agent),
+      browser: String(clearanceSource.browser || "chrome"),
+      flaresolverr_url: String(clearanceSource.flaresolverr_url || ""),
+      timeout_sec: Number(clearanceSource.timeout_sec || 60),
+      refresh_interval: Number(clearanceSource.refresh_interval || 3600),
+      warm_up_on_start: Boolean(clearanceSource.warm_up_on_start),
+      has_cf_cookies: Boolean(clearanceSource.has_cf_cookies),
+      has_cf_clearance: Boolean(clearanceSource.has_cf_clearance),
+    },
+  };
+}
+
 function normalizeConfig(config: SettingsConfig): SettingsConfig {
   const imageStorage = typeof config.image_storage === "object" && config.image_storage
     ? config.image_storage as ImageStorageSettings
@@ -116,6 +184,7 @@ function normalizeConfig(config: SettingsConfig): SettingsConfig {
       webdav_root_path: String(imageStorage.webdav_root_path || "chatgpt2api/images"),
       public_base_url: String(imageStorage.public_base_url || ""),
     },
+    proxy_runtime: normalizeProxyRuntime(config.proxy_runtime),
     backup: {
       ...backup,
       enabled: Boolean(backup.enabled),
@@ -224,6 +293,9 @@ type SettingsStore = {
   setSensitiveWordsText: (value: string) => void;
   setAIReviewField: (key: "enabled" | "base_url" | "api_key" | "model" | "prompt", value: string | boolean) => void;
   setImageStorageField: (key: keyof ImageStorageSettings, value: string | boolean) => void;
+  setProxyRuntimeField: <K extends keyof ProxyRuntimeSettings>(key: K, value: ProxyRuntimeSettings[K]) => void;
+  setProxyRuntimeClearanceField: <K extends keyof ProxyRuntimeSettings["clearance"]>(key: K, value: ProxyRuntimeSettings["clearance"][K]) => void;
+  setProxyRuntimeStatusCodesText: (value: string) => void;
   testImageStorage: () => Promise<void>;
   syncImagesToWebDAV: () => Promise<void>;
   setBackupField: (key: keyof BackupSettings, value: string | boolean) => void;
@@ -379,6 +451,26 @@ export const useSettingsStore = create<SettingsStore>((set, get) => ({
           webdav_root_path: String(config.image_storage?.webdav_root_path || "chatgpt2api/images").trim(),
           public_base_url: String(config.image_storage?.public_base_url || "").trim(),
         },
+        proxy_runtime: {
+          ...normalizeProxyRuntime(config.proxy_runtime),
+          proxy_url: String(config.proxy_runtime?.proxy_url || "").trim(),
+          resource_proxy_url: String(config.proxy_runtime?.resource_proxy_url || "").trim(),
+          reset_session_status_codes: normalizeProxyRuntime({
+            reset_session_status_codes: (config.proxy_runtime?.reset_session_status_codes || [403])
+              .map((item) => Number(item))
+              .filter((item) => Number.isInteger(item) && item >= 100 && item <= 599),
+          }).reset_session_status_codes,
+          clearance: {
+            ...normalizeProxyRuntime(config.proxy_runtime).clearance,
+            cf_cookies: String(config.proxy_runtime?.clearance?.cf_cookies || "").trim(),
+            cf_clearance: String(config.proxy_runtime?.clearance?.cf_clearance || "").trim(),
+            user_agent: String(config.proxy_runtime?.clearance?.user_agent || DEFAULT_PROXY_RUNTIME.clearance.user_agent).trim(),
+            browser: String(config.proxy_runtime?.clearance?.browser || "chrome").trim(),
+            flaresolverr_url: String(config.proxy_runtime?.clearance?.flaresolverr_url || "").trim(),
+            timeout_sec: Math.max(1, Number(config.proxy_runtime?.clearance?.timeout_sec) || 60),
+            refresh_interval: Math.max(60, Number(config.proxy_runtime?.clearance?.refresh_interval) || 3600),
+          },
+        },
         backup: {
           ...(config.backup as BackupSettings),
           account_id: String(config.backup?.account_id || "").trim(),
@@ -532,6 +624,69 @@ export const useSettingsStore = create<SettingsStore>((set, get) => ({
     });
   },
 
+  setProxyRuntimeField: (key, value) => {
+    set((state) => {
+      if (!state.config) {
+        return {};
+      }
+      const runtime = normalizeProxyRuntime(state.config.proxy_runtime);
+      const nextRuntime = normalizeProxyRuntime({
+        ...runtime,
+        [key]: value,
+      });
+      return {
+        config: {
+          ...state.config,
+          proxy_runtime: nextRuntime,
+        },
+      };
+    });
+  },
+
+  setProxyRuntimeClearanceField: (key, value) => {
+    set((state) => {
+      if (!state.config) {
+        return {};
+      }
+      const runtime = normalizeProxyRuntime(state.config.proxy_runtime);
+      const nextRuntime = normalizeProxyRuntime({
+        ...runtime,
+        clearance: {
+          ...runtime.clearance,
+          [key]: value,
+        },
+      });
+      return {
+        config: {
+          ...state.config,
+          proxy_runtime: nextRuntime,
+        },
+      };
+    });
+  },
+
+  setProxyRuntimeStatusCodesText: (value) => {
+    const codes = value
+      .split(/[,\s]+/)
+      .map((item) => Number(item.trim()))
+      .filter((item) => Number.isInteger(item) && item >= 100 && item <= 599);
+    set((state) => {
+      if (!state.config) {
+        return {};
+      }
+      const runtime = normalizeProxyRuntime(state.config.proxy_runtime);
+      return {
+        config: {
+          ...state.config,
+          proxy_runtime: normalizeProxyRuntime({
+            ...runtime,
+            reset_session_status_codes: codes.length > 0 ? codes : [403],
+          }),
+        },
+      };
+    });
+  },
+
   testImageStorage: async () => {
     set({ isTestingImageStorage: true });
     try {

web/src/lib/api.ts

@@ -101,6 +101,50 @@ type AccountUpdateResponse = {
   items: Account[];
 };
 
+export type ProxyRuntimeEgressMode = "direct" | "single_proxy";
+export type ProxyRuntimeClearanceMode = "none" | "manual" | "flaresolverr";
+
+export type ProxyRuntimeClearanceSettings = {
+  enabled: boolean;
+  mode: ProxyRuntimeClearanceMode;
+  cf_cookies: string;
+  cf_clearance: string;
+  user_agent: string;
+  browser: string;
+  flaresolverr_url: string;
+  timeout_sec: number | string;
+  refresh_interval: number | string;
+  warm_up_on_start: boolean;
+  has_cf_cookies?: boolean;
+  has_cf_clearance?: boolean;
+};
+
+export type ProxyRuntimeSettings = {
+  enabled: boolean;
+  egress_mode: ProxyRuntimeEgressMode;
+  proxy_url: string;
+  resource_proxy_url: string;
+  skip_ssl_verify: boolean;
+  reset_session_status_codes: number[];
+  clearance: ProxyRuntimeClearanceSettings;
+};
+
+export type ProxyRuntimeStatus = {
+  enabled: boolean;
+  egress_mode: ProxyRuntimeEgressMode | string;
+  proxy_source: string;
+  has_proxy: boolean;
+  clearance_enabled: boolean;
+  clearance_mode: ProxyRuntimeClearanceMode | string;
+  has_clearance_bundle: boolean;
+  cached_clearance_hosts: string[];
+};
+
+export type ProxyRuntimeResponse = {
+  runtime: ProxyRuntimeSettings;
+  status: ProxyRuntimeStatus;
+};
+
 export type SettingsConfig = {
   proxy: string;
   base_url?: string;
@@ -127,6 +171,7 @@ export type SettingsConfig = {
   auto_relogin_after_refresh?: boolean;
   log_levels?: string[];
   image_storage?: ImageStorageSettings;
+  proxy_runtime?: ProxyRuntimeSettings;
   backup?: BackupSettings;
   backup_state?: BackupState;
   [key: string]: unknown;
@@ -892,6 +937,18 @@ export type ProxyTestResult = {
   status: number;
   latency_ms: number;
   error: string | null;
+  proxy_source?: string;
+  has_proxy?: boolean;
+};
+
+export type ClearanceTestResult = {
+  ok: boolean;
+  status: string;
+  latency_ms: number;
+  has_cookies: boolean;
+  user_agent: string;
+  error: string | null;
+  runtime: ProxyRuntimeStatus;
 };
 
 export async function fetchProxy() {
@@ -911,3 +968,21 @@ export async function testProxy(url?: string) {
     body: { url: url ?? "" },
   });
 }
+
+export async function fetchProxyRuntime() {
+  return httpRequest<ProxyRuntimeResponse>("/api/proxy/runtime");
+}
+
+export async function updateProxyRuntime(runtime: ProxyRuntimeSettings) {
+  return httpRequest<ProxyRuntimeResponse>("/api/proxy/runtime", {
+    method: "POST",
+    body: runtime,
+  });
+}
+
+export async function testProxyClearance(targetUrl?: string) {
+  return httpRequest<{ result: ClearanceTestResult }>("/api/proxy/clearance/test", {
+    method: "POST",
+    body: { target_url: targetUrl ?? "https://chatgpt.com" },
+  });
+}