3 * Copyright 2014 gRPC authors.
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
35 "github.com/golang/protobuf/proto"
36 "golang.org/x/net/http2"
37 "golang.org/x/net/http2/hpack"
38 spb "google.golang.org/genproto/googleapis/rpc/status"
39 "google.golang.org/grpc/codes"
40 "google.golang.org/grpc/status"
44 // http2MaxFrameLen specifies the max length of a HTTP2 frame.
45 http2MaxFrameLen = 16384 // 16KB frame
46 // http://http2.github.io/http2-spec/#SettingValues
47 http2InitHeaderTableSize = 4096
48 // baseContentType is the base content-type for gRPC. This is a valid
49 // content-type on it's own, but can also include a content-subtype such as
50 // "proto" as a suffix after "+" or ";". See
51 // https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests
53 baseContentType = "application/grpc"
57 clientPreface = []byte(http2.ClientPreface)
58 http2ErrConvTab = map[http2.ErrCode]codes.Code{
59 http2.ErrCodeNo: codes.Internal,
60 http2.ErrCodeProtocol: codes.Internal,
61 http2.ErrCodeInternal: codes.Internal,
62 http2.ErrCodeFlowControl: codes.ResourceExhausted,
63 http2.ErrCodeSettingsTimeout: codes.Internal,
64 http2.ErrCodeStreamClosed: codes.Internal,
65 http2.ErrCodeFrameSize: codes.Internal,
66 http2.ErrCodeRefusedStream: codes.Unavailable,
67 http2.ErrCodeCancel: codes.Canceled,
68 http2.ErrCodeCompression: codes.Internal,
69 http2.ErrCodeConnect: codes.Internal,
70 http2.ErrCodeEnhanceYourCalm: codes.ResourceExhausted,
71 http2.ErrCodeInadequateSecurity: codes.PermissionDenied,
72 http2.ErrCodeHTTP11Required: codes.Internal,
74 statusCodeConvTab = map[codes.Code]http2.ErrCode{
75 codes.Internal: http2.ErrCodeInternal,
76 codes.Canceled: http2.ErrCodeCancel,
77 codes.Unavailable: http2.ErrCodeRefusedStream,
78 codes.ResourceExhausted: http2.ErrCodeEnhanceYourCalm,
79 codes.PermissionDenied: http2.ErrCodeInadequateSecurity,
81 httpStatusConvTab = map[int]codes.Code{
82 // 400 Bad Request - INTERNAL.
83 http.StatusBadRequest: codes.Internal,
84 // 401 Unauthorized - UNAUTHENTICATED.
85 http.StatusUnauthorized: codes.Unauthenticated,
86 // 403 Forbidden - PERMISSION_DENIED.
87 http.StatusForbidden: codes.PermissionDenied,
88 // 404 Not Found - UNIMPLEMENTED.
89 http.StatusNotFound: codes.Unimplemented,
90 // 429 Too Many Requests - UNAVAILABLE.
91 http.StatusTooManyRequests: codes.Unavailable,
92 // 502 Bad Gateway - UNAVAILABLE.
93 http.StatusBadGateway: codes.Unavailable,
94 // 503 Service Unavailable - UNAVAILABLE.
95 http.StatusServiceUnavailable: codes.Unavailable,
96 // 504 Gateway timeout - UNAVAILABLE.
97 http.StatusGatewayTimeout: codes.Unavailable,
101 // Records the states during HPACK decoding. Must be reset once the
102 // decoding of the entire headers are finished.
103 type decodeState struct {
105 // statusGen caches the stream status received from the trailer the server
106 // sent. Client side only. Do not access directly. After all trailers are
107 // parsed, use the status method to retrieve the status.
108 statusGen *status.Status
109 // rawStatusCode and rawStatusMsg are set from the raw trailer fields and are not
110 // intended for direct access outside of parsing.
114 // Server side only fields.
116 timeout time.Duration
118 // key-value metadata map from the peer.
119 mdata map[string][]string
122 contentSubtype string
123 // whether decoding on server side or not
127 // isReservedHeader checks whether hdr belongs to HTTP2 headers
128 // reserved by gRPC protocol. Any other headers are classified as the
129 // user-specified metadata.
130 func isReservedHeader(hdr string) bool {
131 if hdr != "" && hdr[0] == ':' {
142 "grpc-status-details-bin",
143 // Intentionally exclude grpc-previous-rpc-attempts and
144 // grpc-retry-pushback-ms, which are "reserved", but their API
145 // intentionally works via metadata.
153 // isWhitelistedHeader checks whether hdr should be propagated into metadata
154 // visible to users, even though it is classified as "reserved", above.
155 func isWhitelistedHeader(hdr string) bool {
157 case ":authority", "user-agent":
164 // contentSubtype returns the content-subtype for the given content-type. The
165 // given content-type must be a valid content-type that starts with
166 // "application/grpc". A content-subtype will follow "application/grpc" after a
168 // https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests for
171 // If contentType is not a valid content-type for gRPC, the boolean
172 // will be false, otherwise true. If content-type == "application/grpc",
173 // "application/grpc+", or "application/grpc;", the boolean will be true,
174 // but no content-subtype will be returned.
176 // contentType is assumed to be lowercase already.
177 func contentSubtype(contentType string) (string, bool) {
178 if contentType == baseContentType {
181 if !strings.HasPrefix(contentType, baseContentType) {
184 // guaranteed since != baseContentType and has baseContentType prefix
185 switch contentType[len(baseContentType)] {
187 // this will return true for "application/grpc+" or "application/grpc;"
188 // which the previous validContentType function tested to be valid, so we
189 // just say that no content-subtype is specified in this case
190 return contentType[len(baseContentType)+1:], true
196 // contentSubtype is assumed to be lowercase
197 func contentType(contentSubtype string) string {
198 if contentSubtype == "" {
199 return baseContentType
201 return baseContentType + "+" + contentSubtype
204 func (d *decodeState) status() *status.Status {
205 if d.statusGen == nil {
206 // No status-details were provided; generate status using code/msg.
207 d.statusGen = status.New(codes.Code(int32(*(d.rawStatusCode))), d.rawStatusMsg)
212 const binHdrSuffix = "-bin"
214 func encodeBinHeader(v []byte) string {
215 return base64.RawStdEncoding.EncodeToString(v)
218 func decodeBinHeader(v string) ([]byte, error) {
220 // Input was padded, or padding was not necessary.
221 return base64.StdEncoding.DecodeString(v)
223 return base64.RawStdEncoding.DecodeString(v)
226 func encodeMetadataHeader(k, v string) string {
227 if strings.HasSuffix(k, binHdrSuffix) {
228 return encodeBinHeader(([]byte)(v))
233 func decodeMetadataHeader(k, v string) (string, error) {
234 if strings.HasSuffix(k, binHdrSuffix) {
235 b, err := decodeBinHeader(v)
236 return string(b), err
241 func (d *decodeState) decodeHeader(frame *http2.MetaHeadersFrame) error {
242 // frame.Truncated is set to true when framer detects that the current header
243 // list size hits MaxHeaderListSize limit.
245 return status.Error(codes.Internal, "peer header list size exceeded limit")
247 for _, hf := range frame.Fields {
248 if err := d.processHeaderField(hf); err != nil {
257 // If grpc status exists, no need to check further.
258 if d.rawStatusCode != nil || d.statusGen != nil {
262 // If grpc status doesn't exist and http status doesn't exist,
263 // then it's a malformed header.
264 if d.httpStatus == nil {
265 return status.Error(codes.Internal, "malformed header: doesn't contain status(gRPC or HTTP)")
268 if *(d.httpStatus) != http.StatusOK {
269 code, ok := httpStatusConvTab[*(d.httpStatus)]
273 return status.Error(code, http.StatusText(*(d.httpStatus)))
276 // gRPC status doesn't exist and http status is OK.
277 // Set rawStatusCode to be unknown and return nil error.
278 // So that, if the stream has ended this Unknown status
279 // will be propagated to the user.
280 // Otherwise, it will be ignored. In which case, status from
281 // a later trailer, that has StreamEnded flag set, is propagated.
282 code := int(codes.Unknown)
283 d.rawStatusCode = &code
287 func (d *decodeState) addMetadata(k, v string) {
289 d.mdata = make(map[string][]string)
291 d.mdata[k] = append(d.mdata[k], v)
294 func (d *decodeState) processHeaderField(f hpack.HeaderField) error {
297 contentSubtype, validContentType := contentSubtype(f.Value)
298 if !validContentType {
299 return status.Errorf(codes.Internal, "transport: received the unexpected content-type %q", f.Value)
301 d.contentSubtype = contentSubtype
302 // TODO: do we want to propagate the whole content-type in the metadata,
303 // or come up with a way to just propagate the content-subtype if it was set?
304 // ie {"content-type": "application/grpc+proto"} or {"content-subtype": "proto"}
306 d.addMetadata(f.Name, f.Value)
307 case "grpc-encoding":
310 code, err := strconv.Atoi(f.Value)
312 return status.Errorf(codes.Internal, "transport: malformed grpc-status: %v", err)
314 d.rawStatusCode = &code
316 d.rawStatusMsg = decodeGrpcMessage(f.Value)
317 case "grpc-status-details-bin":
318 v, err := decodeBinHeader(f.Value)
320 return status.Errorf(codes.Internal, "transport: malformed grpc-status-details-bin: %v", err)
323 if err := proto.Unmarshal(v, s); err != nil {
324 return status.Errorf(codes.Internal, "transport: malformed grpc-status-details-bin: %v", err)
326 d.statusGen = status.FromProto(s)
330 if d.timeout, err = decodeTimeout(f.Value); err != nil {
331 return status.Errorf(codes.Internal, "transport: malformed time-out: %v", err)
336 code, err := strconv.Atoi(f.Value)
338 return status.Errorf(codes.Internal, "transport: malformed http-status: %v", err)
341 case "grpc-tags-bin":
342 v, err := decodeBinHeader(f.Value)
344 return status.Errorf(codes.Internal, "transport: malformed grpc-tags-bin: %v", err)
347 d.addMetadata(f.Name, string(v))
348 case "grpc-trace-bin":
349 v, err := decodeBinHeader(f.Value)
351 return status.Errorf(codes.Internal, "transport: malformed grpc-trace-bin: %v", err)
354 d.addMetadata(f.Name, string(v))
356 if isReservedHeader(f.Name) && !isWhitelistedHeader(f.Name) {
359 v, err := decodeMetadataHeader(f.Name, f.Value)
361 errorf("Failed to decode metadata header (%q, %q): %v", f.Name, f.Value, err)
364 d.addMetadata(f.Name, v)
369 type timeoutUnit uint8
372 hour timeoutUnit = 'H'
373 minute timeoutUnit = 'M'
374 second timeoutUnit = 'S'
375 millisecond timeoutUnit = 'm'
376 microsecond timeoutUnit = 'u'
377 nanosecond timeoutUnit = 'n'
380 func timeoutUnitToDuration(u timeoutUnit) (d time.Duration, ok bool) {
383 return time.Hour, true
385 return time.Minute, true
387 return time.Second, true
389 return time.Millisecond, true
391 return time.Microsecond, true
393 return time.Nanosecond, true
399 const maxTimeoutValue int64 = 100000000 - 1
401 // div does integer division and round-up the result. Note that this is
402 // equivalent to (d+r-1)/r but has less chance to overflow.
403 func div(d, r time.Duration) int64 {
404 if m := d % r; m > 0 {
405 return int64(d/r + 1)
410 // TODO(zhaoq): It is the simplistic and not bandwidth efficient. Improve it.
411 func encodeTimeout(t time.Duration) string {
415 if d := div(t, time.Nanosecond); d <= maxTimeoutValue {
416 return strconv.FormatInt(d, 10) + "n"
418 if d := div(t, time.Microsecond); d <= maxTimeoutValue {
419 return strconv.FormatInt(d, 10) + "u"
421 if d := div(t, time.Millisecond); d <= maxTimeoutValue {
422 return strconv.FormatInt(d, 10) + "m"
424 if d := div(t, time.Second); d <= maxTimeoutValue {
425 return strconv.FormatInt(d, 10) + "S"
427 if d := div(t, time.Minute); d <= maxTimeoutValue {
428 return strconv.FormatInt(d, 10) + "M"
430 // Note that maxTimeoutValue * time.Hour > MaxInt64.
431 return strconv.FormatInt(div(t, time.Hour), 10) + "H"
434 func decodeTimeout(s string) (time.Duration, error) {
437 return 0, fmt.Errorf("transport: timeout string is too short: %q", s)
440 // Spec allows for 8 digits plus the unit.
441 return 0, fmt.Errorf("transport: timeout string is too long: %q", s)
443 unit := timeoutUnit(s[size-1])
444 d, ok := timeoutUnitToDuration(unit)
446 return 0, fmt.Errorf("transport: timeout unit is not recognized: %q", s)
448 t, err := strconv.ParseInt(s[:size-1], 10, 64)
452 const maxHours = math.MaxInt64 / int64(time.Hour)
453 if d == time.Hour && t > maxHours {
454 // This timeout would overflow math.MaxInt64; clamp it.
455 return time.Duration(math.MaxInt64), nil
457 return d * time.Duration(t), nil
466 // encodeGrpcMessage is used to encode status code in header field
467 // "grpc-message". It does percent encoding and also replaces invalid utf-8
468 // characters with Unicode replacement character.
470 // It checks to see if each individual byte in msg is an allowable byte, and
471 // then either percent encoding or passing it through. When percent encoding,
472 // the byte is converted into hexadecimal notation with a '%' prepended.
473 func encodeGrpcMessage(msg string) string {
478 for i := 0; i < lenMsg; i++ {
480 if !(c >= spaceByte && c <= tildeByte && c != percentByte) {
481 return encodeGrpcMessageUnchecked(msg)
487 func encodeGrpcMessageUnchecked(msg string) string {
490 r, size := utf8.DecodeRuneInString(msg)
491 for _, b := range []byte(string(r)) {
493 // If size > 1, r is not ascii. Always do percent encoding.
494 buf.WriteString(fmt.Sprintf("%%%02X", b))
498 // The for loop is necessary even if size == 1. r could be
501 // fmt.Sprintf("%%%02X", utf8.RuneError) gives "%FFFD".
502 if b >= spaceByte && b <= tildeByte && b != percentByte {
505 buf.WriteString(fmt.Sprintf("%%%02X", b))
513 // decodeGrpcMessage decodes the msg encoded by encodeGrpcMessage.
514 func decodeGrpcMessage(msg string) string {
519 for i := 0; i < lenMsg; i++ {
520 if msg[i] == percentByte && i+2 < lenMsg {
521 return decodeGrpcMessageUnchecked(msg)
527 func decodeGrpcMessageUnchecked(msg string) string {
530 for i := 0; i < lenMsg; i++ {
532 if c == percentByte && i+2 < lenMsg {
533 parsed, err := strconv.ParseUint(msg[i+1:i+3], 16, 8)
537 buf.WriteByte(byte(parsed))
547 type bufWriter struct {
557 func newBufWriter(conn net.Conn, batchSize int) *bufWriter {
559 buf: make([]byte, batchSize*2),
560 batchSize: batchSize,
565 func (w *bufWriter) Write(b []byte) (n int, err error) {
569 if w.batchSize == 0 { // Buffer has been disabled.
570 return w.conn.Write(b)
573 nn := copy(w.buf[w.offset:], b)
577 if w.offset >= w.batchSize {
584 func (w *bufWriter) Flush() error {
591 if w.onFlush != nil {
594 _, w.err = w.conn.Write(w.buf[:w.offset])
604 func newFramer(conn net.Conn, writeBufferSize, readBufferSize int, maxHeaderListSize uint32) *framer {
605 if writeBufferSize < 0 {
608 var r io.Reader = conn
609 if readBufferSize > 0 {
610 r = bufio.NewReaderSize(r, readBufferSize)
612 w := newBufWriter(conn, writeBufferSize)
615 fr: http2.NewFramer(w, r),
617 // Opt-in to Frame reuse API on framer to reduce garbage.
618 // Frames aren't safe to read from after a subsequent call to ReadFrame.
619 f.fr.SetReuseFrames()
620 f.fr.MaxHeaderListSize = maxHeaderListSize
621 f.fr.ReadMetaHeaders = hpack.NewDecoder(http2InitHeaderTableSize, nil)