476 lines
12 KiB
Go
476 lines
12 KiB
Go
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)
|
|
}
|
|
}
|