fix: sanitize structured metadata values during query time in pipeline code

pkg/logql/log/pipeline.go

@@ -1,8 +1,11 @@
 package log
 
 import (
+	"bytes"
 	"context"
+	"strings"
 	"sync"
+	"unicode/utf8"
 	"unsafe"
 
 	"github.com/prometheus/prometheus/storage/remote/otlptranslator/prometheus"
@@ -227,6 +230,9 @@ func (p *streamPipeline) Process(ts int64, line []byte, structuredMetadata ...la
 
 	for i, lb := range structuredMetadata {
 		structuredMetadata[i].Name = prometheus.NormalizeLabel(lb.Name)
+		if strings.ContainsRune(structuredMetadata[i].Value, utf8.RuneError) {
+			structuredMetadata[i].Value = string(bytes.Map(removeInvalidUtf, []byte(structuredMetadata[i].Value)))
+		}
 	}
 
 	p.builder.Add(StructuredMetadataLabel, structuredMetadata...)

pkg/logql/log/pipeline_test.go

@@ -194,6 +194,17 @@ func TestPipelineWithStructuredMetadata(t *testing.T) {
 	require.Equal(t, expectedLabelsResults.String(), lbr.String())
 	require.Equal(t, true, matches)
 
+	// test structured metadata with disallowed label values
+	withBadLabelValue := append(structuredMetadata, labels.Label{Name: "z_badValue", Value: "test�"})
+	expectedStructuredMetadata = append(structuredMetadata, labels.Label{Name: "z_badValue", Value: "test "})
+	expectedLabelsResults = append(lbs, expectedStructuredMetadata...)
+
+	_, lbr, matches = p.ForStream(lbs).Process(0, []byte(""), withBadLabelValue...)
+	require.Equal(t, NewLabelsResult(expectedLabelsResults.String(), expectedLabelsResults.Hash(), lbs, expectedStructuredMetadata, labels.EmptyLabels()), lbr)
+	require.Equal(t, expectedLabelsResults.Hash(), lbr.Hash())
+	require.Equal(t, expectedLabelsResults.String(), lbr.String())
+	require.Equal(t, true, matches)
+
 	// Reset caches
 	p.baseBuilder.del = []string{"foo", "bar"}
 	p.baseBuilder.add = [numValidCategories]labels.Labels{

pkg/logql/syntax/parser_test.go

@@ -2,6 +2,7 @@ package syntax
 
 import (
 	"errors"
+	"fmt"
 	"testing"
 	"time"
 
@@ -3587,3 +3588,68 @@ func TestParseSampleExpr_String(t *testing.T) {
 		require.Equal(t, "{cluster=\"beep\", namespace=\"boop\"} | msg=~`\\w.*`", expr.String())
 	})
 }
+
+func newMustLineFormatter(tmpl string) *log.LineFormatter {
+	l, err := log.NewFormatter(tmpl)
+	if err != nil {
+		panic(err)
+	}
+	return l
+}
+
+// ensure we can properly sanitize structured metadata from a pipeline that has added the SM as labels
+func TestParseLabels_StructuredMetadata(t *testing.T) {
+	lbs := labels.FromStrings("foo", "bar")
+	structuredMetadata := labels.FromStrings("user", "asdf bob�")
+	expectedLabelsResults := append(lbs, labels.FromStrings("user", "asdf bob ")...)
+
+	// regex check for the test near the invalid character
+	p := log.NewPipeline([]log.Stage{
+		log.NewStringLabelFilter(labels.MustNewMatcher(labels.MatchEqual, "foo", "bar")),
+		log.NewStringLabelFilter(labels.MustNewMatcher(labels.MatchRegexp, "user", "bob.*")),
+		newMustLineFormatter("lbs {{.foo}} {{.user}}"),
+	})
+
+	l, lbr, matches := p.ForStream(lbs).Process(0, []byte("line"), structuredMetadata...)
+	fmt.Println("l :", string(l))
+	require.Equal(t, []byte("lbs bar asdf bob "), l)
+	require.Equal(t, log.NewLabelsResult(expectedLabelsResults.String(), expectedLabelsResults.Hash(), lbs, structuredMetadata, labels.EmptyLabels()), lbr)
+	require.Equal(t, expectedLabelsResults.Hash(), lbr.Hash())
+	require.Equal(t, expectedLabelsResults.String(), lbr.String())
+	require.Equal(t, true, matches)
+	_, err := ParseLabels(lbr.String())
+	require.NoError(t, err)
+
+	// equal check for the whole contents
+	p = log.NewPipeline([]log.Stage{
+		log.NewStringLabelFilter(labels.MustNewMatcher(labels.MatchEqual, "foo", "bar")),
+		log.NewStringLabelFilter(labels.MustNewMatcher(labels.MatchEqual, "user", "asdf bob ")),
+		newMustLineFormatter("lbs {{.foo}} {{.user}}"),
+	})
+	l, lbr, matches = p.ForStream(lbs).Process(0, []byte("line"), structuredMetadata...)
+	require.Equal(t, []byte("lbs bar asdf bob "), l)
+	require.Equal(t, log.NewLabelsResult(expectedLabelsResults.String(), expectedLabelsResults.Hash(), lbs, structuredMetadata, labels.EmptyLabels()), lbr)
+	require.Equal(t, expectedLabelsResults.Hash(), lbr.Hash())
+	require.Equal(t, expectedLabelsResults.String(), lbr.String())
+	require.Equal(t, true, matches)
+	_, err = ParseLabels(lbr.String())
+	require.NoError(t, err)
+
+	// check that it works for line filter
+	f, err := log.NewFilter("asdf bob ", log.LineMatchEqual)
+	require.NoError(t, err)
+	equalLineFilterStage := f.ToStage()
+	p = log.NewPipeline([]log.Stage{
+		log.NewStringLabelFilter(labels.MustNewMatcher(labels.MatchEqual, "foo", "bar")),
+		newMustLineFormatter("lbs {{.foo}} {{.user}}"),
+		equalLineFilterStage,
+	})
+	l, lbr, matches = p.ForStream(lbs).Process(0, []byte("line"), structuredMetadata...)
+	require.Equal(t, []byte("lbs bar asdf bob "), l)
+	require.Equal(t, log.NewLabelsResult(expectedLabelsResults.String(), expectedLabelsResults.Hash(), lbs, structuredMetadata, labels.EmptyLabels()), lbr)
+	require.Equal(t, expectedLabelsResults.Hash(), lbr.Hash())
+	require.Equal(t, expectedLabelsResults.String(), lbr.String())
+	require.Equal(t, true, matches)
+	_, err = ParseLabels(lbr.String())
+	require.NoError(t, err)
+}