rpc: use tenant client/server certs where appropriate

With this PR, tenant SQL servers use tenant client certificates to
connect to the tenant server endpoint at the KV layer. They were
previously using the KV-internal node certs.

No validation is performed as of this PR, but this is the obvious
next step.

Follow-up work will assertions that make sure that we don't see
tenants "accidentally" use the node certs for some operations
when they are available (as is typically the case during testing).

Finally, there will be some work on the heartbeats exchanged by
the RPC context. We don't want a SQL tenant's time signal to
ever trigger KV nodes to crash, for example.

Touches cockroachdb#47898.

Release note: None

pkg/rpc/context.go

@@ -142,19 +142,46 @@ func requireSuperUser(ctx context.Context) error {
 	return nil
 }
 
-// NewServer is a thin wrapper around grpc.NewServer that registers a heartbeat
-// service.
-func NewServer(ctx *Context) *grpc.Server {
-	return NewServerWithInterceptor(ctx, nil)
-}
-
-// NewServerWithInterceptor is like NewServer, but accepts an additional
-// interceptor which is called before streaming and unary RPCs and may inject an
-// error.
-func NewServerWithInterceptor(
-	ctx *Context, interceptor func(fullMethod string) error,
-) *grpc.Server {
-	opts := []grpc.ServerOption{
+type serverOpts struct {
+	interceptor func(fullMethod string) error
+	tenant      bool
+}
+
+// ServerOption is a configuration option passed to NewServer.
+type ServerOption func(*serverOpts)
+
+// ForTenant is an option to NewServer that results in the server being set
+// up to validate incoming tenants. Without this option, the created server will
+// use the KV-internal node certificates. With it, it uses tenant server
+// certificates.
+func ForTenant(opts *serverOpts) {
+	opts.tenant = true
+}
+
+var _ ServerOption = ForTenant
+
+// WithInterceptor adds an additional interceptor. The interceptor is called before
+// streaming and unary RPCs and may inject an error.
+//
+// This option can only be used once (i.e. interceptors can not be chained).
+func WithInterceptor(f func(fullMethod string) error) ServerOption {
+	return func(opts *serverOpts) {
+		if opts.interceptor != nil {
+			panic("interceptor can only be set once")
+		}
+		opts.interceptor = f
+	}
+}
+
+// NewServer sets up an RPC server. Depending on the ServerOptions, the Server
+// either expects incoming connections from KV nodes, or from tenant SQL
+// servers.
+func NewServer(ctx *Context, opts ...ServerOption) *grpc.Server {
+	var o serverOpts
+	for _, f := range opts {
+		f(&o)
+	}
+	grpcOpts := []grpc.ServerOption{
 		// The limiting factor for lowering the max message size is the fact
 		// that a single large kv can be sent over the network in one message.
 		// Our maximum kv size is unlimited, so we need this to be very large.
@@ -178,11 +205,17 @@ func NewServerWithInterceptor(
 		grpc.StatsHandler(&ctx.stats),
 	}
 	if !ctx.Config.Insecure {
-		tlsConfig, err := ctx.GetServerTLSConfig()
+		var tlsConfig *tls.Config
+		var err error
+		if !o.tenant {
+			tlsConfig, err = ctx.GetServerTLSConfig()
+		} else {
+			tlsConfig, err = ctx.GetTenantServerTLSConfig()
+		}
 		if err != nil {
 			panic(err)
 		}
-		opts = append(opts, grpc.Creds(credentials.NewTLS(tlsConfig)))
+		grpcOpts = append(grpcOpts, grpc.Creds(credentials.NewTLS(tlsConfig)))
 	}
 
 	// These interceptors will be called in the order in which they appear, i.e.
@@ -191,29 +224,42 @@ func NewServerWithInterceptor(
 	var streamInterceptor []grpc.StreamServerInterceptor
 
 	if !ctx.Config.Insecure {
+		var authHook func(ctx context.Context) error
+		if !o.tenant {
+			authHook = requireSuperUser
+		} else {
+			authHook = func(ctx context.Context) error {
+				// TODO(tbg): pull the tenant ID from the incoming
+				// certificate and validate that the request is
+				// admissible.
+				return nil
+			}
+		}
+
+		// TODO(tbg): do something else if `o.tenant`.
 		unaryInterceptor = append(unaryInterceptor, func(
 			ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler,
 		) (interface{}, error) {
-			if err := requireSuperUser(ctx); err != nil {
+			if err := authHook(ctx); err != nil {
 				return nil, err
 			}
 			return handler(ctx, req)
 		})
 		streamInterceptor = append(streamInterceptor, func(
 			srv interface{}, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler,
 		) error {
-			if err := requireSuperUser(stream.Context()); err != nil {
+			if err := authHook(stream.Context()); err != nil {
 				return err
 			}
 			return handler(srv, stream)
 		})
 	}
 
-	if interceptor != nil {
+	if o.interceptor != nil {
 		unaryInterceptor = append(unaryInterceptor, func(
 			ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler,
 		) (interface{}, error) {
-			if err := interceptor(info.FullMethod); err != nil {
+			if err := o.interceptor(info.FullMethod); err != nil {
 				return nil, err
 			}
 			return handler(ctx, req)
@@ -222,7 +268,7 @@ func NewServerWithInterceptor(
 		streamInterceptor = append(streamInterceptor, func(
 			srv interface{}, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler,
 		) error {
-			if err := interceptor(info.FullMethod); err != nil {
+			if err := o.interceptor(info.FullMethod); err != nil {
 				return err
 			}
 			return handler(srv, stream)
@@ -251,10 +297,10 @@ func NewServerWithInterceptor(
 		// https://github.com/grpc-ecosystem/go-grpc-middleware/tree/master/tracing/opentracing
 	}
 
-	opts = append(opts, grpc.ChainUnaryInterceptor(unaryInterceptor...))
-	opts = append(opts, grpc.ChainStreamInterceptor(streamInterceptor...))
+	grpcOpts = append(grpcOpts, grpc.ChainUnaryInterceptor(unaryInterceptor...))
+	grpcOpts = append(grpcOpts, grpc.ChainStreamInterceptor(streamInterceptor...))
 
-	s := grpc.NewServer(opts...)
+	s := grpc.NewServer(grpcOpts...)
 	RegisterHeartbeatServer(s, &HeartbeatService{
 		clock:                                 ctx.Clock,
 		remoteClockMonitor:                    ctx.RemoteClocks,
@@ -721,7 +767,7 @@ func (ctx *Context) grpcDialOptions(
 		var tlsConfig *tls.Config
 		// TODO(tbg): remove this override when the KV layer can authenticate tenant
 		// client certs.
-		const override = true
+		const override = false
 		if override || ctx.tenID == roachpb.SystemTenantID {
 			tlsConfig, err = ctx.GetClientTLSConfig()
 		} else {

pkg/rpc/tls.go

@@ -124,6 +124,27 @@ func (ctx *SecurityContext) GetServerTLSConfig() (*tls.Config, error) {
 	return tlsCfg, nil
 }
 
+// GetTenantServerTLSConfig returns the tenant server TLS config, initializing
+// it if needed. If Insecure is true, return a nil config, otherwise asks the
+// certificate manager for the tenant server TLS config.
+func (ctx *SecurityContext) GetTenantServerTLSConfig() (*tls.Config, error) {
+	// Early out.
+	if ctx.config.Insecure {
+		return nil, nil
+	}
+
+	cm, err := ctx.GetCertificateManager()
+	if err != nil {
+		return nil, wrapError(err)
+	}
+
+	tlsCfg, err := cm.GetTenantServerTLSConfig()
+	if err != nil {
+		return nil, wrapError(err)
+	}
+	return tlsCfg, nil
+}
+
 // GetClientTLSConfig returns the client TLS config, initializing it if needed.
 // If Insecure is true, return a nil config, otherwise ask the certificate
 // manager for a TLS config using certs for the config.User.

pkg/server/grpc_server.go

@@ -29,12 +29,13 @@ type grpcServer struct {
 	mode serveMode
 }
 
-func newGRPCServer(rpcCtx *rpc.Context) *grpcServer {
+func newGRPCServer(rpcCtx *rpc.Context, opts ...rpc.ServerOption) *grpcServer {
 	s := &grpcServer{}
 	s.mode.set(modeInitializing)
-	s.Server = rpc.NewServerWithInterceptor(rpcCtx, func(path string) error {
+	opts = append(opts, rpc.WithInterceptor(func(path string) error {
 		return s.intercept(path)
-	})
+	}))
+	s.Server = rpc.NewServer(rpcCtx, opts...)
 	return s
 }
 

pkg/server/server.go

@@ -332,7 +332,7 @@ func NewServer(cfg Config, stopper *stop.Stopper) (*Server, error) {
 	// TODO(tbg): pass a different rpcContext here.
 	var tenantGRPC *grpcServer
 	if cfg.SplitListenTenant {
-		tenantGRPC = newGRPCServer(rpcContext)
+		tenantGRPC = newGRPCServer(rpcContext, rpc.ForTenant)
 	}
 
 	g := gossip.New(

pkg/server/testserver.go

@@ -631,7 +631,7 @@ func (ts *TestServer) StartTenant(params base.TestTenantArgs) (pgAddr string, _
 			ClusterSettingsUpdater: st.MakeUpdater(),
 		}
 	}
-	sqlCfg.TenantKVAddrs = []string{ts.RPCAddr()}
+	sqlCfg.TenantKVAddrs = []string{ts.TenantAddr()}
 	return StartTenant(
 		ctx,
 		ts.Stopper(),