receive: validate grpc remote write tenant

CHANGELOG.md

@@ -24,6 +24,7 @@ It is recommend to upgrade the storage components first (Receive, Store, etc.) a
 - [#8799](https://github.com/thanos-io/thanos/pull/8799): *: Set a `KeepaliveEnforcementPolicy` with `MinTime: 10s` on all gRPC servers, matching the client keepalive interval.
 - [#8806](https://github.com/thanos-io/thanos/pull/8806): Receive: Validate tenant IDs extracted from split-tenant labels to prevent path traversal.
 - [#8810](https://github.com/thanos-io/thanos/pull/8810): Ruler: correctly pass query partial response for gRPC.
+- [#8812](https://github.com/thanos-io/thanos/issues/8812): Receive: Validate tenant IDs from gRPC RemoteWrite requests to prevent path traversal.
 
 ### Added
 

pkg/api/query/v1.go

@@ -1730,7 +1730,7 @@ func (qapi *QueryAPI) tsdbStatus(r *http.Request) (any, []error, *api.ApiError,
 	// Append tenant matcher when tenancy is enabled.
 	var tenant string
 	if qapi.enforceTenancy {
-		tenant, err = tenancy.GetTenantFromHTTP(r, qapi.tenantHeader, qapi.defaultTenant, qapi.tenantCertField)
+		tenant, err = tenancy.GetTenantFromHTTP(r.Header, r.TLS, qapi.tenantHeader, qapi.defaultTenant, qapi.tenantCertField)
 		if err != nil {
 			return nil, nil, &api.ApiError{Typ: api.ErrorBadData, Err: err}, func() {}
 		}

pkg/receive/handler.go

@@ -154,6 +154,15 @@
 	Limiter *Limiter
 }
 
+type writeRequestTenantHeader string
+
+func (h writeRequestTenantHeader) Get(header string) string {
+	if header == tenancy.DefaultTenantHeader {
+		return string(h)
+	}
+	return ""
+}
+
 func NewHandler(logger log.Logger, o *Options) *Handler {
 	if logger == nil {
 		logger = log.NewNopLogger()
@@ -508,21 +517,21 @@
 		er:        er,
 	}
 }
 
 func (h *Handler) receiveHTTP(w http.ResponseWriter, r *http.Request) {
 	var err error
 	span, ctx := tracing.StartSpan(r.Context(), "receive_http")
 	span.SetTag("receiver.mode", string(h.receiverMode))
 	defer span.Finish()
 
-	tenantHTTP, err := tenancy.GetTenantFromHTTP(r, h.options.TenantHeader, h.options.DefaultTenantID, h.options.TenantField)
+	tenantHTTP, err := tenancy.GetTenantFromHTTP(r.Header, r.TLS, h.options.TenantHeader, h.options.DefaultTenantID, h.options.TenantField)
 	if err != nil {
 		level.Error(h.logger).Log("msg", "error getting tenant from HTTP", "err", err)
 		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
 	}
 
 	tLogger := log.With(h.logger, "tenant", tenantHTTP)
 	span.SetTag("tenant", tenantHTTP)
 
 	writeGate := h.Limiter.WriteGate()
@@ -558,7 +567,7 @@
 		}
 		compressed.Grow(int(r.ContentLength))
 	} else {
 		compressed.Grow(512)
 	}
 	_, err = io.Copy(&compressed, r.Body)
 	if err != nil {
@@ -636,7 +645,7 @@
 		switch errors.Cause(err) {
 		case errNotReady:
 			responseStatusCode = http.StatusServiceUnavailable
 		case errUnavailable:
 			responseStatusCode = http.StatusServiceUnavailable
 		case errConflict:
 			responseStatusCode = http.StatusConflict
@@ -786,7 +795,7 @@
 			cancel()
 		}
 	}()
 
 	logTags := []any{"tenant", params.tenant}
 	if id, ok := middleware.RequestIDFromContext(ctx); ok {
 		logTags = append(logTags, "request-id", id)
@@ -1168,14 +1177,20 @@
 	span, ctx := tracing.StartSpan(ctx, "receive_grpc")
 	defer span.Finish()
 
+	tenant, err := h.tenantFromWriteRequest(r)
+	if err != nil {
+		level.Debug(h.logger).Log("msg", "invalid tenant in write request", "err", err)
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+
 	h.pendingWriteRequests.Set(float64(h.pendingWriteRequestsCounter.Add(1)))
 	defer h.pendingWriteRequestsCounter.Add(-1)
 
 	// Fast path for IngestorOnly mode: write directly to local TSDB.
 	// This skips distributeTimeseriesToReplicas and sendLocalWrite since
 	// the Router already determined this data belongs to this node.
 	if h.receiverMode == IngestorOnly {
-		err := h.writer.Write(ctx, r.Tenant, r.Timeseries)
+		err := h.writer.Write(ctx, tenant, r.Timeseries)
 		if err != nil {
 			level.Debug(h.logger).Log("msg", "failed to write to local TSDB", "err", err)
 		}
@@ -1193,7 +1208,7 @@
 		}
 	}
 
-	_, err := h.handleRequest(ctx, uint64(r.Replica), r.Tenant, &prompb.WriteRequest{Timeseries: r.Timeseries})
+	_, err = h.handleRequest(ctx, uint64(r.Replica), tenant, &prompb.WriteRequest{Timeseries: r.Timeseries})
 	if err != nil {
 		level.Debug(h.logger).Log("msg", "failed to handle request", "err", err)
 	}
@@ -1213,6 +1228,18 @@
 	}
 }
 
+func (h *Handler) tenantFromWriteRequest(r *storepb.WriteRequest) (string, error) {
+	if r == nil {
+		return "", errors.New("write request is nil")
+	}
+	defaultTenantID := h.options.DefaultTenantID
+	if defaultTenantID == "" {
+		defaultTenantID = tenancy.DefaultTenant
+	}
+
+	return tenancy.GetTenantFromHTTP(writeRequestTenantHeader(r.Tenant), nil, tenancy.DefaultTenantHeader, defaultTenantID, "")
+}
+
 // relabel relabels the time series labels in the remote write request.
 func (h *Handler) relabel(wreq *prompb.WriteRequest) {
 	if len(h.options.RelabelConfigs) == 0 {

pkg/receive/handler_otlp.go

@@ -25,7 +25,7 @@ func (h *Handler) receiveOTLPHTTP(w http.ResponseWriter, r *http.Request) {
 	span.SetTag("receiver.mode", string(h.receiverMode))
 	defer span.Finish()
 
-	tenant, err := tenancy.GetTenantFromHTTP(r, h.options.TenantHeader, h.options.DefaultTenantID, h.options.TenantField)
+	tenant, err := tenancy.GetTenantFromHTTP(r.Header, r.TLS, h.options.TenantHeader, h.options.DefaultTenantID, h.options.TenantField)
 	if err != nil {
 		level.Error(h.logger).Log("msg", "error getting tenant from HTTP", "err", err)
 		http.Error(w, err.Error(), http.StatusBadRequest)

pkg/receive/handler_test.go

@@ -44,6 +44,8 @@ import (
 	"github.com/prometheus/prometheus/tsdb"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"google.golang.org/grpc/test/bufconn"
 
 	"github.com/thanos-io/thanos/pkg/block/metadata"
@@ -1118,6 +1120,82 @@ func TestReceiveTenantValidation(t *testing.T) {
 	}
 }
 
+func TestReceiveGRPCTenantValidation(t *testing.T) {
+	t.Parallel()
+
+	h := NewHandler(log.NewNopLogger(), &Options{
+		DefaultTenantID: tenancy.DefaultTenant,
+		ReceiverMode:    IngestorOnly,
+		Writer: NewWriter(
+			log.NewNopLogger(),
+			newFakeTenantAppendable(&fakeAppendable{
+				appender: newFakeAppender(nil, nil, nil),
+			}),
+			&WriterOptions{},
+		),
+	})
+	t.Cleanup(h.Close)
+
+	_, err := h.RemoteWrite(context.Background(), &storepb.WriteRequest{
+		Tenant:     "../malicious",
+		Timeseries: makeSeriesWithValues(1),
+	})
+	require.Error(t, err)
+	require.Equal(t, codes.InvalidArgument, status.Code(err))
+	require.Contains(t, err.Error(), "tenant name not valid")
+}
+
+func TestTenantFromWriteRequest(t *testing.T) {
+	t.Parallel()
+
+	for _, tc := range []struct {
+		name            string
+		defaultTenantID string
+		req             *storepb.WriteRequest
+		expected        string
+		expectedErr     string
+	}{
+		{
+			name:     "valid tenant",
+			req:      &storepb.WriteRequest{Tenant: "tenant-a"},
+			expected: "tenant-a",
+		},
+		{
+			name:            "empty tenant uses configured default",
+			defaultTenantID: tenancy.DefaultTenant,
+			req:             &storepb.WriteRequest{},
+			expected:        tenancy.DefaultTenant,
+		},
+		{
+			name:     "empty tenant uses package default",
+			req:      &storepb.WriteRequest{},
+			expected: tenancy.DefaultTenant,
+		},
+		{
+			name:        "invalid tenant",
+			req:         &storepb.WriteRequest{Tenant: "../malicious"},
+			expectedErr: "tenant name not valid",
+		},
+		{
+			name:        "nil request",
+			expectedErr: "write request is nil",
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			h := &Handler{options: &Options{DefaultTenantID: tc.defaultTenantID}}
+			tenant, err := h.tenantFromWriteRequest(tc.req)
+			if tc.expectedErr != "" {
+				require.ErrorContains(t, err, tc.expectedErr)
+				return
+			}
+			require.NoError(t, err)
+			require.Equal(t, tc.expected, tenant)
+		})
+	}
+}
+
 // endpointHit is a helper to determine if a given endpoint in a hashring would be selected
 // for a given time series, tenant, and replication factor.
 func endpointHit(t *testing.T, h Hashring, rf uint64, endpoint, tenant string, timeSeries *prompb.TimeSeries) bool {

pkg/tenancy/tenancy.go

@@ -5,6 +5,7 @@ package tenancy
 
 import (
 	"context"
+	"crypto/tls"
 	"net/http"
 	"path"
 
@@ -45,19 +46,27 @@ func IsTenantValid(tenant string) error {
 	return nil
 }
 
-// GetTenantFromHTTP extracts the tenant from a http.Request object.
-func GetTenantFromHTTP(r *http.Request, tenantHeader string, defaultTenantID string, certTenantField string) (string, error) {
+// HeaderGetter is the minimal header access needed for tenant extraction.
+type HeaderGetter interface {
+	Get(string) string
+}
+
+// GetTenantFromHTTP extracts the tenant from HTTP-compatible headers or client certificates.
+func GetTenantFromHTTP(header HeaderGetter, tlsState *tls.ConnectionState, tenantHeader string, defaultTenantID string, certTenantField string) (string, error) {
 	var err error
-	tenant := r.Header.Get(tenantHeader)
-	if tenant == "" {
-		tenant = r.Header.Get(DefaultTenantHeader)
+	var tenant string
+	if header != nil {
+		tenant = header.Get(tenantHeader)
 		if tenant == "" {
-			tenant = defaultTenantID
+			tenant = header.Get(DefaultTenantHeader)
 		}
 	}
+	if tenant == "" {
+		tenant = defaultTenantID
+	}
 
 	if certTenantField != "" {
-		tenant, err = getTenantFromCertificate(r, certTenantField)
+		tenant, err = getTenantFromCertificate(tlsState, certTenantField)
 		if err != nil {
 			// This must hard fail to ensure hard tenancy when feature is enabled.
 			return "", err
@@ -82,7 +91,7 @@ func (r roundTripperFunc) RoundTrip(request *http.Request) (*http.Response, erro
 // header is configured and present in the request, it will be stripped out.
 func InternalTenancyConversionTripper(customTenantHeader, certTenantField string, next http.RoundTripper) http.RoundTripper {
 	return roundTripperFunc(func(r *http.Request) (*http.Response, error) {
-		tenant, _ := GetTenantFromHTTP(r, customTenantHeader, DefaultTenant, certTenantField)
+		tenant, _ := GetTenantFromHTTP(r.Header, r.TLS, customTenantHeader, DefaultTenant, certTenantField)
 		r.Header.Set(DefaultTenantHeader, tenant)
 		// If the custom tenant header is not the same as the default internal header, we want to exclude the custom
 		// one from the request to keep things simple.
@@ -95,15 +104,15 @@ func InternalTenancyConversionTripper(customTenantHeader, certTenantField string
 
 // getTenantFromCertificate extracts the tenant value from a client's presented certificate. The x509 field to use as
 // value can be configured with Options.TenantField. An error is returned when the extraction has not succeeded.
-func getTenantFromCertificate(r *http.Request, certTenantField string) (string, error) {
+func getTenantFromCertificate(tlsState *tls.ConnectionState, certTenantField string) (string, error) {
 	var tenant string
 
-	if len(r.TLS.PeerCertificates) == 0 {
+	if tlsState == nil || len(tlsState.PeerCertificates) == 0 {
 		return "", errors.New("could not get required certificate field from client cert")
 	}
 
 	// First cert is the leaf authenticated against.
-	cert := r.TLS.PeerCertificates[0]
+	cert := tlsState.PeerCertificates[0]
 
 	switch certTenantField {
 
@@ -202,7 +211,7 @@ func getLabelMatchers(formMatchers []string, tenant string, enforceTenancy bool,
 // - Get tenant from HTTP header and add it to context.
 // - if tenancy is enforced, add a tenant matcher to the promQL expression.
 func RewritePromQL(ctx context.Context, r *http.Request, tenantHeader string, defaultTenantID string, certTenantField string, enforceTenancy bool, tenantLabel string, queryStr string) (string, string, context.Context, error) {
-	tenant, err := GetTenantFromHTTP(r, tenantHeader, defaultTenantID, certTenantField)
+	tenant, err := GetTenantFromHTTP(r.Header, r.TLS, tenantHeader, defaultTenantID, certTenantField)
 	if err != nil {
 		return "", "", ctx, err
 	}
@@ -220,7 +229,7 @@ func RewritePromQL(ctx context.Context, r *http.Request, tenantHeader string, de
 // - Parse all labels matchers provided.
 // - If tenancy is enforced, make sure a tenant matcher is present.
 func RewriteLabelMatchers(ctx context.Context, r *http.Request, tenantHeader string, defaultTenantID string, certTenantField string, enforceTenancy bool, tenantLabel string, formMatchers []string) ([][]*labels.Matcher, context.Context, error) {
-	tenant, err := GetTenantFromHTTP(r, tenantHeader, defaultTenantID, certTenantField)
+	tenant, err := GetTenantFromHTTP(r.Header, r.TLS, tenantHeader, defaultTenantID, certTenantField)
 	if err != nil {
 		return nil, ctx, err
 	}

pkg/tenancy/tenancy_test.go

@@ -5,6 +5,10 @@ package tenancy_test
 
 import (
 	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"net/http"
 	"testing"
 	"time"
 
@@ -51,6 +55,67 @@ func (s *storeSeriesServer) Context() context.Context {
 
 const testTenant = "test-tenant"
 
+func TestGetTenantFromHTTP(t *testing.T) {
+	t.Run("custom-header", func(t *testing.T) {
+		header := http.Header{}
+		header.Set("X-Scope-OrgID", "custom-tenant")
+		header.Set(tenancy.DefaultTenantHeader, "internal-tenant")
+
+		tenant, err := tenancy.GetTenantFromHTTP(header, nil, "X-Scope-OrgID", "fallback-tenant", "")
+		testutil.Ok(t, err)
+		testutil.Equals(t, "custom-tenant", tenant)
+	})
+
+	t.Run("fallback-to-default-header", func(t *testing.T) {
+		header := http.Header{}
+		header.Set(tenancy.DefaultTenantHeader, "internal-tenant")
+
+		tenant, err := tenancy.GetTenantFromHTTP(header, nil, "X-Scope-OrgID", "fallback-tenant", "")
+		testutil.Ok(t, err)
+		testutil.Equals(t, "internal-tenant", tenant)
+	})
+
+	t.Run("fallback-to-default-tenant", func(t *testing.T) {
+		tenant, err := tenancy.GetTenantFromHTTP(http.Header{}, nil, "X-Scope-OrgID", "fallback-tenant", "")
+		testutil.Ok(t, err)
+		testutil.Equals(t, "fallback-tenant", tenant)
+	})
+
+	t.Run("invalid-tenant", func(t *testing.T) {
+		header := http.Header{}
+		header.Set(tenancy.DefaultTenantHeader, "../malicious")
+
+		_, err := tenancy.GetTenantFromHTTP(header, nil, tenancy.DefaultTenantHeader, tenancy.DefaultTenant, "")
+		testutil.NotOk(t, err)
+	})
+
+	t.Run("certificate-overrides-header", func(t *testing.T) {
+		header := http.Header{}
+		header.Set(tenancy.DefaultTenantHeader, "header-tenant")
+		tlsState := &tls.ConnectionState{
+			PeerCertificates: []*x509.Certificate{
+				{
+					Subject: pkix.Name{
+						OrganizationalUnit: []string{"cert-tenant"},
+					},
+				},
+			},
+		}
+
+		tenant, err := tenancy.GetTenantFromHTTP(header, tlsState, tenancy.DefaultTenantHeader, tenancy.DefaultTenant, tenancy.CertificateFieldOrganizationalUnit)
+		testutil.Ok(t, err)
+		testutil.Equals(t, "cert-tenant", tenant)
+	})
+
+	t.Run("missing-certificate", func(t *testing.T) {
+		header := http.Header{}
+		header.Set(tenancy.DefaultTenantHeader, "header-tenant")
+
+		_, err := tenancy.GetTenantFromHTTP(header, nil, tenancy.DefaultTenantHeader, tenancy.DefaultTenant, tenancy.CertificateFieldCommonName)
+		testutil.NotOk(t, err)
+	})
+}
+
 func getAndAssertTenant(ctx context.Context, t *testing.T) {
 	md, ok := metadata.FromOutgoingContext(ctx)
 	if !ok || len(md.Get(tenancy.DefaultTenantHeader)) == 0 {