From 6aae7b65eaee00c9c076bd47d6838cd694ffec81 Mon Sep 17 00:00:00 2001 From: luotianyi Date: Thu, 8 Oct 2026 00:19:39 +0800 Subject: [PATCH] =?UTF-8?q?feat(proxypool):=20=E7=AB=99=E7=82=B9=E7=BA=A7?= =?UTF-8?q?=E5=87=BA=E5=8F=A3=E4=BB=A3=E7=90=86=E6=B1=A0=EF=BC=88Xray=20/?= =?UTF-8?q?=20=E8=AE=A2=E9=98=85=20/=20=E7=AB=AF=E7=82=B9=EF=BC=8C?= =?UTF-8?q?=E7=B2=98=E6=80=A7=E9=80=89=E8=B7=AF=E4=B8=8E=E5=81=A5=E5=BA=B7?= =?UTF-8?q?=E6=8E=A2=E6=B5=8B=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - proxypool: 站点级出口绑定 proxies = [条目名, "sub:订阅名"]，粘性选路、 失败切换、后台健康探测；支持 Xray 分享链接 / 订阅 / 裸 ip:port 与 socks5 端点， 托管 Xray 核心（可自动下载）；站点未绑定时回退 [proxy] 主/备线路 - upstream: RouteSource 抽象 + routesFor/poolReport；求解与请求共用同一出口 - admin: 代理池面板与维护动作；overview 暴露池状态 --- README.md | 7 +- cmd/dsfree2api/main.go | 14 +- config.example.toml | 34 ++ internal/admin/actions.go | 256 +++++++++ internal/admin/server.go | 16 +- internal/admin/web/app.js | 137 +++++ internal/config/config.go | 65 +++ internal/config/config_test.go | 72 +++ internal/proxypool/manager.go | 722 ++++++++++++++++++++++++++ internal/proxypool/manager_test.go | 76 +++ internal/proxypool/parse.go | 517 ++++++++++++++++++ internal/proxypool/parse_test.go | 145 ++++++ internal/proxypool/xray.go | 498 ++++++++++++++++++ internal/proxypool/xray_smoke_test.go | 80 +++ internal/upstream/client.go | 65 ++- internal/upstream/client_test.go | 18 +- 16 files changed, 2701 insertions(+), 21 deletions(-) create mode 100644 internal/proxypool/manager.go create mode 100644 internal/proxypool/manager_test.go create mode 100644 internal/proxypool/parse.go create mode 100644 internal/proxypool/parse_test.go create mode 100644 internal/proxypool/xray.go create mode 100644 internal/proxypool/xray_smoke_test.go diff --git a/README.md b/README.md index ea36bce..18b101e 100644 --- a/README.md +++ b/README.md @@ -35,6 +35,7 @@ Go 实现，编译为**单个静态二进制**，无运行时依赖；内置 **W - **三站点聚合**：单站点每日额度耗尽自动切换到其他站点的同款模型，额度耗尽的站点临时降级 10 分钟 - **站点额度管理**：额度按访客身份按日计算，控制台站点卡片可一键查询当前额度、重置访客身份（等同新开隐私窗口、额度即刻回满）；请求撞额度时也会自动轮换访客身份恢复，无需重新求解 - **代理线路**：主线路 + 多条备用线路，支持 `http://` `https://` `socks5://`；slow-start 检测 + 失败降级 + 流中失败不重放 +- **代理池**：Xray 核心（分享链接 / 订阅 / `ip:port` 端点；核心可自动下载）+ 粘性选路 + 后台健康探测；每个站点可绑定独立出口，Turnstile 求解与请求共用同一出口 - **Turnstile 自动突破**：三种 Token 获取方式 — 求解服务 API（`provider="api"`）、本地浏览器自动求解（`provider="browser"`，CDP 直连 Chrome/Edge，无需 Playwright/Docker）、手动导入 Cookie（`provider="manual"`）；结果按「代理线路 × 站点」缓存，TTL 过期自动重解 - **会话自愈**：nonce / data-config 缓存 + 配额耗尽 / 验证失败时自动刷新重试 - **Web 管理台**：仪表盘、模型与站点、API Key、代理线路、Turnstile、对话调试、实时日志、设置 @@ -139,9 +140,10 @@ dsfree2api -config config.toml -check | `[admin]` | 管理台 `enabled` / `host` / `port` / `password`（留空启动时生成随机密码） | | `[limits]` | `max_concurrent_per_site` 每站点并发、`rate_per_minute` 每 Key 每分钟限流（0 = 不限） | | `[proxy]` | `url` 主线路、`fallback_urls` 备用线路、`slow_start_seconds` 首事件超时 | +| `[proxypool]` | 代理池总开关、健康检查间隔 / 超时 / URL、裸 `ip:port` 默认协议、Xray 路径 / 版本 / 自动下载；`[proxypool.entries.*]` 手动节点（分享链接或端点）、`[proxypool.subscriptions.*]` 订阅自动拉取 | | `[upstream]` | 超时、配置缓存 TTL、会话自动刷新、`cross_site_failover` 跨站切换、`continue_rounds` 工具调用截断续写轮数（0 = 不续写、直接截断补全） | | `[turnstile]` | **默认 `enabled = false` 且不内置任何求解服务**；`provider` 三选一：`api`（填 `api_url` / `api_key`）、`browser`（填 `browser_path` 指向本机 Chrome/Edge）、`manual`（只用控制台导入的 Cookie），另有 Cookie TTL、重试次数 | -| `[sites.*]` | 三个站点的 `base_url` / `ajax_url` / `sitekey` / `language` | +| `[sites.*]` | 三个站点的 `base_url` / `ajax_url` / `sitekey` / `language` / `proxies`（站点级出口绑定，按顺序粘性选路） | | `[models.*]` | 模型到 `site` + `upstream_id` + `bot_id` / `post_id` 的映射 | ### Turnstile 说明 @@ -265,7 +267,7 @@ curl http://127.0.0.1:8000/v1/messages \ | 仪表盘 | 请求数 / 错误 / Token / P50·P95、模型用量、站点状态（含 Cookie TTL 进度条）、最近请求 | | 模型与站点 | 编辑标签、路径、bot_id/post_id，启停模型与站点，修改站点 URL / 语言 | | API Key | 生成 / 删除 / 复制下游 key，立即写回配置 | -| 代理线路 | 主线路、备用线路增删、连通性测试、slow-start / 并发 / 限流 / 跨站切换 | +| 代理线路 | 主线路、备用线路增删、连通性测试、slow-start / 并发 / 限流 / 跨站切换、代理池（节点 / 订阅 / 每站点绑定 / Xray 核心状态） | | Turnstile | 三种 Token 获取方式的模式面板（求解服务 API / 本地浏览器自动获取 / 手动导入 Cookie）、各站 Cookie 状态与 TTL、强制刷新、求解历史 | | 对话调试 | 选模型 / 系统提示 / tools JSON，流式或非流式，带连接计时与取消 | | 实时日志 | 级别过滤 + SSE 实时推送 | @@ -302,6 +304,7 @@ curl http://127.0.0.1:8000/v1/messages \ cmd/dsfree2api 入口：配置加载、日志、双 HTTP server、优雅退出 internal/config TOML 结构 + 默认值 + env 覆盖 + 校验 + 写回 internal/httpx TLS 指纹会话封装（bogdanfinn/tls-client, Chrome_120） +internal/proxypool 代理池：Xray 核心管理 / 订阅拉取 / 分享链接解析 / 粘性选路与健康探测 internal/openai OpenAI 协议类型、prompt 拼接、tool_calls 解析 internal/anthropic Anthropic Messages 协议类型 + 与 OpenAI 消息的互转 internal/turnstile Turnstile 三种求解方式（API / 浏览器 CDP / 手动）+ 按线路/站点的 Cookie 缓存 diff --git a/cmd/dsfree2api/main.go b/cmd/dsfree2api/main.go index 127a38e..be9b393 100644 --- a/cmd/dsfree2api/main.go +++ b/cmd/dsfree2api/main.go @@ -18,6 +18,7 @@ import ( "github.com/nyoungo/dsfree2api/internal/config" "github.com/nyoungo/dsfree2api/internal/logbuf" "github.com/nyoungo/dsfree2api/internal/metrics" + "github.com/nyoungo/dsfree2api/internal/proxypool" "github.com/nyoungo/dsfree2api/internal/turnstile" "github.com/nyoungo/dsfree2api/internal/upstream" ) @@ -75,9 +76,18 @@ func main() { defer met.Close() ts := turnstile.New(cfg.Turnstile, cfg.Upstream.UserAgent) - up := upstream.New(cfg, ts, logger) + + // Proxy pool: per-site egress endpoints (Xray share links, subscriptions, + // plain http/socks5) with sticky selection. No-op unless enabled. + pool := proxypool.New(cfg, dataDir, logger) + poolCtx, poolCancel := context.WithCancel(context.Background()) + defer poolCancel() + pool.Start(poolCtx) + + up := upstream.New(cfg, ts, logger, pool) + apiSrv := api.New(cfg, up, met, logs, ts, logger) - adminSrv := admin.New(cfg, up, met, logs, ts, logger) + adminSrv := admin.New(cfg, up, met, logs, ts, logger, pool) mainAddr := fmt.Sprintf("%s:%d", cfg.Server.Host, cfg.Server.Port) mainHTTP := &http.Server{ diff --git a/config.example.toml b/config.example.toml index f9a12e2..df7bf61 100644 --- a/config.example.toml +++ b/config.example.toml @@ -33,6 +33,37 @@ fallback_urls = [] # 主线路允许的首事件等待秒数（>0 且存在备用线路时生效）。 slow_start_seconds = 8 +[proxypool] +# 代理池：为每个站点绑定独立出口（Xray 分享链接 / 订阅 / ip:port 端点）。 +# 选路为粘性：优先使用绑定列表中第一个健康节点，失败才切换；未绑定或 +# 全部不健康时回退到上面的 [proxy] 主/备线路。Turnstile 求解与请求共用 +# 同一出口（Cookie 与出口 IP 绑定，换出口后会自动重新求解）。 +enabled = false +check_interval_seconds = 120 # 健康检查间隔 +check_timeout_seconds = 10 # 单次探测超时 +check_url = "https://www.gstatic.com/generate_204" +default_scheme = "http" # 裸 ip:port 的默认协议（http / socks5） +# Xray 核心：留空 xray_path 且开启自动下载时，首次使用会下载到数据目录 +# （data/xray.exe；核心配置写到 data/config.json）。分享链接（vless / vmess +# / trojan / ss）由它承载；纯 ip:port / socks5:// 端点无需核心。 +xray_path = "" +xray_version = "" # 留空 = 最新版 +xray_auto_download = true + +# 手动节点（名称限字母 / 数字 / - / _）： +# [proxypool.entries.node-a] +# link = "vless://uuid@host:443?type=ws&security=tls&sni=example.com&path=/ws#node-a" +# link = "socks5://user:pass@1.2.3.4:1080" +# link = "1.2.3.4:8080" # 使用 default_scheme +# link = "1.2.3.4:8080:user:pass" +# enabled = true # 缺省即启用 + +# 订阅（按 interval_minutes 自动刷新；节点名形如 <订阅名>#xxxxxxxx）： +# [proxypool.subscriptions.home] +# url = "https://example.com/api/v1/client/subscribe?token=…" +# interval_minutes = 60 +# enabled = true + [upstream] timeout = 60.0 stream_timeout = 300.0 @@ -80,6 +111,9 @@ base_url = "https://deepseek.de" ajax_url = "https://deepseek.de/wp-admin/admin-ajax.php" sitekey = "0x4AAAAAADlLZ3ljqZP6cQwq" language = "de-DE,de;q=0.9,en-US;q=0.8,en;q=0.7" +# 站点级出口绑定（代理池条目名或 "sub:订阅名"，按顺序粘性选路）； +# 留空 = 使用 [proxy] 主/备线路。三个站点可各自独立配置。 +# proxies = ["node-a", "sub:home"] [sites.es] base_url = "https://deepseek.es" diff --git a/internal/admin/actions.go b/internal/admin/actions.go index ca4ac99..199770b 100644 --- a/internal/admin/actions.go +++ b/internal/admin/actions.go @@ -9,12 +9,14 @@ import ( "fmt" "io" "net/http" + "regexp" "strings" "time" "github.com/nyoungo/dsfree2api/internal/config" "github.com/nyoungo/dsfree2api/internal/httpx" "github.com/nyoungo/dsfree2api/internal/openai" + "github.com/nyoungo/dsfree2api/internal/proxypool" "github.com/nyoungo/dsfree2api/internal/upstream" ) @@ -396,6 +398,218 @@ func (s *Server) handleActions(w http.ResponseWriter, r *http.Request) { _ = json.Unmarshal(body, &p) writeJSON(w, http.StatusOK, s.rotateGuest(p.Site)) + case "pool_set_config": + var p struct { + Enabled *bool `json:"enabled"` + CheckIntervalSeconds *int `json:"check_interval_seconds"` + CheckTimeoutSeconds *int `json:"check_timeout_seconds"` + CheckURL *string `json:"check_url"` + DefaultScheme *string `json:"default_scheme"` + XrayPath *string `json:"xray_path"` + XrayVersion *string `json:"xray_version"` + XrayAutoDownload *bool `json:"xray_auto_download"` + } + _ = json.Unmarshal(body, &p) + s.cfg.Lock() + pc := s.cfg.ProxyPool + if p.Enabled != nil { + pc.Enabled = *p.Enabled + } + if p.CheckIntervalSeconds != nil { + pc.CheckIntervalSeconds = clampInt(*p.CheckIntervalSeconds, 10, 3600) + } + if p.CheckTimeoutSeconds != nil { + pc.CheckTimeoutSeconds = clampInt(*p.CheckTimeoutSeconds, 2, 120) + } + if p.CheckURL != nil { + pc.CheckURL = strings.TrimSpace(*p.CheckURL) + } + if p.DefaultScheme != nil { + pc.DefaultScheme = strings.ToLower(strings.TrimSpace(*p.DefaultScheme)) + } + if p.XrayPath != nil { + pc.XrayPath = strings.TrimSpace(*p.XrayPath) + } + if p.XrayVersion != nil { + pc.XrayVersion = strings.TrimSpace(*p.XrayVersion) + } + if p.XrayAutoDownload != nil { + pc.XrayAutoDownload = *p.XrayAutoDownload + } + s.cfg.ProxyPool = pc + s.cfg.Unlock() + s.poolReload() + s.save(w) + + case "pool_add_entry": + var p struct { + Name string `json:"name"` + Link string `json:"link"` + } + _ = json.Unmarshal(body, &p) + p.Name, p.Link = strings.TrimSpace(p.Name), strings.TrimSpace(p.Link) + if !validPoolName(p.Name) { + fail(w, "name must match [A-Za-z0-9_-]{1,64}") + return + } + if p.Link == "" { + fail(w, "link is required") + return + } + s.cfg.RLock() + scheme := s.cfg.ProxyPool.DefaultScheme + _, exists := s.cfg.ProxyPool.Entries[p.Name] + s.cfg.RUnlock() + if exists { + fail(w, "entry already exists: "+p.Name) + return + } + if _, err := proxypool.ParseNode(p.Link, scheme); err != nil { + fail(w, "node parse: "+err.Error()) + return + } + s.cfg.Lock() + if s.cfg.ProxyPool.Entries == nil { + s.cfg.ProxyPool.Entries = map[string]*config.ProxyEntry{} + } + s.cfg.ProxyPool.Entries[p.Name] = &config.ProxyEntry{Link: p.Link, Enabled: true} + s.cfg.Unlock() + s.poolReload() + s.save(w) + + case "pool_toggle_entry", "pool_toggle_sub": + var p struct { + Name string `json:"name"` + Enabled bool `json:"enabled"` + } + _ = json.Unmarshal(body, &p) + s.cfg.Lock() + ok := false + if req.Action == "pool_toggle_entry" { + if e := s.cfg.ProxyPool.Entries[p.Name]; e != nil { + e.Enabled = p.Enabled + ok = true + } + } else { + if sc := s.cfg.ProxyPool.Subscriptions[p.Name]; sc != nil { + sc.Enabled = p.Enabled + ok = true + } + } + s.cfg.Unlock() + if !ok { + fail(w, "not found: "+p.Name) + return + } + s.poolReload() + s.save(w) + + case "pool_delete_entry", "pool_delete_sub": + var p struct { + Name string `json:"name"` + } + _ = json.Unmarshal(body, &p) + ref := p.Name + if req.Action == "pool_delete_sub" { + ref = "sub:" + p.Name + } + s.cfg.Lock() + if req.Action == "pool_delete_entry" { + delete(s.cfg.ProxyPool.Entries, p.Name) + } else { + delete(s.cfg.ProxyPool.Subscriptions, p.Name) + } + for _, st := range s.cfg.Sites { + if st != nil { + st.Proxies = removeString(st.Proxies, ref) + } + } + s.cfg.Unlock() + s.poolReload() + s.save(w) + + case "pool_add_sub": + var p struct { + Name string `json:"name"` + URL string `json:"url"` + IntervalMinutes int `json:"interval_minutes"` + } + _ = json.Unmarshal(body, &p) + p.Name, p.URL = strings.TrimSpace(p.Name), strings.TrimSpace(p.URL) + if !validPoolName(p.Name) { + fail(w, "name must match [A-Za-z0-9_-]{1,64}") + return + } + if p.URL == "" { + fail(w, "url is required") + return + } + s.cfg.Lock() + if s.cfg.ProxyPool.Subscriptions == nil { + s.cfg.ProxyPool.Subscriptions = map[string]*config.ProxySubscription{} + } + if _, exists := s.cfg.ProxyPool.Subscriptions[p.Name]; exists { + s.cfg.Unlock() + fail(w, "subscription already exists: "+p.Name) + return + } + if p.IntervalMinutes <= 0 { + p.IntervalMinutes = 60 + } + s.cfg.ProxyPool.Subscriptions[p.Name] = &config.ProxySubscription{ + URL: p.URL, Enabled: true, + IntervalMinutes: clampInt(p.IntervalMinutes, 5, 1440), + } + s.cfg.Unlock() + s.poolReload() + s.save(w) + + case "pool_set_site": + var p struct { + Site string `json:"site"` + Proxies []string `json:"proxies"` + } + _ = json.Unmarshal(body, &p) + clean := make([]string, 0, len(p.Proxies)) + for _, item := range p.Proxies { + if item = strings.TrimSpace(item); item != "" { + clean = append(clean, item) + } + } + s.cfg.Lock() + st, ok := s.cfg.Sites[p.Site] + if ok && st != nil { + for _, item := range clean { + if name, isSub := strings.CutPrefix(item, "sub:"); isSub { + if s.cfg.ProxyPool.Subscriptions[name] == nil { + s.cfg.Unlock() + fail(w, "unknown subscription: "+name) + return + } + } else if s.cfg.ProxyPool.Entries[item] == nil { + s.cfg.Unlock() + fail(w, "unknown entry: "+item) + return + } + } + st.Proxies = clean + } + s.cfg.Unlock() + if !ok || st == nil { + fail(w, "unknown site: "+p.Site) + return + } + s.poolReload() + s.save(w) + + case "pool_refresh": + if s.pool == nil { + fail(w, "proxy pool is not available") + return + } + s.pool.RefreshNow(context.Background()) + writeJSON(w, http.StatusOK, map[string]any{"ok": true}) + default: fail(w, "unknown action: "+req.Action) } @@ -673,3 +887,45 @@ func randomToken(n int) string { } return hex.EncodeToString(b) } + +func clampInt(v, lo, hi int) int { + if v < lo { + return lo + } + if v > hi { + return hi + } + return v +} + +func clampFloat(v, lo, hi float64) float64 { + if v < lo { + return lo + } + if v > hi { + return hi + } + return v +} + +var poolNameRe = regexp.MustCompile(`^[A-Za-z0-9_-]{1,64}$`) + +func validPoolName(name string) bool { return poolNameRe.MatchString(name) } + +// removeString drops every occurrence of want from list (in place copy). +func removeString(list []string, want string) []string { + out := list[:0:0] + for _, v := range list { + if v != want { + out = append(out, v) + } + } + return out +} + +// poolReload pushes pool-relevant config changes into the live manager. +func (s *Server) poolReload() { + if s.pool != nil { + s.pool.Reload() + } +} diff --git a/internal/admin/server.go b/internal/admin/server.go index 89e568d..e298e16 100644 --- a/internal/admin/server.go +++ b/internal/admin/server.go @@ -15,6 +15,7 @@ import ( "github.com/nyoungo/dsfree2api/internal/config" "github.com/nyoungo/dsfree2api/internal/logbuf" "github.com/nyoungo/dsfree2api/internal/metrics" + "github.com/nyoungo/dsfree2api/internal/proxypool" "github.com/nyoungo/dsfree2api/internal/turnstile" "github.com/nyoungo/dsfree2api/internal/upstream" ) @@ -31,6 +32,7 @@ type Server struct { met *metrics.Recorder logs *logbuf.Buffer ts *turnstile.Solver + pool *proxypool.Manager log *slog.Logger start time.Time @@ -39,7 +41,7 @@ type Server struct { password string } -func New(cfg *config.Config, up *upstream.Client, met *metrics.Recorder, logs *logbuf.Buffer, ts *turnstile.Solver, log *slog.Logger) *Server { +func New(cfg *config.Config, up *upstream.Client, met *metrics.Recorder, logs *logbuf.Buffer, ts *turnstile.Solver, log *slog.Logger, pool *proxypool.Manager) *Server { if log == nil { log = slog.Default() } @@ -50,7 +52,7 @@ func New(cfg *config.Config, up *upstream.Client, met *metrics.Recorder, logs *l generated = true } s := &Server{ - cfg: cfg, up: up, met: met, logs: logs, ts: ts, log: log, + cfg: cfg, up: up, met: met, logs: logs, ts: ts, pool: pool, log: log, start: time.Now(), tokens: map[string]time.Time{}, password: pw, } if generated { @@ -185,7 +187,7 @@ func (s *Server) handleOverview(w http.ResponseWriter, _ *http.Request) { sites = append(sites, map[string]any{ "code": code, "base_url": st.BaseURL, "ajax_url": st.AJAXURL, "language": st.Language, "sitekey": st.SiteKey, "enabled": st.Enabled, - "verify_action": st.VerifyAction, + "verify_action": st.VerifyAction, "proxies": st.Proxies, }) } proxyCfg := struct { @@ -201,12 +203,17 @@ func (s *Server) handleOverview(w http.ResponseWriter, _ *http.Request) { keys := append([]string(nil), s.cfg.Security.APIKeys...) s.cfg.RUnlock() + var poolStatus any + if s.pool != nil { + poolStatus = s.pool.Status() + } writeJSON(w, http.StatusOK, map[string]any{ "metrics": s.met.Snapshot(), "models": models, "sites": sites, "proxy": proxyCfg, "turnstile": map[string]any{"config": turnstileCfg, "status": s.ts.Status(siteCodes), "history": s.ts.History()}, + "proxypool": poolStatus, "keys": keys, "cache": map[string]any{"chat_config_entries": s.up.ConfigCacheSize(), "routes": routeList(s.up)}, "system": map[string]any{ @@ -269,6 +276,9 @@ func (s *Server) handlePutConfig(w http.ResponseWriter, r *http.Request) { // applyRuntime pushes a freshly edited config into the live subsystems. func (s *Server) applyRuntime() { s.ts.SetConfig(s.currentTurnstile()) + if s.pool != nil { + s.pool.Reload() + } } func (s *Server) currentTurnstile() config.Turnstile { diff --git a/internal/admin/web/app.js b/internal/admin/web/app.js index f6bd42c..b4473cb 100644 --- a/internal/admin/web/app.js +++ b/internal/admin/web/app.js @@ -328,6 +328,7 @@ const mask = k => k.length > 16 ? k.slice(0, 7) + "…" + k.slice(-6) : k; PAGES.proxy = async () => { const ov = await fetchOverview(); const p = ov.proxy; + const pool = ov.proxypool || { xray: {}, entries: [], subscriptions: [], sticky: {} }; view.innerHTML = `

主线路

@@ -355,6 +356,49 @@ PAGES.proxy = async () => {
#地址
当前生效顺序：${ov.cache.routes.map(r => `${esc(r.name)} → ${esc(r.proxy)}`).join(" ")}
+
+
+

代理池 · Xray 核心

+
+
${switchHTML("pool-en", pool.enabled)}
+
+
+
+
+
+
+
+
+
+
+
${switchHTML("pool-xray-auto", pool.xray_auto_download)}
+
+
+ + +
+
Xray：${pool.xray.running ? "运行中 · pid " + pool.xray.pid : "未运行"} + ${pool.xray.error ? ` ${esc(pool.xray.error)}` : ""} + ${pool.last_error ? `
${esc(pool.last_error)}` : ""}
+
+

节点与端点

+
+ +
+
名称类型目标健康延迟
+
分享链接（vless / vmess / trojan / ss）由内置 Xray 核心承载；ip:port / socks5:// 等端点直接使用。
+
+
+

订阅

+
+ + +
+
名称地址间隔节点上次拉取
+
+

站点绑定（按顺序，粘性选路）

+
站点绑定（逗号分隔：名称 或 sub:订阅名）当前出口
+
未绑定、或绑定节点全部不健康时，回退到主线路 / 备用线路。Cookie 与出口 IP 绑定：切换出口后该站点会在下一轮巡检重新求解 Turnstile。
`; const renderFb = list => { @@ -369,6 +413,98 @@ PAGES.proxy = async () => { }; renderFb(p.fallbacks || []); + const renderPool = () => { + const entries = pool.entries || []; + $("#tb-pool").innerHTML = entries.length ? entries.map(e => { + const health = e.parse_error ? `解析失败` + : e.checked ? (e.healthy ? '健康' : `不可用`) + : '未探测'; + return `${esc(e.name)}${e.source ? ` ← ${esc(e.source)}` : ""} + ${esc(e.kind === "url" ? e.scheme : (e.scheme || "xray"))} + ${esc(e.target || "")} + ${health}${e.latency_ms ? e.latency_ms + "ms" : "–"} + + ${e.source ? "" : ``}`; + }).join("") : '尚无节点 —— 添加分享链接或 ip:port 端点'; + $$("[data-ptoggle]").forEach(b => b.onclick = async () => { + try { await api("/api/actions", { body: { action: "pool_toggle_entry", name: b.dataset.ptoggle, enabled: b.dataset.pen === "1" } }); PAGES.proxy(); } + catch (e) { toast(e.message, "err"); } + }); + $$("[data-pdel]").forEach(b => b.onclick = async () => { + if (!confirm("删除节点 " + b.dataset.pdel + "？")) return; + try { await api("/api/actions", { body: { action: "pool_delete_entry", name: b.dataset.pdel } }); PAGES.proxy(); } + catch (e) { toast(e.message, "err"); } + }); + + const subs = pool.subscriptions || []; + $("#tb-subs").innerHTML = subs.length ? subs.map(s => ` + ${esc(s.name)}${esc(s.url)} + ${s.interval_minutes || 60}m${s.nodes || 0} + ${s.fetched_at ? timeStr(s.fetched_at * 1000) : "未拉取"}${s.last_error ? ` ${esc(String(s.last_error).slice(0, 60))}` : ""} + + `).join("") + : '尚无订阅'; + $$("[data-stoggle]").forEach(b => b.onclick = async () => { + try { await api("/api/actions", { body: { action: "pool_toggle_sub", name: b.dataset.stoggle, enabled: b.dataset.sen === "1" } }); PAGES.proxy(); } + catch (e) { toast(e.message, "err"); } + }); + $$("[data-sdel]").forEach(b => b.onclick = async () => { + if (!confirm("删除订阅 " + b.dataset.sdel + "？")) return; + try { await api("/api/actions", { body: { action: "pool_delete_sub", name: b.dataset.sdel } }); PAGES.proxy(); } + catch (e) { toast(e.message, "err"); } + }); + + const sticky = pool.sticky || {}; + $("#tb-bind").innerHTML = (ov.sites || []).map(s => { + const cur = sticky[s.code]; + const curEntry = entries.find(e => e.name === cur); + return `${esc(s.code.toUpperCase())} + + ${cur ? `${esc(cur)} → ${esc(curEntry ? curEntry.target : "?")}` : "全局线路"} + `; + }).join(""); + $$("[data-bsave]").forEach(b => b.onclick = async () => { + const input = $(`[data-bind="${CSS.escape(b.dataset.bsave)}"]`); + const list = input.value.split(",").map(x => x.trim()).filter(Boolean); + try { await api("/api/actions", { body: { action: "pool_set_site", site: b.dataset.bsave, proxies: list } }); toast("已保存", "ok"); PAGES.proxy(); } + catch (e) { toast(e.message, "err"); } + }); + }; + renderPool(); + + $("#pool-save").onclick = async () => { + try { + await api("/api/actions", { body: { + action: "pool_set_config", + enabled: $("#pool-en").checked, + check_interval_seconds: +$("#pool-check-int").value || 120, + check_timeout_seconds: +$("#pool-check-to").value || 10, + check_url: $("#pool-check-url").value.trim(), + default_scheme: $("#pool-scheme").value.trim() || "http", + xray_path: $("#pool-xray-path").value.trim(), + xray_version: $("#pool-xray-ver").value.trim(), + xray_auto_download: $("#pool-xray-auto").checked } }); + toast("池配置已保存", "ok"); PAGES.proxy(); + } catch (e) { toast(e.message, "err"); } + }; + $("#pool-refresh").onclick = async () => { + const b = $("#pool-refresh"); b.disabled = true; b.textContent = "刷新中…"; + try { await api("/api/actions", { body: { action: "pool_refresh" } }); toast("已在后台拉取订阅并检查健康，稍后重进本页查看", "ok"); PAGES.proxy(); } + catch (e) { toast(e.message, "err"); b.disabled = false; b.textContent = "刷新节点与健康检查"; } + }; + $("#pool-add").onclick = async () => { + try { + await api("/api/actions", { body: { action: "pool_add_entry", name: $("#pool-name").value.trim(), link: $("#pool-link").value.trim() } }); + toast("已添加", "ok"); PAGES.proxy(); + } catch (e) { toast(e.message, "err"); } + }; + $("#sub-add").onclick = async () => { + try { + await api("/api/actions", { body: { action: "pool_add_sub", name: $("#sub-name").value.trim(), url: $("#sub-url").value.trim(), interval_minutes: +$("#sub-int").value || 60 } }); + toast("已添加", "ok"); PAGES.proxy(); + } catch (e) { toast(e.message, "err"); } + }; + $("#px-save").onclick = async () => { try { await api("/api/config", { method: "PUT", body: await merged(c => { c.proxy.url = $("#px-primary").value.trim(); }) }); @@ -406,6 +542,7 @@ async function merged(mutate) { } PAGES.turnstile = async () => { + if (overviewTimer) { clearInterval(overviewTimer); overviewTimer = null; } const ov = await fetchOverview(); const t = ov.turnstile.config; view.innerHTML = ` diff --git a/internal/config/config.go b/internal/config/config.go index f63d1f2..fc5debb 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -59,6 +59,41 @@ type Site struct { VerifyAction string `toml:"verify_action" json:"verify_action"` Language string `toml:"language" json:"language"` Enabled bool `toml:"enabled" json:"enabled"` + // Proxies binds this site to the proxy pool: an ordered list of entry + // names and/or "sub:" references. Empty = global [proxy] routes. + Proxies []string `toml:"proxies" json:"proxies"` + // WarmPoolOnly keeps the cookie warmer on the pool routes bound via + // Proxies; the global [proxy] primary/fallback lanes are not pre-solved + // (they remain available for request-time failover). Default false. + WarmPoolOnly bool `toml:"warm_pool_only" json:"warm_pool_only"` +} + +// ProxyEntry is one manually configured pool node: an Xray share link +// (vless/vmess/trojan/ss) or a plain http/https/socks5 endpoint. +type ProxyEntry struct { + Link string `toml:"link" json:"link"` + Enabled bool `toml:"enabled" json:"enabled"` +} + +// ProxySubscription is a remote node list refreshed on an interval. +type ProxySubscription struct { + URL string `toml:"url" json:"url"` + Enabled bool `toml:"enabled" json:"enabled"` + IntervalMinutes int `toml:"interval_minutes" json:"interval_minutes"` +} + +// ProxyPool configures the egress pool and its managed Xray core. +type ProxyPool struct { + Enabled bool `toml:"enabled" json:"enabled"` + CheckIntervalSeconds int `toml:"check_interval_seconds" json:"check_interval_seconds"` + CheckTimeoutSeconds int `toml:"check_timeout_seconds" json:"check_timeout_seconds"` + CheckURL string `toml:"check_url" json:"check_url"` + DefaultScheme string `toml:"default_scheme" json:"default_scheme"` + XrayPath string `toml:"xray_path" json:"xray_path"` + XrayVersion string `toml:"xray_version" json:"xray_version"` + XrayAutoDownload bool `toml:"xray_auto_download" json:"xray_auto_download"` + Entries map[string]*ProxyEntry `toml:"entries" json:"entries"` + Subscriptions map[string]*ProxySubscription `toml:"subscriptions" json:"subscriptions"` } type Model struct { @@ -101,6 +136,7 @@ type Config struct { SlowStartSeconds float64 `toml:"slow_start_seconds" json:"slow_start_seconds"` } `toml:"proxy" json:"proxy"` + ProxyPool ProxyPool `toml:"proxypool" json:"proxypool"` Upstream struct { Timeout float64 `toml:"timeout" json:"timeout"` StreamTimeout float64 `toml:"stream_timeout" json:"stream_timeout"` @@ -125,6 +161,20 @@ type Config struct { path string `toml:"-"` } +// DefaultProxyPool returns the built-in pool settings (feature off, Xray +// auto-download on). +func DefaultProxyPool() ProxyPool { + return ProxyPool{ + CheckIntervalSeconds: 120, + CheckTimeoutSeconds: 10, + CheckURL: "https://www.gstatic.com/generate_204", + DefaultScheme: "http", + XrayAutoDownload: true, + Entries: map[string]*ProxyEntry{}, + Subscriptions: map[string]*ProxySubscription{}, + } +} + func DefaultTurnstile() Turnstile { return Turnstile{ Enabled: false, @@ -201,6 +251,7 @@ func newConfig() *Config { c.Upstream.CrossSiteFailover = true c.Upstream.ContinueRounds = 20 c.Runtime.DataDir = "./data" + c.ProxyPool = DefaultProxyPool() c.Turnstile = DefaultTurnstile() c.Sites = DefaultSites(c.Turnstile.SiteKey) c.Models = DefaultModels() @@ -300,22 +351,36 @@ func (c *Config) applyDefaults(md *toml.MetaData) { c.Turnstile.Enabled = true c.Upstream.AutoRefresh = true c.Upstream.ContinueRounds = 20 + c.ProxyPool.XrayAutoDownload = true for _, s := range c.Sites { s.Enabled = true } for _, m := range c.Models { m.Enabled = true } + for _, e := range c.ProxyPool.Entries { + e.Enabled = true + } + for _, sc := range c.ProxyPool.Subscriptions { + sc.Enabled = true + } return } def(md.IsDefined("turnstile", "enabled"), &c.Turnstile.Enabled, true) def(md.IsDefined("upstream", "auto_refresh"), &c.Upstream.AutoRefresh, true) + def(md.IsDefined("proxypool", "xray_auto_download"), &c.ProxyPool.XrayAutoDownload, true) for code, s := range c.Sites { def(md.IsDefined("sites", code, "enabled"), &s.Enabled, true) } for id, m := range c.Models { def(md.IsDefined("models", id, "enabled"), &m.Enabled, true) } + for name, e := range c.ProxyPool.Entries { + def(md.IsDefined("proxypool", "entries", name, "enabled"), &e.Enabled, true) + } + for name, sc := range c.ProxyPool.Subscriptions { + def(md.IsDefined("proxypool", "subscriptions", name, "enabled"), &sc.Enabled, true) + } } func (c *Config) applyEnv() error { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 26dae15..a3b36a7 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -204,6 +204,78 @@ func TestTurnstileProviderEnv(t *testing.T) { } } +func TestProxyPoolDefaults(t *testing.T) { + d := DefaultProxyPool() + if d.Enabled { + t.Error("proxy pool must be opt-in") + } + if !d.XrayAutoDownload || d.CheckIntervalSeconds != 120 || d.CheckTimeoutSeconds != 10 { + t.Errorf("pool defaults = %+v", d) + } + if d.DefaultScheme != "http" { + t.Errorf("default scheme = %q", d.DefaultScheme) + } + cfg, err := Load(examplePath) + if err != nil { + t.Fatalf("load: %v", err) + } + if !cfg.ProxyPool.XrayAutoDownload { + t.Error("example config should keep xray_auto_download = true") + } + if cfg.ProxyPool.CheckIntervalSeconds != 120 { + t.Errorf("check interval = %d", cfg.ProxyPool.CheckIntervalSeconds) + } +} + +func TestTurnstileSolveTimeoutAndWarmDefaults(t *testing.T) { + d := DefaultTurnstile() + if d.WarmEnabled { + t.Error("cookie pool must be opt-in") + } + if d.WarmRatio != 0.3 || d.WarmCheckSeconds != 60 { + t.Errorf("warm defaults = %v / %v", d.WarmRatio, d.WarmCheckSeconds) + } + z := Turnstile{} + if z.SolveTimeoutValue() != 90 { + t.Errorf("zero timeout should normalize to 90, got %d", z.SolveTimeoutValue()) + } + if z.WarmRatioValue() != 0.3 || z.WarmCheckValue() != 60 { + t.Errorf("zero warm values should normalize to 0.3 / 60") + } + if got := (Turnstile{TimeoutSeconds: 1}).SolveTimeoutValue(); got != 5 { + t.Errorf("timeout floor = %d, want 5", got) + } + if got := (Turnstile{TimeoutSeconds: 9999}).SolveTimeoutValue(); got != 600 { + t.Errorf("timeout ceiling = %d, want 600", got) + } + if got := (Turnstile{WarmRatio: 0.99}).WarmRatioValue(); got != 0.95 { + t.Errorf("warm ratio ceiling = %v, want 0.95", got) + } + if (Turnstile{}).APIStyleValue() != "sync" { + t.Error("zero api_style should normalize to sync") + } + if (Turnstile{APIStyle: "EzSolver"}).APIStyleValue() != "ezsolver" { + t.Error("api_style should normalize EzSolver") + } +} + +func TestTurnstileSolveAndWarmEnvOverrides(t *testing.T) { + t.Setenv("TURNSTILE_TIMEOUT_SECONDS", "45") + t.Setenv("TURNSTILE_WARM_ENABLED", "true") + t.Setenv("TURNSTILE_WARM_RATIO", "0.5") + t.Setenv("TURNSTILE_WARM_CHECK_SECONDS", "120") + cfg, err := Load(examplePath) + if err != nil { + t.Fatalf("load: %v", err) + } + if cfg.Turnstile.TimeoutSeconds != 45 { + t.Errorf("timeout = %d, want 45", cfg.Turnstile.TimeoutSeconds) + } + if !cfg.Turnstile.WarmEnabled || cfg.Turnstile.WarmRatio != 0.5 || cfg.Turnstile.WarmCheckSeconds != 120 { + t.Errorf("warm env = %+v", cfg.Turnstile) + } +} + func TestSaveRoundTrip(t *testing.T) { cfg, err := Load(examplePath) if err != nil { diff --git a/internal/proxypool/manager.go b/internal/proxypool/manager.go new file mode 100644 index 0000000..c738b97 --- /dev/null +++ b/internal/proxypool/manager.go @@ -0,0 +1,722 @@ +package proxypool + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "io" + "log/slog" + "net/http" + "net/url" + "sort" + "strings" + "sync" + "time" + + "github.com/nyoungo/dsfree2api/internal/config" + "github.com/nyoungo/dsfree2api/internal/upstream" +) + +// endpoint is one runtime pool slot. +type endpoint struct { + name string + kind string // "xray" | "url" | "invalid" + scheme string + display string + link string + source string // "" = manual entry, else subscription name + enabled bool + + parseErr string + outbound map[string]any + url string + proxyURL string // resolved egress URL (socks5://127.0.0.1:port or plain) + + checked bool + healthy bool + latencyMs int64 + lastErr string + checkedAt time.Time +} + +type subState struct { + name string + nodes []string + lastErr string + fetchedAt time.Time +} + +// Manager owns the proxy pool: endpoints, health, subscriptions and sticky +// per-site selection. It is safe for concurrent use. +type Manager struct { + cfg *config.Config + log *slog.Logger + dir string + xray *xrayCore + + reloadMu sync.Mutex + + mu sync.Mutex + endpoints map[string]*endpoint + subs map[string]*subState + sticky map[string]string + fetching map[string]bool + started bool + lastCheck time.Time + lastErr string +} + +func New(cfg *config.Config, dataDir string, log *slog.Logger) *Manager { + if log == nil { + log = slog.Default() + } + return &Manager{ + cfg: cfg, + log: log, + dir: dataDir, + xray: newXrayCore(dataDir, log), + endpoints: map[string]*endpoint{}, + subs: map[string]*subState{}, + sticky: map[string]string{}, + fetching: map[string]bool{}, + } +} + +// Start runs the background loop: reload from config, refresh due +// subscriptions, probe endpoint health. +func (m *Manager) Start(ctx context.Context) { + m.mu.Lock() + if m.started { + m.mu.Unlock() + return + } + m.started = true + m.mu.Unlock() + go m.loop(ctx) +} + +func (m *Manager) loop(ctx context.Context) { + ticker := time.NewTicker(5 * time.Second) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + m.xray.stop() + return + case <-ticker.C: + } + m.cycle(ctx) + } +} + +func (m *Manager) cycle(ctx context.Context) { + m.Reload() + + m.cfg.RLock() + enabled := m.cfg.ProxyPool.Enabled + interval := time.Duration(m.cfg.ProxyPool.CheckIntervalSeconds) * time.Second + subs := make(map[string]config.ProxySubscription, len(m.cfg.ProxyPool.Subscriptions)) + for name, sc := range m.cfg.ProxyPool.Subscriptions { + if sc != nil { + subs[name] = *sc + } + } + m.cfg.RUnlock() + if !enabled { + return + } + if interval <= 0 { + interval = 120 * time.Second + } + + m.mu.Lock() + due := m.lastCheck.IsZero() || time.Since(m.lastCheck) >= interval + m.mu.Unlock() + if due { + m.checkAll(ctx) + } + + now := time.Now() + for name, sc := range subs { + if !sc.Enabled || strings.TrimSpace(sc.URL) == "" { + continue + } + m.mu.Lock() + st := m.subs[name] + busy := m.fetching[name] + m.mu.Unlock() + if busy { + continue + } + iv := time.Duration(sc.IntervalMinutes) * time.Minute + if iv <= 0 { + iv = time.Hour + } + if st != nil && !st.fetchedAt.IsZero() && now.Sub(st.fetchedAt) < iv { + continue + } + go m.fetchSub(ctx, name, sc) + } +} + +// Reload rebuilds the runtime endpoint table from the live config plus the +// cached subscription nodes, (re)configures the Xray core and prunes sticky +// picks that no longer resolve. +func (m *Manager) Reload() { + m.reloadMu.Lock() + defer m.reloadMu.Unlock() + + m.cfg.RLock() + poolCfg := m.cfg.ProxyPool + entries := make(map[string]config.ProxyEntry, len(poolCfg.Entries)) + for name, e := range poolCfg.Entries { + if e != nil { + entries[name] = *e + } + } + subsCfg := make(map[string]config.ProxySubscription, len(poolCfg.Subscriptions)) + for name, sc := range poolCfg.Subscriptions { + if sc != nil { + subsCfg[name] = *sc + } + } + m.cfg.RUnlock() + + m.mu.Lock() + subNodes := make(map[string][]string, len(m.subs)) + for name, st := range m.subs { + subNodes[name] = append([]string(nil), st.nodes...) + } + old := m.endpoints + m.mu.Unlock() + + desired := map[string]*endpoint{} + for name, e := range entries { + if strings.TrimSpace(e.Link) == "" { + continue + } + ep := newEndpoint(name, e.Link, poolCfg.DefaultScheme, "") + ep.enabled = e.Enabled + desired[name] = ep + } + for subName, nodes := range subNodes { + sc, ok := subsCfg[subName] + if !ok { + continue + } + for _, line := range nodes { + name := subName + "#" + shortHash(line) + if _, dup := desired[name]; dup { + continue + } + ep := newEndpoint(name, line, poolCfg.DefaultScheme, subName) + ep.enabled = sc.Enabled + desired[name] = ep + } + } + for name, ep := range desired { + if o := old[name]; o != nil && o.link == ep.link { + ep.checked = o.checked + ep.healthy = o.healthy + ep.latencyMs = o.latencyMs + ep.lastErr = o.lastErr + ep.checkedAt = o.checkedAt + } + } + + nodes := make([]xrayNode, 0, len(desired)) + for _, ep := range desired { + if ep.kind == "xray" && ep.enabled && ep.parseErr == "" && ep.outbound != nil { + nodes = append(nodes, xrayNode{Name: ep.name, Outbound: ep.outbound}) + } + } + sort.Slice(nodes, func(i, j int) bool { return nodes[i].Name < nodes[j].Name }) + + if len(nodes) > 0 { + if err := m.xray.ensure(poolCfg.XrayPath, poolCfg.XrayAutoDownload, poolCfg.XrayVersion); err != nil { + m.setErr(err) + } else if err := m.xray.configure(nodes); err != nil { + m.setErr(err) + } else { + m.setErr(nil) + } + } else { + _ = m.xray.configure(nil) + m.setErr(nil) + } + + xrayUp := m.xray.status().Running + for _, ep := range desired { + switch ep.kind { + case "url": + ep.proxyURL = ep.url + case "xray": + if ep.parseErr == "" && xrayUp { + if port := m.xray.port(ep.name); port > 0 { + ep.proxyURL = fmt.Sprintf("socks5://127.0.0.1:%d", port) + } + } + } + } + + m.mu.Lock() + m.endpoints = desired + for site, name := range m.sticky { + ep := desired[name] + if ep == nil || !ep.enabled || ep.proxyURL == "" { + delete(m.sticky, site) + } + } + m.mu.Unlock() +} + +// RefreshNow refreshes every subscription and probes every endpoint, then +// reloads. Used by the console button. +func (m *Manager) RefreshNow(ctx context.Context) { + go func() { + m.Reload() + m.cfg.RLock() + enabled := m.cfg.ProxyPool.Enabled + subs := make(map[string]config.ProxySubscription, len(m.cfg.ProxyPool.Subscriptions)) + for name, sc := range m.cfg.ProxyPool.Subscriptions { + if sc != nil { + subs[name] = *sc + } + } + m.cfg.RUnlock() + for name, sc := range subs { + if sc.Enabled && strings.TrimSpace(sc.URL) != "" { + m.fetchSub(ctx, name, sc) + } + } + if enabled { + m.checkAll(ctx) + } + m.Reload() + }() +} + +func (m *Manager) fetchSub(ctx context.Context, name string, sc config.ProxySubscription) { + m.mu.Lock() + m.fetching[name] = true + m.mu.Unlock() + defer func() { + m.mu.Lock() + delete(m.fetching, name) + m.mu.Unlock() + }() + + client, err := m.fetchClient() + if err == nil { + var req *http.Request + req, err = http.NewRequestWithContext(ctx, http.MethodGet, sc.URL, nil) + if err == nil { + var resp *http.Response + resp, err = client.Do(req) + if err == nil { + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + err = fmt.Errorf("subscription http %d", resp.StatusCode) + } else { + var raw []byte + raw, err = io.ReadAll(io.LimitReader(resp.Body, 8<<20)) + if err == nil { + lines := ParseSubscriptionBody(raw) + if len(lines) == 0 { + err = fmt.Errorf("subscription returned no nodes") + } else { + err = nil + m.subResult(name, lines, nil) + m.log.Info("subscription refreshed", "name", name, "nodes", len(lines)) + m.Reload() + return + } + } + } + } + } + } + m.subResult(name, nil, err) + m.log.Warn("subscription refresh failed", "name", name, "error", err) +} + +func (m *Manager) subResult(name string, nodes []string, err error) { + m.mu.Lock() + defer m.mu.Unlock() + st := m.subs[name] + if st == nil { + st = &subState{name: name} + m.subs[name] = st + } + if err != nil { + st.lastErr = err.Error() + return + } + st.nodes = nodes + st.lastErr = "" + st.fetchedAt = time.Now() +} + +func (m *Manager) fetchClient() (*http.Client, error) { + m.cfg.RLock() + proxy := strings.TrimSpace(m.cfg.Proxy.URL) + m.cfg.RUnlock() + tr := &http.Transport{ + TLSHandshakeTimeout: 20 * time.Second, + ResponseHeaderTimeout: 30 * time.Second, + } + if proxy != "" { + u, err := url.Parse(proxy) + if err != nil { + return nil, fmt.Errorf("proxy url: %w", err) + } + tr.Proxy = http.ProxyURL(u) + } + return &http.Client{Transport: tr, Timeout: 45 * time.Second}, nil +} + +func (m *Manager) checkAll(ctx context.Context) { + m.mu.Lock() + list := make([]*endpoint, 0, len(m.endpoints)) + for _, ep := range m.endpoints { + list = append(list, ep) + } + m.mu.Unlock() + + m.cfg.RLock() + timeout := time.Duration(m.cfg.ProxyPool.CheckTimeoutSeconds) * time.Second + checkURL := strings.TrimSpace(m.cfg.ProxyPool.CheckURL) + m.cfg.RUnlock() + if timeout <= 0 { + timeout = 10 * time.Second + } + if checkURL == "" { + checkURL = "https://www.gstatic.com/generate_204" + } + + sem := make(chan struct{}, 4) + var wg sync.WaitGroup + for _, ep := range list { + if !ep.enabled || ep.proxyURL == "" { + continue + } + wg.Add(1) + sem <- struct{}{} + go func(ep *endpoint) { + defer wg.Done() + defer func() { <-sem }() + if ctx.Err() != nil { + return + } + ok, ms, errMsg := probeURL(ctx, ep.proxyURL, checkURL, timeout) + m.mu.Lock() + ep.checked = true + ep.checkedAt = time.Now() + ep.healthy = ok + ep.latencyMs = ms + ep.lastErr = errMsg + m.mu.Unlock() + }(ep) + } + wg.Wait() + + m.mu.Lock() + m.lastCheck = time.Now() + m.mu.Unlock() +} + +func probeURL(ctx context.Context, proxyURL, checkURL string, timeout time.Duration) (bool, int64, string) { + u, err := url.Parse(proxyURL) + if err != nil { + return false, 0, "bad proxy url" + } + tr := &http.Transport{ + Proxy: http.ProxyURL(u), + TLSHandshakeTimeout: timeout, + ResponseHeaderTimeout: timeout, + DisableKeepAlives: true, + } + client := &http.Client{Transport: tr, Timeout: timeout} + req, err := http.NewRequestWithContext(ctx, http.MethodGet, checkURL, nil) + if err != nil { + return false, 0, err.Error() + } + start := time.Now() + resp, err := client.Do(req) + ms := time.Since(start).Milliseconds() + if err != nil { + return false, ms, trimErr(err) + } + defer resp.Body.Close() + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4096)) + if resp.StatusCode >= 400 { + return false, ms, fmt.Sprintf("http %d", resp.StatusCode) + } + return true, ms, "" +} + +// Candidates resolves the ordered route candidates for one site: its bound +// entries (or subscriptions) filtered to healthy ones, sticky pick first. +// It implements upstream.RouteSource. +func (m *Manager) Candidates(site string) []upstream.RouteCandidate { + m.cfg.RLock() + enabled := m.cfg.ProxyPool.Enabled + var items []string + if st, ok := m.cfg.Sites[site]; ok && st != nil { + items = append(items, st.Proxies...) + } + m.cfg.RUnlock() + if !enabled || len(items) == 0 { + return nil + } + + m.mu.Lock() + defer m.mu.Unlock() + var all []*endpoint + for _, item := range items { + all = append(all, m.resolveItemLocked(item)...) + } + usable := make([]*endpoint, 0, len(all)) + for _, ep := range all { + if ep.enabled && ep.proxyURL != "" && ep.parseErr == "" { + usable = append(usable, ep) + } + } + pick := make([]*endpoint, 0, len(usable)) + for _, ep := range usable { + if !ep.checked || ep.healthy { + pick = append(pick, ep) + } + } + if len(pick) == 0 { + pick = usable + } + if len(pick) == 0 { + return nil + } + if chosen := m.sticky[site]; chosen != "" { + for i, ep := range pick { + if ep.name == chosen && i > 0 { + pick = append([]*endpoint{ep}, append(pick[:i], pick[i+1:]...)...) + break + } + } + } + m.sticky[site] = pick[0].name + out := make([]upstream.RouteCandidate, 0, len(pick)) + for _, ep := range pick { + out = append(out, upstream.RouteCandidate{Name: ep.name, Proxy: ep.proxyURL}) + } + return out +} + +func (m *Manager) resolveItemLocked(item string) []*endpoint { + item = strings.TrimSpace(item) + if item == "" { + return nil + } + if name, ok := strings.CutPrefix(item, "sub:"); ok { + var out []*endpoint + for _, ep := range m.endpoints { + if ep.source == name { + out = append(out, ep) + } + } + sort.Slice(out, func(i, j int) bool { return out[i].name < out[j].name }) + return out + } + if ep := m.endpoints[item]; ep != nil { + return []*endpoint{ep} + } + return nil +} + +// Report records how a request through `proxy` went, so a failing endpoint +// loses its sticky pick immediately and a healthy one is trusted again. +func (m *Manager) Report(proxy string, ok bool) { + if m == nil || proxy == "" { + return + } + m.mu.Lock() + defer m.mu.Unlock() + for _, ep := range m.endpoints { + if ep.proxyURL != proxy { + continue + } + ep.checked = true + ep.checkedAt = time.Now() + if ok { + ep.healthy = true + ep.lastErr = "" + } else { + ep.healthy = false + ep.lastErr = "request failed" + for site, name := range m.sticky { + if name == ep.name { + delete(m.sticky, site) + } + } + } + return + } +} + +// Status is the console snapshot. +type Status struct { + Enabled bool `json:"enabled"` + CheckIntervalS int `json:"check_interval_seconds"` + CheckTimeoutS int `json:"check_timeout_seconds"` + CheckURL string `json:"check_url"` + DefaultScheme string `json:"default_scheme"` + XrayPath string `json:"xray_path"` + XrayVersion string `json:"xray_version"` + XrayAutoDownload bool `json:"xray_auto_download"` + Xray XrayStatus `json:"xray"` + Entries []EntryStatus `json:"entries"` + Subscriptions []SubStatus `json:"subscriptions"` + Sticky map[string]string `json:"sticky"` + LastError string `json:"last_error,omitempty"` +} + +type EntryStatus struct { + Name string `json:"name"` + Kind string `json:"kind"` + Scheme string `json:"scheme"` + Target string `json:"target"` + Source string `json:"source,omitempty"` + Enabled bool `json:"enabled"` + Healthy bool `json:"healthy"` + Checked bool `json:"checked"` + LatencyMs int64 `json:"latency_ms"` + LastErr string `json:"last_error,omitempty"` + ParseError string `json:"parse_error,omitempty"` + CheckedAt int64 `json:"checked_at"` +} + +type SubStatus struct { + Name string `json:"name"` + URL string `json:"url"` + Enabled bool `json:"enabled"` + IntervalMinutes int `json:"interval_minutes"` + Nodes int `json:"nodes"` + FetchedAt int64 `json:"fetched_at"` + LastErr string `json:"last_error,omitempty"` +} + +func (m *Manager) Status() Status { + m.cfg.RLock() + poolCfg := m.cfg.ProxyPool + subsCfg := make(map[string]config.ProxySubscription, len(poolCfg.Subscriptions)) + for name, sc := range poolCfg.Subscriptions { + if sc != nil { + subsCfg[name] = *sc + } + } + m.cfg.RUnlock() + + m.mu.Lock() + endpoints := make([]*endpoint, 0, len(m.endpoints)) + for _, ep := range m.endpoints { + endpoints = append(endpoints, ep) + } + sticky := make(map[string]string, len(m.sticky)) + for k, v := range m.sticky { + sticky[k] = v + } + subs := make(map[string]*subState, len(m.subs)) + for name, st := range m.subs { + cp := *st + subs[name] = &cp + } + lastErr := m.lastErr + m.mu.Unlock() + + st := Status{ + Enabled: poolCfg.Enabled, + CheckIntervalS: poolCfg.CheckIntervalSeconds, + CheckTimeoutS: poolCfg.CheckTimeoutSeconds, + CheckURL: poolCfg.CheckURL, + DefaultScheme: poolCfg.DefaultScheme, + XrayPath: poolCfg.XrayPath, + XrayVersion: poolCfg.XrayVersion, + XrayAutoDownload: poolCfg.XrayAutoDownload, + Xray: m.xray.status(), + Sticky: sticky, + LastError: lastErr, + } + for _, ep := range endpoints { + es := EntryStatus{ + Name: ep.name, Kind: ep.kind, Scheme: ep.scheme, Target: ep.display, + Source: ep.source, Enabled: ep.enabled, Healthy: ep.healthy, + Checked: ep.checked, LatencyMs: ep.latencyMs, LastErr: ep.lastErr, + ParseError: ep.parseErr, + } + if !ep.checkedAt.IsZero() { + es.CheckedAt = ep.checkedAt.Unix() + } + st.Entries = append(st.Entries, es) + } + sort.Slice(st.Entries, func(i, j int) bool { + a, b := st.Entries[i], st.Entries[j] + if (a.Source == "") != (b.Source == "") { + return a.Source == "" + } + if a.Source != b.Source { + return a.Source < b.Source + } + return a.Name < b.Name + }) + for name, sc := range subsCfg { + ss := SubStatus{Name: name, URL: sc.URL, Enabled: sc.Enabled, IntervalMinutes: sc.IntervalMinutes} + if stt := subs[name]; stt != nil { + ss.Nodes = len(stt.nodes) + ss.LastErr = stt.lastErr + if !stt.fetchedAt.IsZero() { + ss.FetchedAt = stt.fetchedAt.Unix() + } + } + st.Subscriptions = append(st.Subscriptions, ss) + } + sort.Slice(st.Subscriptions, func(i, j int) bool { return st.Subscriptions[i].Name < st.Subscriptions[j].Name }) + return st +} + +func newEndpoint(name, link, defaultScheme, source string) *endpoint { + ep := &endpoint{name: name, source: source, link: strings.TrimSpace(link)} + node, err := ParseNode(ep.link, defaultScheme) + if err != nil { + ep.kind = "invalid" + ep.parseErr = err.Error() + return ep + } + ep.kind = node.Kind + ep.scheme = node.Scheme + ep.display = node.Display + ep.outbound = node.Outbound + ep.url = node.URL + return ep +} + +func (m *Manager) setErr(err error) { + m.mu.Lock() + defer m.mu.Unlock() + if err == nil { + m.lastErr = "" + return + } + m.lastErr = err.Error() +} + +func shortHash(s string) string { + sum := sha256.Sum256([]byte(s)) + return hex.EncodeToString(sum[:4]) +} + +func trimErr(err error) string { + s := err.Error() + if len(s) > 160 { + return s[:160] + "..." + } + return s +} diff --git a/internal/proxypool/manager_test.go b/internal/proxypool/manager_test.go new file mode 100644 index 0000000..24aaa26 --- /dev/null +++ b/internal/proxypool/manager_test.go @@ -0,0 +1,76 @@ +package proxypool + +import ( + "testing" + + "github.com/nyoungo/dsfree2api/internal/config" +) + +const examplePath = "../../config.example.toml" + +func TestCandidatesStickySwitchOnFailure(t *testing.T) { + cfg, err := config.Load(examplePath) + if err != nil { + t.Fatalf("load config: %v", err) + } + cfg.ProxyPool.Enabled = true + cfg.ProxyPool.Entries = map[string]*config.ProxyEntry{ + "a": {Link: "http://127.0.0.1:9", Enabled: true}, + "b": {Link: "socks5://127.0.0.1:10", Enabled: true}, + } + cfg.Sites["de"].Proxies = []string{"a", "b"} + + m := New(cfg, t.TempDir(), nil) + m.Reload() + + cands := m.Candidates("de") + if len(cands) != 2 || cands[0].Name != "a" || cands[0].Proxy != "http://127.0.0.1:9" { + t.Fatalf("candidates = %+v", cands) + } + + // Both healthy: the sticky pick stays a. + m.mu.Lock() + for _, ep := range m.endpoints { + ep.checked = true + ep.healthy = true + } + m.mu.Unlock() + cands = m.Candidates("de") + if cands[0].Name != "a" { + t.Fatalf("sticky pick changed without failure: %+v", cands) + } + + // a fails: the request path must immediately switch to b. + m.Report("http://127.0.0.1:9", false) + cands = m.Candidates("de") + if len(cands) != 1 || cands[0].Name != "b" { + t.Fatalf("after failure = %+v", cands) + } + + // Everything unhealthy: fall back to the bound order so requests still try. + m.Report("socks5://127.0.0.1:10", false) + cands = m.Candidates("de") + if len(cands) != 2 || cands[0].Name != "a" { + t.Fatalf("all-unhealthy fallback = %+v", cands) + } +} + +func TestCandidatesDisabledOrUnbound(t *testing.T) { + cfg, err := config.Load(examplePath) + if err != nil { + t.Fatalf("load config: %v", err) + } + cfg.ProxyPool.Enabled = false + cfg.Sites["de"].Proxies = []string{"a"} + m := New(cfg, t.TempDir(), nil) + m.Reload() + if got := m.Candidates("de"); got != nil { + t.Fatalf("disabled pool returned %+v", got) + } + + cfg.ProxyPool.Enabled = true + m.Reload() + if got := m.Candidates("de"); got != nil { + t.Fatalf("unbound/unknown entry returned %+v", got) + } +} diff --git a/internal/proxypool/parse.go b/internal/proxypool/parse.go new file mode 100644 index 0000000..4b201cd --- /dev/null +++ b/internal/proxypool/parse.go @@ -0,0 +1,517 @@ +// Package proxypool manages the egress endpoints used to reach the upstream +// sites: Xray share links driven by a managed Xray core, plus plain +// http/https/socks5 endpoints. Selection is sticky per site and health is +// probed in the background. +package proxypool + +import ( + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "net" + "net/url" + "strconv" + "strings" +) + +// Node is a parsed proxy node. +type Node struct { + Kind string // "xray" (share link) or "url" (plain endpoint) + Scheme string // vless / vmess / trojan / ss / http / https / socks5 + Display string // host:port, for the console + Link string // original input line + URL string // Kind == "url": ready-to-use proxy URL + Outbound map[string]any // Kind == "xray": Xray outbound object (tag added later) +} + +// ParseNode parses one line: an Xray share link (vless/vmess/trojan/ss) or a +// plain endpoint. Accepted plain forms: +// +// scheme://user:pass@host:port +// user:pass@host:port (scheme = defaultScheme) +// host:port (scheme = defaultScheme) +// host:port:user:pass (scheme = defaultScheme) +func ParseNode(line, defaultScheme string) (*Node, error) { + line = strings.TrimSpace(line) + if line == "" { + return nil, errors.New("empty node") + } + switch { + case strings.HasPrefix(line, "vless://"): + return parseVless(line) + case strings.HasPrefix(line, "vmess://"): + return parseVmess(line) + case strings.HasPrefix(line, "trojan://"): + return parseTrojan(line) + case strings.HasPrefix(line, "ss://"): + return parseShadowsocks(line) + default: + return parsePlainEndpoint(line, defaultScheme) + } +} + +func parseVless(raw string) (*Node, error) { + u, err := url.Parse(raw) + if err != nil { + return nil, fmt.Errorf("vless: %w", err) + } + id := u.User.Username() + host, port := u.Hostname(), u.Port() + if id == "" || host == "" || port == "" { + return nil, errors.New("vless: missing uuid / host / port") + } + q := u.Query() + user := map[string]any{"id": id, "encryption": "none"} + if flow := q.Get("flow"); flow != "" { + user["flow"] = flow + } + out := map[string]any{ + "protocol": "vless", + "settings": map[string]any{ + "vnext": []any{map[string]any{ + "address": host, "port": portInt(port), "users": []any{user}, + }}, + }, + "streamSettings": streamSettings(firstNonEmptyStr(q.Get("type"), "tcp"), q.Get("security"), q), + } + return &Node{Kind: "xray", Scheme: "vless", Display: net.JoinHostPort(host, port), Link: raw, Outbound: out}, nil +} + +func parseVmess(raw string) (*Node, error) { + body := strings.TrimPrefix(raw, "vmess://") + if i := strings.IndexAny(body, "#"); i >= 0 { + body = body[:i] + } + dec, err := b64Decode(body) + if err != nil { + return nil, fmt.Errorf("vmess: invalid base64 payload: %w", err) + } + var v struct { + Add string `json:"add"` + Port any `json:"port"` + ID string `json:"id"` + Aid any `json:"aid"` + Scy string `json:"scy"` + Net string `json:"net"` + Type string `json:"type"` + Host string `json:"host"` + Path string `json:"path"` + TLS string `json:"tls"` + SNI string `json:"sni"` + FP string `json:"fp"` + ALPN string `json:"alpn"` + } + if err := json.Unmarshal([]byte(dec), &v); err != nil { + return nil, fmt.Errorf("vmess: bad json: %w", err) + } + port := toInt(v.Port) + if v.Add == "" || port == 0 || v.ID == "" { + return nil, errors.New("vmess: missing add / port / id") + } + security := strings.TrimSpace(v.Scy) + if security == "" { + security = "auto" + } + out := map[string]any{ + "protocol": "vmess", + "settings": map[string]any{ + "vnext": []any{map[string]any{ + "address": v.Add, "port": port, + "users": []any{map[string]any{"id": v.ID, "alterId": toInt(v.Aid), "security": security}}, + }}, + }, + } + network := strings.ToLower(strings.TrimSpace(v.Net)) + if network == "" { + network = "tcp" + } + ss := map[string]any{"network": network} + switch network { + case "ws": + ws := map[string]any{} + if v.Path != "" { + ws["path"] = v.Path + } + if v.Host != "" { + ws["headers"] = map[string]any{"Host": v.Host} + } + ss["wsSettings"] = ws + case "h2": + h2 := map[string]any{} + if v.Path != "" { + h2["path"] = v.Path + } + if v.Host != "" { + h2["host"] = []any{v.Host} + } + ss["h2Settings"] = h2 + case "grpc": + g := map[string]any{} + if v.Path != "" { + g["serviceName"] = v.Path + } + ss["grpcSettings"] = g + case "tcp": + if strings.EqualFold(v.Type, "http") { + req := map[string]any{"path": []any{firstNonEmptyStr(v.Path, "/")}} + if v.Host != "" { + req["headers"] = map[string]any{"Host": []any{v.Host}} + } + ss["tcpSettings"] = map[string]any{"header": map[string]any{"type": "http", "request": req}} + } + case "httpupgrade", "xhttp": + sub := map[string]any{} + if v.Path != "" { + sub["path"] = v.Path + } + if v.Host != "" { + sub["host"] = v.Host + } + ss[network+"Settings"] = sub + } + if strings.EqualFold(v.TLS, "tls") { + ss["security"] = "tls" + t := map[string]any{} + if sni := firstNonEmptyStr(v.SNI, v.Host); sni != "" { + t["serverName"] = sni + } + if v.FP != "" { + t["fingerprint"] = v.FP + } + if v.ALPN != "" { + t["alpn"] = strings.Split(v.ALPN, ",") + } + ss["tlsSettings"] = t + } + out["streamSettings"] = ss + return &Node{Kind: "xray", Scheme: "vmess", Display: net.JoinHostPort(v.Add, strconv.Itoa(port)), Link: raw, Outbound: out}, nil +} + +func parseTrojan(raw string) (*Node, error) { + u, err := url.Parse(raw) + if err != nil { + return nil, fmt.Errorf("trojan: %w", err) + } + pass := u.User.Username() + host, port := u.Hostname(), u.Port() + if pass == "" || host == "" || port == "" { + return nil, errors.New("trojan: missing password / host / port") + } + q := u.Query() + security := q.Get("security") + if security == "" { + security = "tls" + } + out := map[string]any{ + "protocol": "trojan", + "settings": map[string]any{ + "servers": []any{map[string]any{"address": host, "port": portInt(port), "password": pass}}, + }, + "streamSettings": streamSettings(firstNonEmptyStr(q.Get("type"), "tcp"), security, q), + } + return &Node{Kind: "xray", Scheme: "trojan", Display: net.JoinHostPort(host, port), Link: raw, Outbound: out}, nil +} + +func parseShadowsocks(raw string) (*Node, error) { + body := strings.TrimPrefix(raw, "ss://") + if i := strings.Index(body, "#"); i >= 0 { + body = body[:i] + } + var method, pass, hostport string + if at := strings.LastIndex(body, "@"); at >= 0 { + userinfo, hp := body[:at], body[at+1:] + if i := strings.IndexAny(hp, "?"); i >= 0 { + hp = hp[:i] + } + if strings.Contains(userinfo, ":") { + // plain method:password (uncommon) + } else { + dec, err := b64Decode(userinfo) + if err != nil { + return nil, errors.New("ss: cannot decode credentials") + } + userinfo = dec + } + mp := strings.SplitN(userinfo, ":", 2) + if len(mp) != 2 || mp[0] == "" || mp[1] == "" { + return nil, errors.New("ss: bad credentials") + } + method, pass, hostport = mp[0], mp[1], hp + } else { + // legacy: ss://base64(method:password@host:port) + dec, err := b64Decode(body) + if err != nil { + return nil, errors.New("ss: invalid base64 payload") + } + at := strings.LastIndex(dec, "@") + if at < 0 { + return nil, errors.New("ss: missing @host") + } + mp := strings.SplitN(dec[:at], ":", 2) + if len(mp) != 2 { + return nil, errors.New("ss: bad credentials") + } + method, pass, hostport = mp[0], mp[1], dec[at+1:] + } + host, port, err := splitHostPort(hostport) + if err != nil { + return nil, fmt.Errorf("ss: %w", err) + } + out := map[string]any{ + "protocol": "shadowsocks", + "settings": map[string]any{ + "servers": []any{map[string]any{"address": host, "port": port, "method": method, "password": pass}}, + }, + } + return &Node{Kind: "xray", Scheme: "ss", Display: net.JoinHostPort(host, strconv.Itoa(port)), Link: raw, Outbound: out}, nil +} + +func parsePlainEndpoint(line, defaultScheme string) (*Node, error) { + if defaultScheme = strings.ToLower(strings.TrimSpace(defaultScheme)); defaultScheme == "" { + defaultScheme = "http" + } + build := func(scheme, host, port, user, pass string) (*Node, error) { + scheme = normalizeScheme(scheme) + if scheme == "" { + return nil, fmt.Errorf("unsupported proxy scheme %q (http/https/socks5)", scheme) + } + if host == "" || port == "" { + return nil, errors.New("missing host or port") + } + u := &url.URL{Scheme: scheme, Host: net.JoinHostPort(host, port)} + if user != "" { + u.User = url.UserPassword(user, pass) + } + return &Node{ + Kind: "url", Scheme: scheme, + Display: net.JoinHostPort(host, port), + Link: line, URL: u.String(), + }, nil + } + + switch { + case strings.Contains(line, "://"): + u, err := url.Parse(line) + if err != nil { + return nil, fmt.Errorf("bad proxy url: %w", err) + } + user, pass := "", "" + if u.User != nil { + user = u.User.Username() + pass, _ = u.User.Password() + } + return build(u.Scheme, u.Hostname(), u.Port(), user, pass) + case strings.Contains(line, "@"): + at := strings.LastIndex(line, "@") + userinfo, hostport := line[:at], line[at+1:] + mp := strings.SplitN(userinfo, ":", 2) + user, pass := mp[0], "" + if len(mp) == 2 { + pass = mp[1] + } + host, port, err := splitHostPort(hostport) + if err != nil { + return nil, err + } + return build(defaultScheme, host, strconv.Itoa(port), user, pass) + default: + parts := strings.Split(line, ":") + switch len(parts) { + case 2: + return build(defaultScheme, parts[0], parts[1], "", "") + case 4: + // ip:port:user:pass + return build(defaultScheme, parts[0], parts[1], parts[2], parts[3]) + default: + return nil, fmt.Errorf("unrecognized endpoint %q", line) + } + } +} + +// ParseSubscriptionBody turns a subscription response into node lines. The +// body may be plain text (one node per line) or a base64 blob of that text. +func ParseSubscriptionBody(raw []byte) []string { + s := strings.TrimSpace(string(raw)) + if s == "" { + return nil + } + if !strings.Contains(s, "://") { + if dec, err := b64Decode(strings.ReplaceAll(s, "\n", "")); err == nil { + s = dec + } + } + out := make([]string, 0, 16) + for _, line := range strings.Split(s, "\n") { + line = strings.TrimSpace(line) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + out = append(out, line) + } + return out +} + +// ── helpers ────────────────────────────────────────────────────── + +// streamSettings builds Xray streamSettings from share-link query params. +func streamSettings(network, security string, q url.Values) map[string]any { + network = strings.ToLower(strings.TrimSpace(network)) + if network == "" { + network = "tcp" + } + ss := map[string]any{"network": network} + switch network { + case "ws": + ws := map[string]any{} + if p := q.Get("path"); p != "" { + ws["path"] = p + } + if h := q.Get("host"); h != "" { + ws["headers"] = map[string]any{"Host": h} + } + ss["wsSettings"] = ws + case "grpc": + g := map[string]any{} + if s := q.Get("serviceName"); s != "" { + g["serviceName"] = s + } + ss["grpcSettings"] = g + case "httpupgrade": + h := map[string]any{} + if p := q.Get("path"); p != "" { + h["path"] = p + } + if hh := q.Get("host"); hh != "" { + h["host"] = hh + } + ss["httpupgradeSettings"] = h + case "xhttp": + x := map[string]any{} + if p := q.Get("path"); p != "" { + x["path"] = p + } + if hh := q.Get("host"); hh != "" { + x["host"] = hh + } + if m := q.Get("mode"); m != "" { + x["mode"] = m + } + ss["xhttpSettings"] = x + case "tcp": + if strings.EqualFold(q.Get("headerType"), "http") { + req := map[string]any{} + if h := q.Get("host"); h != "" { + req["headers"] = map[string]any{"Host": []any{h}} + } + if p := q.Get("path"); p != "" { + req["path"] = []any{p} + } + ss["tcpSettings"] = map[string]any{"header": map[string]any{"type": "http", "request": req}} + } + } + switch strings.ToLower(security) { + case "tls": + t := map[string]any{} + if sni := firstNonEmptyStr(q.Get("sni"), q.Get("host")); sni != "" { + t["serverName"] = sni + } + if fp := q.Get("fp"); fp != "" { + t["fingerprint"] = fp + } + if alpn := q.Get("alpn"); alpn != "" { + t["alpn"] = strings.Split(alpn, ",") + } + ss["security"] = "tls" + ss["tlsSettings"] = t + case "reality": + r := map[string]any{} + if sni := q.Get("sni"); sni != "" { + r["serverName"] = sni + } + if fp := q.Get("fp"); fp != "" { + r["fingerprint"] = fp + } + if pbk := q.Get("pbk"); pbk != "" { + r["publicKey"] = pbk + } + if sid := q.Get("sid"); sid != "" { + r["shortId"] = sid + } + if spx := q.Get("spx"); spx != "" { + r["spiderX"] = spx + } + ss["security"] = "reality" + ss["realitySettings"] = r + } + return ss +} + +func normalizeScheme(s string) string { + switch strings.ToLower(strings.TrimSpace(s)) { + case "http": + return "http" + case "https": + return "https" + case "socks", "socks5", "socks5h": + return "socks5" + default: + return "" + } +} + +func b64Decode(s string) (string, error) { + s = strings.TrimSpace(s) + trimmed := strings.TrimRight(s, "=") + for _, enc := range []*base64.Encoding{ + base64.RawURLEncoding, base64.URLEncoding, + base64.RawStdEncoding, base64.StdEncoding, + } { + if b, err := enc.DecodeString(trimmed); err == nil { + return string(b), nil + } + } + return "", errors.New("invalid base64") +} + +func firstNonEmptyStr(vals ...string) string { + for _, v := range vals { + if strings.TrimSpace(v) != "" { + return v + } + } + return "" +} + +func portInt(s string) int { + n, _ := strconv.Atoi(strings.TrimSpace(s)) + return n +} + +func toInt(v any) int { + switch x := v.(type) { + case float64: + return int(x) + case int: + return x + case int64: + return int(x) + case string: + n, _ := strconv.Atoi(strings.TrimSpace(x)) + return n + default: + return 0 + } +} + +func splitHostPort(hostport string) (string, int, error) { + host, port, err := net.SplitHostPort(strings.TrimSpace(hostport)) + if err != nil { + return "", 0, fmt.Errorf("bad host:port %q", hostport) + } + n, err := strconv.Atoi(port) + if err != nil || n < 1 || n > 65535 { + return "", 0, fmt.Errorf("bad port in %q", hostport) + } + return host, n, nil +} diff --git a/internal/proxypool/parse_test.go b/internal/proxypool/parse_test.go new file mode 100644 index 0000000..125d92a --- /dev/null +++ b/internal/proxypool/parse_test.go @@ -0,0 +1,145 @@ +package proxypool + +import ( + "encoding/base64" + "testing" +) + +func TestParseVless(t *testing.T) { + line := "vless://11111111-2222-3333-4444-555555555555@example.com:443" + + "?type=ws&security=tls&sni=sni.example.com&path=%2Fws&host=cdn.example.com&fp=chrome#node" + n, err := ParseNode(line, "http") + if err != nil { + t.Fatalf("parse: %v", err) + } + if n.Kind != "xray" || n.Scheme != "vless" || n.Display != "example.com:443" { + t.Fatalf("node = %+v", n) + } + if n.Outbound["protocol"] != "vless" { + t.Fatalf("protocol = %v", n.Outbound["protocol"]) + } + settings := n.Outbound["settings"].(map[string]any) + vnext := settings["vnext"].([]any)[0].(map[string]any) + if vnext["address"] != "example.com" || vnext["port"] != 443 { + t.Fatalf("vnext = %+v", vnext) + } + users := vnext["users"].([]any)[0].(map[string]any) + if users["id"] != "11111111-2222-3333-4444-555555555555" || users["encryption"] != "none" { + t.Fatalf("users = %+v", users) + } + ss := n.Outbound["streamSettings"].(map[string]any) + if ss["network"] != "ws" || ss["security"] != "tls" { + t.Fatalf("stream = %+v", ss) + } + ws := ss["wsSettings"].(map[string]any) + if ws["path"] != "/ws" { + t.Fatalf("ws = %+v", ws) + } + tls := ss["tlsSettings"].(map[string]any) + if tls["serverName"] != "sni.example.com" { + t.Fatalf("tls = %+v", tls) + } +} + +func TestParseVmess(t *testing.T) { + payload := `{"v":"2","ps":"n","add":"vm.example.com","port":"443","id":"uuid-1",` + + `"aid":"0","scy":"auto","net":"ws","type":"none","host":"vm.example.com",` + + `"path":"/vm","tls":"tls","sni":"vm.example.com"}` + line := "vmess://" + base64.StdEncoding.EncodeToString([]byte(payload)) + n, err := ParseNode(line, "") + if err != nil { + t.Fatalf("parse: %v", err) + } + if n.Kind != "xray" || n.Scheme != "vmess" || n.Display != "vm.example.com:443" { + t.Fatalf("node = %+v", n) + } + ss := n.Outbound["streamSettings"].(map[string]any) + if ss["network"] != "ws" || ss["security"] != "tls" { + t.Fatalf("stream = %+v", ss) + } + if ss["wsSettings"].(map[string]any)["path"] != "/vm" { + t.Fatalf("ws = %+v", ss["wsSettings"]) + } +} + +func TestParseTrojan(t *testing.T) { + n, err := ParseNode("trojan://secret@tr.example.com:443?security=tls&sni=tr.example.com#t", "") + if err != nil { + t.Fatalf("parse: %v", err) + } + if n.Scheme != "trojan" || n.Display != "tr.example.com:443" { + t.Fatalf("node = %+v", n) + } + settings := n.Outbound["settings"].(map[string]any) + srv := settings["servers"].([]any)[0].(map[string]any) + if srv["address"] != "tr.example.com" || srv["port"] != 443 || srv["password"] != "secret" { + t.Fatalf("server = %+v", srv) + } +} + +func TestParseShadowsocks(t *testing.T) { + // SIP002: base64(method:password)@host:port + ui := base64.RawURLEncoding.EncodeToString([]byte("aes-256-gcm:pass:word")) + n, err := ParseNode("ss://"+ui+"@ss.example.com:8388#x", "") + if err != nil { + t.Fatalf("sip002 parse: %v", err) + } + srv := n.Outbound["settings"].(map[string]any)["servers"].([]any)[0].(map[string]any) + if srv["method"] != "aes-256-gcm" || srv["password"] != "pass:word" || srv["address"] != "ss.example.com" { + t.Fatalf("server = %+v", srv) + } + + // legacy: base64(method:password@host:port) + legacy := base64.RawURLEncoding.EncodeToString([]byte("aes-256-gcm:secret@ss.example.com:8388")) + n2, err := ParseNode("ss://"+legacy, "") + if err != nil { + t.Fatalf("legacy parse: %v", err) + } + srv2 := n2.Outbound["settings"].(map[string]any)["servers"].([]any)[0].(map[string]any) + if srv2["password"] != "secret" || srv2["address"] != "ss.example.com" || srv2["port"] != 8388 { + t.Fatalf("legacy server = %+v", srv2) + } +} + +func TestParsePlainEndpoints(t *testing.T) { + cases := []struct { + line, def, wantURL, wantDisplay string + }{ + {"socks5://u:p@1.2.3.4:1080", "http", "socks5://u:p@1.2.3.4:1080", "1.2.3.4:1080"}, + {"http://1.2.3.4:8080", "socks5", "http://1.2.3.4:8080", "1.2.3.4:8080"}, + {"1.2.3.4:8080", "http", "http://1.2.3.4:8080", "1.2.3.4:8080"}, + {"1.2.3.4:8080", "socks5", "socks5://1.2.3.4:8080", "1.2.3.4:8080"}, + {"u:p@1.2.3.4:1080", "http", "http://u:p@1.2.3.4:1080", "1.2.3.4:1080"}, + {"1.2.3.4:8080:u:p", "http", "http://u:p@1.2.3.4:8080", "1.2.3.4:8080"}, + } + for _, c := range cases { + n, err := ParseNode(c.line, c.def) + if err != nil { + t.Fatalf("%q: %v", c.line, err) + } + if n.Kind != "url" || n.URL != c.wantURL || n.Display != c.wantDisplay { + t.Errorf("%q => kind=%s url=%q display=%q", c.line, n.Kind, n.URL, n.Display) + } + } +} + +func TestParseSubscriptionBody(t *testing.T) { + plain := "vless://a@b:1\n# comment\nsocks5://x:y@1.2.3.4:5\n\n" + got := ParseSubscriptionBody([]byte(plain)) + if len(got) != 2 || got[0] != "vless://a@b:1" || got[1] != "socks5://x:y@1.2.3.4:5" { + t.Fatalf("plain parse = %v", got) + } + enc := base64.StdEncoding.EncodeToString([]byte(plain)) + got2 := ParseSubscriptionBody([]byte(enc)) + if len(got2) != 2 || got2[0] != got[0] { + t.Fatalf("base64 parse = %v", got2) + } +} + +func TestParseNodeRejectsJunk(t *testing.T) { + for _, bad := range []string{"", "hello world", "vless://no-port", "ss://@@@", "1.2.3.4", "ftp://1.2.3.4:21"} { + if _, err := ParseNode(bad, "http"); err == nil { + t.Errorf("%q should fail to parse", bad) + } + } +} diff --git a/internal/proxypool/xray.go b/internal/proxypool/xray.go new file mode 100644 index 0000000..67d4457 --- /dev/null +++ b/internal/proxypool/xray.go @@ -0,0 +1,498 @@ +package proxypool + +import ( + "archive/zip" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "net" + "net/http" + "os" + "os/exec" + "path/filepath" + "runtime" + "strings" + "sync" + "time" +) + +// xrayNode is one Xray-backed endpoint, ready to be written into the core +// config. +type xrayNode struct { + Name string + Outbound map[string]any +} + +// xrayCore owns the managed Xray process: one socks inbound per pool node, +// each routed to its own outbound. +type xrayCore struct { + dir string + log *slog.Logger + + mu sync.Mutex + path string + proc *exec.Cmd + exited chan struct{} + running bool + pid int + cfgPath string + cfgHash string + ports map[string]int + lastErr string + tail *logTail +} + +// XrayStatus is the console view of the core. +type XrayStatus struct { + Path string `json:"path"` + Running bool `json:"running"` + PID int `json:"pid"` + Error string `json:"error,omitempty"` +} + +func newXrayCore(dir string, log *slog.Logger) *xrayCore { + if log == nil { + log = slog.Default() + } + return &xrayCore{dir: dir, log: log, ports: map[string]int{}} +} + +func (c *xrayCore) status() XrayStatus { + c.mu.Lock() + defer c.mu.Unlock() + return XrayStatus{Path: c.path, Running: c.running, PID: c.pid, Error: c.lastErr} +} + +// ensure resolves the Xray binary: explicit path, then a previously downloaded +// copy, then (optionally) a fresh download. The download runs unlocked so +// status reads never stall behind a slow network. +func (c *xrayCore) ensure(explicitPath string, autoDownload bool, version string) error { + c.mu.Lock() + if c.path != "" { + if _, err := os.Stat(c.path); err == nil { + c.mu.Unlock() + return nil + } + c.path = "" + } + if p := strings.TrimSpace(explicitPath); p != "" { + if _, err := os.Stat(p); err != nil { + c.mu.Unlock() + return fmt.Errorf("xray_path: %w", err) + } + c.path = p + c.mu.Unlock() + return nil + } + target := filepath.Join(c.dir, "xray"+exeSuffix()) + if _, err := os.Stat(target); err == nil { + c.path = target + c.mu.Unlock() + return nil + } + if !autoDownload { + c.mu.Unlock() + return errors.New("xray binary not found — set [proxypool].xray_path, or enable xray_auto_download") + } + c.mu.Unlock() + + if err := c.download(target, version); err != nil { + c.mu.Lock() + c.lastErr = err.Error() + c.mu.Unlock() + return err + } + c.mu.Lock() + c.path = target + c.mu.Unlock() + return nil +} + +func (c *xrayCore) download(target, version string) error { + if err := os.MkdirAll(c.dir, 0o755); err != nil { + return err + } + osName := map[string]string{"windows": "windows", "linux": "linux", "darwin": "macos"}[runtime.GOOS] + arch := map[string]string{"amd64": "64", "arm64": "arm64-v8a", "386": "32", "arm": "arm32-v7a"}[runtime.GOARCH] + if osName == "" || arch == "" { + return fmt.Errorf("xray: unsupported platform %s/%s", runtime.GOOS, runtime.GOARCH) + } + asset := fmt.Sprintf("Xray-%s-%s.zip", osName, arch) + downloadURL := "https://github.com/XTLS/Xray-core/releases/latest/download/" + asset + if v := strings.TrimSpace(version); v != "" { + if !strings.HasPrefix(v, "v") { + v = "v" + v + } + downloadURL = fmt.Sprintf("https://github.com/XTLS/Xray-core/releases/download/%s/%s", v, asset) + } + c.log.Info("downloading xray core", "asset", asset) + zipPath := target + ".download" + if err := httpDownload(downloadURL, zipPath, 180*time.Second); err != nil { + return err + } + defer os.Remove(zipPath) + if want, err := fetchDigest(downloadURL + ".dgst"); err != nil { + c.log.Warn("xray checksum unavailable, skipping verification", "error", err) + } else if want != "" { + got, err := fileSHA256(zipPath) + if err != nil { + return err + } + if !strings.EqualFold(got, want) { + return fmt.Errorf("xray checksum mismatch: got %s want %s", got, want) + } + } + if err := extractZipBinary(zipPath, target); err != nil { + return err + } + _ = os.Chmod(target, 0o755) + c.log.Info("xray core ready", "path", target) + return nil +} + +// configure regenerates the core config for the given nodes and restarts the +// process when something changed. Ports are kept stable across restarts so +// cached cookies stay valid. +func (c *xrayCore) configure(nodes []xrayNode) error { + c.mu.Lock() + defer c.mu.Unlock() + if len(nodes) == 0 { + c.stopLocked() + c.cfgHash = "" + return nil + } + if c.path == "" { + return errors.New("xray binary unavailable") + } + for _, n := range nodes { + if c.ports[n.Name] == 0 { + p, err := freePort() + if err != nil { + return err + } + c.ports[n.Name] = p + } + } + if c.configHash(nodes) == c.cfgHash && c.running { + return nil + } + // A restart is due. Stop first so our own listeners do not count as + // "port busy", then reclaim or reassign every assigned port — another + // local core may have raced us for one, since the OS only guaranteed it + // at allocation time. + c.stopLocked() + for _, n := range nodes { + if !portAvailable(c.ports[n.Name]) { + p, err := freePort() + if err != nil { + return err + } + c.ports[n.Name] = p + } + } + raw, err := json.Marshal(buildXrayConfig(nodes, c.ports)) + if err != nil { + return err + } + sum := sha256.Sum256(raw) + c.cfgHash = hex.EncodeToString(sum[:]) + cfgPath := filepath.Join(c.dir, "config.json") + if err := os.MkdirAll(c.dir, 0o755); err != nil { + return err + } + if err := os.WriteFile(cfgPath, raw, 0o600); err != nil { + return err + } + c.cfgPath = cfgPath + return c.startLocked(nodes) +} + +// configHash is the digest of the config that would be generated for nodes +// with the current port assignments. +func (c *xrayCore) configHash(nodes []xrayNode) string { + raw, err := json.Marshal(buildXrayConfig(nodes, c.ports)) + if err != nil { + return "" + } + sum := sha256.Sum256(raw) + return hex.EncodeToString(sum[:]) +} + +// portAvailable reports whether 127.0.0.1:port can still be claimed. +func portAvailable(port int) bool { + if port <= 0 { + return false + } + l, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", port)) + if err != nil { + return false + } + l.Close() + return true +} + +func (c *xrayCore) startLocked(nodes []xrayNode) error { + cmd := exec.Command(c.path, "run", "-c", c.cfgPath) + c.tail = &logTail{max: 40} + cmd.Stdout = c.tail + cmd.Stderr = c.tail + if err := cmd.Start(); err != nil { + c.lastErr = "start xray: " + err.Error() + return errors.New(c.lastErr) + } + c.proc = cmd + c.pid = cmd.Process.Pid + c.running = true + c.lastErr = "" + exited := make(chan struct{}) + c.exited = exited + go func() { + _ = cmd.Wait() + c.mu.Lock() + if c.proc == cmd { + c.running = false + c.pid = 0 + if t := c.tail.Last(); t != "" { + c.lastErr = "xray exited: " + t + } else { + c.lastErr = "xray exited" + } + } + c.mu.Unlock() + close(exited) + }() + // Wait until the local socks inbounds answer. + deadline := time.Now().Add(8 * time.Second) + for time.Now().Before(deadline) { + ok := true + for _, n := range nodes { + conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", c.ports[n.Name]), 300*time.Millisecond) + if err != nil { + ok = false + break + } + conn.Close() + } + if ok { + return nil + } + if !c.running { + break + } + time.Sleep(200 * time.Millisecond) + } + if !c.running { + if c.lastErr == "" { + c.lastErr = "xray failed to start" + } + return errors.New(c.lastErr) + } + c.lastErr = "xray socks ports did not come up in time" + return errors.New(c.lastErr) +} + +func (c *xrayCore) stopLocked() { + if c.proc != nil && c.proc.Process != nil { + _ = c.proc.Process.Kill() + } + if c.exited != nil { + select { + case <-c.exited: + case <-time.After(3 * time.Second): + } + } + c.proc = nil + c.exited = nil + c.running = false + c.pid = 0 +} + +func (c *xrayCore) stop() { + c.mu.Lock() + defer c.mu.Unlock() + c.stopLocked() +} + +// port returns the local socks port assigned to a node name. +func (c *xrayCore) port(name string) int { + c.mu.Lock() + defer c.mu.Unlock() + return c.ports[name] +} + +func buildXrayConfig(nodes []xrayNode, ports map[string]int) map[string]any { + inbounds := make([]any, 0, len(nodes)) + outbounds := make([]any, 0, len(nodes)+2) + rules := make([]any, 0, len(nodes)) + for i, n := range nodes { + inTag := fmt.Sprintf("in-%d", i) + outTag := fmt.Sprintf("out-%d", i) + inbounds = append(inbounds, map[string]any{ + "tag": inTag, "listen": "127.0.0.1", "port": ports[n.Name], + "protocol": "socks", + "settings": map[string]any{"udp": false, "auth": "noauth"}, + }) + ob := make(map[string]any, len(n.Outbound)+1) + for k, v := range n.Outbound { + ob[k] = v + } + ob["tag"] = outTag + outbounds = append(outbounds, ob) + rules = append(rules, map[string]any{"type": "field", "inboundTag": []any{inTag}, "outboundTag": outTag}) + } + outbounds = append(outbounds, + map[string]any{"tag": "direct", "protocol": "freedom"}, + map[string]any{"tag": "blocked", "protocol": "blackhole"}, + ) + return map[string]any{ + "log": map[string]any{"loglevel": "warning", "access": "none"}, + "inbounds": inbounds, + "outbounds": outbounds, + "routing": map[string]any{"domainStrategy": "AsIs", "rules": rules}, + } +} + +// ── platform / io helpers ──────────────────────────────────────── + +func exeSuffix() string { + if runtime.GOOS == "windows" { + return ".exe" + } + return "" +} + +func freePort() (int, error) { + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + return 0, err + } + defer l.Close() + return l.Addr().(*net.TCPAddr).Port, nil +} + +func httpDownload(url, dst string, timeout time.Duration) error { + client := &http.Client{Timeout: timeout} + resp, err := client.Get(url) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("download %s: http %d", url, resp.StatusCode) + } + f, err := os.Create(dst) + if err != nil { + return err + } + defer f.Close() + if _, err := io.Copy(f, io.LimitReader(resp.Body, 256<<20)); err != nil { + return err + } + return f.Sync() +} + +func fetchDigest(url string) (string, error) { + client := &http.Client{Timeout: 20 * time.Second} + resp, err := client.Get(url) + if err != nil { + return "", err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return "", fmt.Errorf("http %d", resp.StatusCode) + } + raw, err := io.ReadAll(io.LimitReader(resp.Body, 64<<10)) + if err != nil { + return "", err + } + for _, line := range strings.Split(string(raw), "\n") { + key, val, ok := strings.Cut(line, ":") + if !ok { + continue + } + if strings.Contains(strings.ToUpper(key), "SHA2-256") || strings.Contains(strings.ToUpper(key), "SHA256") { + if v := strings.TrimSpace(strings.Fields(val)[0]); v != "" { + return v, nil + } + } + } + return "", errors.New("no sha256 digest found") +} + +func fileSHA256(path string) (string, error) { + f, err := os.Open(path) + if err != nil { + return "", err + } + defer f.Close() + h := sha256.New() + if _, err := io.Copy(h, f); err != nil { + return "", err + } + return hex.EncodeToString(h.Sum(nil)), nil +} + +func extractZipBinary(zipPath, target string) error { + r, err := zip.OpenReader(zipPath) + if err != nil { + return fmt.Errorf("open archive: %w", err) + } + defer r.Close() + want := "xray" + exeSuffix() + for _, f := range r.File { + if !strings.EqualFold(filepath.Base(f.Name), want) { + continue + } + rc, err := f.Open() + if err != nil { + return err + } + defer rc.Close() + out, err := os.OpenFile(target, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o755) + if err != nil { + return err + } + if _, err := io.Copy(out, rc); err != nil { + out.Close() + return err + } + return out.Close() + } + return errors.New("no xray binary inside the archive") +} + +// logTail keeps the last few process log lines for status reporting. +type logTail struct { + mu sync.Mutex + lines []string + max int +} + +func (t *logTail) Write(p []byte) (int, error) { + t.mu.Lock() + defer t.mu.Unlock() + for _, line := range strings.Split(strings.TrimRight(string(p), "\r\n"), "\n") { + line = strings.TrimSpace(line) + if line == "" { + continue + } + t.lines = append(t.lines, line) + } + if len(t.lines) > t.max { + t.lines = t.lines[len(t.lines)-t.max:] + } + return len(p), nil +} + +func (t *logTail) Last() string { + t.mu.Lock() + defer t.mu.Unlock() + if len(t.lines) == 0 { + return "" + } + return t.lines[len(t.lines)-1] +} diff --git a/internal/proxypool/xray_smoke_test.go b/internal/proxypool/xray_smoke_test.go new file mode 100644 index 0000000..e71bed9 --- /dev/null +++ b/internal/proxypool/xray_smoke_test.go @@ -0,0 +1,80 @@ +package proxypool + +import ( + "net" + "os" + "strings" + "testing" + + "github.com/nyoungo/dsfree2api/internal/config" +) + +// TestXrayCoreSmoke drives the managed core against the locally placed +// binary (data/xray.exe) with a syntactically real VLESS node. It never +// connects upstream — it only proves that ensure() picks up the local +// binary, configure() starts the process and the local socks inbound is +// live. Skipped unless opted in: +// +// $env:DSFREE_XRAY_SMOKE=1; go test ./internal/proxypool -run Smoke -v +func TestXrayCoreSmoke(t *testing.T) { + if os.Getenv("DSFREE_XRAY_SMOKE") == "" { + t.Skip("set DSFREE_XRAY_SMOKE=1 to run the local xray smoke test") + } + cfg, err := config.Load(examplePath) + if err != nil { + t.Fatalf("load config: %v", err) + } + cfg.ProxyPool.Enabled = true + cfg.ProxyPool.XrayAutoDownload = false // the local binary must be used as-is + cfg.ProxyPool.Entries = map[string]*config.ProxyEntry{ + "smoke": { + Link: "vless://11111111-2222-3333-4444-555555555555@example.com:443" + + "?security=tls&type=tcp&sni=example.com", + Enabled: true, + }, + } + cfg.Sites["de"].Proxies = []string{"smoke"} + + m := New(cfg, "../../data", nil) + m.Reload() + st := m.Status() + if !st.Xray.Running { + t.Fatalf("xray core did not start: %+v", st.Xray) + } + t.Logf("xray running: pid=%d path=%s", st.Xray.PID, st.Xray.Path) + + cands := m.Candidates("de") + if len(cands) != 1 || !strings.HasPrefix(cands[0].Proxy, "socks5://127.0.0.1:") { + t.Fatalf("candidates = %+v", cands) + } + t.Logf("candidate proxy: %s", cands[0].Proxy) + + m.xray.stop() + if m.Status().Xray.Running { + t.Fatal("xray still running after stop") + } + + // Port stability: a restart re-claims the same socks port, so cached + // cookies (keyed by the local proxy URL) stay valid. + m.Reload() + again := m.Candidates("de") + if len(again) != 1 || again[0].Proxy != cands[0].Proxy { + t.Fatalf("port not stable across restart: %q -> %+v", cands[0].Proxy, again) + } + m.xray.stop() + + // Race self-heal: steal the assigned port; the next start must move away. + port := strings.TrimPrefix(cands[0].Proxy, "socks5://127.0.0.1:") + ln, err := net.Listen("tcp", "127.0.0.1:"+port) + if err != nil { + t.Fatalf("occupy port %s: %v", port, err) + } + defer ln.Close() + m.Reload() + third := m.Candidates("de") + if len(third) != 1 || third[0].Proxy == cands[0].Proxy { + t.Fatalf("stolen port was not reassigned: %+v", third) + } + t.Logf("port moved %s -> %s", cands[0].Proxy, third[0].Proxy) + m.xray.stop() +} diff --git a/internal/upstream/client.go b/internal/upstream/client.go index d068935..6c68c85 100644 --- a/internal/upstream/client.go +++ b/internal/upstream/client.go @@ -33,6 +33,19 @@ func (r Route) Key() string { return r.Proxy } +// RouteCandidate is one egress route supplied by the proxy pool. +type RouteCandidate struct { + Name string + Proxy string +} + +// RouteSource resolves per-site route candidates and records request +// outcomes. Implemented by proxypool.Manager; nil when the pool is unused. +type RouteSource interface { + Candidates(site string) []RouteCandidate + Report(proxy string, ok bool) +} + type chatConfig struct { BotID int PostID int @@ -42,9 +55,10 @@ type chatConfig struct { } type Client struct { - cfg *config.Config - ts *turnstile.Solver - log *slog.Logger + cfg *config.Config + ts *turnstile.Solver + log *slog.Logger + pool RouteSource stateMu sync.Mutex chatCfg map[string]chatConfig @@ -64,7 +78,7 @@ type Client struct { prompt string, route Route, info *ServeInfo, yield func(Event) error) error } -func New(cfg *config.Config, ts *turnstile.Solver, log *slog.Logger) *Client { +func New(cfg *config.Config, ts *turnstile.Solver, log *slog.Logger, pool RouteSource) *Client { if log == nil { log = slog.Default() } @@ -72,6 +86,7 @@ func New(cfg *config.Config, ts *turnstile.Solver, log *slog.Logger) *Client { cfg: cfg, ts: ts, log: log, + pool: pool, chatCfg: map[string]chatConfig{}, cfgLocks: map[string]*sync.Mutex{}, gates: map[string]chan struct{}{}, @@ -239,6 +254,40 @@ func (c *Client) logCacheReject(site, modelID, prompt string) { "head", strconv.Quote(head), "tail", strconv.Quote(tail)) } +// routesFor returns the ordered candidate routes for one site: proxy-pool +// candidates first (sticky, health filtered), then the global primary and +// fallback lines as the safety net. +func (c *Client) routesFor(siteCode string) []Route { + out := make([]Route, 0, 4) + seen := map[string]bool{} + add := func(r Route) { + key := r.Key() + if seen[key] { + return + } + seen[key] = true + out = append(out, r) + } + if c.pool != nil { + for _, cand := range c.pool.Candidates(siteCode) { + add(Route{Name: cand.Name, Proxy: cand.Proxy}) + } + } + for _, r := range c.Routes() { + add(r) + } + return out +} + +// poolReport feeds a route outcome back to the proxy pool; it is a no-op for +// global routes and the direct connection. +func (c *Client) poolReport(proxy string, ok bool) { + if c.pool == nil || proxy == "" { + return + } + c.pool.Report(proxy, ok) +} + // Chat streams the prompt through the upstream site, honouring route // failover, slow-start detection, session refresh and cross-site failover. func (c *Client) Chat(ctx context.Context, modelID string, prompt string, info *ServeInfo, yield func(Event) error) error { @@ -355,7 +404,7 @@ func (c *Client) chatWithRetries(ctx context.Context, modelID string, model conf if !site.Enabled { return errf("site %s is disabled", model.Site) } - routes := c.Routes() + routes := c.routesFor(site.Code) var lastErr error sawQuota := false @@ -369,6 +418,7 @@ func (c *Client) chatWithRetries(ctx context.Context, modelID string, model conf c.clearQuotaCool(site.Code) c.probeBalanceAsync(site, model, route) } + c.poolReport(route.Proxy, true) return nil } lastErr = err @@ -398,6 +448,11 @@ func (c *Client) chatWithRetries(ctx context.Context, modelID string, model conf if isCacheEmptyProblem(err) { return AsUpstream(err) } + // Quota/session failures are site-side, not proxy-side — do not + // blame the egress endpoint for them. + if !isQuota(err) && !isSession(err) { + c.poolReport(route.Proxy, false) + } failed = append(failed, route) c.logUpstreamErr(site.Code, modelID, attempt, route, err) if isQuota(err) { diff --git a/internal/upstream/client_test.go b/internal/upstream/client_test.go index fb445b2..8e1d3df 100644 --- a/internal/upstream/client_test.go +++ b/internal/upstream/client_test.go @@ -32,7 +32,7 @@ func TestRoutesKeepPrimaryFirstAndDedupe(t *testing.T) { cfg.Proxy.URL = "" cfg.Proxy.FallbackURLs = []string{"socks5://fallback:1", "socks5://fallback:1"} - c := New(cfg, nil, nil) + c := New(cfg, nil, nil, nil) got := c.Routes() want := []Route{{Name: "primary", Proxy: ""}, {Name: "fallback-1", Proxy: "socks5://fallback:1"}} if !reflect.DeepEqual(got, want) { @@ -47,7 +47,7 @@ func TestSlowPrimarySwitchesToFallbackBeforeContent(t *testing.T) { cfg.Proxy.SlowStartSeconds = 0.02 var calls []string - c := New(cfg, nil, nil) + c := New(cfg, nil, nil, nil) c.chatOnceOverride = func(ctx context.Context, site config.Site, modelID string, model config.Model, prompt string, route Route, info *ServeInfo, yield func(Event) error) error { calls = append(calls, route.Name) @@ -86,7 +86,7 @@ func TestPrimaryErrorSwitchesToFallbackBeforeContent(t *testing.T) { cfg.Proxy.FallbackURLs = []string{"socks5://fallback:1"} var calls []string - c := New(cfg, nil, nil) + c := New(cfg, nil, nil, nil) c.chatOnceOverride = func(ctx context.Context, site config.Site, modelID string, model config.Model, prompt string, route Route, info *ServeInfo, yield func(Event) error) error { calls = append(calls, route.Name) @@ -119,7 +119,7 @@ func TestNoFallbackAfterContentStarted(t *testing.T) { cfg.Proxy.FallbackURLs = []string{"socks5://fallback:1"} var calls []string - c := New(cfg, nil, nil) + c := New(cfg, nil, nil, nil) c.chatOnceOverride = func(ctx context.Context, site config.Site, modelID string, model config.Model, prompt string, route Route, info *ServeInfo, yield func(Event) error) error { calls = append(calls, route.Name) @@ -211,7 +211,7 @@ func TestTurnstileRequiredWhileDisabledFailsFast(t *testing.T) { cfg.Upstream.RefreshRetries = 4 var calls int - c := New(cfg, nil, nil) + c := New(cfg, nil, nil, nil) c.chatOnceOverride = func(ctx context.Context, site config.Site, modelID string, model config.Model, prompt string, route Route, info *ServeInfo, yield func(Event) error) error { calls++ @@ -237,7 +237,7 @@ func TestNonceFetchFailureFailsFast(t *testing.T) { cfg.Upstream.RefreshRetries = 4 var calls int - c := New(cfg, nil, nil) + c := New(cfg, nil, nil, nil) c.chatOnceOverride = func(ctx context.Context, site config.Site, modelID string, model config.Model, prompt string, route Route, info *ServeInfo, yield func(Event) error) error { calls++ @@ -273,7 +273,7 @@ func TestSSEDeltaAndDoneEvents(t *testing.T) { func TestQuotaSiteCoolsDownAndPrefersSibling(t *testing.T) { cfg := testConfig(t) - c := New(cfg, nil, nil) + c := New(cfg, nil, nil, nil) calls := map[string]int{} c.chatOnceOverride = func(_ context.Context, site config.Site, modelID string, model config.Model, _ string, _ Route, _ *ServeInfo, _ func(Event) error) error { @@ -318,7 +318,7 @@ func TestCacheEmptyFailsOverToMirrorWithoutRefreshCycles(t *testing.T) { cfg := testConfig(t) cfg.Upstream.AutoRefresh = true cfg.Upstream.RefreshRetries = 3 - c := New(cfg, nil, nil) + c := New(cfg, nil, nil, nil) calls := map[string]int{} c.chatOnceOverride = func(_ context.Context, site config.Site, modelID string, model config.Model, _ string, _ Route, _ *ServeInfo, _ func(Event) error) error { @@ -348,7 +348,7 @@ func TestQuotaRotationRecoversWithoutSolve(t *testing.T) { if _, _, err := solver.ImportCookies("de", "cf_clearance=abc; dsgt_gid="+oldGid, ""); err != nil { t.Fatalf("seed cookies: %v", err) } - c := New(cfg, solver, nil) + c := New(cfg, solver, nil, nil) calls := 0 c.chatOnceOverride = func(_ context.Context, site config.Site, modelID string, model config.Model, _ string, _ Route, _ *ServeInfo, _ func(Event) error) error {