From e356b7cc2a744b84bb1442d354342587a1d42db7 Mon Sep 17 00:00:00 2001 From: Rogee Date: Wed, 26 Aug 2026 11:07:30 +0800 Subject: [PATCH 1/2] fix: propagate subscription fetch failures Co-authored-by: multica-agent --- internal/service/subscription.go | 35 +++++------ internal/service/subscription_test.go | 89 ++++++++++++++++++++------- 2 files changed, 83 insertions(+), 41 deletions(-) diff --git a/internal/service/subscription.go b/internal/service/subscription.go index 35e4ee1..2cfbb8e 100644 --- a/internal/service/subscription.go +++ b/internal/service/subscription.go @@ -2,9 +2,9 @@ package service import ( "context" + "errors" "fmt" "io" - "math" "net/http" "net/url" "sort" @@ -87,7 +87,10 @@ func loadProxyNodes(ctx context.Context, opts BuildOptions) ([]model.ProxyNode, i := i sub := sub tasks[i] = func() ([]model.ProxyNode, error) { - raw, meta := loadSubscriptionRaw(ctx, sub, opts) + raw, meta, err := loadSubscriptionRaw(ctx, sub, opts) + if err != nil { + return nil, err + } metadataByIndex[i] = meta nodes := proxy.ParseProxies(raw) originalCounts[i] = len(nodes) @@ -193,14 +196,14 @@ func getSources(opts BuildOptions) []model.SourceRecord { return result } -func loadSubscriptionRaw(ctx context.Context, sub model.SourceRecord, opts BuildOptions) (string, model.SubscriptionResponseMetadata) { +func loadSubscriptionRaw(ctx context.Context, sub model.SourceRecord, opts BuildOptions) (string, model.SubscriptionResponseMetadata, error) { if sub.Type == "local" || sub.Content != "" { - return sub.Content + sub.URL, metadataFromSource(sub) + return sub.Content, metadataFromSource(sub), nil } urls := splitSourceUrls(sub.URL) if len(urls) == 0 { - return "", metadataFromSource(sub) + return "", metadataFromSource(sub), nil } if len(urls) > util.MaxRemoteSourceUrls { urls = urls[:util.MaxRemoteSourceUrls] @@ -215,7 +218,10 @@ func loadSubscriptionRaw(ctx context.Context, sub model.SourceRecord, opts Build } } - results, _ := RunWithConcurrencyT(tasks, getConcurrency(opts.Settings), getConcurrencyWait(opts.Settings)) + results, err := RunWithConcurrency(tasks, getConcurrency(opts.Settings), getConcurrencyWait(opts.Settings)) + if err != nil { + return "", model.SubscriptionResponseMetadata{}, err + } var contents []string var metadata model.SubscriptionResponseMetadata @@ -226,7 +232,7 @@ func loadSubscriptionRaw(ctx context.Context, sub model.SourceRecord, opts Build } } - return strings.Join(contents, "\n"), metadata + return strings.Join(contents, "\n"), metadata, nil } type fetchResult struct { @@ -491,6 +497,7 @@ func RunWithConcurrency[T any](tasks []func() (T, error), concurrency int, wait concurrency = len(tasks) } results := make([]T, len(tasks)) + errs := make([]error, len(tasks)) cursor := 0 var mu sync.Mutex var wg sync.WaitGroup @@ -512,8 +519,8 @@ func RunWithConcurrency[T any](tasks []func() (T, error), concurrency int, wait } val, err := tasks[idx]() if err != nil { - results[idx] = val // zero value - // In non-settled mode, we could cancel, but let's keep it simple + results[idx] = val + errs[idx] = err logrus.WithError(err).Warn("task failed") continue } @@ -522,12 +529,7 @@ func RunWithConcurrency[T any](tasks []func() (T, error), concurrency int, wait }() } wg.Wait() - return results, nil -} - -// RunWithConcurrencyT is a variant for tasks returning a value + metadata. -func RunWithConcurrencyT[T any](tasks []func() (T, error), concurrency int, wait time.Duration) ([]T, error) { - return RunWithConcurrency(tasks, concurrency, wait) + return results, errors.Join(errs...) } // RunSettledWithConcurrency runs tasks with allSettled semantics. @@ -618,9 +620,6 @@ func (aw *AsyncWriter) WaitWithTimeout(timeout time.Duration) { } } -// Prevent unused import -var _ = math.MaxInt32 - // buildHTTPClient creates an *http.Client with optional proxy support. // If proxyURL is empty, a standard client is returned. func buildHTTPClient(timeout time.Duration, proxyURL string) *http.Client { diff --git a/internal/service/subscription_test.go b/internal/service/subscription_test.go index 7a3ca15..dd8108a 100644 --- a/internal/service/subscription_test.go +++ b/internal/service/subscription_test.go @@ -2,7 +2,11 @@ package service import ( "context" + "encoding/json" "errors" + "net/http" + "net/http/httptest" + "strings" "sync/atomic" "testing" "time" @@ -38,6 +42,50 @@ func TestBuildSubscriptionResult_LocalSource(t *testing.T) { } } +func TestBuildSubscriptionResult_RealityVlessSubscription(t *testing.T) { + content := strings.Join([]string{ + "vless://00000000-0000-0000-0000-000000000001@8.220.220.41:443?encryption=none&flow=xtls-rprx-vision&security=reality&sni=dl.google.com&fp=chrome&pbk=public-key-1&sid=short-id-1&spx=%2F&type=tcp&headerType=none#ali-seoul", + "vless://00000000-0000-0000-0000-000000000002@8.216.16.28:443?encryption=none&flow=xtls-rprx-vision&security=reality&sni=www.yahoo.co.jp&fp=chrome&pbk=public-key-2&sid=short-id-2&spx=%2F&type=tcp&headerType=none#Ali-Tokyo", + "vless://00000000-0000-0000-0000-000000000003@67.215.229.50:443?encryption=none&flow=xtls-rprx-vision&security=reality&sni=www.cloudflare.com&fp=chrome&pbk=public-key-3&sid=short-id-3&spx=%2F&type=tcp&headerType=none#racknerd-la", + }, "\n") + source := &model.SourceRecord{ID: "reality", Type: "local", Content: content, Enabled: true} + + result, err := BuildSubscriptionResult(context.Background(), BuildOptions{Source: source, Target: "json"}) + if err != nil { + t.Fatalf("build failed: %v", err) + } + if result.OriginalNodes != 3 || result.Nodes != 3 { + t.Fatalf("nodes = %d/%d, want 3/3", result.OriginalNodes, result.Nodes) + } + var output struct { + Proxies []model.ProxyNode `json:"proxies"` + } + if err := json.Unmarshal([]byte(result.Body), &output); err != nil { + t.Fatalf("decode output: %v", err) + } + for _, node := range output.Proxies { + if node["type"] != "vless" || node["flow"] != "xtls-rprx-vision" || node["network"] != "tcp" { + t.Fatalf("VLESS fields lost: %#v", node) + } + if reality, ok := node["reality-opts"].(map[string]any); !ok || reality["public-key"] == "" || reality["short-id"] == "" { + t.Fatalf("Reality fields lost: %#v", node["reality-opts"]) + } + } +} + +func TestBuildSubscriptionResult_RemoteFetchError(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + http.Error(w, "blocked", http.StatusForbidden) + })) + defer server.Close() + source := &model.SourceRecord{ID: "blocked", Name: "blocked", Type: "remote", URL: server.URL, Enabled: true} + + _, err := BuildSubscriptionResult(context.Background(), BuildOptions{Source: source, Target: "json"}) + if err == nil || !strings.Contains(err.Error(), "Remote source blocked failed: 403") { + t.Fatalf("error = %v, want upstream 403", err) + } +} + func TestBuildSubscriptionResult_NoEnabledSources(t *testing.T) { source := &model.SourceRecord{ ID: "test-src", @@ -520,18 +568,19 @@ func TestRunWithConcurrency(t *testing.T) { } func TestRunWithConcurrency_WithErrors(t *testing.T) { + taskErr := errors.New("task error") tasks := make([]func() (int, error), 5) for i := range tasks { i := i if i == 2 { - tasks[i] = func() (int, error) { return 0, errors.New("task error") } + tasks[i] = func() (int, error) { return 0, taskErr } } else { tasks[i] = func() (int, error) { return i, nil } } } results, err := RunWithConcurrency(tasks, 2, 0) - if err != nil { - t.Fatalf("unexpected error: %v", err) + if !errors.Is(err, taskErr) { + t.Fatalf("error = %v, want task error", err) } if len(results) != 5 { t.Fatalf("expected 5 results, got %d", len(results)) @@ -567,21 +616,6 @@ func TestRunWithConcurrency_ConcurrencyOne(t *testing.T) { } } -func TestRunWithConcurrencyT(t *testing.T) { - tasks := make([]func() (string, error), 3) - for i := range tasks { - i := i - tasks[i] = func() (string, error) { return string(rune('a' + i)), nil } - } - results, err := RunWithConcurrencyT(tasks, 2, 0) - if err != nil { - t.Fatalf("unexpected error: %v", err) - } - if len(results) != 3 { - t.Errorf("expected 3 results, got %d", len(results)) - } -} - func TestRunSettledWithConcurrency(t *testing.T) { tasks := make([]func() (int, error), 5) for i := range tasks { @@ -668,7 +702,10 @@ func TestLoadSubscriptionRaw_LocalSource(t *testing.T) { Type: "local", Content: "ss://pass@host:80#Node1", } - raw, meta := loadSubscriptionRaw(context.Background(), sub, BuildOptions{}) + raw, meta, err := loadSubscriptionRaw(context.Background(), sub, BuildOptions{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } if raw != "ss://pass@host:80#Node1" { t.Errorf("expected content, got %s", raw) } @@ -682,7 +719,10 @@ func TestLoadSubscriptionRaw_RemoteNoUrls(t *testing.T) { Type: "remote", URL: "", } - raw, _ := loadSubscriptionRaw(context.Background(), sub, BuildOptions{}) + raw, _, err := loadSubscriptionRaw(context.Background(), sub, BuildOptions{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } if raw != "" { t.Errorf("expected empty, got %s", raw) } @@ -695,9 +735,12 @@ func TestLoadSubscriptionRaw_ContentPresent(t *testing.T) { Content: "ss://pass@host:80#Node1", URL: "https://example.com/sub", } - raw, _ := loadSubscriptionRaw(context.Background(), sub, BuildOptions{}) - if raw != "ss://pass@host:80#Node1https://example.com/sub" { - t.Errorf("expected content+url, got %s", raw) + raw, _, err := loadSubscriptionRaw(context.Background(), sub, BuildOptions{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if raw != "ss://pass@host:80#Node1" { + t.Errorf("expected content only, got %s", raw) } } -- 2.54.0 From 3b059b974c92c0f9aff4c24f1b3465f74b0f4938 Mon Sep 17 00:00:00 2001 From: Rogee Date: Wed, 26 Aug 2026 14:00:34 +0800 Subject: [PATCH 2/2] fix: sanitize subscription download errors Co-authored-by: multica-agent --- internal/handler/download_tools.go | 7 +++++-- internal/handler/handler_test.go | 31 ++++++++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 2 deletions(-) diff --git a/internal/handler/download_tools.go b/internal/handler/download_tools.go index a1086f9..1ef3cac 100644 --- a/internal/handler/download_tools.go +++ b/internal/handler/download_tools.go @@ -13,6 +13,7 @@ import ( "time" "github.com/gofiber/fiber/v3" + "github.com/sirupsen/logrus" "github.com/peterqiu0516/sub-store/internal/middleware" "github.com/peterqiu0516/sub-store/internal/model" @@ -55,7 +56,8 @@ func (d *Deps) HandleDownloadCollection(c fiber.Ctx) error { ProxyURL: d.Cfg.Fetcher.ProxyURL, }) if err != nil { - return failed(c, err.Error(), 500) + logrus.WithError(err).Error("Failed to build subscription") + return failed(c, "Failed to build subscription", 500) } return d.sendDownloadResponse(c, result, target) } @@ -86,7 +88,8 @@ func (d *Deps) HandleDownloadSource(c fiber.Ctx) error { ProxyURL: d.Cfg.Fetcher.ProxyURL, }) if err != nil { - return failed(c, err.Error(), 500) + logrus.WithError(err).Error("Failed to build subscription") + return failed(c, "Failed to build subscription", 500) } return d.sendDownloadResponse(c, result, target) } diff --git a/internal/handler/handler_test.go b/internal/handler/handler_test.go index 88c418c..87fa4cf 100644 --- a/internal/handler/handler_test.go +++ b/internal/handler/handler_test.go @@ -1836,6 +1836,37 @@ func TestHandleDownloadSourceDisabled(t *testing.T) { assertStatus(t, "DownloadSource disabled", code, 404) } +func TestHandleDownloadBuildErrorDoesNotLeakUpstreamURL(t *testing.T) { + for _, kind := range []string{"source", "collection"} { + t.Run(kind, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})) + upstreamURL := server.URL + "/subscription?token=upstream-secret" + server.Close() + + deps := newTestDeps(t) + app := newApp(deps) + deps.SourceRepo.Upsert(model.SourceRecord{ID: "remote", Name: "Remote", Type: "remote", URL: upstreamURL, Enabled: true}) + + path := "/sources/remote/dl-tok?target=json" + if kind == "collection" { + deps.CollectionRepo.Upsert(model.CollectionRecord{ID: "remote", Name: "Remote", SourceIds: []string{"remote"}, Enabled: true}) + path = "/collections/remote/dl-tok?target=json" + } + + code, body := doRequest(t, app, "GET", path, "", nil) + assertStatus(t, kind, code, http.StatusInternalServerError) + errorBody, ok := body["error"].(map[string]any) + if !ok || errorBody["message"] != "Failed to build subscription" { + t.Fatalf("response = %v, want fixed generic error", body) + } + response, _ := json.Marshal(body) + if strings.Contains(string(response), "upstream-secret") || strings.Contains(string(response), upstreamURL) { + t.Fatalf("response leaked upstream URL: %s", response) + } + }) + } +} + // --------------------------------------------------------------------------- // Preview handlers // --------------------------------------------------------------------------- -- 2.54.0