metadata: fix validation issues

internal/metadata/metadata.go

@@ -84,6 +84,9 @@ func Set(addr resolver.Address, md metadata.MD) resolver.Address {
 // - otherwise, the header value must contain one or more characters from the set [%x20-%x7E].
 func Validate(md metadata.MD) error {
 	for k, vals := range md {
+		if k == "" {
+			return fmt.Errorf("there is an empty key in the header")
+		}
 		// pseudo-header will be ignored
 		if k[0] == ':' {
 			continue

internal/metadata/metadata_test.go

@@ -100,6 +100,10 @@ func TestValidate(t *testing.T) {
 			md:   map[string][]string{"test": {string(rune(0x19))}},
 			want: errors.New("header key \"test\" contains value with non-printable ASCII characters"),
 		},
+		{
+			md:   map[string][]string{"": {string(rune(0x19))}},
+			want: errors.New("there is an empty key in the header"),
+		},
 		{
 			md:   map[string][]string{"test-bin": {string(rune(0x19))}},
 			want: nil,

stream.go

@@ -168,8 +168,24 @@ func NewClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 }
 
 func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (_ ClientStream, err error) {
-	if md, _, ok := metadata.FromOutgoingContextRaw(ctx); ok {
-		if err := imetadata.Validate(md); err != nil {
+	// validate original input
+	if md, added, ok := metadata.FromOutgoingContextRaw(ctx); ok {
+		// copy the original MD to avoid changing it
+		mdSize := len(md)
+		for i := range added {
+			mdSize += len(added[i]) / 2
+		}
+		merged := make(metadata.MD, mdSize)
+		for k, v := range md {
+			merged[k] = v
+		}
+		for _, pairs := range added {
+			for i := 0; i < len(pairs); i += 2 {
+				k := pairs[i]
+				merged[k] = append(merged[k], pairs[i+1])
+			}
+		}
+		if err := imetadata.Validate(merged); err != nil {
 			return nil, status.Error(codes.Internal, err.Error())
 		}
 	}

test/metadata_test.go

@@ -36,12 +36,13 @@ import (
 )
 
 func (s) TestInvalidMetadata(t *testing.T) {
-	grpctest.TLogger.ExpectErrorN("stream: failed to validate md when setting trailer", 2)
+	grpctest.TLogger.ExpectErrorN("stream: failed to validate md when setting trailer", 5)
 
 	tests := []struct {
-		md   metadata.MD
-		want error
-		recv error
+		md       metadata.MD
+		appendMD []string
+		want     error
+		recv     error
 	}{
 		{
 			md:   map[string][]string{string(rune(0x19)): {"testVal"}},
@@ -53,6 +54,23 @@ func (s) TestInvalidMetadata(t *testing.T) {
 			want: status.Error(codes.Internal, "header key \"test\" contains value with non-printable ASCII characters"),
 			recv: status.Error(codes.Internal, "invalid header field"),
 		},
+		{
+			md:       map[string][]string{"test": {"test"}},
+			appendMD: []string{"/", "value"},
+			want:     status.Error(codes.Internal, "header key \"/\" contains illegal characters not in [0-9a-z-_.]"),
+			recv:     status.Error(codes.Internal, "invalid header field"),
+		},
+		{
+			md:       map[string][]string{"test": {"test"}},
+			appendMD: []string{"", "value"},
+			want:     status.Error(codes.Internal, "there is an empty key in the header"),
+			recv:     status.Error(codes.Internal, "invalid header field"),
+		},
+		{
+			md:   map[string][]string{"": {"test"}},
+			want: status.Error(codes.Internal, "there is an empty key in the header"),
+			recv: status.Error(codes.Internal, "invalid header field"),
+		},
 		{
 			md:   map[string][]string{"test-bin": {string(rune(0x19))}},
 			want: nil,
@@ -77,13 +95,22 @@ func (s) TestInvalidMetadata(t *testing.T) {
 			}
 			test := tests[testNum]
 			testNum++
-			if err := stream.SetHeader(test.md); !reflect.DeepEqual(test.want, err) {
-				return fmt.Errorf("call stream.SendHeader(md) validate metadata which is %v got err :%v, want err :%v", test.md, err, test.want)
+			// merge md
+			md := make(metadata.MD)
+			for k, v := range test.md {
+				md[k] = v
+			}
+			for i := 0; i < len(test.appendMD); i += 2 {
+				md[test.appendMD[i]] = append(md[test.appendMD[i]], test.appendMD[i+1])
 			}
-			if err := stream.SendHeader(test.md); !reflect.DeepEqual(test.want, err) {
-				return fmt.Errorf("call stream.SendHeader(md) validate metadata which is %v got err :%v, want err :%v", test.md, err, test.want)
+
+			if err := stream.SetHeader(md); !reflect.DeepEqual(test.want, err) {
+				return fmt.Errorf("call stream.SendHeader(md) validate metadata which is %v got err :%v, want err :%v", md, err, test.want)
 			}
-			stream.SetTrailer(test.md)
+			if err := stream.SendHeader(md); !reflect.DeepEqual(test.want, err) {
+				return fmt.Errorf("call stream.SendHeader(md) validate metadata which is %v got err :%v, want err :%v", md, err, test.want)
+			}
+			stream.SetTrailer(md)
 			return nil
 		},
 	}
@@ -94,21 +121,21 @@ func (s) TestInvalidMetadata(t *testing.T) {
 
 	for _, test := range tests {
 		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
-		defer cancel()
-
 		ctx = metadata.NewOutgoingContext(ctx, test.md)
+		ctx = metadata.AppendToOutgoingContext(ctx, test.appendMD...)
 		if _, err := ss.Client.EmptyCall(ctx, &testpb.Empty{}); !reflect.DeepEqual(test.want, err) {
 			t.Errorf("call ss.Client.EmptyCall() validate metadata which is %v got err :%v, want err :%v", test.md, err, test.want)
 		}
+		cancel()
 	}
 
 	// call the stream server's api to drive the server-side unit testing
 	for _, test := range tests {
 		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
 		stream, err := ss.Client.FullDuplexCall(ctx)
-		defer cancel()
 		if err != nil {
 			t.Errorf("call ss.Client.FullDuplexCall(context.Background()) will success but got err :%v", err)
+			cancel()
 			continue
 		}
 		if err := stream.Send(&testpb.StreamingOutputCallRequest{}); err != nil {
@@ -117,5 +144,6 @@ func (s) TestInvalidMetadata(t *testing.T) {
 		if _, err := stream.Recv(); status.Code(err) != status.Code(test.recv) || !strings.Contains(err.Error(), test.recv.Error()) {
 			t.Errorf("stream.Recv() = _, get err :%v, want err :%v", err, test.recv)
 		}
+		cancel()
 	}
 }