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

Code before:

// NewStream creates a stream and registers it into the transport as "active"
// streams.  All non-nil errors returned will be *NewStreamError.
func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (*ClientStream, error) {
	ctx = peer.NewContext(ctx, t.getPeer())

	// ServerName field of the resolver returned address takes precedence over
	// Host field of CallHdr to determine the :authority header. This is because,
	// the ServerName field takes precedence for server authentication during
	// TLS handshake, and the :authority header should match the value used
	// for server authentication.
	if t.address.ServerName != "" {
		newCallHdr := *callHdr
		newCallHdr.Host = t.address.ServerName
		callHdr = &newCallHdr
	}

	// Authority specified in the `CallAuthority` CallOption takes the highest
	// precedence to determine the :authority header. Any value in Host field of
	// CallHdr is overwritten. But before overriding, we validate the authority
	// string against the peer certificates and fail the RPC with `UNAVAILABLE`
	// status code if either of the condition fails.
	if callHdr.Authority != "" {
		auth, ok := t.authInfo.(credentials.AuthorityValidator)
		if !ok {
			return nil, &NewStreamError{Err: status.Error(codes.Unavailable, fmt.Sprintf("credentials type %s does not implement the AuthorityValidator interface", t.authInfo.AuthType())), AllowTransparentRetry: false}
		}
		if err := auth.ValidateAuthority(callHdr.Authority); err != nil {
			return nil, &NewStreamError{Err: status.Error(codes.Unavailable, err.Error()), AllowTransparentRetry: false}
		}
		newCallHdr := *callHdr
		newCallHdr.Host = callHdr.Authority
		callHdr = &newCallHdr
	}

	headerFields, err := t.createHeaderFields(ctx, callHdr)
	if err != nil {
		return nil, &NewStreamError{Err: err, AllowTransparentRetry: false}
	}
	s := t.newStream(ctx, callHdr)
	cleanup := func(err error) {
		if s.swapState(streamDone) == streamDone {
			// If it was already done, return.
			return
		}
		// The stream was unprocessed by the server.
		s.unprocessed.Store(true)
		s.write(recvMsg{err: err})
		close(s.done)
		// If headerChan isn't closed, then close it.
		if atomic.CompareAndSwapUint32(&s.headerChanClosed, 0, 1) {

Review comment (security):
Extend the error message like

credentials type %s does not implement the AuthorityValidator interface, but authority override specified with CallAuthority call option

Code after:

// NewStream creates a stream and registers it into the transport as "active"
// streams.  All non-nil errors returned will be *NewStreamError.
func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (*ClientStream, error) {
	ctx = peer.NewContext(ctx, t.getPeer())

	// ServerName field of the resolver returned address takes precedence over
	// Host field of CallHdr to determine the :authority header. This is because,
	// the ServerName field takes precedence for server authentication during
	// TLS handshake, and the :authority header should match the value used
	// for server authentication.
	if t.address.ServerName != "" {
		newCallHdr := *callHdr
		newCallHdr.Host = t.address.ServerName
		callHdr = &newCallHdr
	}

	// The authority specified via the `CallAuthority` CallOption takes the
	// highest precedence when determining the `:authority` header. It overrides
	// any value present in the Host field of CallHdr. Before applying this
	// override, the authority string is validated. If the credentials do not
	// implement the AuthorityValidator interface, or if validation fails, the
	// RPC is failed with a status code of `UNAVAILABLE`.
	if callHdr.Authority != "" {
		auth, ok := t.authInfo.(credentials.AuthorityValidator)
		if !ok {
			return nil, &NewStreamError{Err: status.Errorf(codes.Unavailable, "credentials type %q does not implement the AuthorityValidator interface, but authority override specified with CallAuthority call option", t.authInfo.AuthType())}
		}
		if err := auth.ValidateAuthority(callHdr.Authority); err != nil {
			return nil, &NewStreamError{Err: status.Errorf(codes.Unavailable, "failed to validate authority %q : %v", callHdr.Authority, err)}
		}
		newCallHdr := *callHdr
		newCallHdr.Host = callHdr.Authority
		callHdr = &newCallHdr
	}

	headerFields, err := t.createHeaderFields(ctx, callHdr)
	if err != nil {
		return nil, &NewStreamError{Err: err, AllowTransparentRetry: false}
	}
	s := t.newStream(ctx, callHdr)
	cleanup := func(err error) {
		if s.swapState(streamDone) == streamDone {
			// If it was already done, return.
			return
		}
		// The stream was unprocessed by the server.
		s.unprocessed.Store(true)
		s.write(recvMsg{err: err})
		close(s.done)
		// If headerChan isn't closed, then close it.