fix TestCompressOK and client reserved HTTP header handling

This commit is contained in:
Michal Witkowski
2016-05-17 14:29:48 +01:00
parent e54a726f0a
commit 1ef2c5293f
5 changed files with 46 additions and 19 deletions

View File

@ -460,6 +460,10 @@ func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.
} }
}() }()
} }
if s.opts.cp != nil {
// NOTE: this needs to be ahead of all handling, https://github.com/grpc/grpc-go/issues/686.
stream.SetSendCompress(s.opts.cp.Type())
}
p := &parser{r: stream} p := &parser{r: stream}
for { for {
pf, req, err := p.recvMsg() pf, req, err := p.recvMsg()
@ -545,9 +549,6 @@ func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.
Last: true, Last: true,
Delay: false, Delay: false,
} }
if s.opts.cp != nil {
stream.SetSendCompress(s.opts.cp.Type())
}
if err := s.sendResponse(t, stream, reply, s.opts.cp, opts); err != nil { if err := s.sendResponse(t, stream, reply, s.opts.cp, opts); err != nil {
switch err := err.(type) { switch err := err.(type) {
case transport.ConnectionError: case transport.ConnectionError:

View File

@ -130,6 +130,9 @@ func newPayload(t testpb.PayloadType, size int32) (*testpb.Payload, error) {
func (s *testServer) UnaryCall(ctx context.Context, in *testpb.SimpleRequest) (*testpb.SimpleResponse, error) { func (s *testServer) UnaryCall(ctx context.Context, in *testpb.SimpleRequest) (*testpb.SimpleResponse, error) {
md, ok := metadata.FromContext(ctx) md, ok := metadata.FromContext(ctx)
if ok { if ok {
if _, exists := md[":authority"]; !exists {
return nil, grpc.Errorf(codes.DataLoss, "expected an :authority metadata: %v", md)
}
if err := grpc.SendHeader(ctx, md); err != nil { if err := grpc.SendHeader(ctx, md); err != nil {
return nil, fmt.Errorf("grpc.SendHeader(%v, %v) = %v, want %v", ctx, md, err, nil) return nil, fmt.Errorf("grpc.SendHeader(%v, %v) = %v, want %v", ctx, md, err, nil)
} }
@ -167,6 +170,7 @@ func (s *testServer) UnaryCall(ctx context.Context, in *testpb.SimpleRequest) (*
if err != nil { if err != nil {
return nil, err return nil, err
} }
return &testpb.SimpleResponse{ return &testpb.SimpleResponse{
Payload: payload, Payload: payload,
}, nil }, nil
@ -174,8 +178,11 @@ func (s *testServer) UnaryCall(ctx context.Context, in *testpb.SimpleRequest) (*
func (s *testServer) StreamingOutputCall(args *testpb.StreamingOutputCallRequest, stream testpb.TestService_StreamingOutputCallServer) error { func (s *testServer) StreamingOutputCall(args *testpb.StreamingOutputCallRequest, stream testpb.TestService_StreamingOutputCallServer) error {
if md, ok := metadata.FromContext(stream.Context()); ok { if md, ok := metadata.FromContext(stream.Context()); ok {
// For testing purpose, returns an error if there is attached metadata. if _, exists := md[":authority"]; !exists {
if len(md) > 0 { return grpc.Errorf(codes.DataLoss, "expected an :authority metadata: %v", md)
}
// For testing purpose, returns an error if there is attached metadata except for authority.
if len(md) > 1 {
return grpc.Errorf(codes.DataLoss, "got extra metadata") return grpc.Errorf(codes.DataLoss, "got extra metadata")
} }
} }
@ -1678,7 +1685,8 @@ func testCompressOK(t *testing.T, e env) {
ResponseSize: proto.Int32(respSize), ResponseSize: proto.Int32(respSize),
Payload: payload, Payload: payload,
} }
if _, err := tc.UnaryCall(context.Background(), req); err != nil { ctx := metadata.NewContext(context.Background(), metadata.Pairs("something", "something"))
if _, err := tc.UnaryCall(ctx, req); err != nil {
t.Fatalf("TestService/UnaryCall(_, _) = _, %v, want _, <nil>", err) t.Fatalf("TestService/UnaryCall(_, _) = _, %v, want _, <nil>", err)
} }
// Streaming RPC // Streaming RPC

View File

@ -92,9 +92,12 @@ func NewServerHandlerTransport(w http.ResponseWriter, r *http.Request) (ServerTr
} }
var metakv []string var metakv []string
if r.Host != "" {
metakv = append(metakv, ":authority", r.Host)
}
for k, vv := range r.Header { for k, vv := range r.Header {
k = strings.ToLower(k) k = strings.ToLower(k)
if isReservedHeader(k) { if isReservedHeader(k) && !isWhitelistedHttp2Header(k){
continue continue
} }
for _, v := range vv { for _, v := range vv {
@ -108,7 +111,6 @@ func NewServerHandlerTransport(w http.ResponseWriter, r *http.Request) (ServerTr
} }
} }
metakv = append(metakv, k, v) metakv = append(metakv, k, v)
} }
} }
st.headerMD = metadata.Pairs(metakv...) st.headerMD = metadata.Pairs(metakv...)
@ -196,6 +198,10 @@ func (ht *serverHandlerTransport) WriteStatus(s *Stream, statusCode codes.Code,
} }
if md := s.Trailer(); len(md) > 0 { if md := s.Trailer(); len(md) > 0 {
for k, vv := range md { for k, vv := range md {
// Clients don't tolerate reading restricted headers after some non restricted ones were sent.
if isReservedHeader(k) {
continue
}
for _, v := range vv { for _, v := range vv {
// http2 ResponseWriter mechanism to // http2 ResponseWriter mechanism to
// send undeclared Trailers after the // send undeclared Trailers after the
@ -249,6 +255,10 @@ func (ht *serverHandlerTransport) WriteHeader(s *Stream, md metadata.MD) error {
ht.writeCommonHeaders(s) ht.writeCommonHeaders(s)
h := ht.rw.Header() h := ht.rw.Header()
for k, vv := range md { for k, vv := range md {
// Clients don't tolerate reading restricted headers after some non restricted ones were sent.
if isReservedHeader(k) {
continue
}
for _, v := range vv { for _, v := range vv {
h.Add(k, v) h.Add(k, v)
} }

View File

@ -460,6 +460,10 @@ func (t *http2Server) WriteHeader(s *Stream, md metadata.MD) error {
t.hEnc.WriteField(hpack.HeaderField{Name: "grpc-encoding", Value: s.sendCompress}) t.hEnc.WriteField(hpack.HeaderField{Name: "grpc-encoding", Value: s.sendCompress})
} }
for k, v := range md { for k, v := range md {
if isReservedHeader(k) {
// Clients don't tolerate reading restricted headers after some non restricted ones were sent.
continue
}
for _, entry := range v { for _, entry := range v {
t.hEnc.WriteField(hpack.HeaderField{Name: k, Value: entry}) t.hEnc.WriteField(hpack.HeaderField{Name: k, Value: entry})
} }
@ -502,6 +506,10 @@ func (t *http2Server) WriteStatus(s *Stream, statusCode codes.Code, statusDesc s
t.hEnc.WriteField(hpack.HeaderField{Name: "grpc-message", Value: statusDesc}) t.hEnc.WriteField(hpack.HeaderField{Name: "grpc-message", Value: statusDesc})
// Attach the trailer metadata. // Attach the trailer metadata.
for k, v := range s.trailer { for k, v := range s.trailer {
// Clients don't tolerate reading restricted headers after some non restricted ones were sent.
if isReservedHeader(k) {
continue
}
for _, entry := range v { for _, entry := range v {
t.hEnc.WriteField(hpack.HeaderField{Name: k, Value: entry}) t.hEnc.WriteField(hpack.HeaderField{Name: k, Value: entry})
} }

View File

@ -106,17 +106,6 @@ type decodeState struct {
mdata map[string][]string mdata map[string][]string
} }
// isWhitelistedHttp2Header checks whether hdr belongs to HTTP2 headers
// that should be propagated into metadata visible to users.
func isWhitelistedHttp2Header(hdr string) bool {
switch hdr {
case ":authority":
return true
default:
return false
}
}
// isReservedHeader checks whether hdr belongs to HTTP2 headers // isReservedHeader checks whether hdr belongs to HTTP2 headers
// reserved by gRPC protocol. Any other headers are classified as the // reserved by gRPC protocol. Any other headers are classified as the
// user-specified metadata. // user-specified metadata.
@ -138,6 +127,17 @@ func isReservedHeader(hdr string) bool {
} }
} }
// isWhitelistedHttp2Header checks whether hdr belongs to HTTP2 headers
// that should be propagated into metadata visible to users.
func isWhitelistedHttp2Header(hdr string) bool {
switch hdr {
case ":authority", "Authority":
return true
default:
return false
}
}
func (d *decodeState) setErr(err error) { func (d *decodeState) setErr(err error) {
if d.err == nil { if d.err == nil {
d.err = err d.err = err