diff --git a/internal/server/server.go b/internal/server/server.go index 24a1477..3c26ad6 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -228,7 +228,7 @@ type Server struct { subscriptionGraphReadWaiter chan<- struct{} subscriptionGraphWriteWaiter chan<- struct{} subscriptionGraphPruneSkipped chan<- struct{} - subscriptionRender func(context.Context, model.SubscriptionShare, string, string, model.SubscriptionSnapshot) (renderedSubscription, error) + subscriptionRender func(context.Context, model.SubscriptionShare, string, string, shareRenderVariant, model.SubscriptionSnapshot) (renderedSubscription, error) subscriptionBeforeCacheExtend func() subscriptionCacheLookupWaiter chan<- struct{} subscriptionCacheExtendWaiter chan<- struct{} diff --git a/internal/server/server_subscription_share.go b/internal/server/server_subscription_share.go index f3ddb98..b8ce5e7 100644 --- a/internal/server/server_subscription_share.go +++ b/internal/server/server_subscription_share.go @@ -37,6 +37,66 @@ var shareSlugRe = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,62}$`) // client screenshots. Authorization rests entirely on the token, which is why // this function validates the slug's shape but never treats a correct slug as // evidence of anything. +// subscriptionShareTargets is the bounded set of client targets a share URL +// may name via ?target= — the Sub-Store URL-parity contract. Bounded on +// purpose: the target participates in the render cache key, and an unbounded +// caller-chosen string there is a cache-exhaustion lever on an +// unauthenticated-by-design endpoint. +var subscriptionShareTargets = map[string]bool{ + "URI": true, "Stash": true, "ClashMeta": true, "Egern": true, + "Surfboard": true, "Surge": true, "SurgeMac": true, "Loon": true, + "Shadowrocket": true, "QX": true, "sing-box": true, "V2Ray": true, + "Clash": true, "JSON": true, +} + +// shareRenderVariant carries the explicit render parameters of one request, +// under Sub-Store's own query-parameter names (target, includeUnsupportedProxy, +// prettyYaml, noFlow). The zero value means "no parameters", which renders and +// caches exactly as requests did before the parameters existed. +type shareRenderVariant struct { + Target string + IncludeUnsupported bool + PrettyYAML bool + // NoFlow suppresses the Subscription-Userinfo response header — upstream's + // "不查询订阅流量信息". It affects only the response envelope, never the + // rendered body, so it deliberately stays OUT of the cache key. + NoFlow bool +} + +// cacheToken is the canonical cache-key fragment. Empty for the zero variant +// so pre-existing cache keys are unchanged. NoFlow is absent by design: the +// cached body is identical either way. +func (v shareRenderVariant) cacheToken() string { + if v.Target == "" && !v.IncludeUnsupported && !v.PrettyYAML { + return "" + } + token := "t=" + v.Target + if v.IncludeUnsupported { + token += ";iup=1" + } + if v.PrettyYAML { + token += ";py=1" + } + return token +} + +// options is the produce() flag map handed to the plugin, under the flag +// names the embedded Sub-Store core reads. Nil when nothing is set, so old +// plugins see the payload they always saw. +func (v shareRenderVariant) options() map[string]bool { + if !v.IncludeUnsupported && !v.PrettyYAML { + return nil + } + opts := map[string]bool{} + if v.IncludeUnsupported { + opts["include-unsupported-proxy"] = true + } + if v.PrettyYAML { + opts["pretty-yaml"] = true + } + return opts +} + func sharePathFromRequest(value string) (string, string, bool) { rest, ok := strings.CutPrefix(value, "/sub/") if !ok { @@ -150,7 +210,23 @@ func (s *Server) handleSubscriptionShare(w http.ResponseWriter, r *http.Request) } uaClass := classifyClientUA(r.Header.Get("User-Agent")) - key := subscriptionCacheKey{ShareID: share.ID, Format: format, UAClass: uaClass} + + // Explicit render parameters, Sub-Store URL style: ?target= names the + // client outright (validated against the bounded target set — this string + // enters the cache key), and includeUnsupportedProxy rides through to + // produce() under upstream's own flag name. + variant := shareRenderVariant{ + Target: strings.TrimSpace(r.URL.Query().Get("target")), + IncludeUnsupported: requestBool(r, "includeUnsupportedProxy"), + PrettyYAML: requestBool(r, "prettyYaml") || requestBool(r, "pretty-yaml"), + NoFlow: requestBool(r, "noFlow"), + } + if variant.Target != "" && !subscriptionShareTargets[variant.Target] { + deny("invalid subscription target", map[string]string{"slug": slug, "token_sha256": tokenHash, "share_id": share.ID}) + return + } + + key := subscriptionCacheKey{ShareID: share.ID, Format: format, UAClass: uaClass, Variant: variant.cacheToken()} var cacheEntry subscriptionCacheEntry var cached bool @@ -210,7 +286,7 @@ func (s *Server) handleSubscriptionShare(w http.ResponseWriter, r *http.Request) } accepted := false for attempt := 0; attempt < attempts; attempt++ { - rendered, renderErr := s.renderShare(r.Context(), share, format, uaClass) + rendered, renderErr := s.renderShare(r.Context(), share, format, uaClass, variant) if renderErr != nil { s.logger.Printf("subscription share: render failed for share %s (%s)", share.ID, subscriptionDiagnosticSummary(renderErr)) deny("subscription_render_failed", map[string]string{"slug": slug, "token_sha256": tokenHash, "share_id": share.ID}) @@ -250,7 +326,9 @@ func (s *Server) handleSubscriptionShare(w http.ResponseWriter, r *http.Request) } w.Header().Set("Cache-Control", "no-store") w.Header().Set("Content-Type", contentType) - if userinfo != "" { + // ?noFlow=1 keeps quota headers off the wire (upstream's 不查询订阅流量) — + // some clients probe aggressively when they see one. + if userinfo != "" && !variant.NoFlow { w.Header().Set("Subscription-Userinfo", userinfo) } if staleResponse { @@ -392,7 +470,7 @@ func (s *Server) invalidateSharesForSource(pluginID, subscriptionID string) { // renderShare asks the share's source for content. It never shows the source the // token and never lets it influence the response beyond the bytes and a content // type. -func (s *Server) renderShare(ctx context.Context, share model.SubscriptionShare, format, uaClass string) (renderedSubscription, error) { +func (s *Server) renderShare(ctx context.Context, share model.SubscriptionShare, format, uaClass string, variant shareRenderVariant) (renderedSubscription, error) { switch share.Source.Kind { case model.ShareSourceCoreProxyUser: user, ok := s.store.ProxyUser(share.Source.ProxyUserID) @@ -423,16 +501,25 @@ func (s *Server) renderShare(ctx context.Context, share model.SubscriptionShare, return renderedSubscription{}, errors.New("subscription source changed during render capture") } if s.subscriptionRender != nil { - rendered, err := s.subscriptionRender(ctx, share, format, uaClass, snap) + rendered, err := s.subscriptionRender(ctx, share, format, uaClass, variant, snap) rendered.SourceEpoch = epoch return rendered, err } - payload, err := json.Marshal(map[string]string{ + payloadFields := map[string]any{ "subscription_id": share.Source.SubscriptionID, "format": format, "ua_class": uaClass, "raw": snap.Raw, - }) + } + // Explicit render parameters ride only when set, so a plugin built + // before they existed receives the exact payload it always did. + if variant.Target != "" { + payloadFields["target"] = variant.Target + } + if opts := variant.options(); opts != nil { + payloadFields["options"] = opts + } + payload, err := json.Marshal(payloadFields) if err != nil { return renderedSubscription{}, err } diff --git a/internal/server/server_subscription_share_test.go b/internal/server/server_subscription_share_test.go index 5fb26e1..669bb8a 100644 --- a/internal/server/server_subscription_share_test.go +++ b/internal/server/server_subscription_share_test.go @@ -302,7 +302,7 @@ func TestSubscriptionShareStaleCacheHitRevalidatesAndClearsHeaderOnRecovery(t *t calls++ return model.SubscriptionSnapshot{Raw: "last-good", Userinfo: "upload=2"}, nil } - s.subscriptionRender = func(_ context.Context, _ model.SubscriptionShare, _, _ string, snap model.SubscriptionSnapshot) (renderedSubscription, error) { + s.subscriptionRender = func(_ context.Context, _ model.SubscriptionShare, _, _ string, _ shareRenderVariant, snap model.SubscriptionSnapshot) (renderedSubscription, error) { epoch, ok := s.subscriptionSnapshotEpoch("p", "graph", snap) if !ok { return renderedSubscription{}, errors.New("snapshot changed") @@ -372,7 +372,7 @@ func TestSubscriptionSharePropagatesStaleAndRecoveryAcrossSiblingShares(t *testi fetchCalls++ return model.SubscriptionSnapshot{}, errors.New("provider down") } - s.subscriptionRender = func(_ context.Context, share model.SubscriptionShare, _, _ string, snap model.SubscriptionSnapshot) (renderedSubscription, error) { + s.subscriptionRender = func(_ context.Context, share model.SubscriptionShare, _, _ string, _ shareRenderVariant, snap model.SubscriptionSnapshot) (renderedSubscription, error) { return renderedSubscription{Body: []byte("rendered-" + share.ID), ContentType: "text/plain", Userinfo: snap.Userinfo, Stale: snap.Stale, RevalidationVersion: subscriptionRevalidationVersion(snap), SourceVersion: snap.SourceVersion, FetchedAt: snap.FetchedAt}, nil } @@ -428,7 +428,7 @@ func TestSubscriptionShareRevisionMismatchRendersInsteadOfStampingReplacement(t s.subscriptionCache.PutSnapshot(key, []byte("replacement"), "text/plain", "replacement-ui", "new-version", "", false, now, now) s.subscriptionBeforeCacheExtend = nil } - s.subscriptionRender = func(_ context.Context, _ model.SubscriptionShare, _, _ string, snap model.SubscriptionSnapshot) (renderedSubscription, error) { + s.subscriptionRender = func(_ context.Context, _ model.SubscriptionShare, _, _ string, _ shareRenderVariant, snap model.SubscriptionSnapshot) (renderedSubscription, error) { return renderedSubscription{Body: []byte("rerendered"), ContentType: "text/plain", Userinfo: snap.Userinfo, RevalidationVersion: subscriptionRevalidationVersion(snap), FetchedAt: snap.FetchedAt}, nil } @@ -457,7 +457,7 @@ func TestSubscriptionShareRejectsLateRenderCachePutAfterSourceTransition(t *test started, release := make(chan struct{}), make(chan struct{}) var first sync.Once renderCalls := 0 - s.subscriptionRender = func(_ context.Context, _ model.SubscriptionShare, _, _ string, snap model.SubscriptionSnapshot) (renderedSubscription, error) { + s.subscriptionRender = func(_ context.Context, _ model.SubscriptionShare, _, _ string, _ shareRenderVariant, snap model.SubscriptionSnapshot) (renderedSubscription, error) { renderCalls++ blocked := false first.Do(func() { @@ -521,7 +521,7 @@ func TestSubscriptionShareFailsClosedWhenSourceChangesDuringBothRenderAttempts(t t.Fatal(err) } calls := 0 - s.subscriptionRender = func(_ context.Context, _ model.SubscriptionShare, _, _ string, snap model.SubscriptionSnapshot) (renderedSubscription, error) { + s.subscriptionRender = func(_ context.Context, _ model.SubscriptionShare, _, _ string, _ shareRenderVariant, snap model.SubscriptionSnapshot) (renderedSubscription, error) { calls++ publication := s.subscriptionPublicationStateFor(subscriptionRefreshKey{pluginID: "p", subscriptionID: "graph"}) publication.mu.Lock() @@ -574,7 +574,7 @@ func TestSubscriptionShareCacheHitCannotObservePartialSourcePublication(t *testi return model.SubscriptionSnapshot{Raw: "new"}, nil } } - s.subscriptionRender = func(_ context.Context, _ model.SubscriptionShare, _, _ string, snap model.SubscriptionSnapshot) (renderedSubscription, error) { + s.subscriptionRender = func(_ context.Context, _ model.SubscriptionShare, _, _ string, _ shareRenderVariant, snap model.SubscriptionSnapshot) (renderedSubscription, error) { body := "new-render" if snap.Stale { body = "stale-current" @@ -717,7 +717,7 @@ func TestSubscriptionShareRevalidationCannotExtendAcrossPartialSourcePublication } extendStarted := make(chan struct{}, 1) s.subscriptionCacheExtendWaiter = extendStarted - s.subscriptionRender = func(_ context.Context, _ model.SubscriptionShare, _, _ string, snap model.SubscriptionSnapshot) (renderedSubscription, error) { + s.subscriptionRender = func(_ context.Context, _ model.SubscriptionShare, _, _ string, _ shareRenderVariant, snap model.SubscriptionSnapshot) (renderedSubscription, error) { return renderedSubscription{Body: []byte("render-" + snap.Raw), ContentType: "text/plain", RevalidationVersion: subscriptionRevalidationVersion(snap), FetchedAt: snap.FetchedAt}, nil } @@ -755,7 +755,7 @@ func TestPluginInvalidationRejectsBlockedOldRender(t *testing.T) { started, release := make(chan struct{}), make(chan struct{}) var first sync.Once calls := 0 - s.subscriptionRender = func(_ context.Context, _ model.SubscriptionShare, _, _ string, snap model.SubscriptionSnapshot) (renderedSubscription, error) { + s.subscriptionRender = func(_ context.Context, _ model.SubscriptionShare, _, _ string, _ shareRenderVariant, snap model.SubscriptionSnapshot) (renderedSubscription, error) { calls++ blocked := false first.Do(func() { blocked = true; close(started) }) @@ -809,7 +809,7 @@ func TestSubscriptionFailureDiagnosticsAreSanitizedAtRestAndInAudit(t *testing.T Source: model.ShareSource{Kind: model.ShareSourcePlugin, PluginID: "p", SubscriptionID: "s"}} mustUpsertShare(t, st, share) s.subscriptionFetch = nil - s.subscriptionRender = func(context.Context, model.SubscriptionShare, string, string, model.SubscriptionSnapshot) (renderedSubscription, error) { + s.subscriptionRender = func(context.Context, model.SubscriptionShare, string, string, shareRenderVariant, model.SubscriptionSnapshot) (renderedSubscription, error) { return renderedSubscription{}, errors.New(canary) } rec := httptest.NewRecorder() @@ -924,3 +924,84 @@ var requestIDInBody = regexp.MustCompile(`"request_id":"[^"]*"`) func stripRequestID(body string) string { return requestIDInBody.ReplaceAllString(body, `"request_id":""`) } + +// The Sub-Store URL parity contract on the serve path: ?target= names the +// client explicitly and reaches the plugin render, distinct targets cache +// separately, an unknown target is denied like any other bad input, and +// includeUnsupportedProxy rides through as a produce flag. +func TestSubscriptionShareExplicitTargetParameter(t *testing.T) { + s, st := newShareTestServer(t) + now := time.Unix(1_700_000_000, 0).UTC() + s.now = func() time.Time { return now } + token := strings.Repeat("a", 32) + mustUpsertShare(t, st, model.SubscriptionShare{ID: "s1", Slug: "team", Token: token, Enabled: true, DefaultFormat: "plain", + Source: model.ShareSource{Kind: model.ShareSourcePlugin, PluginID: "p", SubscriptionID: "graph"}}) + if err := st.UpsertSubscriptionSnapshot(model.SubscriptionSnapshot{PluginID: "p", SubscriptionID: "graph", Raw: "nodes", FetchedAt: now}); err != nil { + t.Fatal(err) + } + s.subscriptionFetch = func(context.Context, string, string) (model.SubscriptionSnapshot, error) { + return model.SubscriptionSnapshot{Raw: "nodes", FetchedAt: now}, nil + } + var variants []shareRenderVariant + s.subscriptionRender = func(_ context.Context, _ model.SubscriptionShare, _, _ string, variant shareRenderVariant, snap model.SubscriptionSnapshot) (renderedSubscription, error) { + variants = append(variants, variant) + epoch, _ := s.subscriptionSnapshotEpoch("p", "graph", snap) + return renderedSubscription{Body: []byte("for-" + variant.Target), ContentType: "text/plain", + RevalidationVersion: subscriptionRevalidationVersion(snap), SourceVersion: snap.SourceVersion, SourceEpoch: epoch, FetchedAt: snap.FetchedAt}, nil + } + + // Stash and sing-box render and cache independently. + recA := httptest.NewRecorder() + s.handleSubscriptionShare(recA, shareRequest("/sub/team/"+token+"?target=Stash&includeUnsupportedProxy=1", "curl/8")) + recB := httptest.NewRecorder() + s.handleSubscriptionShare(recB, shareRequest("/sub/team/"+token+"?target=sing-box", "curl/8")) + if recA.Code != http.StatusOK || recA.Body.String() != "for-Stash" { + t.Fatalf("target=Stash response = %d %q", recA.Code, recA.Body.String()) + } + if recB.Code != http.StatusOK || recB.Body.String() != "for-sing-box" { + t.Fatalf("target=sing-box response = %d %q", recB.Code, recB.Body.String()) + } + if len(variants) != 2 { + t.Fatalf("expected two renders (distinct cache keys), got %d", len(variants)) + } + if variants[0].Target != "Stash" || !variants[0].IncludeUnsupported { + t.Fatalf("first render variant = %+v", variants[0]) + } + if variants[1].Target != "sing-box" || variants[1].IncludeUnsupported { + t.Fatalf("second render variant = %+v", variants[1]) + } + + // A repeat hit with the same target is served from cache: no third render. + recC := httptest.NewRecorder() + s.handleSubscriptionShare(recC, shareRequest("/sub/team/"+token+"?target=Stash&includeUnsupportedProxy=1", "curl/8")) + if recC.Code != http.StatusOK || recC.Body.String() != "for-Stash" || len(variants) != 2 { + t.Fatalf("cache miss on identical variant: code=%d body=%q renders=%d", recC.Code, recC.Body.String(), len(variants)) + } + + // An unknown target is denied without reaching a render. + recD := httptest.NewRecorder() + s.handleSubscriptionShare(recD, shareRequest("/sub/team/"+token+"?target=EvilClient", "curl/8")) + if recD.Code == http.StatusOK || len(variants) != 2 { + t.Fatalf("unknown target must be denied before rendering: code=%d renders=%d", recD.Code, len(variants)) + } + + // prettyYaml is its own cache dimension and reaches produce as pretty-yaml. + recE := httptest.NewRecorder() + s.handleSubscriptionShare(recE, shareRequest("/sub/team/"+token+"?target=Stash&prettyYaml=1", "curl/8")) + if recE.Code != http.StatusOK || len(variants) != 3 { + t.Fatalf("prettyYaml variant should render separately: code=%d renders=%d", recE.Code, len(variants)) + } + if !variants[2].PrettyYAML || variants[2].options()["pretty-yaml"] != true { + t.Fatalf("prettyYaml did not reach the produce options: %+v", variants[2]) + } + + // noFlow suppresses the quota header without splitting the cache. + recF := httptest.NewRecorder() + s.handleSubscriptionShare(recF, shareRequest("/sub/team/"+token+"?target=Stash&includeUnsupportedProxy=1&noFlow=1", "curl/8")) + if recF.Code != http.StatusOK || len(variants) != 3 { + t.Fatalf("noFlow must not add a cache dimension: code=%d renders=%d", recF.Code, len(variants)) + } + if recF.Header().Get("Subscription-Userinfo") != "" { + t.Fatalf("noFlow response still carried Subscription-Userinfo: %q", recF.Header().Get("Subscription-Userinfo")) + } +} diff --git a/internal/server/subscription_cache.go b/internal/server/subscription_cache.go index b241063..00b8be9 100644 --- a/internal/server/subscription_cache.go +++ b/internal/server/subscription_cache.go @@ -14,6 +14,12 @@ type subscriptionCacheKey struct { ShareID string Format string UAClass string + // Variant is the canonical token for explicit render parameters (?target= + // and produce flags). Empty for a parameterless request, so existing keys + // are byte-identical to before the parameters existed. The parameter space + // is bounded by an allowlist at the handler, which is what keeps this from + // becoming a cache-exhaustion lever. + Variant string } type subscriptionCacheEntry struct { @@ -172,7 +178,7 @@ func (c *subscriptionCache) PutSnapshot(key subscriptionCacheKey, body []byte, c c.mu.Lock() defer c.mu.Unlock() c.nextRevision++ - storedKey := subscriptionCacheKey{ShareID: strings.Clone(key.ShareID), Format: strings.Clone(key.Format), UAClass: strings.Clone(key.UAClass)} + storedKey := subscriptionCacheKey{ShareID: strings.Clone(key.ShareID), Format: strings.Clone(key.Format), UAClass: strings.Clone(key.UAClass), Variant: strings.Clone(key.Variant)} entry := &subscriptionCacheEntry{ key: storedKey, body: append([]byte(nil), body...), revision: c.nextRevision, contentType: strings.Clone(contentType), userinfo: strings.Clone(userinfo), revalidationVersion: strings.Clone(revalidationVersion), publicSourceVersion: strings.Clone(publicSourceVersion),