feat(proxypool): 站点级出口代理池（Xray / 订阅 / 端点，粘性选路与健康探测）
- proxypool: 站点级出口绑定 proxies = [条目名, "sub:订阅名"]，粘性选路、 失败切换、后台健康探测；支持 Xray 分享链接 / 订阅 / 裸 ip:port 与 socks5 端点， 托管 Xray 核心（可自动下载）；站点未绑定时回退 [proxy] 主/备线路 - upstream: RouteSource 抽象 + routesFor/poolReport；求解与请求共用同一出口 - admin: 代理池面板与维护动作；overview 暴露池状态
This commit is contained in:
luotianyi committed
1 parent
4cc4504a18
commit
6aae7b65ea
16 files changed
+2701
-21
No files matched your search
@@ -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 缓存
|
||||
|
||||
+12
-2
@@ -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{
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -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 = `
|
||||
<div class="grid g2">
|
||||
<div class="card"><h3>主线路</h3>
|
||||
@@ -355,6 +356,49 @@ PAGES.proxy = async () => {
|
||||
<div class="row"><input id="px-fb" placeholder="socks5://..." style="width:280px"><button class="btn" id="px-add">添加</button></div></div>
|
||||
<div class="table-wrap mt"><table><thead><tr><th>#</th><th>地址</th><th></th></tr></thead><tbody id="tb-fb"></tbody></table></div>
|
||||
<div class="dim mt">当前生效顺序：${ov.cache.routes.map(r => `<span class="tag">${esc(r.name)} → ${esc(r.proxy)}</span>`).join(" ")}</div>
|
||||
</div>
|
||||
<div class="grid g2 mt">
|
||||
<div class="card"><h3>代理池 · Xray 核心</h3>
|
||||
<div class="row">
|
||||
<div class="field"><label>池开关</label>${switchHTML("pool-en", pool.enabled)}</div>
|
||||
<div class="field"><label>检查间隔（秒）</label><input id="pool-check-int" type="number" min="10" value="${pool.check_interval_seconds || 120}"></div>
|
||||
<div class="field"><label>检查超时（秒）</label><input id="pool-check-to" type="number" min="2" value="${pool.check_timeout_seconds || 10}"></div>
|
||||
</div>
|
||||
<div class="row">
|
||||
<div class="field"><label>健康检查 URL</label><input id="pool-check-url" value="${esc(pool.check_url || "")}" placeholder="https://www.gstatic.com/generate_204"></div>
|
||||
<div class="field"><label>裸 ip:port 默认协议</label><input id="pool-scheme" value="${esc(pool.default_scheme || "http")}" placeholder="http / socks5"></div>
|
||||
</div>
|
||||
<div class="row">
|
||||
<div class="field"><label>Xray 路径（留空 = 自动）</label><input id="pool-xray-path" value="${esc(pool.xray_path || "")}" placeholder="留空自动下载到数据目录"></div>
|
||||
<div class="field"><label>版本（留空 = latest）</label><input id="pool-xray-ver" value="${esc(pool.xray_version || "")}"></div>
|
||||
<div class="field"><label>自动下载</label>${switchHTML("pool-xray-auto", pool.xray_auto_download)}</div>
|
||||
</div>
|
||||
<div class="row mt">
|
||||
<button class="btn primary" id="pool-save">保存池配置</button>
|
||||
<button class="btn" id="pool-refresh">刷新节点与健康检查</button>
|
||||
</div>
|
||||
<div class="dim mt">Xray：<span class="pill ${pool.xray.running ? "on" : "off"}">${pool.xray.running ? "运行中 · pid " + pool.xray.pid : "未运行"}</span>
|
||||
${pool.xray.error ? `<span style="color:var(--err)"> ${esc(pool.xray.error)}</span>` : ""}
|
||||
${pool.last_error ? `<br><span style="color:var(--err)">${esc(pool.last_error)}</span>` : ""}</div>
|
||||
</div>
|
||||
<div class="card"><h3>节点与端点</h3>
|
||||
<div class="row"><input id="pool-name" placeholder="名称（字母数字-_）" style="width:180px">
|
||||
<input id="pool-link" placeholder="vless://… 或 ip:port 或 socks5://user:pass@host:port" style="flex:1;min-width:200px">
|
||||
<button class="btn" id="pool-add">添加</button></div>
|
||||
<div class="table-wrap mt"><table><thead><tr><th>名称</th><th>类型</th><th>目标</th><th>健康</th><th class="num">延迟</th><th></th></tr></thead><tbody id="tb-pool"></tbody></table></div>
|
||||
<div class="dim mt">分享链接（vless / vmess / trojan / ss）由内置 Xray 核心承载；ip:port / socks5:// 等端点直接使用。</div>
|
||||
</div>
|
||||
</div>
|
||||
<div class="card mt"><div class="toolbar"><h3 style="margin:0">订阅</h3>
|
||||
<div class="row"><input id="sub-name" placeholder="名称" style="width:150px">
|
||||
<input id="sub-url" placeholder="https://…/sub?token=…" style="width:360px">
|
||||
<input id="sub-int" type="number" min="5" value="60" title="刷新间隔（分钟）" style="width:90px">
|
||||
<button class="btn" id="sub-add">添加订阅</button></div></div>
|
||||
<div class="table-wrap mt"><table><thead><tr><th>名称</th><th>地址</th><th class="num">间隔</th><th class="num">节点</th><th>上次拉取</th><th></th></tr></thead><tbody id="tb-subs"></tbody></table></div>
|
||||
</div>
|
||||
<div class="card mt"><h3>站点绑定（按顺序，粘性选路）</h3>
|
||||
<div class="table-wrap"><table><thead><tr><th>站点</th><th>绑定（逗号分隔：名称 或 sub:订阅名）</th><th>当前出口</th><th></th></tr></thead><tbody id="tb-bind"></tbody></table></div>
|
||||
<div class="dim mt">未绑定、或绑定节点全部不健康时，回退到主线路 / 备用线路。Cookie 与出口 IP 绑定：切换出口后该站点会在下一轮巡检重新求解 Turnstile。</div>
|
||||
</div>`;
|
||||
|
||||
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 ? `<span class="pill off" title="${esc(e.parse_error)}">解析失败</span>`
|
||||
: e.checked ? (e.healthy ? '<span class="pill on">健康</span>' : `<span class="pill off" title="${esc(e.last_error || "")}">不可用</span>`)
|
||||
: '<span class="pill">未探测</span>';
|
||||
return `<tr><td><code>${esc(e.name)}</code>${e.source ? ` <span class="dim">← ${esc(e.source)}</span>` : ""}</td>
|
||||
<td>${esc(e.kind === "url" ? e.scheme : (e.scheme || "xray"))}</td>
|
||||
<td class="dim">${esc(e.target || "")}</td>
|
||||
<td>${health}</td><td class="num">${e.latency_ms ? e.latency_ms + "ms" : "–"}</td>
|
||||
<td class="right"><button class="btn small" data-ptoggle="${esc(e.name)}" data-pen="${e.enabled ? "0" : "1"}">${e.enabled ? "停用" : "启用"}</button>
|
||||
${e.source ? "" : `<button class="btn small danger" data-pdel="${esc(e.name)}">删除</button>`}</td></tr>`;
|
||||
}).join("") : '<tr><td colspan="6" class="empty">尚无节点 —— 添加分享链接或 ip:port 端点</td></tr>';
|
||||
$$("[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 => `
|
||||
<tr><td><code>${esc(s.name)}</code></td><td class="dim">${esc(s.url)}</td>
|
||||
<td class="num">${s.interval_minutes || 60}m</td><td class="num">${s.nodes || 0}</td>
|
||||
<td class="dim">${s.fetched_at ? timeStr(s.fetched_at * 1000) : "未拉取"}${s.last_error ? ` <span style="color:var(--err)">${esc(String(s.last_error).slice(0, 60))}</span>` : ""}</td>
|
||||
<td class="right"><button class="btn small" data-stoggle="${esc(s.name)}" data-sen="${s.enabled ? "0" : "1"}">${s.enabled ? "停用" : "启用"}</button>
|
||||
<button class="btn small danger" data-sdel="${esc(s.name)}">删除</button></td></tr>`).join("")
|
||||
: '<tr><td colspan="6" class="empty">尚无订阅</td></tr>';
|
||||
$$("[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 `<tr><td><b>${esc(s.code.toUpperCase())}</b></td>
|
||||
<td><input data-bind="${esc(s.code)}" value="${esc((s.proxies || []).join(", "))}" placeholder="留空 = 全局线路"></td>
|
||||
<td class="dim">${cur ? `${esc(cur)} → ${esc(curEntry ? curEntry.target : "?")}` : "全局线路"}</td>
|
||||
<td class="right"><button class="btn small" data-bsave="${esc(s.code)}">保存</button></td></tr>`;
|
||||
}).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 = `
|
||||
|
||||
@@ -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:<name>" 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 {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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]
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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) {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in new issue
Block a user