Plugins: Remove connection/hop-by-hop request/response headers for call resource (#60077)

Removes request/response connection/hop headers for call resource in similar 
manner as Go's reverse proxy functions. Also removes Prometheus datasource 
custom call resource header manipulation in regards to hop-by-hop headers.

Fixes #60076
Ref #58646

Co-authored-by: Will Browne <wbrowne@users.noreply.github.com>

pkg/plugins/manager/client/client.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/textproto"
+	"strings"
 
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
 
@@ -82,7 +84,19 @@ func (s *Service) CallResource(ctx context.Context, req *backend.CallResourceReq
 		return backendplugin.ErrPluginNotRegistered
 	}
 	err := instrumentation.InstrumentCallResourceRequest(ctx, &req.PluginContext, s.cfg, func() error {
-		if err := p.CallResource(ctx, req, sender); err != nil {
+		removeConnectionHeaders(req.Headers)
+		removeHopByHopHeaders(req.Headers)
+
+		wrappedSender := callResourceResponseSenderFunc(func(res *backend.CallResourceResponse) error {
+			if res != nil && len(res.Headers) > 0 {
+				removeConnectionHeaders(res.Headers)
+				removeHopByHopHeaders(res.Headers)
+			}
+
+			return sender.Send(res)
+		})
+
+		if err := p.CallResource(ctx, req, wrappedSender); err != nil {
 			return err
 		}
 		return nil
@@ -204,3 +218,63 @@ func (s *Service) plugin(ctx context.Context, pluginID string) (*plugins.Plugin,
 
 	return p, true
 }
+
+// removeConnectionHeaders removes hop-by-hop headers listed in the "Connection" header of h.
+// See RFC 7230, section 6.1
+//
+// Based on https://github.com/golang/go/blob/dc04f3ba1f25313bc9c97e728620206c235db9ee/src/net/http/httputil/reverseproxy.go#L411-L421
+func removeConnectionHeaders(h map[string][]string) {
+	for _, f := range h["Connection"] {
+		for _, sf := range strings.Split(f, ",") {
+			if sf = textproto.TrimString(sf); sf != "" {
+				for k := range h {
+					if textproto.CanonicalMIMEHeaderKey(sf) == textproto.CanonicalMIMEHeaderKey(k) {
+						delete(h, k)
+						break
+					}
+				}
+			}
+		}
+	}
+}
+
+// Hop-by-hop headers. These are removed when sent to the backend.
+// As of RFC 7230, hop-by-hop headers are required to appear in the
+// Connection header field. These are the headers defined by the
+// obsoleted RFC 2616 (section 13.5.1) and are used for backward
+// compatibility.
+//
+// Copied from https://github.com/golang/go/blob/dc04f3ba1f25313bc9c97e728620206c235db9ee/src/net/http/httputil/reverseproxy.go#L171-L186
+var hopHeaders = []string{
+	"Connection",
+	"Proxy-Connection", // non-standard but still sent by libcurl and rejected by e.g. google
+	"Keep-Alive",
+	"Proxy-Authenticate",
+	"Proxy-Authorization",
+	"Te",      // canonicalized version of "TE"
+	"Trailer", // not Trailers per URL above; https://www.rfc-editor.org/errata_search.php?eid=4522
+	"Transfer-Encoding",
+	"Upgrade",
+}
+
+// removeHopByHopHeaders removes hop-by-hop headers. Especially
+// important is "Connection" because we want a persistent
+// connection, regardless of what the client sent to us.
+//
+// Based on https://github.com/golang/go/blob/dc04f3ba1f25313bc9c97e728620206c235db9ee/src/net/http/httputil/reverseproxy.go#L276-L281
+func removeHopByHopHeaders(h map[string][]string) {
+	for _, hh := range hopHeaders {
+		for k := range h {
+			if hh == textproto.CanonicalMIMEHeaderKey(k) {
+				delete(h, k)
+				break
+			}
+		}
+	}
+}
+
+type callResourceResponseSenderFunc func(res *backend.CallResourceResponse) error
+
+func (fn callResourceResponseSenderFunc) Send(res *backend.CallResourceResponse) error {
+	return fn(res)
+}

pkg/plugins/manager/client/client_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/http"
 	"testing"
 
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
@@ -71,8 +72,142 @@ func TestQueryData(t *testing.T) {
 	})
 }
 
+func TestCallResource(t *testing.T) {
+	registry := fakes.NewFakePluginRegistry()
+	p := &plugins.Plugin{
+		JSONData: plugins.JSONData{
+			ID: "pid",
+		},
+	}
+
+	const backendResponse = "I am the backend"
+
+	t.Run("Should strip request and response hop-by-hop headers", func(t *testing.T) {
+		reqHeaders := map[string][]string{
+			"Connection":       {"close, TE"},
+			"Te":               {"foo", "bar, trailers"},
+			"Proxy-Connection": {"should be deleted"},
+			"Upgrade":          {"foo"},
+			"X-Custom":         {"should not be deleted"},
+		}
+
+		resHeaders := make(map[string][]string, len(reqHeaders))
+		for k, v := range reqHeaders {
+			resHeaders[k] = v
+		}
+
+		req := &backend.CallResourceRequest{
+			PluginContext: backend.PluginContext{
+				PluginID: "pid",
+			},
+			Headers: reqHeaders,
+		}
+
+		responses := []*backend.CallResourceResponse{}
+		sender := callResourceResponseSenderFunc(func(res *backend.CallResourceResponse) error {
+			responses = append(responses, res)
+			return nil
+		})
+
+		var actualReq *backend.CallResourceRequest
+		p.RegisterClient(&fakePluginBackend{
+			crr: func(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error {
+				actualReq = req
+				return sender.Send(&backend.CallResourceResponse{
+					Headers: resHeaders,
+					Status:  http.StatusOK,
+					Body:    []byte(backendResponse),
+				})
+			},
+		})
+		err := registry.Add(context.Background(), p)
+		require.NoError(t, err)
+
+		client := ProvideService(registry, &config.Cfg{})
+
+		err = client.CallResource(context.Background(), req, sender)
+		require.NoError(t, err)
+
+		require.NotNil(t, actualReq)
+		require.Len(t, actualReq.Headers, 1)
+		require.Equal(t, "should not be deleted", actualReq.Headers["X-Custom"][0])
+
+		require.Len(t, responses, 1)
+		res := responses[0]
+		require.Equal(t, http.StatusOK, res.Status)
+		require.Equal(t, []byte(backendResponse), res.Body)
+		require.Len(t, res.Headers, 1)
+		require.Equal(t, "should not be deleted", actualReq.Headers["X-Custom"][0])
+	})
+
+	t.Run("Should strip request and response headers present in Connection", func(t *testing.T) {
+		//nolint:gosec
+		const fakeConnectionToken = "X-fake-Connection-Token"
+
+		// someConnHeader is some arbitrary header to be declared as a hop-by-hop header
+		// in the Request's Connection header.
+		const someConnHeader = "X-Some-Conn-Header"
+
+		reqHeaders := map[string][]string{
+			"Connection":        {"Upgrade, " + fakeConnectionToken, someConnHeader},
+			someConnHeader:      {"should be deleted"},
+			fakeConnectionToken: {"should be deleted"},
+			"X-Custom":          {"should not be deleted"},
+		}
+
+		resHeaders := make(map[string][]string, len(reqHeaders))
+		for k, v := range reqHeaders {
+			resHeaders[k] = v
+		}
+
+		req := &backend.CallResourceRequest{
+			PluginContext: backend.PluginContext{
+				PluginID: "pid",
+			},
+			Headers: reqHeaders,
+		}
+
+		responses := []*backend.CallResourceResponse{}
+		sender := callResourceResponseSenderFunc(func(res *backend.CallResourceResponse) error {
+			responses = append(responses, res)
+			return nil
+		})
+
+		var actualReq *backend.CallResourceRequest
+		p.RegisterClient(&fakePluginBackend{
+			crr: func(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error {
+				actualReq = req
+				return sender.Send(&backend.CallResourceResponse{
+					Headers: resHeaders,
+					Status:  http.StatusOK,
+					Body:    []byte(backendResponse),
+				})
+			},
+		})
+		err := registry.Add(context.Background(), p)
+		require.NoError(t, err)
+
+		client := ProvideService(registry, &config.Cfg{})
+
+		err = client.CallResource(context.Background(), req, sender)
+		require.NoError(t, err)
+
+		require.NotNil(t, actualReq)
+		require.Len(t, actualReq.Headers, 1)
+		require.Equal(t, "should not be deleted", actualReq.Headers["X-Custom"][0])
+
+		require.Len(t, responses, 1)
+		res := responses[0]
+		require.Equal(t, http.StatusOK, res.Status)
+		require.Equal(t, []byte(backendResponse), res.Body)
+		require.Len(t, res.Headers, 1)
+		require.Equal(t, "should not be deleted", actualReq.Headers["X-Custom"][0])
+	})
+}
+
 type fakePluginBackend struct {
 	qdr backend.QueryDataHandlerFunc
+	crr backend.CallResourceHandlerFunc
 
 	backendplugin.Plugin
 }
@@ -84,6 +219,14 @@ func (f *fakePluginBackend) QueryData(ctx context.Context, req *backend.QueryDat
 	return backend.NewQueryDataResponse(), nil
 }
 
+func (f *fakePluginBackend) CallResource(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error {
+	if f.crr != nil {
+		return f.crr(ctx, req, sender)
+	}
+
+	return nil
+}
+
 func (f *fakePluginBackend) IsDecommissioned() bool {
 	return false
 }

pkg/plugins/manager/client/decorator_test.go

@@ -220,10 +220,4 @@ func (m *TestMiddleware) RunStream(ctx context.Context, req *backend.RunStreamRe
 	return err
 }
 
-type callResourceResponseSenderFunc func(res *backend.CallResourceResponse) error
-
-func (fn callResourceResponseSenderFunc) Send(res *backend.CallResourceResponse) error {
-	return fn(res)
-}
-
 var _ plugins.Client = &TestClient{}

pkg/tests/api/plugins/backendplugin/backendplugin_test.go

@@ -405,6 +405,12 @@ func (tsCtx *testScenarioContext) runCallResourceTest(t *testing.T) {
 			}
 			outReq, err := http.NewRequestWithContext(ctx, http.MethodGet, tsCtx.outgoingServer.URL, nil)
 			require.NoError(t, err)
+			for k, vals := range req.Headers {
+				for _, v := range vals {
+					outReq.Header.Add(k, v)
+				}
+			}
+
 			resp, err := c.Do(outReq)
 			if err != nil {
 				return err
@@ -420,8 +426,18 @@ func (tsCtx *testScenarioContext) runCallResourceTest(t *testing.T) {
 				tsCtx.testEnv.Server.HTTPServer.Cfg.Logger.Error("Failed to discard body", "error", err)
 			}
 
+			responseHeaders := map[string][]string{
+				"Connection":       {"close, TE"},
+				"Te":               {"foo", "bar, trailers"},
+				"Proxy-Connection": {"should be deleted"},
+				"Upgrade":          {"foo"},
+				"Set-Cookie":       {"should be deleted"},
+				"X-Custom":         {"should not be deleted"},
+			}
+
 			err = sender.Send(&backend.CallResourceResponse{
-				Status: http.StatusOK,
+				Status:  http.StatusOK,
+				Headers: responseHeaders,
 			})
 
 			return err
@@ -431,6 +447,10 @@ func (tsCtx *testScenarioContext) runCallResourceTest(t *testing.T) {
 
 		req, err := http.NewRequest(http.MethodGet, u, nil)
 		req.Header.Set("Content-Type", "application/json")
+		req.Header.Set("Connection", "X-Some-Conn-Header")
+		req.Header.Set("X-Some-Conn-Header", "should be deleted")
+		req.Header.Set("Proxy-Connection", "should be deleted")
+		req.Header.Set("X-Custom", "custom")
 		req.AddCookie(&http.Cookie{
 			Name: "cookie1",
 		})
@@ -457,9 +477,20 @@ func (tsCtx *testScenarioContext) runCallResourceTest(t *testing.T) {
 		_, err = io.ReadAll(resp.Body)
 		require.NoError(t, err)
 
+		require.Empty(t, resp.Header.Get("Connection"))
+		require.Empty(t, resp.Header.Get("Te"))
+		require.Empty(t, resp.Header.Get("Proxy-Connection"))
+		require.Empty(t, resp.Header.Get("Upgrade"))
+		require.Empty(t, resp.Header.Get("Set-Cookie"))
+		require.Equal(t, "should not be deleted", resp.Header.Get("X-Custom"))
+
 		// backend query data request
 		require.NotNil(t, received)
 		require.Equal(t, "cookie1=; cookie3=", received.req.Headers["Cookie"][0])
+		require.Empty(t, received.req.Headers["Connection"])
+		require.Empty(t, received.req.Headers["X-Some-Conn-Header"])
+		require.Empty(t, received.req.Headers["Proxy-Connection"])
+		require.Equal(t, "custom", received.req.Headers["X-Custom"][0])
 
 		token := tsCtx.testEnv.OAuthTokenService.Token
 
@@ -477,6 +508,10 @@ func (tsCtx *testScenarioContext) runCallResourceTest(t *testing.T) {
 		// outgoing HTTP request
 		require.NotNil(t, tsCtx.outgoingRequest)
 		require.Equal(t, "cookie1=; cookie3=", tsCtx.outgoingRequest.Header.Get("Cookie"))
+		require.Empty(t, tsCtx.outgoingRequest.Header.Get("Connection"))
+		require.Empty(t, tsCtx.outgoingRequest.Header.Get("X-Some-Conn-Header"))
+		require.Empty(t, tsCtx.outgoingRequest.Header.Get("Proxy-Connection"))
+		require.Equal(t, "custom", tsCtx.outgoingRequest.Header.Get("X-Custom"))
 		require.Equal(t, "custom-header-value", tsCtx.outgoingRequest.Header.Get("X-CUSTOM-HEADER"))
 
 		if token == nil {

pkg/tsdb/prometheus/resource/resource.go

@@ -19,37 +19,6 @@ type Resource struct {
 	log        log.Logger
 }
 
-// Hop-by-hop headers. These are removed when sent to the backend.
-// http://www.w3.org/Protocols/rfc2616/rfc2616-sec13.html
-var hopHeaders = []string{
-	"Connection",
-	"Keep-Alive",
-	"Proxy-Authenticate",
-	"Proxy-Authorization",
-	"Te", // canonicalized version of "TE"
-	"Trailers",
-	"Transfer-Encoding",
-	"Upgrade",
-}
-
-// The following headers will be removed from the request
-var stopHeaders = []string{
-	"cookie",
-	"Cookie",
-}
-
-func delHopHeaders(header http.Header) {
-	for _, h := range hopHeaders {
-		header.Del(h)
-	}
-}
-
-func delStopHeaders(header http.Header) {
-	for _, h := range stopHeaders {
-		header.Del(h)
-	}
-}
-
 func New(
 	httpClient *http.Client,
 	settings backend.DataSourceInstanceSettings,
@@ -68,9 +37,6 @@ func New(
 }
 
 func (r *Resource) Execute(ctx context.Context, req *backend.CallResourceRequest) (*backend.CallResourceResponse, error) {
-	delHopHeaders(req.Headers)
-	delStopHeaders(req.Headers)
-
 	r.log.FromContext(ctx).Debug("Sending resource query", "URL", req.URL)
 	resp, err := r.promClient.QueryResource(ctx, req)
 	if err != nil {