vendor/google.golang.org/grpc/transport/http_util.go

   1 /*
   2  *
   3  * Copyright 2014 gRPC authors.
   4  *
   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
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  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.
  16  *
  17  */
  18
  19 package transport
  20
  21 import (
  22         "bufio"
  23         "bytes"
  24         "encoding/base64"
  25         "fmt"
  26         "io"
  27         "net"
  28         "net/http"
  29         "strconv"
  30         "strings"
  31         "time"
  32
  33         "github.com/golang/protobuf/proto"
  34         "golang.org/x/net/http2"
  35         "golang.org/x/net/http2/hpack"
  36         spb "google.golang.org/genproto/googleapis/rpc/status"
  37         "google.golang.org/grpc/codes"
  38         "google.golang.org/grpc/status"
  39 )
  40
  41 const (
  42         // http2MaxFrameLen specifies the max length of a HTTP2 frame.
  43         http2MaxFrameLen = 16384 // 16KB frame
  44         // http://http2.github.io/http2-spec/#SettingValues
  45         http2InitHeaderTableSize = 4096
  46         // http2IOBufSize specifies the buffer size for sending frames.
  47         defaultWriteBufSize = 32 * 1024
  48         defaultReadBufSize  = 32 * 1024
  49 )
  50
  51 var (
  52         clientPreface   = []byte(http2.ClientPreface)
  53         http2ErrConvTab = map[http2.ErrCode]codes.Code{
  54                 http2.ErrCodeNo:                 codes.Internal,
  55                 http2.ErrCodeProtocol:           codes.Internal,
  56                 http2.ErrCodeInternal:           codes.Internal,
  57                 http2.ErrCodeFlowControl:        codes.ResourceExhausted,
  58                 http2.ErrCodeSettingsTimeout:    codes.Internal,
  59                 http2.ErrCodeStreamClosed:       codes.Internal,
  60                 http2.ErrCodeFrameSize:          codes.Internal,
  61                 http2.ErrCodeRefusedStream:      codes.Unavailable,
  62                 http2.ErrCodeCancel:             codes.Canceled,
  63                 http2.ErrCodeCompression:        codes.Internal,
  64                 http2.ErrCodeConnect:            codes.Internal,
  65                 http2.ErrCodeEnhanceYourCalm:    codes.ResourceExhausted,
  66                 http2.ErrCodeInadequateSecurity: codes.PermissionDenied,
  67                 http2.ErrCodeHTTP11Required:     codes.FailedPrecondition,
  68         }
  69         statusCodeConvTab = map[codes.Code]http2.ErrCode{
  70                 codes.Internal:          http2.ErrCodeInternal,
  71                 codes.Canceled:          http2.ErrCodeCancel,
  72                 codes.Unavailable:       http2.ErrCodeRefusedStream,
  73                 codes.ResourceExhausted: http2.ErrCodeEnhanceYourCalm,
  74                 codes.PermissionDenied:  http2.ErrCodeInadequateSecurity,
  75         }
  76         httpStatusConvTab = map[int]codes.Code{
  77                 // 400 Bad Request - INTERNAL.
  78                 http.StatusBadRequest: codes.Internal,
  79                 // 401 Unauthorized  - UNAUTHENTICATED.
  80                 http.StatusUnauthorized: codes.Unauthenticated,
  81                 // 403 Forbidden - PERMISSION_DENIED.
  82                 http.StatusForbidden: codes.PermissionDenied,
  83                 // 404 Not Found - UNIMPLEMENTED.
  84                 http.StatusNotFound: codes.Unimplemented,
  85                 // 429 Too Many Requests - UNAVAILABLE.
  86                 http.StatusTooManyRequests: codes.Unavailable,
  87                 // 502 Bad Gateway - UNAVAILABLE.
  88                 http.StatusBadGateway: codes.Unavailable,
  89                 // 503 Service Unavailable - UNAVAILABLE.
  90                 http.StatusServiceUnavailable: codes.Unavailable,
  91                 // 504 Gateway timeout - UNAVAILABLE.
  92                 http.StatusGatewayTimeout: codes.Unavailable,
  93         }
  94 )
  95
  96 // Records the states during HPACK decoding. Must be reset once the
  97 // decoding of the entire headers are finished.
  98 type decodeState struct {
  99         encoding string
 100         // statusGen caches the stream status received from the trailer the server
 101         // sent.  Client side only.  Do not access directly.  After all trailers are
 102         // parsed, use the status method to retrieve the status.
 103         statusGen *status.Status
 104         // rawStatusCode and rawStatusMsg are set from the raw trailer fields and are not
 105         // intended for direct access outside of parsing.
 106         rawStatusCode *int
 107         rawStatusMsg  string
 108         httpStatus    *int
 109         // Server side only fields.
 110         timeoutSet bool
 111         timeout    time.Duration
 112         method     string
 113         // key-value metadata map from the peer.
 114         mdata      map[string][]string
 115         statsTags  []byte
 116         statsTrace []byte
 117 }
 118
 119 // isReservedHeader checks whether hdr belongs to HTTP2 headers
 120 // reserved by gRPC protocol. Any other headers are classified as the
 121 // user-specified metadata.
 122 func isReservedHeader(hdr string) bool {
 123         if hdr != "" && hdr[0] == ':' {
 124                 return true
 125         }
 126         switch hdr {
 127         case "content-type",
 128                 "grpc-message-type",
 129                 "grpc-encoding",
 130                 "grpc-message",
 131                 "grpc-status",
 132                 "grpc-timeout",
 133                 "grpc-status-details-bin",
 134                 "te":
 135                 return true
 136         default:
 137                 return false
 138         }
 139 }
 140
 141 // isWhitelistedPseudoHeader checks whether hdr belongs to HTTP2 pseudoheaders
 142 // that should be propagated into metadata visible to users.
 143 func isWhitelistedPseudoHeader(hdr string) bool {
 144         switch hdr {
 145         case ":authority":
 146                 return true
 147         default:
 148                 return false
 149         }
 150 }
 151
 152 func validContentType(t string) bool {
 153         e := "application/grpc"
 154         if !strings.HasPrefix(t, e) {
 155                 return false
 156         }
 157         // Support variations on the content-type
 158         // (e.g. "application/grpc+blah", "application/grpc;blah").
 159         if len(t) > len(e) && t[len(e)] != '+' && t[len(e)] != ';' {
 160                 return false
 161         }
 162         return true
 163 }
 164
 165 func (d *decodeState) status() *status.Status {
 166         if d.statusGen == nil {
 167                 // No status-details were provided; generate status using code/msg.
 168                 d.statusGen = status.New(codes.Code(int32(*(d.rawStatusCode))), d.rawStatusMsg)
 169         }
 170         return d.statusGen
 171 }
 172
 173 const binHdrSuffix = "-bin"
 174
 175 func encodeBinHeader(v []byte) string {
 176         return base64.RawStdEncoding.EncodeToString(v)
 177 }
 178
 179 func decodeBinHeader(v string) ([]byte, error) {
 180         if len(v)%4 == 0 {
 181                 // Input was padded, or padding was not necessary.
 182                 return base64.StdEncoding.DecodeString(v)
 183         }
 184         return base64.RawStdEncoding.DecodeString(v)
 185 }
 186
 187 func encodeMetadataHeader(k, v string) string {
 188         if strings.HasSuffix(k, binHdrSuffix) {
 189                 return encodeBinHeader(([]byte)(v))
 190         }
 191         return v
 192 }
 193
 194 func decodeMetadataHeader(k, v string) (string, error) {
 195         if strings.HasSuffix(k, binHdrSuffix) {
 196                 b, err := decodeBinHeader(v)
 197                 return string(b), err
 198         }
 199         return v, nil
 200 }
 201
 202 func (d *decodeState) decodeResponseHeader(frame *http2.MetaHeadersFrame) error {
 203         for _, hf := range frame.Fields {
 204                 if err := d.processHeaderField(hf); err != nil {
 205                         return err
 206                 }
 207         }
 208
 209         // If grpc status exists, no need to check further.
 210         if d.rawStatusCode != nil || d.statusGen != nil {
 211                 return nil
 212         }
 213
 214         // If grpc status doesn't exist and http status doesn't exist,
 215         // then it's a malformed header.
 216         if d.httpStatus == nil {
 217                 return streamErrorf(codes.Internal, "malformed header: doesn't contain status(gRPC or HTTP)")
 218         }
 219
 220         if *(d.httpStatus) != http.StatusOK {
 221                 code, ok := httpStatusConvTab[*(d.httpStatus)]
 222                 if !ok {
 223                         code = codes.Unknown
 224                 }
 225                 return streamErrorf(code, http.StatusText(*(d.httpStatus)))
 226         }
 227
 228         // gRPC status doesn't exist and http status is OK.
 229         // Set rawStatusCode to be unknown and return nil error.
 230         // So that, if the stream has ended this Unknown status
 231         // will be propogated to the user.
 232         // Otherwise, it will be ignored. In which case, status from
 233         // a later trailer, that has StreamEnded flag set, is propogated.
 234         code := int(codes.Unknown)
 235         d.rawStatusCode = &code
 236         return nil
 237
 238 }
 239
 240 func (d *decodeState) addMetadata(k, v string) {
 241         if d.mdata == nil {
 242                 d.mdata = make(map[string][]string)
 243         }
 244         d.mdata[k] = append(d.mdata[k], v)
 245 }
 246
 247 func (d *decodeState) processHeaderField(f hpack.HeaderField) error {
 248         switch f.Name {
 249         case "content-type":
 250                 if !validContentType(f.Value) {
 251                         return streamErrorf(codes.FailedPrecondition, "transport: received the unexpected content-type %q", f.Value)
 252                 }
 253         case "grpc-encoding":
 254                 d.encoding = f.Value
 255         case "grpc-status":
 256                 code, err := strconv.Atoi(f.Value)
 257                 if err != nil {
 258                         return streamErrorf(codes.Internal, "transport: malformed grpc-status: %v", err)
 259                 }
 260                 d.rawStatusCode = &code
 261         case "grpc-message":
 262                 d.rawStatusMsg = decodeGrpcMessage(f.Value)
 263         case "grpc-status-details-bin":
 264                 v, err := decodeBinHeader(f.Value)
 265                 if err != nil {
 266                         return streamErrorf(codes.Internal, "transport: malformed grpc-status-details-bin: %v", err)
 267                 }
 268                 s := &spb.Status{}
 269                 if err := proto.Unmarshal(v, s); err != nil {
 270                         return streamErrorf(codes.Internal, "transport: malformed grpc-status-details-bin: %v", err)
 271                 }
 272                 d.statusGen = status.FromProto(s)
 273         case "grpc-timeout":
 274                 d.timeoutSet = true
 275                 var err error
 276                 if d.timeout, err = decodeTimeout(f.Value); err != nil {
 277                         return streamErrorf(codes.Internal, "transport: malformed time-out: %v", err)
 278                 }
 279         case ":path":
 280                 d.method = f.Value
 281         case ":status":
 282                 code, err := strconv.Atoi(f.Value)
 283                 if err != nil {
 284                         return streamErrorf(codes.Internal, "transport: malformed http-status: %v", err)
 285                 }
 286                 d.httpStatus = &code
 287         case "grpc-tags-bin":
 288                 v, err := decodeBinHeader(f.Value)
 289                 if err != nil {
 290                         return streamErrorf(codes.Internal, "transport: malformed grpc-tags-bin: %v", err)
 291                 }
 292                 d.statsTags = v
 293                 d.addMetadata(f.Name, string(v))
 294         case "grpc-trace-bin":
 295                 v, err := decodeBinHeader(f.Value)
 296                 if err != nil {
 297                         return streamErrorf(codes.Internal, "transport: malformed grpc-trace-bin: %v", err)
 298                 }
 299                 d.statsTrace = v
 300                 d.addMetadata(f.Name, string(v))
 301         default:
 302                 if isReservedHeader(f.Name) && !isWhitelistedPseudoHeader(f.Name) {
 303                         break
 304                 }
 305                 v, err := decodeMetadataHeader(f.Name, f.Value)
 306                 if err != nil {
 307                         errorf("Failed to decode metadata header (%q, %q): %v", f.Name, f.Value, err)
 308                         return nil
 309                 }
 310                 d.addMetadata(f.Name, string(v))
 311         }
 312         return nil
 313 }
 314
 315 type timeoutUnit uint8
 316
 317 const (
 318         hour        timeoutUnit = 'H'
 319         minute      timeoutUnit = 'M'
 320         second      timeoutUnit = 'S'
 321         millisecond timeoutUnit = 'm'
 322         microsecond timeoutUnit = 'u'
 323         nanosecond  timeoutUnit = 'n'
 324 )
 325
 326 func timeoutUnitToDuration(u timeoutUnit) (d time.Duration, ok bool) {
 327         switch u {
 328         case hour:
 329                 return time.Hour, true
 330         case minute:
 331                 return time.Minute, true
 332         case second:
 333                 return time.Second, true
 334         case millisecond:
 335                 return time.Millisecond, true
 336         case microsecond:
 337                 return time.Microsecond, true
 338         case nanosecond:
 339                 return time.Nanosecond, true
 340         default:
 341         }
 342         return
 343 }
 344
 345 const maxTimeoutValue int64 = 100000000 - 1
 346
 347 // div does integer division and round-up the result. Note that this is
 348 // equivalent to (d+r-1)/r but has less chance to overflow.
 349 func div(d, r time.Duration) int64 {
 350         if m := d % r; m > 0 {
 351                 return int64(d/r + 1)
 352         }
 353         return int64(d / r)
 354 }
 355
 356 // TODO(zhaoq): It is the simplistic and not bandwidth efficient. Improve it.
 357 func encodeTimeout(t time.Duration) string {
 358         if t <= 0 {
 359                 return "0n"
 360         }
 361         if d := div(t, time.Nanosecond); d <= maxTimeoutValue {
 362                 return strconv.FormatInt(d, 10) + "n"
 363         }
 364         if d := div(t, time.Microsecond); d <= maxTimeoutValue {
 365                 return strconv.FormatInt(d, 10) + "u"
 366         }
 367         if d := div(t, time.Millisecond); d <= maxTimeoutValue {
 368                 return strconv.FormatInt(d, 10) + "m"
 369         }
 370         if d := div(t, time.Second); d <= maxTimeoutValue {
 371                 return strconv.FormatInt(d, 10) + "S"
 372         }
 373         if d := div(t, time.Minute); d <= maxTimeoutValue {
 374                 return strconv.FormatInt(d, 10) + "M"
 375         }
 376         // Note that maxTimeoutValue * time.Hour > MaxInt64.
 377         return strconv.FormatInt(div(t, time.Hour), 10) + "H"
 378 }
 379
 380 func decodeTimeout(s string) (time.Duration, error) {
 381         size := len(s)
 382         if size < 2 {
 383                 return 0, fmt.Errorf("transport: timeout string is too short: %q", s)
 384         }
 385         unit := timeoutUnit(s[size-1])
 386         d, ok := timeoutUnitToDuration(unit)
 387         if !ok {
 388                 return 0, fmt.Errorf("transport: timeout unit is not recognized: %q", s)
 389         }
 390         t, err := strconv.ParseInt(s[:size-1], 10, 64)
 391         if err != nil {
 392                 return 0, err
 393         }
 394         return d * time.Duration(t), nil
 395 }
 396
 397 const (
 398         spaceByte   = ' '
 399         tildaByte   = '~'
 400         percentByte = '%'
 401 )
 402
 403 // encodeGrpcMessage is used to encode status code in header field
 404 // "grpc-message".
 405 // It checks to see if each individual byte in msg is an
 406 // allowable byte, and then either percent encoding or passing it through.
 407 // When percent encoding, the byte is converted into hexadecimal notation
 408 // with a '%' prepended.
 409 func encodeGrpcMessage(msg string) string {
 410         if msg == "" {
 411                 return ""
 412         }
 413         lenMsg := len(msg)
 414         for i := 0; i < lenMsg; i++ {
 415                 c := msg[i]
 416                 if !(c >= spaceByte && c < tildaByte && c != percentByte) {
 417                         return encodeGrpcMessageUnchecked(msg)
 418                 }
 419         }
 420         return msg
 421 }
 422
 423 func encodeGrpcMessageUnchecked(msg string) string {
 424         var buf bytes.Buffer
 425         lenMsg := len(msg)
 426         for i := 0; i < lenMsg; i++ {
 427                 c := msg[i]
 428                 if c >= spaceByte && c < tildaByte && c != percentByte {
 429                         buf.WriteByte(c)
 430                 } else {
 431                         buf.WriteString(fmt.Sprintf("%%%02X", c))
 432                 }
 433         }
 434         return buf.String()
 435 }
 436
 437 // decodeGrpcMessage decodes the msg encoded by encodeGrpcMessage.
 438 func decodeGrpcMessage(msg string) string {
 439         if msg == "" {
 440                 return ""
 441         }
 442         lenMsg := len(msg)
 443         for i := 0; i < lenMsg; i++ {
 444                 if msg[i] == percentByte && i+2 < lenMsg {
 445                         return decodeGrpcMessageUnchecked(msg)
 446                 }
 447         }
 448         return msg
 449 }
 450
 451 func decodeGrpcMessageUnchecked(msg string) string {
 452         var buf bytes.Buffer
 453         lenMsg := len(msg)
 454         for i := 0; i < lenMsg; i++ {
 455                 c := msg[i]
 456                 if c == percentByte && i+2 < lenMsg {
 457                         parsed, err := strconv.ParseUint(msg[i+1:i+3], 16, 8)
 458                         if err != nil {
 459                                 buf.WriteByte(c)
 460                         } else {
 461                                 buf.WriteByte(byte(parsed))
 462                                 i += 2
 463                         }
 464                 } else {
 465                         buf.WriteByte(c)
 466                 }
 467         }
 468         return buf.String()
 469 }
 470
 471 type framer struct {
 472         numWriters int32
 473         reader     io.Reader
 474         writer     *bufio.Writer
 475         fr         *http2.Framer
 476 }
 477
 478 func newFramer(conn net.Conn, writeBufferSize, readBufferSize int) *framer {
 479         f := &framer{
 480                 reader: bufio.NewReaderSize(conn, readBufferSize),
 481                 writer: bufio.NewWriterSize(conn, writeBufferSize),
 482         }
 483         f.fr = http2.NewFramer(f.writer, f.reader)
 484         // Opt-in to Frame reuse API on framer to reduce garbage.
 485         // Frames aren't safe to read from after a subsequent call to ReadFrame.
 486         f.fr.SetReuseFrames()
 487         f.fr.ReadMetaHeaders = hpack.NewDecoder(http2InitHeaderTableSize, nil)
 488         return f
 489 }