buildx/vendor/github.com/moby/buildkit/client/client.go

package client

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"net"
	"net/url"
	"os"
	"strings"

	contentapi "github.com/containerd/containerd/api/services/content/v1"
	"github.com/containerd/containerd/defaults"
	controlapi "github.com/moby/buildkit/api/services/control"
	"github.com/moby/buildkit/client/connhelper"
	"github.com/moby/buildkit/session"
	"github.com/moby/buildkit/session/grpchijack"
	"github.com/moby/buildkit/util/appdefaults"
	"github.com/moby/buildkit/util/grpcerrors"
	"github.com/moby/buildkit/util/tracing/otlptracegrpc"
	"github.com/pkg/errors"
	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
	"go.opentelemetry.io/otel/exporters/otlp/otlptrace"
	"go.opentelemetry.io/otel/propagation"
	sdktrace "go.opentelemetry.io/otel/sdk/trace"
	"go.opentelemetry.io/otel/trace"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/credentials/insecure"
)

type Client struct {
	conn          *grpc.ClientConn
	sessionDialer func(ctx context.Context, proto string, meta map[string][]string) (net.Conn, error)
}

type ClientOpt interface {
	isClientOpt()
}

// New returns a new buildkit client. Address can be empty for the system-default address.
func New(ctx context.Context, address string, opts ...ClientOpt) (*Client, error) {
	gopts := []grpc.DialOption{
		grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(defaults.DefaultMaxRecvMsgSize)),
		grpc.WithDefaultCallOptions(grpc.MaxCallSendMsgSize(defaults.DefaultMaxSendMsgSize)),
	}
	needDialer := true

	var unary []grpc.UnaryClientInterceptor
	var stream []grpc.StreamClientInterceptor

	var customTracer bool // allows manually setting disabling tracing even if tracer in context
	var tracerProvider trace.TracerProvider
	var tracerDelegate TracerDelegate
	var sessionDialer func(context.Context, string, map[string][]string) (net.Conn, error)
	var customDialOptions []grpc.DialOption
	var creds *withCredentials

	for _, o := range opts {
		if _, ok := o.(*withFailFast); ok {
			gopts = append(gopts, grpc.FailOnNonTempDialError(true))
		}
		if credInfo, ok := o.(*withCredentials); ok {
			if creds == nil {
				creds = &withCredentials{}
			}
			creds = creds.merge(credInfo)
		}
		if wt, ok := o.(*withTracer); ok {
			customTracer = true
			tracerProvider = wt.tp
		}
		if wd, ok := o.(*withDialer); ok {
			gopts = append(gopts, grpc.WithContextDialer(wd.dialer))
			needDialer = false
		}
		if wt, ok := o.(*withTracerDelegate); ok {
			tracerDelegate = wt
		}
		if sd, ok := o.(*withSessionDialer); ok {
			sessionDialer = sd.dialer
		}
		if opt, ok := o.(*withGRPCDialOption); ok {
			customDialOptions = append(customDialOptions, opt.opt)
		}
	}

	if creds == nil {
		gopts = append(gopts, grpc.WithTransportCredentials(insecure.NewCredentials()))
	} else {
		credOpts, err := loadCredentials(creds)
		if err != nil {
			return nil, err
		}
		gopts = append(gopts, credOpts)
	}

	if !customTracer {
		if span := trace.SpanFromContext(ctx); span.SpanContext().IsValid() {
			tracerProvider = span.TracerProvider()
		}
	}

	if tracerProvider != nil {
		var propagators = propagation.NewCompositeTextMapPropagator(propagation.TraceContext{}, propagation.Baggage{})
		unary = append(unary, filterInterceptor(otelgrpc.UnaryClientInterceptor(otelgrpc.WithTracerProvider(tracerProvider), otelgrpc.WithPropagators(propagators))))
		stream = append(stream, otelgrpc.StreamClientInterceptor(otelgrpc.WithTracerProvider(tracerProvider), otelgrpc.WithPropagators(propagators)))
	}

	if needDialer {
		dialFn, err := resolveDialer(address)
		if err != nil {
			return nil, err
		}
		gopts = append(gopts, grpc.WithContextDialer(dialFn))
	}
	if address == "" {
		address = appdefaults.Address
	}

	// Setting :authority pseudo header
	// - HTTP/2 (RFC7540) defines :authority pseudo header includes
	//   the authority portion of target URI but it must not include
	//   userinfo part (i.e. url.Host).
	//   ref: https://datatracker.ietf.org/doc/html/rfc7540#section-8.1.2.3
	// - However, when TLS specified, grpc-go requires it must match
	//   with its servername specified for certificate validation.
	var authority string
	if creds != nil && creds.serverName != "" {
		authority = creds.serverName
	}
	if authority == "" {
		// authority as hostname from target address
		uri, err := url.Parse(address)
		if err != nil {
			return nil, err
		}
		authority = uri.Host
	}
	gopts = append(gopts, grpc.WithAuthority(authority))

	unary = append(unary, grpcerrors.UnaryClientInterceptor)
	stream = append(stream, grpcerrors.StreamClientInterceptor)

	gopts = append(gopts, grpc.WithChainUnaryInterceptor(unary...))
	gopts = append(gopts, grpc.WithChainStreamInterceptor(stream...))
	gopts = append(gopts, customDialOptions...)

	conn, err := grpc.DialContext(ctx, address, gopts...)
	if err != nil {
		return nil, errors.Wrapf(err, "failed to dial %q . make sure buildkitd is running", address)
	}

	c := &Client{
		conn:          conn,
		sessionDialer: sessionDialer,
	}

	if tracerDelegate != nil {
		_ = c.setupDelegatedTracing(ctx, tracerDelegate) // ignore error
	}

	return c, nil
}

func (c *Client) setupDelegatedTracing(ctx context.Context, td TracerDelegate) error {
	pd := otlptracegrpc.NewClient(c.conn)
	e, err := otlptrace.New(ctx, pd)
	if err != nil {
		return nil
	}
	return td.SetSpanExporter(ctx, e)
}

func (c *Client) ControlClient() controlapi.ControlClient {
	return controlapi.NewControlClient(c.conn)
}

func (c *Client) ContentClient() contentapi.ContentClient {
	return contentapi.NewContentClient(c.conn)
}

func (c *Client) Dialer() session.Dialer {
	return grpchijack.Dialer(c.ControlClient())
}

func (c *Client) Close() error {
	return c.conn.Close()
}

type withFailFast struct{}

func (*withFailFast) isClientOpt() {}

func WithFailFast() ClientOpt {
	return &withFailFast{}
}

type withDialer struct {
	dialer func(context.Context, string) (net.Conn, error)
}

func (*withDialer) isClientOpt() {}

func WithContextDialer(df func(context.Context, string) (net.Conn, error)) ClientOpt {
	return &withDialer{dialer: df}
}

type withCredentials struct {
	// server options
	serverName   string
	caCert       string
	caCertSystem bool

	// client options
	cert string
	key  string
}

func (opts *withCredentials) merge(opts2 *withCredentials) *withCredentials {
	result := *opts
	if opts2 == nil {
		return &result
	}

	// server options
	if opts2.serverName != "" {
		result.serverName = opts2.serverName
	}
	if opts2.caCert != "" {
		result.caCert = opts2.caCert
	}
	if opts2.caCertSystem {
		result.caCertSystem = opts2.caCertSystem
	}

	// client options
	if opts2.cert != "" {
		result.cert = opts2.cert
	}
	if opts2.key != "" {
		result.key = opts2.key
	}

	return &result
}

func (*withCredentials) isClientOpt() {}

// WithCredentials configures the TLS parameters of the client.
// Arguments:
// * cert:	specifies the filepath of the client certificate
// * key:	specifies the filepath of the client key
func WithCredentials(cert, key string) ClientOpt {
	return &withCredentials{
		cert: cert,
		key:  key,
	}
}

// WithServerConfig configures the TLS parameters to connect to the server.
// Arguments:
// * serverName:	specifies the server name to verify the hostname
// * caCert:		specifies the filepath of the CA certificate
func WithServerConfig(serverName, caCert string) ClientOpt {
	return &withCredentials{
		serverName: serverName,
		caCert:     caCert,
	}
}

// WithServerConfigSystem configures the TLS parameters to connect to the
// server, using the system's certificate pool.
func WithServerConfigSystem(serverName string) ClientOpt {
	return &withCredentials{
		serverName:   serverName,
		caCertSystem: true,
	}
}

func loadCredentials(opts *withCredentials) (grpc.DialOption, error) {
	cfg := &tls.Config{}

	if opts.caCertSystem {
		cfg.RootCAs, _ = x509.SystemCertPool()
	}
	if cfg.RootCAs == nil {
		cfg.RootCAs = x509.NewCertPool()
	}

	if opts.caCert != "" {
		ca, err := os.ReadFile(opts.caCert)
		if err != nil {
			return nil, errors.Wrap(err, "could not read ca certificate")
		}
		if ok := cfg.RootCAs.AppendCertsFromPEM(ca); !ok {
			return nil, errors.New("failed to append ca certs")
		}
	}

	if opts.serverName != "" {
		cfg.ServerName = opts.serverName
	}

	// we will produce an error if the user forgot about either cert or key if at least one is specified
	if opts.cert != "" || opts.key != "" {
		cert, err := tls.LoadX509KeyPair(opts.cert, opts.key)
		if err != nil {
			return nil, errors.Wrap(err, "could not read certificate/key")
		}
		cfg.Certificates = append(cfg.Certificates, cert)
	}

	return grpc.WithTransportCredentials(credentials.NewTLS(cfg)), nil
}

func WithTracerProvider(t trace.TracerProvider) ClientOpt {
	return &withTracer{t}
}

type withTracer struct {
	tp trace.TracerProvider
}

func (w *withTracer) isClientOpt() {}

type TracerDelegate interface {
	SetSpanExporter(context.Context, sdktrace.SpanExporter) error
}

func WithTracerDelegate(td TracerDelegate) ClientOpt {
	return &withTracerDelegate{
		TracerDelegate: td,
	}
}

type withTracerDelegate struct {
	TracerDelegate
}

func (w *withTracerDelegate) isClientOpt() {}

func WithSessionDialer(dialer func(context.Context, string, map[string][]string) (net.Conn, error)) ClientOpt {
	return &withSessionDialer{dialer}
}

type withSessionDialer struct {
	dialer func(context.Context, string, map[string][]string) (net.Conn, error)
}

func (w *withSessionDialer) isClientOpt() {}

func resolveDialer(address string) (func(context.Context, string) (net.Conn, error), error) {
	ch, err := connhelper.GetConnectionHelper(address)
	if err != nil {
		return nil, err
	}
	if ch != nil {
		return ch.ContextDialer, nil
	}
	// basic dialer
	return dialer, nil
}

func filterInterceptor(intercept grpc.UnaryClientInterceptor) grpc.UnaryClientInterceptor {
	return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
		if strings.HasSuffix(method, "opentelemetry.proto.collector.trace.v1.TraceService/Export") {
			return invoker(ctx, method, req, reply, cc, opts...)
		}
		return intercept(ctx, method, req, reply, cc, invoker, opts...)
	}
}

type withGRPCDialOption struct {
	opt grpc.DialOption
}

func (*withGRPCDialOption) isClientOpt() {}

func WithGRPCDialOption(opt grpc.DialOption) ClientOpt {
	return &withGRPCDialOption{opt}
}