feat(panel): reskin backend — node sortIndex, presence/speed, /api/system, credentials, subscription aggregation (Node+Go parity)
This commit is contained in:
aaaamoooo
@@ -0,0 +1,475 @@
|
||||
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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user