cmd/bpa-operator/vendor/google.golang.org/grpc/internal/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         "math"
  28         "net"
  29         "net/http"
  30         "strconv"
  31         "strings"
  32         "time"
  33         "unicode/utf8"
  34
  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"
  41 )
  42
  43 const (
  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
  52         // for more details.
  53         baseContentType = "application/grpc"
  54 )
  55
  56 var (
  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,
  73         }
  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,
  80         }
  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,
  98         }
  99 )
 100
 101 // Records the states during HPACK decoding. Must be reset once the
 102 // decoding of the entire headers are finished.
 103 type decodeState struct {
 104         encoding string
 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.
 111         rawStatusCode *int
 112         rawStatusMsg  string
 113         httpStatus    *int
 114         // Server side only fields.
 115         timeoutSet bool
 116         timeout    time.Duration
 117         method     string
 118         // key-value metadata map from the peer.
 119         mdata          map[string][]string
 120         statsTags      []byte
 121         statsTrace     []byte
 122         contentSubtype string
 123         // whether decoding on server side or not
 124         serverSide bool
 125 }
 126
 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] == ':' {
 132                 return true
 133         }
 134         switch hdr {
 135         case "content-type",
 136                 "user-agent",
 137                 "grpc-message-type",
 138                 "grpc-encoding",
 139                 "grpc-message",
 140                 "grpc-status",
 141                 "grpc-timeout",
 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.
 146                 "te":
 147                 return true
 148         default:
 149                 return false
 150         }
 151 }
 152
 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 {
 156         switch hdr {
 157         case ":authority", "user-agent":
 158                 return true
 159         default:
 160                 return false
 161         }
 162 }
 163
 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
 167 // "+" or ";". See
 168 // https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests for
 169 // more details.
 170 //
 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.
 175 //
 176 // contentType is assumed to be lowercase already.
 177 func contentSubtype(contentType string) (string, bool) {
 178         if contentType == baseContentType {
 179                 return "", true
 180         }
 181         if !strings.HasPrefix(contentType, baseContentType) {
 182                 return "", false
 183         }
 184         // guaranteed since != baseContentType and has baseContentType prefix
 185         switch contentType[len(baseContentType)] {
 186         case '+', ';':
 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
 191         default:
 192                 return "", false
 193         }
 194 }
 195
 196 // contentSubtype is assumed to be lowercase
 197 func contentType(contentSubtype string) string {
 198         if contentSubtype == "" {
 199                 return baseContentType
 200         }
 201         return baseContentType + "+" + contentSubtype
 202 }
 203
 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)
 208         }
 209         return d.statusGen
 210 }
 211
 212 const binHdrSuffix = "-bin"
 213
 214 func encodeBinHeader(v []byte) string {
 215         return base64.RawStdEncoding.EncodeToString(v)
 216 }
 217
 218 func decodeBinHeader(v string) ([]byte, error) {
 219         if len(v)%4 == 0 {
 220                 // Input was padded, or padding was not necessary.
 221                 return base64.StdEncoding.DecodeString(v)
 222         }
 223         return base64.RawStdEncoding.DecodeString(v)
 224 }
 225
 226 func encodeMetadataHeader(k, v string) string {
 227         if strings.HasSuffix(k, binHdrSuffix) {
 228                 return encodeBinHeader(([]byte)(v))
 229         }
 230         return v
 231 }
 232
 233 func decodeMetadataHeader(k, v string) (string, error) {
 234         if strings.HasSuffix(k, binHdrSuffix) {
 235                 b, err := decodeBinHeader(v)
 236                 return string(b), err
 237         }
 238         return v, nil
 239 }
 240
 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.
 244         if frame.Truncated {
 245                 return status.Error(codes.Internal, "peer header list size exceeded limit")
 246         }
 247         for _, hf := range frame.Fields {
 248                 if err := d.processHeaderField(hf); err != nil {
 249                         return err
 250                 }
 251         }
 252
 253         if d.serverSide {
 254                 return nil
 255         }
 256
 257         // If grpc status exists, no need to check further.
 258         if d.rawStatusCode != nil || d.statusGen != nil {
 259                 return nil
 260         }
 261
 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)")
 266         }
 267
 268         if *(d.httpStatus) != http.StatusOK {
 269                 code, ok := httpStatusConvTab[*(d.httpStatus)]
 270                 if !ok {
 271                         code = codes.Unknown
 272                 }
 273                 return status.Error(code, http.StatusText(*(d.httpStatus)))
 274         }
 275
 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
 284         return nil
 285 }
 286
 287 func (d *decodeState) addMetadata(k, v string) {
 288         if d.mdata == nil {
 289                 d.mdata = make(map[string][]string)
 290         }
 291         d.mdata[k] = append(d.mdata[k], v)
 292 }
 293
 294 func (d *decodeState) processHeaderField(f hpack.HeaderField) error {
 295         switch f.Name {
 296         case "content-type":
 297                 contentSubtype, validContentType := contentSubtype(f.Value)
 298                 if !validContentType {
 299                         return status.Errorf(codes.Internal, "transport: received the unexpected content-type %q", f.Value)
 300                 }
 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"}
 305                 // in the metadata?
 306                 d.addMetadata(f.Name, f.Value)
 307         case "grpc-encoding":
 308                 d.encoding = f.Value
 309         case "grpc-status":
 310                 code, err := strconv.Atoi(f.Value)
 311                 if err != nil {
 312                         return status.Errorf(codes.Internal, "transport: malformed grpc-status: %v", err)
 313                 }
 314                 d.rawStatusCode = &code
 315         case "grpc-message":
 316                 d.rawStatusMsg = decodeGrpcMessage(f.Value)
 317         case "grpc-status-details-bin":
 318                 v, err := decodeBinHeader(f.Value)
 319                 if err != nil {
 320                         return status.Errorf(codes.Internal, "transport: malformed grpc-status-details-bin: %v", err)
 321                 }
 322                 s := &spb.Status{}
 323                 if err := proto.Unmarshal(v, s); err != nil {
 324                         return status.Errorf(codes.Internal, "transport: malformed grpc-status-details-bin: %v", err)
 325                 }
 326                 d.statusGen = status.FromProto(s)
 327         case "grpc-timeout":
 328                 d.timeoutSet = true
 329                 var err error
 330                 if d.timeout, err = decodeTimeout(f.Value); err != nil {
 331                         return status.Errorf(codes.Internal, "transport: malformed time-out: %v", err)
 332                 }
 333         case ":path":
 334                 d.method = f.Value
 335         case ":status":
 336                 code, err := strconv.Atoi(f.Value)
 337                 if err != nil {
 338                         return status.Errorf(codes.Internal, "transport: malformed http-status: %v", err)
 339                 }
 340                 d.httpStatus = &code
 341         case "grpc-tags-bin":
 342                 v, err := decodeBinHeader(f.Value)
 343                 if err != nil {
 344                         return status.Errorf(codes.Internal, "transport: malformed grpc-tags-bin: %v", err)
 345                 }
 346                 d.statsTags = v
 347                 d.addMetadata(f.Name, string(v))
 348         case "grpc-trace-bin":
 349                 v, err := decodeBinHeader(f.Value)
 350                 if err != nil {
 351                         return status.Errorf(codes.Internal, "transport: malformed grpc-trace-bin: %v", err)
 352                 }
 353                 d.statsTrace = v
 354                 d.addMetadata(f.Name, string(v))
 355         default:
 356                 if isReservedHeader(f.Name) && !isWhitelistedHeader(f.Name) {
 357                         break
 358                 }
 359                 v, err := decodeMetadataHeader(f.Name, f.Value)
 360                 if err != nil {
 361                         errorf("Failed to decode metadata header (%q, %q): %v", f.Name, f.Value, err)
 362                         return nil
 363                 }
 364                 d.addMetadata(f.Name, v)
 365         }
 366         return nil
 367 }
 368
 369 type timeoutUnit uint8
 370
 371 const (
 372         hour        timeoutUnit = 'H'
 373         minute      timeoutUnit = 'M'
 374         second      timeoutUnit = 'S'
 375         millisecond timeoutUnit = 'm'
 376         microsecond timeoutUnit = 'u'
 377         nanosecond  timeoutUnit = 'n'
 378 )
 379
 380 func timeoutUnitToDuration(u timeoutUnit) (d time.Duration, ok bool) {
 381         switch u {
 382         case hour:
 383                 return time.Hour, true
 384         case minute:
 385                 return time.Minute, true
 386         case second:
 387                 return time.Second, true
 388         case millisecond:
 389                 return time.Millisecond, true
 390         case microsecond:
 391                 return time.Microsecond, true
 392         case nanosecond:
 393                 return time.Nanosecond, true
 394         default:
 395         }
 396         return
 397 }
 398
 399 const maxTimeoutValue int64 = 100000000 - 1
 400
 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)
 406         }
 407         return int64(d / r)
 408 }
 409
 410 // TODO(zhaoq): It is the simplistic and not bandwidth efficient. Improve it.
 411 func encodeTimeout(t time.Duration) string {
 412         if t <= 0 {
 413                 return "0n"
 414         }
 415         if d := div(t, time.Nanosecond); d <= maxTimeoutValue {
 416                 return strconv.FormatInt(d, 10) + "n"
 417         }
 418         if d := div(t, time.Microsecond); d <= maxTimeoutValue {
 419                 return strconv.FormatInt(d, 10) + "u"
 420         }
 421         if d := div(t, time.Millisecond); d <= maxTimeoutValue {
 422                 return strconv.FormatInt(d, 10) + "m"
 423         }
 424         if d := div(t, time.Second); d <= maxTimeoutValue {
 425                 return strconv.FormatInt(d, 10) + "S"
 426         }
 427         if d := div(t, time.Minute); d <= maxTimeoutValue {
 428                 return strconv.FormatInt(d, 10) + "M"
 429         }
 430         // Note that maxTimeoutValue * time.Hour > MaxInt64.
 431         return strconv.FormatInt(div(t, time.Hour), 10) + "H"
 432 }
 433
 434 func decodeTimeout(s string) (time.Duration, error) {
 435         size := len(s)
 436         if size < 2 {
 437                 return 0, fmt.Errorf("transport: timeout string is too short: %q", s)
 438         }
 439         if size > 9 {
 440                 // Spec allows for 8 digits plus the unit.
 441                 return 0, fmt.Errorf("transport: timeout string is too long: %q", s)
 442         }
 443         unit := timeoutUnit(s[size-1])
 444         d, ok := timeoutUnitToDuration(unit)
 445         if !ok {
 446                 return 0, fmt.Errorf("transport: timeout unit is not recognized: %q", s)
 447         }
 448         t, err := strconv.ParseInt(s[:size-1], 10, 64)
 449         if err != nil {
 450                 return 0, err
 451         }
 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
 456         }
 457         return d * time.Duration(t), nil
 458 }
 459
 460 const (
 461         spaceByte   = ' '
 462         tildeByte   = '~'
 463         percentByte = '%'
 464 )
 465
 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.
 469 //
 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 {
 474         if msg == "" {
 475                 return ""
 476         }
 477         lenMsg := len(msg)
 478         for i := 0; i < lenMsg; i++ {
 479                 c := msg[i]
 480                 if !(c >= spaceByte && c <= tildeByte && c != percentByte) {
 481                         return encodeGrpcMessageUnchecked(msg)
 482                 }
 483         }
 484         return msg
 485 }
 486
 487 func encodeGrpcMessageUnchecked(msg string) string {
 488         var buf bytes.Buffer
 489         for len(msg) > 0 {
 490                 r, size := utf8.DecodeRuneInString(msg)
 491                 for _, b := range []byte(string(r)) {
 492                         if size > 1 {
 493                                 // If size > 1, r is not ascii. Always do percent encoding.
 494                                 buf.WriteString(fmt.Sprintf("%%%02X", b))
 495                                 continue
 496                         }
 497
 498                         // The for loop is necessary even if size == 1. r could be
 499                         // utf8.RuneError.
 500                         //
 501                         // fmt.Sprintf("%%%02X", utf8.RuneError) gives "%FFFD".
 502                         if b >= spaceByte && b <= tildeByte && b != percentByte {
 503                                 buf.WriteByte(b)
 504                         } else {
 505                                 buf.WriteString(fmt.Sprintf("%%%02X", b))
 506                         }
 507                 }
 508                 msg = msg[size:]
 509         }
 510         return buf.String()
 511 }
 512
 513 // decodeGrpcMessage decodes the msg encoded by encodeGrpcMessage.
 514 func decodeGrpcMessage(msg string) string {
 515         if msg == "" {
 516                 return ""
 517         }
 518         lenMsg := len(msg)
 519         for i := 0; i < lenMsg; i++ {
 520                 if msg[i] == percentByte && i+2 < lenMsg {
 521                         return decodeGrpcMessageUnchecked(msg)
 522                 }
 523         }
 524         return msg
 525 }
 526
 527 func decodeGrpcMessageUnchecked(msg string) string {
 528         var buf bytes.Buffer
 529         lenMsg := len(msg)
 530         for i := 0; i < lenMsg; i++ {
 531                 c := msg[i]
 532                 if c == percentByte && i+2 < lenMsg {
 533                         parsed, err := strconv.ParseUint(msg[i+1:i+3], 16, 8)
 534                         if err != nil {
 535                                 buf.WriteByte(c)
 536                         } else {
 537                                 buf.WriteByte(byte(parsed))
 538                                 i += 2
 539                         }
 540                 } else {
 541                         buf.WriteByte(c)
 542                 }
 543         }
 544         return buf.String()
 545 }
 546
 547 type bufWriter struct {
 548         buf       []byte
 549         offset    int
 550         batchSize int
 551         conn      net.Conn
 552         err       error
 553
 554         onFlush func()
 555 }
 556
 557 func newBufWriter(conn net.Conn, batchSize int) *bufWriter {
 558         return &bufWriter{
 559                 buf:       make([]byte, batchSize*2),
 560                 batchSize: batchSize,
 561                 conn:      conn,
 562         }
 563 }
 564
 565 func (w *bufWriter) Write(b []byte) (n int, err error) {
 566         if w.err != nil {
 567                 return 0, w.err
 568         }
 569         if w.batchSize == 0 { // Buffer has been disabled.
 570                 return w.conn.Write(b)
 571         }
 572         for len(b) > 0 {
 573                 nn := copy(w.buf[w.offset:], b)
 574                 b = b[nn:]
 575                 w.offset += nn
 576                 n += nn
 577                 if w.offset >= w.batchSize {
 578                         err = w.Flush()
 579                 }
 580         }
 581         return n, err
 582 }
 583
 584 func (w *bufWriter) Flush() error {
 585         if w.err != nil {
 586                 return w.err
 587         }
 588         if w.offset == 0 {
 589                 return nil
 590         }
 591         if w.onFlush != nil {
 592                 w.onFlush()
 593         }
 594         _, w.err = w.conn.Write(w.buf[:w.offset])
 595         w.offset = 0
 596         return w.err
 597 }
 598
 599 type framer struct {
 600         writer *bufWriter
 601         fr     *http2.Framer
 602 }
 603
 604 func newFramer(conn net.Conn, writeBufferSize, readBufferSize int, maxHeaderListSize uint32) *framer {
 605         if writeBufferSize < 0 {
 606                 writeBufferSize = 0
 607         }
 608         var r io.Reader = conn
 609         if readBufferSize > 0 {
 610                 r = bufio.NewReaderSize(r, readBufferSize)
 611         }
 612         w := newBufWriter(conn, writeBufferSize)
 613         f := &framer{
 614                 writer: w,
 615                 fr:     http2.NewFramer(w, r),
 616         }
 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)
 622         return f
 623 }