package main // External subscription aggregation — mirrors lib/extsub.js (design: // SUB-AGGREGATION-DESIGN.md). SSRF guard, pinned-IP dialing, redirect // re-validation, 10s timeout, 1MiB cap, 200 items per source. import ( "context" "crypto/sha256" "crypto/tls" "encoding/base64" "encoding/hex" "errors" "fmt" "io" "math" "net" "net/http" "net/url" "regexp" "strings" "time" ) var LinkSchemes = map[string]bool{ "vless:": true, "ss:": true, "hysteria2:": true, "anytls:": true, "socks5:": true, } const maxSubBodyBytes = 1024 * 1024 const maxSubItems = 200 const fetchTimeout = 10 * time.Second const defaultSubUA = "v2rayN/6.55" func ValidateLinkUri(uri string) bool { if uri == "" || len(uri) > 2048 { return false } parsed, err := url.Parse(uri) if err != nil { return false } return LinkSchemes[parsed.Scheme+":"] && parsed.Hostname() != "" } var base64CharsetRe = regexp.MustCompile(`^[A-Za-z0-9+/=_-]+$`) // ParseLinkList mirrors parseLinkList in lib/extsub.js. func ParseLinkList(body string) ([]SubCacheItem, int) { text := body compact := strings.Join(strings.Fields(text), "") if len(compact) >= 16 && base64CharsetRe.MatchString(compact) { normalized := strings.NewReplacer("-", "+", "_", "/").Replace(compact) if pad := len(normalized) % 4; pad != 0 { normalized += strings.Repeat("=", 4-pad) } if decoded, err := base64.StdEncoding.DecodeString(normalized); err == nil && strings.Contains(string(decoded), "://") { text = string(decoded) } } items := []SubCacheItem{} dropped := 0 for _, rawLine := range strings.Split(text, "\n") { line := strings.TrimSpace(rawLine) if line == "" { continue } if len(items) >= maxSubItems { dropped++ continue } if !ValidateLinkUri(line) { dropped++ continue } name := "" if idx := strings.LastIndex(line, "#"); idx >= 0 { rawName := line[idx+1:] if decoded, err := url.PathUnescape(rawName); err == nil { name = decoded } else { name = rawName } } items = append(items, SubCacheItem{URI: line, Name: truncateRunes(name, 120)}) } return items, dropped } func isGlobalIPString(ip string) bool { parsed := net.ParseIP(strings.Trim(ip, "[]")) if parsed == nil { return false } if parsed.To4() != nil { return isGlobalIPv4(parsed) } return isGlobalIPv6(parsed) } func resolveCheckedIP(ctx context.Context, hostname string, allowPrivate bool) (string, error) { var addresses []string if ip := net.ParseIP(hostname); ip != nil { addresses = []string{hostname} } else { records, err := net.DefaultResolver.LookupIPAddr(ctx, hostname) if err != nil { return "", errors.New("域名解析无结果") } for _, record := range records { addresses = append(addresses, record.IP.String()) } } if len(addresses) == 0 { return "", errors.New("域名解析无结果") } if !allowPrivate { for _, ip := range addresses { if !isGlobalIPString(ip) { return "", errors.New("订阅源指向非公网地址,已拒绝") } } } return addresses[0], nil } func fetchSourceBody(source ExtSource, settings *Settings) (string, error) { parsed, err := url.Parse(source.URL) if err != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") { return "", errors.New("仅支持 http/https 订阅源") } allowPrivate := settings != nil && settings.AllowPrivateSubSources dialer := &net.Dialer{Timeout: 5 * time.Second} pin := func(ctx context.Context, addr string) (string, string, error) { host, port, err := net.SplitHostPort(addr) if err != nil { return "", "", err } ip, err := resolveCheckedIP(ctx, host, allowPrivate) if err != nil { return "", "", err } return ip, port, nil } transport := &http.Transport{ DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { ip, port, err := pin(ctx, addr) if err != nil { return nil, err } return dialer.DialContext(ctx, network, net.JoinHostPort(ip, port)) }, DialTLSContext: func(ctx context.Context, network, addr string) (net.Conn, error) { host, _, _ := net.SplitHostPort(addr) ip, port, err := pin(ctx, addr) if err != nil { return nil, err } tlsDialer := &tls.Dialer{NetDialer: dialer, Config: &tls.Config{ServerName: host, MinVersion: tls.VersionTLS12}} return tlsDialer.DialContext(ctx, network, net.JoinHostPort(ip, port)) }, ResponseHeaderTimeout: fetchTimeout, } client := &http.Client{ Transport: transport, Timeout: fetchTimeout, CheckRedirect: func(req *http.Request, via []*http.Request) error { if len(via) >= 3 { return errors.New("订阅源重定向次数过多") } return nil }, } ua := source.UserAgent if ua == "" { ua = defaultSubUA } req, err := http.NewRequest(http.MethodGet, source.URL, nil) if err != nil { return "", err } req.Header.Set("User-Agent", ua) req.Header.Set("Accept", "*/*") resp, err := client.Do(req) if err != nil { if errors.Is(err, context.DeadlineExceeded) { return "", errors.New("订阅源请求超时") } return "", err } defer resp.Body.Close() if resp.StatusCode != 200 { return "", fmt.Errorf("订阅源返回 HTTP %d", resp.StatusCode) } body, err := io.ReadAll(io.LimitReader(resp.Body, maxSubBodyBytes+1)) if err != nil { return "", err } if len(body) > maxSubBodyBytes { return "", errors.New("订阅源响应超过 1MiB") } return string(body), nil } type FetchResult struct { OK bool Items []SubCacheItem Dropped int Error string } // FetchExternalSource mirrors fetchExternalSource: never fails hard. func FetchExternalSource(source ExtSource, settings *Settings) FetchResult { body, err := fetchSourceBody(source, settings) if err != nil { message := err.Error() if len([]rune(message)) > 240 { message = truncateRunes(message, 240) } return FetchResult{OK: false, Items: []SubCacheItem{}, Error: message} } items, dropped := ParseLinkList(body) return FetchResult{OK: true, Items: items, Dropped: dropped} } func RefreshMinutes(settings *Settings) float64 { if settings == nil || settings.ExternalSubRefreshMinutes <= 0 { return 60 } return math.Min(720, math.Max(15, settings.ExternalSubRefreshMinutes)) } func jitterFactor(sourceID string) float64 { hash := sha256.Sum256([]byte(sourceID)) return 1 + float64(int(hash[0]%21)-10)/100 } type dueSource struct { CustomerID string Source ExtSource } // DueSources mirrors dueSources in lib/extsub.js. func DueSources(d *Data, nowMs int64, limit int) []dueSource { out := []dueSource{} for _, customer := range d.Customers { for _, source := range customer.ExternalSources { if !source.Enabled { continue } entry := d.SubSourceCache[source.ID] if entry == nil || entry.FetchedAt == nil { out = append(out, dueSource{customer.ID, source}) continue } base := RefreshMinutes(&d.Settings) * 60000 failCount := entry.FailCount if failCount > 10 { failCount = 10 } backoff := math.Min(math.Pow(2, float64(failCount))*base, 24*3600*1000) fetchedMs, _ := ParseISOMs(*entry.FetchedAt) if float64(nowMs) >= float64(fetchedMs)+backoff*jitterFactor(source.ID) { out = append(out, dueSource{customer.ID, source}) } if len(out) >= limit { return out } } if len(out) >= limit { break } } if len(out) > limit { out = out[:limit] } return out } // ApplyFetchResult mirrors applyFetchResult in lib/extsub.js. func ApplyFetchResult(d *Data, sourceID string, result FetchResult, nowISO string) *SubSourceCacheEntry { if d.SubSourceCache == nil { d.SubSourceCache = map[string]*SubSourceCacheEntry{} } entry := d.SubSourceCache[sourceID] if entry == nil { entry = &SubSourceCacheEntry{Items: []SubCacheItem{}} } entry.FetchedAt = StrPtr(nowISO) if result.OK { entry.OkAt = StrPtr(nowISO) entry.FailCount = 0 entry.LastError = nil entry.DroppedCount = result.Dropped entry.Items = result.Items if entry.Items == nil { entry.Items = []SubCacheItem{} } } else { entry.FailCount++ entry.LastError = StrPtr(result.Error) } d.SubSourceCache[sourceID] = entry return entry } // ---------- save-time normalization ---------- func NormalizeExternalSources(value any) ([]ExtSource, error) { if value == nil { return nil, nil } arr, ok := value.([]any) if !ok { return nil, httpErr(400, "外部订阅格式无效") } if len(arr) > 5 { return nil, httpErr(400, "每个客户最多 5 个外部订阅") } out := []ExtSource{} for _, raw := range arr { item, _ := raw.(map[string]any) rawURL := CleanText(item["url"], 2048) parsed, err := url.Parse(rawURL) if err != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Hostname() == "" { return nil, httpErr(400, "外部订阅地址无效") } idValue := "" if s, isStr := item["id"].(string); isStr && s != "" { idValue = truncateRunes(s, 80) } else { idValue = NewID("src") } next := ExtSource{ ID: idValue, URL: rawURL, Remark: CleanText(item["remark"], 80), Enabled: true, } if enabled, isBool := item["enabled"].(bool); isBool && !enabled { next.Enabled = false } if ua := CleanText(item["userAgent"], 120); ua != "" { next.UserAgent = ua } out = append(out, next) } return out, nil } func NormalizeManualLinks(value any) ([]ManualLink, error) { if value == nil { return nil, nil } arr, ok := value.([]any) if !ok { return nil, httpErr(400, "外部链接格式无效") } if len(arr) > 50 { return nil, httpErr(400, "每个客户最多 50 条外部链接") } out := []ManualLink{} for _, raw := range arr { item, _ := raw.(map[string]any) uri := CleanText(item["uri"], 2048) if !ValidateLinkUri(uri) { return nil, httpErr(400, "外部链接无效或协议不受支持") } idValue := "" if s, isStr := item["id"].(string); isStr && s != "" { idValue = truncateRunes(s, 80) } else { idValue = NewID("lnk") } next := ManualLink{ ID: idValue, URI: uri, Remark: CleanText(item["remark"], 80), Enabled: true, } if enabled, isBool := item["enabled"].(bool); isBool && !enabled { next.Enabled = false } out = append(out, next) } return out, nil } // ---------- merge helpers (used by subscriptions.go) ---------- func DedupKey(uri string) string { parsed, err := url.Parse(uri) if err != nil { return "" } username, password := "", "" if parsed.User != nil { username = parsed.User.Username() password, _ = parsed.User.Password() } hash := sha256.Sum256([]byte(username + ":" + password)) return fmt.Sprintf("%s//%s:%s#%s", parsed.Scheme+":", strings.ToLower(parsed.Hostname()), parsed.Port(), hex.EncodeToString(hash[:])[:16]) } func WithName(uri, name string) string { base := uri if idx := strings.LastIndex(uri, "#"); idx >= 0 { base = uri[:idx] } return base + "#" + encodeURIComponent(name) } func NameFromURI(uri string) string { idx := strings.LastIndex(uri, "#") if idx < 0 { return "" } raw := uri[idx+1:] if decoded, err := url.PathUnescape(raw); err == nil { return decoded } return raw } func strOrNil(s string) any { if s == "" { return nil } return s } // ---------- App-level refresh machinery ---------- // RefreshOneSource fetches one source now and persists the outcome. func (a *App) RefreshOneSource(customerID, sourceID string) *FetchResult { data := a.store.Snapshot() customer := findCustomer(data, customerID) if customer == nil { return nil } var source *ExtSource for i := range customer.ExternalSources { if customer.ExternalSources[i].ID == sourceID { source = &customer.ExternalSources[i] break } } if source == nil { return nil } settings := data.Settings result := FetchExternalSource(*source, &settings) _, err := a.store.Transaction(func(d *Data) (any, error) { ApplyFetchResult(d, sourceID, result, NowISO()) return nil, nil }) if err != nil { return nil } return &result } func (a *App) RefreshSourcesInBackground(customerID string, sourceIDs []string) { for _, sourceID := range sourceIDs { go func(id string) { _ = a.RefreshOneSource(customerID, id) }(sourceID) } } func (a *App) RefreshDueSources() { data := a.store.Snapshot() for _, due := range DueSources(data, time.Now().UnixMilli(), 3) { _ = a.RefreshOneSource(due.CustomerID, due.Source.ID) } }