Flexible / sane header forwarding (cadence-workflow#5103)

Blindly forwarding all inbound headers to outbound requests is unambiguously wrong, and causing problems.

This adds a semi-flexible system to allow controlling what headers are forwarded, via static and dynamic config.
Dynamic config is currently only loaded at startup, but at least internally it's still quicker than changing static config.  We should consider making this truly dynamic.

As this has historically been copying ?? to outbounds, leading to ?? total headers: the current version does not change anything.  We'll likely experiment internally to make sure our new sets of headers are safe with our setup, and will make a separate PR to tighten down the list somewhat when we've confirmed that it works + is likely to work in other clusters.

---

Why this kind of add/remove/regex list structure?  No strong reason, I've just used it in previous projects pretty happily.  Only-inclusive or only-exclusive has been insufficient every time, but I've had this in a couple services for years without needing more.

Longer-term we probably want this to be truly dynamically updatable and/or fully pluggable.  If we're lucky we might not need more than this though, and driving it via config is pretty convenient and easily understood.

Makefile

@@ -453,12 +453,13 @@ bins: $(BINS) ## Make all binaries
 
 tools: $(TOOLS)
 
-go-generate: $(BIN)/mockgen $(BIN)/enumer $(BIN)/mockery  $(BIN)/gowrap
-	$Q echo "running go generate ./..., this takes almost 5 minutes..."
+go-generate: $(BIN)/mockgen $(BIN)/enumer $(BIN)/mockery  $(BIN)/gowrap ## run go generate to regen mocks, enums, etc
+	$Q echo "running go generate ./..., this takes a minute or more..."
 	$Q # add our bins to PATH so `go generate` can find them
-	$Q $(BIN_PATH) go generate ./...
+	$Q $(BIN_PATH) go generate $(if $(verbose),-v) ./...
 	$Q echo "updating copyright headers"
 	$Q $(MAKE) --no-print-directory copyright
+	$Q $(MAKE) --no-print-directory fmt
 
 release: ## Re-generate generated code and run tests
 	$(MAKE) --no-print-directory go-generate

common/config/config.go

@@ -24,6 +24,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"log"
+	"regexp"
 	"time"
 
 	"github.com/uber-go/tally/m3"
@@ -72,6 +73,13 @@ type (
 		Blobstore Blobstore `yaml:"blobstore"`
 		// Authorization is the config for setting up authorization
 		Authorization Authorization `yaml:"authorization"`
+		// HeaderForwardingRules defines which inbound headers to include or exclude on outbound calls
+		HeaderForwardingRules []HeaderRule `yaml:"headerForwardingRules"`
+	}
+
+	HeaderRule struct {
+		Add   bool // if false, matching headers are removed if previously matched.
+		Match *regexp.Regexp
 	}
 
 	Authorization struct {

common/dynamicconfig/clientInterface.go

@@ -47,6 +47,7 @@ type Client interface {
 	GetStringValue(name StringKey, filters map[Filter]interface{}) (string, error)
 	GetMapValue(name MapKey, filters map[Filter]interface{}) (map[string]interface{}, error)
 	GetDurationValue(name DurationKey, filters map[Filter]interface{}) (time.Duration, error)
+	GetListValue(name ListKey, filters map[Filter]interface{}) ([]interface{}, error)
 	// UpdateValue takes value as map and updates by overriding. It doesn't support update with filters.
 	UpdateValue(name Key, value interface{}) error
 	RestoreValue(name Key, filters map[Filter]interface{}) error

common/dynamicconfig/config.go

@@ -170,6 +170,9 @@ type IntPropertyFnWithWorkflowTypeFilter func(domainName string, workflowType st
 // DurationPropertyFnWithDomainFilter is a wrapper to get duration property from dynamic config with domain as filter
 type DurationPropertyFnWithWorkflowTypeFilter func(domainName string, workflowType string) time.Duration
 
+// ListPropertyFn is a wrapper to get a list property from dynamic config
+type ListPropertyFn func(opts ...FilterOption) []interface{}
+
 // GetProperty gets a interface property and returns defaultValue if property is not found
 func (c *Collection) GetProperty(key Key) PropertyFn {
 	return func() interface{} {
@@ -558,6 +561,22 @@ func (c *Collection) GetBoolPropertyFilteredByTaskListInfo(key BoolKey) BoolProp
 	}
 }
 
+func (c *Collection) GetListProperty(key ListKey) ListPropertyFn {
+	return func(opts ...FilterOption) []interface{} {
+		filters := c.toFilterMap(opts...)
+		val, err := c.client.GetListValue(
+			key,
+			filters,
+		)
+		if err != nil {
+			c.logError(key, filters, err)
+			return key.DefaultList()
+		}
+		c.logValue(key, filters, val, key.DefaultValue(), reflect.DeepEqual)
+		return val
+	}
+}
+
 func (c *Collection) toFilterMap(opts ...FilterOption) map[Filter]interface{} {
 	l := len(opts)
 	m := make(map[Filter]interface{}, l)

common/dynamicconfig/config/testConfig.yaml

@@ -1,3 +1,8 @@
+admin.HeaderForwardingRules:
+- value:
+  - Add: true
+    Match: (?i)key
+  constraints: {}
 frontend.validSearchAttributes:
 - value:
     DomainID: 1

common/dynamicconfig/configstore/config_store_client.go

@@ -238,6 +238,17 @@ func (csc *configStoreClient) GetDurationValue(name dc.DurationKey, filters map[
 
 	return durVal, nil
 }
+func (csc *configStoreClient) GetListValue(name dc.ListKey, filters map[dc.Filter]interface{}) ([]interface{}, error) {
+	defaultValue := name.DefaultList()
+	val, err := csc.getValueWithFilters(name, filters, defaultValue)
+	if err != nil {
+		return defaultValue, err
+	}
+	if listVal, ok := val.([]interface{}); ok {
+		return listVal, nil
+	}
+	return defaultValue, errors.New("value type is not list")
+}
 
 func (csc *configStoreClient) UpdateValue(name dc.Key, value interface{}) error {
 	dcValues, ok := value.([]*types.DynamicConfigValue)

common/dynamicconfig/constants.go

@@ -67,12 +67,19 @@ type (
 		DefaultValue map[string]interface{}
 	}
 
+	DynamicList struct {
+		KeyName      string
+		Description  string
+		DefaultValue []interface{}
+	}
+
 	IntKey      int
 	BoolKey     int
 	FloatKey    int
 	StringKey   int
 	DurationKey int
 	MapKey      int
+	ListKey     int
 
 	Key interface {
 		String() string
@@ -102,6 +109,9 @@ func ListAllProductionKeys() []Key {
 	for i := TestGetMapPropertyKey + 1; i < LastMapKey; i++ {
 		result = append(result, i)
 	}
+	for i := TestGetListPropertyKey + 1; i < LastListKey; i++ {
+		result = append(result, i)
+	}
 	return result
 }
 
@@ -149,6 +159,10 @@ func ValidateKeyValuePair(key Key, value interface{}) error {
 		if _, ok := value.(map[string]interface{}); !ok {
 			return err
 		}
+	case ListKey:
+		if _, ok := value.([]interface{}); !ok {
+			return err
+		}
 	default:
 		return fmt.Errorf("unknown key type: %T", key)
 	}
@@ -251,6 +265,22 @@ func (k MapKey) DefaultMap() map[string]interface{} {
 	return MapKeys[k].DefaultValue
 }
 
+func (k ListKey) String() string {
+	return ListKeys[k].KeyName
+}
+
+func (k ListKey) Description() string {
+	return ListKeys[k].Description
+}
+
+func (k ListKey) DefaultValue() interface{} {
+	return ListKeys[k].DefaultValue
+}
+
+func (k ListKey) DefaultList() []interface{} {
+	return ListKeys[k].DefaultValue
+}
+
 // UnlimitedRPS represents an integer to use for "unlimited" RPS values.
 //
 // Since our ratelimiters do int/float conversions, and zero or negative values
@@ -2508,6 +2538,23 @@ const (
 	LastMapKey
 )
 
+const (
+	UnknownListKey ListKey = iota
+	TestGetListPropertyKey
+
+	// HeaderForwardingRules defines which headers are forwarded from inbound calls to outbound.
+	// This value is only loaded at startup.
+	//
+	// Regexes and header names are used as-is, you are strongly encouraged to use `(?i)` to make your regex case-insensitive.
+	//
+	// KeyName: admin.HeaderForwardingRules
+	// Value type: []rpc.HeaderRule or an []interface{} containing `map[string]interface{}{"Add":bool,"Match":string}` values.
+	// Default value: forward all headers.  (this is a problematic value, and it will be changing as we reduce to a list of known values)
+	HeaderForwardingRules
+
+	LastListKey
+)
+
 var IntKeys = map[IntKey]DynamicInt{
 	TestGetIntPropertyKey: DynamicInt{
 		KeyName:      "testGetIntPropertyKey",
@@ -4377,6 +4424,23 @@ var MapKeys = map[MapKey]DynamicMap{
 	},
 }
 
+var ListKeys = map[ListKey]DynamicList{
+	HeaderForwardingRules: {
+		KeyName: "admin.HeaderForwardingRules", // make a new scope for global?
+		Description: "Only loaded at startup.  " +
+			"A list of rpc.HeaderRule values that define which headers to include or exclude for all requests, applied in order.  " +
+			"Regexes and header names are used as-is, you are strongly encouraged to use `(?i)` to make your regex case-insensitive.",
+		DefaultValue: []interface{}{
+			// historical behavior: include literally everything.
+			// this alone is quite problematic, and is strongly recommended against.
+			map[string]interface{}{ // config imports dynamicconfig, sadly
+				"Add":   true,
+				"Match": "",
+			},
+		},
+	},
+}
+
 var _keyNames map[string]Key
 
 func init() {
@@ -4413,4 +4477,8 @@ func init() {
 		panicIfKeyInvalid(v.KeyName, k)
 		_keyNames[v.KeyName] = k
 	}
+	for k, v := range ListKeys {
+		panicIfKeyInvalid(v.KeyName, k)
+		_keyNames[v.KeyName] = k
+	}
 }

common/dynamicconfig/file_based_client.go

@@ -188,6 +188,18 @@ func (fc *fileBasedClient) GetDurationValue(name DurationKey, filters map[Filter
 	return durationVal, nil
 }
 
+func (fc *fileBasedClient) GetListValue(name ListKey, filters map[Filter]interface{}) ([]interface{}, error) {
+	defaultValue := name.DefaultList()
+	val, err := fc.getValueWithFilters(name, filters, defaultValue)
+	if err != nil {
+		return defaultValue, err
+	}
+	if listVal, ok := val.([]interface{}); ok {
+		return listVal, nil
+	}
+	return defaultValue, fmt.Errorf("value type is not list but is: %T", val)
+}
+
 func (fc *fileBasedClient) UpdateValue(name Key, value interface{}) error {
 	if err := ValidateKeyValuePair(name, value); err != nil {
 		return err

common/dynamicconfig/inMemoryClient.go

@@ -125,6 +125,16 @@ func (mc *inMemoryClient) GetDurationValue(name DurationKey, filters map[Filter]
 	return name.DefaultDuration(), NotFoundError
 }
 
+func (mc *inMemoryClient) GetListValue(name ListKey, filters map[Filter]interface{}) ([]interface{}, error) {
+	mc.RLock()
+	defer mc.RUnlock()
+
+	if val, ok := mc.globalValues[name]; ok {
+		return val.([]interface{}), nil
+	}
+	return name.DefaultList(), NotFoundError
+}
+
 func (mc *inMemoryClient) UpdateValue(key Key, value interface{}) error {
 	if err := ValidateKeyValuePair(key, value); err != nil {
 		return err

common/dynamicconfig/nopClient.go

@@ -66,6 +66,10 @@ func (mc *nopClient) GetDurationValue(name DurationKey, filters map[Filter]inter
 	return name.DefaultDuration(), NotFoundError
 }
 
+func (mc *nopClient) GetListValue(name ListKey, filters map[Filter]interface{}) ([]interface{}, error) {
+	return name.DefaultList(), NotFoundError
+}
+
 func (mc *nopClient) UpdateValue(name Key, value interface{}) error {
 	return errors.New("not supported for nop client")
 }

common/rpc/middleware.go

@@ -25,6 +25,7 @@ import (
 	"io"
 
 	"github.com/uber/cadence/common"
+	"github.com/uber/cadence/common/config"
 	"github.com/uber/cadence/common/metrics"
 
 	"go.uber.org/cadence/worker"
@@ -122,19 +123,56 @@ func (m *overrideCallerMiddleware) Call(ctx context.Context, request *transport.
 }
 
 // HeaderForwardingMiddleware forwards headers from current inbound RPC call that is being handled to new outbound calls being made.
-// If new value for the same header key is provided in the outbound request, keep it instead.
-type HeaderForwardingMiddleware struct{}
+// As this does NOT differentiate between transports or purposes, it generally assumes we are not acting as a true proxy,
+// so things like content lengths and encodings should not be forwarded - they will be provided by the outbound RPC library as needed.
+//
+// Duplicated headers retain the first value only, matching how browsers and Go (afaict) generally behave.
+//
+// This uses overly-simplified rules for choosing which headers are forwarded and which are not, intended to be lightly configurable.
+// For a more in-depth logic review if it becomes needed, check:
+//   - How Go's ReverseProxy deals with headers, e.g. per-protocol and a list of exclusions: https://cs.opensource.google/go/go/+/refs/tags/go1.20.1:src/net/http/httputil/reverseproxy.go;l=332
+//   - HTTP's spec for headers, namely how duplicates and Connection work: https://www.rfc-editor.org/rfc/rfc9110.html#name-header-fields
+//   - Many browsers prefer first-value-wins for unexpected duplicates: https://bugzilla.mozilla.org/show_bug.cgi?id=376756
+//   - But there are MANY map-like implementations that choose last-value wins, and this mismatch is a source of frequent security problems.
+//   - YARPC's `With` only retains the last call's value: https://github.com/yarpc/yarpc-go/blob/8ccd79a2ca696150213faac1d35011c5be52e5fb/api/transport/header.go#L69-L77
+//   - Go's MIMEHeader's Get (used by YARPC) only returns the first value, and does not join duplicates: https://pkg.go.dev/net/textproto#MIMEHeader.Get
+//
+// There is likely no correct choice, as it depends on the recipients' behavior.
+// If we need to support more complex logic, it's likely worth jumping to a fully-controllable thing.
+// Middle-grounds will probably just need to be changed again later.
+type HeaderForwardingMiddleware struct {
+	// Rules are applied in order to add or remove headers by regex.
+	//
+	// There are no default rules, so by default no headers are copied.
+	// To include headers by default, Add with a permissive regex and then remove specific ones.
+	Rules []config.HeaderRule
+}
 
 func (m *HeaderForwardingMiddleware) Call(ctx context.Context, request *transport.Request, out transport.UnaryOutbound) (*transport.Response, error) {
-	if inboundCall := yarpc.CallFromContext(ctx); inboundCall != nil && request != nil {
+	if inboundCall := yarpc.CallFromContext(ctx); inboundCall != nil {
 		outboundHeaders := request.Headers
-		for _, key := range inboundCall.HeaderNames() {
-			if _, exists := outboundHeaders.Get(key); !exists {
-				value := inboundCall.Header(key)
-				outboundHeaders = outboundHeaders.With(key, value)
+		pending := make(map[string][]string, len(inboundCall.HeaderNames()))
+		names := inboundCall.HeaderNames()
+		for _, rule := range m.Rules {
+			for _, key := range names {
+				if !rule.Match.MatchString(key) {
+					continue
+				}
+				if _, ok := outboundHeaders.Get(key); ok {
+					continue // do not overwrite existing headers
+				}
+				if rule.Add {
+					pending[key] = append(pending[key], inboundCall.Header(key))
+				} else {
+					delete(pending, key)
+				}
 			}
 		}
-		request.Headers = outboundHeaders
+		for k, vs := range pending {
+			// yarpc's Headers.With keeps the LAST value, but we (and browsers) prefer the FIRST,
+			// and we do not canonicalize duplicates.
+			request.Headers = request.Headers.With(k, vs[0])
+		}
 	}
 
 	return out.Call(ctx, request)