Repository: grpc/grpc-go
File: internal/transport/http2_server.go (Go)

Code before:
			if err != nil {
				headerError = status.Newf(codes.Internal, "malformed binary metadata %q in header %q: %v", hf.Value, hf.Name, err)
				t.logger.Warningf("Failed to decode metadata header (%q, %q): %v", hf.Name, hf.Value, err)
				break
			}
			mdata[hf.Name] = append(mdata[hf.Name], v)
		}
	}

	// "If multiple Host headers or multiple :authority headers are present, the
	// request must be rejected with an HTTP status code 400 as required by Host
	// validation in RFC 7230 §5.4, gRPC status code INTERNAL, or RST_STREAM
	// with HTTP/2 error code PROTOCOL_ERROR." - A41. Since this is a HTTP/2
	// error, this takes precedence over a client not speaking gRPC.
	if len(mdata[":authority"]) > 1 || len(mdata["host"]) > 1 {
		errMsg := fmt.Sprintf("num values of :authority: %v, num values of host: %v, both must only have 1 value as per HTTP/2 spec", len(mdata[":authority"]), len(mdata["host"]))
		if t.logger.V(logLevel) {
			t.logger.Infof("Aborting the stream early: %v", errMsg)
		}
		t.controlBuf.put(&earlyAbortStream{
			httpStatus:            http.StatusBadRequest,
			streamID:              streamID,
			contentSubtype:        s.contentSubtype,
			status:                status.New(codes.Internal, errMsg),
			rst:                   !frame.StreamEnded(),
			maxSendHeaderListSize: t.maxSendHeaderListSize,
		})
		return nil
	}

	if protocolError {
		t.controlBuf.put(&cleanupStream{
			streamID: streamID,
			rst:      true,
			rstCode:  http2.ErrCodeProtocol,
			onWrite:  func() {},
		})
		return nil
	}
	if !isGRPC {
		t.controlBuf.put(&earlyAbortStream{
			httpStatus:            http.StatusUnsupportedMediaType,
			streamID:              streamID,
			contentSubtype:        s.contentSubtype,
			status:                status.Newf(codes.InvalidArgument, "invalid gRPC request content-type %q", contentType),
			rst:                   !frame.StreamEnded(),
			maxSendHeaderListSize: t.maxSendHeaderListSize,
		})
		return nil
	}
	if headerError != nil {

Review comment (bug):
I believe the `controlBuf.mu` mutex needs to be held while accessing  `t.maxSendHeaderListSize` since the value can be updated while handling an HTTP/2 `Settings` frame from the peer. For the `Settings` to take effect, the server need to send a `Settings ACK` frame to the peer. The FIFO ordering of the `controlbuf` ensures that last ACKed settings are always respected.

I suggest using `t.controlBuf.executeAndPut` with a callback that validates the header size and enqueues `earlyAbortStream` if validation passes. If validation fails, call `t.closeStream` to close the stream with a RST frame. Let me know if this adds too much complexity.

Code after:
			if err != nil {
				headerError = status.Newf(codes.Internal, "malformed binary metadata %q in header %q: %v", hf.Value, hf.Name, err)
				t.logger.Warningf("Failed to decode metadata header (%q, %q): %v", hf.Name, hf.Value, err)
				break
			}
			mdata[hf.Name] = append(mdata[hf.Name], v)
		}
	}

	// "If multiple Host headers or multiple :authority headers are present, the
	// request must be rejected with an HTTP status code 400 as required by Host
	// validation in RFC 7230 §5.4, gRPC status code INTERNAL, or RST_STREAM
	// with HTTP/2 error code PROTOCOL_ERROR." - A41. Since this is a HTTP/2
	// error, this takes precedence over a client not speaking gRPC.
	if len(mdata[":authority"]) > 1 || len(mdata["host"]) > 1 {
		errMsg := fmt.Sprintf("num values of :authority: %v, num values of host: %v, both must only have 1 value as per HTTP/2 spec", len(mdata[":authority"]), len(mdata["host"]))
		if t.logger.V(logLevel) {
			t.logger.Infof("Aborting the stream early: %v", errMsg)
		}
		t.writeEarlyAbort(streamID, s.contentSubtype, status.New(codes.Internal, errMsg), http.StatusBadRequest, !frame.StreamEnded())
		return nil
	}

	if protocolError {
		t.controlBuf.put(&cleanupStream{
			streamID: streamID,
			rst:      true,
			rstCode:  http2.ErrCodeProtocol,
			onWrite:  func() {},
		})
		return nil
	}
	if !isGRPC {
		t.writeEarlyAbort(streamID, s.contentSubtype, status.Newf(codes.InvalidArgument, "invalid gRPC request content-type %q", contentType), http.StatusUnsupportedMediaType, !frame.StreamEnded())
		return nil
	}
	if headerError != nil {
		t.writeEarlyAbort(streamID, s.contentSubtype, headerError, http.StatusBadRequest, !frame.StreamEnded())
		return nil
	}

	// "If :authority is missing, Host must be renamed to :authority." - A41
	if len(mdata[":authority"]) == 0 {
		// No-op if host isn't present, no eventual :authority header is a valid
		// RPC.
		if host, ok := mdata["host"]; ok {
			mdata[":authority"] = host
			delete(mdata, "host")
		}
	} else {
		// "If :authority is present, Host must be discarded" - A41