From f2f3ddd90cd026f5098d56447544d576ecad045c Mon Sep 17 00:00:00 2001 From: Ri-go Date: Fri, 18 Sep 2026 20:50:39 +0800 Subject: [PATCH] Keep mixed listener CORS policy authoritative --- internal/server/cosmos_grpc/web.go | 3 +- internal/server/cosmos_grpc/web_cors.go | 96 +++++++ internal/server/cosmos_grpc/web_cors_test.go | 287 +++++++++++++++++++ internal/server/cosmos_grpc/web_test.go | 2 +- 4 files changed, 386 insertions(+), 2 deletions(-) create mode 100644 internal/server/cosmos_grpc/web_cors.go create mode 100644 internal/server/cosmos_grpc/web_cors_test.go diff --git a/internal/server/cosmos_grpc/web.go b/internal/server/cosmos_grpc/web.go index 6781f0c..9bd8fe2 100644 --- a/internal/server/cosmos_grpc/web.go +++ b/internal/server/cosmos_grpc/web.go @@ -122,8 +122,9 @@ func (s *Server) WebHandler(opts WebOptions, fallback http.Handler) http.Handler // preflights without x-grpc-web use the same explicit origin policy. wrapped.HandleGrpcWebRequest(w, r) }) - withCORS := corsHandler.Handler(route) + withCORS := corsHandler.Handler(authoritativeWebCORS(route)) return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + clearWebCORSHeaders(w.Header()) if origin := r.Header.Get("Origin"); origin != "" && !allowedOrigin(origin) { w.Header().Add("Vary", "Origin") http.Error(w, "origin is not allowed", http.StatusForbidden) diff --git a/internal/server/cosmos_grpc/web_cors.go b/internal/server/cosmos_grpc/web_cors.go new file mode 100644 index 0000000..40c3b0f --- /dev/null +++ b/internal/server/cosmos_grpc/web_cors.go @@ -0,0 +1,96 @@ +package cosmos_grpc + +import ( + "net/http" + "strings" +) + +// Run inside the CORS middleware: snapshot its decision before the REST +// upstream or gRPC adapter adds headers, then restore it before headers leave +// the server. In particular, upstream wildcard/credential policies must not +// widen this listener's policy or produce multiple Allow-Origin values. +func authoritativeWebCORS(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + policy := make(http.Header) + for name, values := range w.Header() { + if isWebCORSHeader(name) { + policy[name] = append([]string(nil), values...) + } + } + protected := &webCORSResponseWriter{ResponseWriter: w, policy: policy} + var writer http.ResponseWriter = protected + if flusher, ok := w.(http.Flusher); ok { + writer = &flushingWebCORSResponseWriter{webCORSResponseWriter: protected, flusher: flusher} + } + next.ServeHTTP(writer, r) + // A handler may return without explicitly writing a status or body. + protected.restoreHeaders() + }) +} + +func isWebCORSHeader(name string) bool { + return strings.HasPrefix(strings.ToLower(name), "access-control-") +} + +func clearWebCORSHeaders(headers http.Header) { + for name := range headers { + if isWebCORSHeader(name) { + delete(headers, name) + } + } +} + +type webCORSResponseWriter struct { + http.ResponseWriter + policy http.Header + wroteHeader bool +} + +func (w *webCORSResponseWriter) restoreHeaders() { + headers := w.Header() + clearWebCORSHeaders(headers) + for name, values := range w.policy { + headers[name] = append([]string(nil), values...) + } + // Keep upstream cache dimensions while preventing it from dropping Origin. + for _, value := range headers.Values("Vary") { + for _, token := range strings.Split(value, ",") { + if token = strings.TrimSpace(token); token == "*" || strings.EqualFold(token, "Origin") { + return + } + } + } + headers.Add("Vary", "Origin") +} + +func (w *webCORSResponseWriter) WriteHeader(code int) { + if w.wroteHeader { + return + } + w.restoreHeaders() + if code >= 200 || code == http.StatusSwitchingProtocols { + w.wroteHeader = true + } + w.ResponseWriter.WriteHeader(code) +} + +func (w *webCORSResponseWriter) Write(body []byte) (int, error) { + if !w.wroteHeader { + w.WriteHeader(http.StatusOK) + } + return w.ResponseWriter.Write(body) +} + +func (w *webCORSResponseWriter) Unwrap() http.ResponseWriter { return w.ResponseWriter } + +type flushingWebCORSResponseWriter struct { + *webCORSResponseWriter + flusher http.Flusher +} + +func (w *flushingWebCORSResponseWriter) Flush() { + if !w.wroteHeader { + w.WriteHeader(http.StatusOK) + } + w.flusher.Flush() +} diff --git a/internal/server/cosmos_grpc/web_cors_test.go b/internal/server/cosmos_grpc/web_cors_test.go new file mode 100644 index 0000000..13a7789 --- /dev/null +++ b/internal/server/cosmos_grpc/web_cors_test.go @@ -0,0 +1,287 @@ +package cosmos_grpc + +import ( + "context" + "io" + "net/http" + "net/http/httptest" + "slices" + "strings" + "sync/atomic" + "testing" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/metadata" + "google.golang.org/protobuf/types/known/emptypb" + + "github.com/InjectiveLabs/stitch/internal/backend" + "github.com/InjectiveLabs/stitch/internal/circuit" + "github.com/InjectiveLabs/stitch/internal/forwarder" + healthreg "github.com/InjectiveLabs/stitch/internal/health" + "github.com/InjectiveLabs/stitch/internal/pool" + "github.com/InjectiveLabs/stitch/internal/selector" + "github.com/InjectiveLabs/stitch/internal/server/cosmos_rest" + "github.com/InjectiveLabs/stitch/internal/types" +) + +func conflictingCORSHeaders() http.Header { + return http.Header{ + "Access-Control-Allow-Origin": {"*", "https://upstream.example"}, + "Access-Control-Allow-Credentials": {"true"}, + "Access-Control-Allow-Methods": {"DELETE"}, + "Access-Control-Allow-Headers": {"*"}, + "Access-Control-Allow-Private-Network": {"true"}, + "Access-Control-Expose-Headers": {"*"}, + "Access-Control-Max-Age": {"86400"}, + "Vary": {"Accept-Encoding"}, + "X-Upstream": {"kept"}, + } +} + +func assertCORSHeaders(t *testing.T, headers http.Header, origin string) { + t.Helper() + want := []string{origin} + if origin == "" { + want = nil + } + if got := headers.Values("Access-Control-Allow-Origin"); !slices.Equal(got, want) { + t.Fatalf("Allow-Origin = %v; want %v", got, want) + } + for _, name := range []string{"Allow-Credentials", "Allow-Methods", "Allow-Headers", "Allow-Private-Network", "Max-Age"} { + if got := headers.Values("Access-Control-" + name); len(got) != 0 { + t.Fatalf("upstream %s escaped policy: %v", name, got) + } + } + exposed := strings.ToLower(strings.Join(headers.Values("Access-Control-Expose-Headers"), ",")) + if origin == "" { + if exposed != "" { + t.Fatalf("unexpected exposed headers without Origin: %s", exposed) + } + } else if exposed != "grpc-status, grpc-message, grpc-status-details-bin, x-cosmos-block-height, x-request-id" { + t.Fatalf("exposed headers changed: %s", exposed) + } + vary := strings.ToLower(strings.Join(headers.Values("Vary"), ",")) + if !strings.Contains(vary, "origin") || !strings.Contains(vary, "accept-encoding") { + t.Fatalf("cache dimensions lost: %v", headers.Values("Vary")) + } + if headers.Get("X-Upstream") != "kept" { + t.Fatalf("non-CORS upstream header lost: %v", headers) + } +} + +func TestWebCORSPolicyOverridesRESTUpstream(t *testing.T) { + var hits atomic.Int64 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits.Add(1) + if r.Header.Get(HeightHeader) != "112637000" || r.URL.RawQuery != "height=112637000" { + t.Errorf("upstream request changed: %s %v", r.URL, r.Header) + } + for name, values := range conflictingCORSHeaders() { + w.Header()[name] = values + } + w.Header().Set("Content-Type", "application/json") + w.Header().Set(HeightHeader, "112637000") + _, _ = io.WriteString(w, `{"height":"112637000"}`) + })) + defer upstream.Close() + reg := backend.NewRegistry([]*backend.Backend{{Name: "archive", Weight: 100, + Coverage: backend.Coverage{Kind: backend.CovArchive}, + Endpoints: map[types.Protocol]string{types.ProtoAPI: upstream.URL}, + }}) + health := healthreg.NewRegistry() + health.Update(healthreg.Snapshot{Backend: "archive", Protocol: types.ProtoRPC, Healthy: true, LatestHeight: 200000000}) + cm := circuit.NewManager(circuit.Policy{ErrorThreshold: 0.5, MinRequests: 10, OpenDuration: time.Minute}) + fwd := forwarder.NewHTTP(selector.NewRangeSelector(reg, health, cm, 0), pool.NewHTTPPool(), cm, forwarder.Policy{MaxAttempts: 1}) + rest := cosmos_rest.New("", fwd).Handler() + for _, tc := range []struct { + name string + origins []string + origin string + denied bool + }{ + {"allowed", []string{"https://client.example"}, "https://client.example", false}, + {"wildcard", []string{"*"}, "https://client.example", false}, + {"no origin", nil, "", false}, + {"disallowed", []string{"https://client.example"}, "https://other.example", true}, + {"default deny", nil, "https://client.example", true}, + } { + t.Run(tc.name, func(t *testing.T) { + srv := newWebBackend(t, WebOptions{AllowedOrigins: tc.origins}, rest, func(grpc.ServerStream) error { + t.Error("REST reached gRPC backend") + return nil + }) + before := hits.Load() + req, _ := http.NewRequest(http.MethodGet, srv.URL+"/cosmos/bank/v1beta1/supply?height=112637000", nil) + req.Header.Set(HeightHeader, "112637000") + if tc.origin != "" { + req.Header.Set("Origin", tc.origin) + } + resp, err := srv.Client().Do(req) + if err != nil { + t.Fatal(err) + } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if tc.denied { + if resp.StatusCode != http.StatusForbidden || hits.Load() != before || len(resp.Header.Values("Access-Control-Allow-Origin")) != 0 { + t.Fatalf("denied origin reached upstream or got CORS: %d %v", resp.StatusCode, resp.Header) + } + return + } + if resp.StatusCode != http.StatusOK || string(body) != `{"height":"112637000"}` || resp.Header.Get(HeightHeader) != "112637000" || hits.Load() != before+1 { + t.Fatalf("REST forwarding changed: %d %s %v", resp.StatusCode, body, resp.Header) + } + assertCORSHeaders(t, resp.Header, tc.origin) + }) + } +} + +func TestWebCORSPolicyOverridesGRPCUpstream(t *testing.T) { + for _, format := range []string{"native", "application/grpc-web+proto", "application/grpc-web-text+proto"} { + for _, origin := range []string{"", "https://client.example"} { + t.Run(format+"/origin="+origin, func(t *testing.T) { + srv := newWebBackend(t, WebOptions{AllowedOrigins: []string{"https://client.example"}}, nil, func(stream grpc.ServerStream) error { + md := metadata.MD{} + for name, values := range conflictingCORSHeaders() { + md.Set(name, values...) + } + if err := stream.SendHeader(md); err != nil { + return err + } + stream.SetTrailer(metadata.Pairs("x-test-trailer", "kept")) + return stream.SendMsg(&emptypb.Empty{}) + }) + if format == "native" { + conn, err := grpc.NewClient(strings.TrimPrefix(srv.URL, "http://"), grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if origin != "" { + ctx = metadata.AppendToOutgoingContext(ctx, "origin", origin) + } + var headers, trailers metadata.MD + if err := conn.Invoke(ctx, webBlockMethod, &emptypb.Empty{}, &emptypb.Empty{}, grpc.Header(&headers), grpc.Trailer(&trailers)); err != nil { + t.Fatal(err) + } + canonical := http.Header{} + for name, values := range headers { + canonical[http.CanonicalHeaderKey(name)] = values + } + assertCORSHeaders(t, canonical, origin) + if got := trailers.Get("x-test-trailer"); len(got) != 1 || got[0] != "kept" { + t.Fatalf("native trailer lost: %v", trailers) + } + return + } + req := webRequest(t, srv.URL+webBlockMethod, format, nil) + if origin != "" { + req.Header.Set("Origin", origin) + } + resp, err := srv.Client().Do(req) + if err != nil { + t.Fatal(err) + } + messages, trailers := readWebResponse(t, resp) + if len(messages) != 1 || trailers.Get("Grpc-Status") != "0" || trailers.Get("X-Test-Trailer") != "kept" { + t.Fatalf("gRPC-Web framing or trailers changed: messages=%d trailers=%v", len(messages), trailers) + } + assertCORSHeaders(t, resp.Header, origin) + }) + } + } +} + +func TestWebCORSPolicySurvivesLateHeaderWrites(t *testing.T) { + for _, mode := range []string{"set", "add", "flush", "empty", "error"} { + t.Run(mode, func(t *testing.T) { + fallback := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + for name, values := range conflictingCORSHeaders() { + if mode == "add" { + for _, value := range values { + w.Header().Add(name, value) + } + } else { + w.Header()[name] = values + } + } + // Direct map writes must not bypass case-insensitive filtering. + w.Header()["access-control-allow-origin"] = []string{"https://bypass.example"} + if mode == "flush" { + w.(http.Flusher).Flush() + } + if mode == "error" { + w.WriteHeader(http.StatusBadGateway) + } + if mode != "empty" { + _, _ = io.WriteString(w, "body") + } + }) + srv := newWebBackend(t, WebOptions{AllowedOrigins: []string{"https://client.example"}}, fallback, func(grpc.ServerStream) error { return nil }) + req, _ := http.NewRequest(http.MethodGet, srv.URL+"/rest", nil) + req.Header.Set("Origin", "https://client.example") + resp, err := srv.Client().Do(req) + if err != nil { + t.Fatal(err) + } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if mode != "empty" && string(body) != "body" { + t.Fatalf("body changed: %q", body) + } + wantStatus := http.StatusOK + if mode == "error" { + wantStatus = http.StatusBadGateway + } + if resp.StatusCode != wantStatus { + t.Fatalf("status changed: %d; want %d", resp.StatusCode, wantStatus) + } + assertCORSHeaders(t, resp.Header, "https://client.example") + }) + } +} + +func TestWebCORSPolicyPreservesRESTStreaming(t *testing.T) { + release := make(chan struct{}) + fallback := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + for name, values := range conflictingCORSHeaders() { + w.Header()[name] = values + } + w.Header().Set("Trailer", "X-Test-Trailer") + _, _ = io.WriteString(w, "first") + w.(http.Flusher).Flush() + <-release + _, _ = io.WriteString(w, "last") + w.Header().Set("X-Test-Trailer", "kept") + }) + srv := newWebBackend(t, WebOptions{AllowedOrigins: []string{"https://client.example"}}, fallback, func(grpc.ServerStream) error { return nil }) + // Always release the server, including when an assertion fails. + defer func() { + select { + case <-release: + default: + close(release) + } + }() + req, _ := http.NewRequest(http.MethodGet, srv.URL+"/rest", nil) + req.Header.Set("Origin", "https://client.example") + resp, err := srv.Client().Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + first := make([]byte, len("first")) + if _, err := io.ReadFull(resp.Body, first); err != nil || string(first) != "first" { + t.Fatalf("first chunk was not flushed: %q %v", first, err) + } + assertCORSHeaders(t, resp.Header, "https://client.example") + close(release) + last, err := io.ReadAll(resp.Body) + if err != nil || string(last) != "last" || resp.Trailer.Get("X-Test-Trailer") != "kept" { + t.Fatalf("stream or trailers changed: %q %v %v", last, resp.Trailer, err) + } +} diff --git a/internal/server/cosmos_grpc/web_test.go b/internal/server/cosmos_grpc/web_test.go index dd91414..53e868c 100644 --- a/internal/server/cosmos_grpc/web_test.go +++ b/internal/server/cosmos_grpc/web_test.go @@ -247,7 +247,7 @@ func newWebBackend(t *testing.T, opts WebOptions, fallback http.Handler, handler t.Fatal(err) } t.Cleanup(func() { _ = front.lis.Close(); front.srv.Stop() }) - httpServer := httptest.NewServer(front.WebHandler(opts, fallback)) + httpServer := httptest.NewServer(h2c.NewHandler(front.WebHandler(opts, fallback), &http2.Server{})) httpServer.Client().Timeout = 4 * time.Second t.Cleanup(httpServer.Close) return httpServer