onesvm-browser-server/server/internal/proxymanager/subscription.go

// subscription.go：多订阅容灾拉取 + TTL 缓存 + stale-on-error（design §5.1）。
//
// 安全边界（与 bench/proxy/lib.py / up.sh 对齐）：
//   - 仅 http/https、超时 45s、UA clash-meta/browser-server、capped read 8MB；
//   - 订阅 URL 只进环境变量，解析文本只在内存；URL 与凭据永不入日志。
package proxymanager

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"strings"
	"sync"
	"time"
)

// 订阅拉取参数（design §5.1 / up.sh 同值）。
const (
	subFetchTimeout = 45 * time.Second
	subMaxBytes     = 8 * 1024 * 1024 // 8MB capped read 防 OOM
	subUA           = "clash-meta/browser-server"
	subTTLCache     = 10 * time.Minute
)

// subClient 订阅拉取客户端（独立于 httpx：订阅是机场站点非目标站，
// 不走 Chrome UA 与 Content-Type 白名单）。
type subClient struct {
	hc *http.Client
}

func newSubClient() *subClient {
	hc := &http.Client{
		Timeout: subFetchTimeout,
		Transport: &http.Transport{
			TLSHandshakeTimeout:   10 * time.Second,
			ResponseHeaderTimeout: subFetchTimeout,
			MaxIdleConns:          4,
			IdleConnTimeout:       60 * time.Second,
		},
		// 不跟随跨协议跳转；订阅源应直接给最终地址（up.sh 用 curl -L 语义）。
		CheckRedirect: func(req *http.Request, via []*http.Request) error {
			if len(via) >= 5 {
				return errors.New("subscription: 重定向超 5 跳")
			}
			if req.URL.Scheme != "http" && req.URL.Scheme != "https" {
				return errors.New("subscription: 仅允许 http/https 跳转")
			}
			return nil
		},
	}
	// 订阅拉取专用 bootstrap 出口（非密钥；缺省空=直连）。2026-09-02 实测定时
	// mgr1 直连机场订阅被墙（Mac 可通），复用 mgr3 central-proxy :7890 作
	// bootstrap（deploy-preset §1 扩展期布局预告）；探活仍走自家 mihomo，
	// 此代理仅用于订阅拉取一跳，不污染其它出站。
	if proxyURL := os.Getenv("BROWSER_SERVER_SUB_FETCH_PROXY"); proxyURL != "" {
		if u, err := url.Parse(proxyURL); err == nil && (u.Scheme == "http" || u.Scheme == "https") {
			hc.Transport.(*http.Transport).Proxy = http.ProxyURL(u)
		}
	}
	return &subClient{hc: hc}
}

// fetch 拉取单个订阅 URL：校验 scheme → GET → capped read。
func (s *subClient) fetch(ctx context.Context, rawURL string) ([]byte, error) {
	if !strings.HasPrefix(rawURL, "http://") && !strings.HasPrefix(rawURL, "https://") {
		return nil, fmt.Errorf("subscription: URL 必须为 http(s)：已拒绝")
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, rawURL, nil)
	if err != nil {
		return nil, fmt.Errorf("subscription: 请求构造失败: %w", err)
	}
	req.Header.Set("User-Agent", subUA)
	req.Header.Set("Accept", "*/*")
	resp, err := s.hc.Do(req)
	if err != nil {
		return nil, fmt.Errorf("subscription: 拉取失败（URL 不记日志）: %w", err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("subscription: HTTP %d", resp.StatusCode)
	}
	body, err := io.ReadAll(io.LimitReader(resp.Body, subMaxBytes+1))
	if err != nil {
		return nil, fmt.Errorf("subscription: 读取失败: %w", err)
	}
	if int64(len(body)) > subMaxBytes {
		return nil, fmt.Errorf("subscription: 响应超上限")
	}
	return body, nil
}

// subscriptionCache 多订阅容灾 + TTL 缓存 + stale-on-error。
//
// 容灾语义：按 PROXY_SUB_URLS 顺序逐个拉取，第一个成功者生效；
// 全部失败时若有 stale 缓存则续用（stale-on-error），否则 fail-closed。
type subscriptionCache struct {
	mu       sync.Mutex
	client   *subClient
	urls     []string
	text     []byte    // 最近一次成功正文（stale-on-error 备份）
	fetched  time.Time // 正文拉取时刻
	stale    bool      // 当前正文是否已过 TTL（stale 状态）
	lastErrs []string  // 最近一轮各订阅失败摘要（不含 URL）
}

func newSubscriptionCache(urls []string) *subscriptionCache {
	return &subscriptionCache{client: newSubClient(), urls: urls}
}

// Get 返回订阅正文；fresh（TTL 内）直接用缓存；
// 过期则重拉（全部失败时回退 stale 正文，stale=true）；
// 无任何正文且失败 → fail-closed 错误。
//
// 脱敏：错误摘要只保留序号 + 首行原因；Go http.Client 错误会内嵌完整
// URL（含 query token），必须剥离，订阅 URL 永不出本类型。
func (c *subscriptionCache) Get(ctx context.Context) (text []byte, stale bool, err error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.text != nil && time.Since(c.fetched) < subTTLCache {
		return c.text, false, nil
	}
	var lastErrs []string
	for _, u := range c.urls {
		body, ferr := c.client.fetch(ctx, u)
		if ferr == nil {
			c.text, c.fetched, c.stale, c.lastErrs = body, time.Now(), false, nil
			return c.text, false, nil
		}
		lastErrs = append(lastErrs, sanitizeSubErr(ferr, len(lastErrs)+1))
	}
	c.lastErrs = lastErrs
	if c.text != nil {
		c.stale = true
		return c.text, true, nil
	}
	return nil, false, fmt.Errorf("subscription: 全部订阅拉取失败且无缓存（fail-closed）: %s",
		strings.Join(lastErrs, "; "))
}

// sanitizeSubErr 失败摘要脱敏：仅保留错误类型词与订阅序号，
// 剥离 URL/网络细节（含 Go 标准库内嵌的完整 URL 与 token）。
func sanitizeSubErr(err error, idx int) string {
	msg := err.Error()
	switch {
	case strings.Contains(msg, "context deadline exceeded"), strings.Contains(msg, "Client.Timeout"):
		return fmt.Sprintf("订阅#%d 失败: 超时", idx)
	case strings.Contains(msg, "HTTP "):
		// 仅保留状态码
		i := strings.Index(msg, "HTTP ")
		code := msg[i+5:]
		if len(code) > 3 {
			code = code[:3]
		}
		return fmt.Sprintf("订阅#%d 失败: HTTP %s", idx, strings.TrimSpace(code))
	case strings.Contains(msg, "超上限"):
		return fmt.Sprintf("订阅#%d 失败: 响应超 8MB 上限", idx)
	case strings.Contains(msg, "connection refused"), strings.Contains(msg, "no such host"),
		strings.Contains(msg, "i/o timeout"), strings.Contains(msg, "reset"):
		return fmt.Sprintf("订阅#%d 失败: 网络不可达", idx)
	case strings.Contains(msg, "必须为 http(s)"):
		return fmt.Sprintf("订阅#%d 失败: URL scheme 非法", idx)
	case strings.Contains(msg, "重定向"):
		return fmt.Sprintf("订阅#%d 失败: 重定向超限", idx)
	default:
		return fmt.Sprintf("订阅#%d 失败: 拉取异常", idx)
	}
}

// Stale 当前缓存是否处于 stale-on-error 状态。
func (c *subscriptionCache) Stale() bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.stale
}

// LastErrors 最近一轮拉取失败摘要（无 URL）。
func (c *subscriptionCache) LastErrors() []string {
	c.mu.Lock()
	defer c.mu.Unlock()
	return append([]string(nil), c.lastErrs...)
}