diff --git a/.changelog/12787.txt b/.changelog/12787.txt
new file mode 100644
index 000000000..0e6d7fc6c
--- /dev/null
+++ b/.changelog/12787.txt
@@ -0,0 +1,3 @@
+```release-note:feature
+ca: Leaf certificates can now be obtained via the gRPC API: `Sign`
+```
diff --git a/agent/connect/csr.go b/agent/connect/csr.go
index cc01f991e..f699a5879 100644
--- a/agent/connect/csr.go
+++ b/agent/connect/csr.go
@@ -9,6 +9,7 @@ import (
 	"crypto/x509/pkix"
 	"encoding/asn1"
 	"encoding/pem"
+	"fmt"
 	"net"
 	"net/url"
 )
@@ -100,3 +101,24 @@ func CreateCAExtension() (pkix.Extension, error) {
 		Value:    bitstr,
 	}, nil
 }
+
+// InvalidCSRError returns an error with the given fmt.Sprintf-formatted message
+// indicating certificate signing failed because the user supplied an invalid CSR.
+//
+// See: IsInvalidCSRError.
+func InvalidCSRError(format string, args ...interface{}) error {
+	return invalidCSRError{fmt.Sprintf(format, args...)}
+}
+
+// IsInvalidCSRError returns whether the given error indicates that certificate
+// signing failed because the user supplied an invalid CSR.
+func IsInvalidCSRError(err error) bool {
+	_, ok := err.(invalidCSRError)
+	return ok
+}
+
+type invalidCSRError struct {
+	s string
+}
+
+func (e invalidCSRError) Error() string { return e.s }
diff --git a/agent/consul/connect_ca_endpoint.go b/agent/consul/connect_ca_endpoint.go
index c325ff123..29cfc38be 100644
--- a/agent/consul/connect_ca_endpoint.go
+++ b/agent/consul/connect_ca_endpoint.go
@@ -24,7 +24,7 @@ var (
 	// consul.ErrRateLimited.Error()` which is very sad. Short of replacing our
 	// RPC mechanism it's hard to know how to make that much better though.
 	ErrConnectNotEnabled    = errors.New("Connect must be enabled in order to use this endpoint")
-	ErrRateLimited          = errors.New("Rate limit reached, try again later")
+	ErrRateLimited          = errors.New("Rate limit reached, try again later") // Note: we depend on this error message in the gRPC ConnectCA.Sign endpoint (see: isRateLimitError).
 	ErrNotPrimaryDatacenter = errors.New("not the primary datacenter")
 	ErrStateReadOnly        = errors.New("CA Provider State is read-only")
 )
diff --git a/agent/consul/grpc_integration_test.go b/agent/consul/grpc_integration_test.go
new file mode 100644
index 000000000..c243ebfee
--- /dev/null
+++ b/agent/consul/grpc_integration_test.go
@@ -0,0 +1,84 @@
+package consul
+
+import (
+	"context"
+	"net"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+
+	"github.com/hashicorp/consul/agent/connect"
+	"github.com/hashicorp/consul/proto-public/pbconnectca"
+	"github.com/hashicorp/consul/testrpc"
+)
+
+func TestGRPCIntegration_ConnectCA_Sign(t *testing.T) {
+	if testing.Short() {
+		t.Skip("too slow for testing.Short")
+	}
+
+	t.Parallel()
+
+	// The gRPC endpoint itself well-tested with mocks. This test checks we're
+	// correctly wiring everything up in the server by:
+	//
+	//	* Starting a cluster with multiple servers.
+	//	* Making a request to a follower's public gRPC port.
+	//	* Ensuring that the request is correctly forwarded to the leader.
+	//	* Ensuring we get a valid certificate back (so it went through the CAManager).
+	dir1, server1 := testServerWithConfig(t, func(c *Config) {
+		c.Bootstrap = false
+		c.BootstrapExpect = 2
+	})
+	defer os.RemoveAll(dir1)
+	defer server1.Shutdown()
+
+	dir2, server2 := testServerWithConfig(t, func(c *Config) {
+		c.Bootstrap = false
+	})
+	defer os.RemoveAll(dir2)
+	defer server2.Shutdown()
+
+	joinLAN(t, server2, server1)
+
+	testrpc.WaitForLeader(t, server1.RPC, "dc1")
+
+	var follower *Server
+	if server1.IsLeader() {
+		follower = server2
+	} else {
+		follower = server1
+	}
+
+	// publicGRPCServer is bound to a listener by the wrapping agent code, so we
+	// need to do it ourselves here.
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	go func() {
+		require.NoError(t, follower.publicGRPCServer.Serve(lis))
+	}()
+	t.Cleanup(follower.publicGRPCServer.Stop)
+
+	conn, err := grpc.Dial(lis.Addr().String(), grpc.WithInsecure())
+	require.NoError(t, err)
+
+	client := pbconnectca.NewConnectCAServiceClient(conn)
+
+	csr, _ := connect.TestCSR(t, &connect.SpiffeIDService{
+		Host:       connect.TestClusterID + ".consul",
+		Namespace:  "default",
+		Datacenter: "dc1",
+		Service:    "foo",
+	})
+
+	// This would fail if it wasn't forwarded to the leader.
+	rsp, err := client.Sign(context.Background(), &pbconnectca.SignRequest{
+		Csr: csr,
+	})
+	require.NoError(t, err)
+
+	_, err = connect.ParseCert(rsp.CertPem)
+	require.NoError(t, err)
+}
diff --git a/agent/consul/leader_connect_ca.go b/agent/consul/leader_connect_ca.go
index 88d3c5d42..91da428ac 100644
--- a/agent/consul/leader_connect_ca.go
+++ b/agent/consul/leader_connect_ca.go
@@ -1382,7 +1382,7 @@ func (l *connectSignRateLimiter) getCSRRateLimiterWithLimit(limit rate.Limit) *r
 func (c *CAManager) AuthorizeAndSignCertificate(csr *x509.CertificateRequest, authz acl.Authorizer) (*structs.IssuedCert, error) {
 	// Parse the SPIFFE ID from the CSR SAN.
 	if len(csr.URIs) == 0 {
-		return nil, errors.New("CSR SAN does not contain a SPIFFE ID")
+		return nil, connect.InvalidCSRError("CSR SAN does not contain a SPIFFE ID")
 	}
 	spiffeID, err := connect.ParseCertURI(csr.URIs[0])
 	if err != nil {
@@ -1403,7 +1403,7 @@ func (c *CAManager) AuthorizeAndSignCertificate(csr *x509.CertificateRequest, au
 		// requirement later but being restrictive for now is safer.
 		dc := c.serverConf.Datacenter
 		if v.Datacenter != dc {
-			return nil, fmt.Errorf("SPIFFE ID in CSR from a different datacenter: %s, "+
+			return nil, connect.InvalidCSRError("SPIFFE ID in CSR from a different datacenter: %s, "+
 				"we are %s", v.Datacenter, dc)
 		}
 	case *connect.SpiffeIDAgent:
@@ -1412,7 +1412,7 @@ func (c *CAManager) AuthorizeAndSignCertificate(csr *x509.CertificateRequest, au
 			return nil, err
 		}
 	default:
-		return nil, errors.New("SPIFFE ID in CSR must be a service or agent ID")
+		return nil, connect.InvalidCSRError("SPIFFE ID in CSR must be a service or agent ID")
 	}
 
 	return c.SignCertificate(csr, spiffeID)
@@ -1436,13 +1436,13 @@ func (c *CAManager) SignCertificate(csr *x509.CertificateRequest, spiffeID conne
 	serviceID, isService := spiffeID.(*connect.SpiffeIDService)
 	agentID, isAgent := spiffeID.(*connect.SpiffeIDAgent)
 	if !isService && !isAgent {
-		return nil, fmt.Errorf("SPIFFE ID in CSR must be a service or agent ID")
+		return nil, connect.InvalidCSRError("SPIFFE ID in CSR must be a service or agent ID")
 	}
 
 	var entMeta acl.EnterpriseMeta
 	if isService {
 		if !signingID.CanSign(spiffeID) {
-			return nil, fmt.Errorf("SPIFFE ID in CSR from a different trust domain: %s, "+
+			return nil, connect.InvalidCSRError("SPIFFE ID in CSR from a different trust domain: %s, "+
 				"we are %s", serviceID.Host, signingID.Host())
 		}
 		entMeta.Merge(serviceID.GetEnterpriseMeta())
diff --git a/agent/consul/server.go b/agent/consul/server.go
index e278c2011..d9b4aed6b 100644
--- a/agent/consul/server.go
+++ b/agent/consul/server.go
@@ -238,6 +238,11 @@ type Server struct {
 	// is only ever closed.
 	leaveCh chan struct{}
 
+	// publicConnectCAServer serves the Connect CA service exposed on the public
+	// gRPC port. It is also exposed on the private multiplexed "server" port to
+	// enable RPC forwarding.
+	publicConnectCAServer *connectca.Server
+
 	// publicGRPCServer is the gRPC server exposed on the dedicated gRPC port, as
 	// opposed to the multiplexed "server" port which is served by grpcHandler.
 	publicGRPCServer *grpc.Server
@@ -657,6 +662,29 @@ func NewServer(config *Config, flat Deps, publicGRPCServer *grpc.Server) (*Serve
 	s.overviewManager = NewOverviewManager(s.logger, s.fsm, s.config.MetricsReportingInterval)
 	go s.overviewManager.Run(&lib.StopChannelContext{StopCh: s.shutdownCh})
 
+	// Initialize public gRPC server - register services on public gRPC server.
+	s.publicConnectCAServer = connectca.NewServer(connectca.Config{
+		Publisher:   s.publisher,
+		GetStore:    func() connectca.StateStore { return s.FSM().State() },
+		Logger:      logger.Named("grpc-api.connect-ca"),
+		ACLResolver: plainACLResolver{s.ACLResolver},
+		CAManager:   s.caManager,
+		ForwardRPC: func(info structs.RPCInfo, fn func(*grpc.ClientConn) error) (bool, error) {
+			return s.ForwardGRPC(s.grpcConnPool, info, fn)
+		},
+		ConnectEnabled: s.config.ConnectEnabled,
+	})
+	s.publicConnectCAServer.Register(s.publicGRPCServer)
+
+	dataplane.NewServer(dataplane.Config{
+		Logger:      logger.Named("grpc-api.dataplane"),
+		ACLResolver: plainACLResolver{s.ACLResolver},
+	}).Register(s.publicGRPCServer)
+
+	// Initialize private gRPC server.
+	//
+	// Note: some "public" gRPC services are also exposed on the private gRPC server
+	// to enable RPC forwarding.
 	s.grpcHandler = newGRPCHandlerFromConfig(flat, config, s)
 	s.grpcLeaderForwarder = flat.LeaderForwarder
 	go s.trackLeaderChanges()
@@ -669,18 +697,6 @@ func NewServer(config *Config, flat Deps, publicGRPCServer *grpc.Server) (*Serve
 	// since it can fire events when leadership is obtained.
 	go s.monitorLeadership()
 
-	// Initialize public gRPC server - register services on public gRPC server.
-	connectca.NewServer(connectca.Config{
-		Publisher:   s.publisher,
-		GetStore:    func() connectca.StateStore { return s.FSM().State() },
-		Logger:      logger.Named("grpc-api.connect-ca"),
-		ACLResolver: plainACLResolver{s.ACLResolver},
-	}).Register(s.publicGRPCServer)
-	dataplane.NewServer(dataplane.Config{
-		Logger:      logger.Named("grpc-api.dataplane"),
-		ACLResolver: plainACLResolver{s.ACLResolver},
-	}).Register(s.publicGRPCServer)
-
 	// Start listening for RPC requests.
 	go func() {
 		if err := s.grpcHandler.Run(); err != nil {
@@ -712,6 +728,10 @@ func newGRPCHandlerFromConfig(deps Deps, config *Config, s *Server) connHandler
 				deps.Logger.Named("grpc-api.subscription")))
 		}
 		s.registerEnterpriseGRPCServices(deps, srv)
+
+		// Note: this public gRPC service is also exposed on the private server to
+		// enable RPC forwarding.
+		s.publicConnectCAServer.Register(srv)
 	}
 
 	return agentgrpc.NewHandler(deps.Logger, config.RPCAddr, register)
diff --git a/agent/grpc/public/services/connectca/mock_ACLResolver.go b/agent/grpc/public/services/connectca/mock_ACLResolver.go
index 6b6a6a771..ce21ffdeb 100644
--- a/agent/grpc/public/services/connectca/mock_ACLResolver.go
+++ b/agent/grpc/public/services/connectca/mock_ACLResolver.go
@@ -3,9 +3,8 @@
 package connectca
 
 import (
-	mock "github.com/stretchr/testify/mock"
-
 	acl "github.com/hashicorp/consul/acl"
+	mock "github.com/stretchr/testify/mock"
 )
 
 // MockACLResolver is an autogenerated mock type for the ACLResolver type
@@ -13,13 +12,13 @@ type MockACLResolver struct {
 	mock.Mock
 }
 
-// ResolveTokenAndDefaultMeta provides a mock function with given fields: _a0, _a1, _a2
-func (_m *MockACLResolver) ResolveTokenAndDefaultMeta(_a0 string, _a1 *acl.EnterpriseMeta, _a2 *acl.AuthorizerContext) (acl.Authorizer, error) {
-	ret := _m.Called(_a0, _a1, _a2)
+// ResolveTokenAndDefaultMeta provides a mock function with given fields: token, entMeta, authzContext
+func (_m *MockACLResolver) ResolveTokenAndDefaultMeta(token string, entMeta *acl.EnterpriseMeta, authzContext *acl.AuthorizerContext) (acl.Authorizer, error) {
+	ret := _m.Called(token, entMeta, authzContext)
 
 	var r0 acl.Authorizer
 	if rf, ok := ret.Get(0).(func(string, *acl.EnterpriseMeta, *acl.AuthorizerContext) acl.Authorizer); ok {
-		r0 = rf(_a0, _a1, _a2)
+		r0 = rf(token, entMeta, authzContext)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).(acl.Authorizer)
@@ -28,7 +27,7 @@ func (_m *MockACLResolver) ResolveTokenAndDefaultMeta(_a0 string, _a1 *acl.Enter
 
 	var r1 error
 	if rf, ok := ret.Get(1).(func(string, *acl.EnterpriseMeta, *acl.AuthorizerContext) error); ok {
-		r1 = rf(_a0, _a1, _a2)
+		r1 = rf(token, entMeta, authzContext)
 	} else {
 		r1 = ret.Error(1)
 	}
diff --git a/agent/grpc/public/services/connectca/mock_CAManager.go b/agent/grpc/public/services/connectca/mock_CAManager.go
new file mode 100644
index 000000000..1034c4b97
--- /dev/null
+++ b/agent/grpc/public/services/connectca/mock_CAManager.go
@@ -0,0 +1,40 @@
+// Code generated by mockery v1.0.0. DO NOT EDIT.
+
+package connectca
+
+import (
+	acl "github.com/hashicorp/consul/acl"
+	mock "github.com/stretchr/testify/mock"
+
+	structs "github.com/hashicorp/consul/agent/structs"
+
+	x509 "crypto/x509"
+)
+
+// MockCAManager is an autogenerated mock type for the CAManager type
+type MockCAManager struct {
+	mock.Mock
+}
+
+// AuthorizeAndSignCertificate provides a mock function with given fields: csr, authz
+func (_m *MockCAManager) AuthorizeAndSignCertificate(csr *x509.CertificateRequest, authz acl.Authorizer) (*structs.IssuedCert, error) {
+	ret := _m.Called(csr, authz)
+
+	var r0 *structs.IssuedCert
+	if rf, ok := ret.Get(0).(func(*x509.CertificateRequest, acl.Authorizer) *structs.IssuedCert); ok {
+		r0 = rf(csr, authz)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*structs.IssuedCert)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(*x509.CertificateRequest, acl.Authorizer) error); ok {
+		r1 = rf(csr, authz)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
diff --git a/agent/grpc/public/services/connectca/server.go b/agent/grpc/public/services/connectca/server.go
index 86edfdb54..1407e42d6 100644
--- a/agent/grpc/public/services/connectca/server.go
+++ b/agent/grpc/public/services/connectca/server.go
@@ -1,7 +1,11 @@
 package connectca
 
 import (
+	"crypto/x509"
+
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-memdb"
@@ -17,10 +21,13 @@ type Server struct {
 }
 
 type Config struct {
-	Publisher   EventPublisher
-	GetStore    func() StateStore
-	Logger      hclog.Logger
-	ACLResolver ACLResolver
+	Publisher      EventPublisher
+	GetStore       func() StateStore
+	Logger         hclog.Logger
+	ACLResolver    ACLResolver
+	CAManager      CAManager
+	ForwardRPC     func(structs.RPCInfo, func(*grpc.ClientConn) error) (bool, error)
+	ConnectEnabled bool
 }
 
 type EventPublisher interface {
@@ -34,7 +41,12 @@ type StateStore interface {
 
 //go:generate mockery -name ACLResolver -inpkg
 type ACLResolver interface {
-	ResolveTokenAndDefaultMeta(string, *acl.EnterpriseMeta, *acl.AuthorizerContext) (acl.Authorizer, error)
+	ResolveTokenAndDefaultMeta(token string, entMeta *acl.EnterpriseMeta, authzContext *acl.AuthorizerContext) (acl.Authorizer, error)
+}
+
+//go:generate mockery -name CAManager -inpkg
+type CAManager interface {
+	AuthorizeAndSignCertificate(csr *x509.CertificateRequest, authz acl.Authorizer) (*structs.IssuedCert, error)
 }
 
 func NewServer(cfg Config) *Server {
@@ -44,3 +56,10 @@ func NewServer(cfg Config) *Server {
 func (s *Server) Register(grpcServer *grpc.Server) {
 	pbconnectca.RegisterConnectCAServiceServer(grpcServer, s)
 }
+
+func (s *Server) requireConnect() error {
+	if s.ConnectEnabled {
+		return nil
+	}
+	return status.Error(codes.FailedPrecondition, "Connect must be enabled in order to use this endpoint")
+}
diff --git a/agent/grpc/public/services/connectca/server_test.go b/agent/grpc/public/services/connectca/server_test.go
index e74b7c094..def654bf8 100644
--- a/agent/grpc/public/services/connectca/server_test.go
+++ b/agent/grpc/public/services/connectca/server_test.go
@@ -12,9 +12,14 @@ import (
 
 	"github.com/hashicorp/consul/agent/consul/state"
 	"github.com/hashicorp/consul/agent/consul/stream"
+	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/proto-public/pbconnectca"
 )
 
+func noopForwardRPC(structs.RPCInfo, func(*grpc.ClientConn) error) (bool, error) {
+	return false, nil
+}
+
 func testStateStore(t *testing.T, publisher state.EventPublisher) *state.Store {
 	t.Helper()
 
diff --git a/agent/grpc/public/services/connectca/sign.go b/agent/grpc/public/services/connectca/sign.go
new file mode 100644
index 000000000..d6a21d616
--- /dev/null
+++ b/agent/grpc/public/services/connectca/sign.go
@@ -0,0 +1,95 @@
+package connectca
+
+import (
+	"context"
+	"strings"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"github.com/hashicorp/consul/acl"
+	"github.com/hashicorp/consul/agent/connect"
+	"github.com/hashicorp/consul/agent/grpc/public"
+	"github.com/hashicorp/consul/agent/structs"
+	"github.com/hashicorp/consul/proto-public/pbconnectca"
+)
+
+// Sign a leaf certificate for the service or agent identified by the SPIFFE
+// ID in the given CSR's SAN.
+func (s *Server) Sign(ctx context.Context, req *pbconnectca.SignRequest) (*pbconnectca.SignResponse, error) {
+	if err := s.requireConnect(); err != nil {
+		return nil, err
+	}
+
+	logger := s.Logger.Named("sign").With("request_id", traceID())
+	logger.Trace("request received")
+
+	token := public.TokenFromContext(ctx)
+
+	if req.Csr == "" {
+		return nil, status.Error(codes.InvalidArgument, "CSR is required")
+	}
+
+	// For private/internal gRPC handlers, protoc-gen-rpc-glue generates the
+	// requisite methods to satisfy the structs.RPCInfo interface using fields
+	// from the pbcommon package. This service is public, so we can't use those
+	// fields in our proto definition. Instead, we construct our RPCInfo manually.
+	//
+	// Embedding WriteRequest ensures RPCs are forwarded to the leader, embedding
+	// DCSpecificRequest adds the RequestDatacenter method (but as we're not
+	// setting Datacenter it has the effect of *not* doing DC forwarding).
+	var rpcInfo struct {
+		structs.WriteRequest
+		structs.DCSpecificRequest
+	}
+	rpcInfo.Token = token
+
+	var rsp *pbconnectca.SignResponse
+	handled, err := s.ForwardRPC(&rpcInfo, func(conn *grpc.ClientConn) error {
+		logger.Trace("forwarding RPC")
+		var err error
+		rsp, err = pbconnectca.NewConnectCAServiceClient(conn).Sign(ctx, req)
+		return err
+	})
+	if handled || err != nil {
+		return rsp, err
+	}
+
+	csr, err := connect.ParseCSR(req.Csr)
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+
+	authz, err := s.ACLResolver.ResolveTokenAndDefaultMeta(token, nil, nil)
+	if err != nil {
+		return nil, status.Error(codes.Unauthenticated, err.Error())
+	}
+
+	cert, err := s.CAManager.AuthorizeAndSignCertificate(csr, authz)
+	switch {
+	case connect.IsInvalidCSRError(err):
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	case acl.IsErrPermissionDenied(err):
+		return nil, status.Error(codes.PermissionDenied, err.Error())
+	case isRateLimitError(err):
+		return nil, status.Error(codes.ResourceExhausted, err.Error())
+	case err != nil:
+		logger.Error("failed to sign leaf certificate", "error", err.Error())
+		return nil, status.Error(codes.Internal, "failed to sign leaf certificate")
+	}
+
+	return &pbconnectca.SignResponse{
+		CertPem: cert.CertPEM,
+	}, nil
+}
+
+// TODO(agentless): CAManager currently lives in the `agent/consul` package and
+// returns ErrRateLimited which we can't reference directly here because it'd
+// create an import cycle. Checking the error message like this is fragile, but
+// because of net/rpc's limited error handling support it's what we already do
+// on the client. We should either move the error constant so that can use it
+// here, or perhaps make it a typed error?
+func isRateLimitError(err error) bool {
+	return err != nil && strings.Contains(err.Error(), "limit reached")
+}
diff --git a/agent/grpc/public/services/connectca/sign_test.go b/agent/grpc/public/services/connectca/sign_test.go
new file mode 100644
index 000000000..600b1056c
--- /dev/null
+++ b/agent/grpc/public/services/connectca/sign_test.go
@@ -0,0 +1,252 @@
+package connectca
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"github.com/hashicorp/go-hclog"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+
+	acl "github.com/hashicorp/consul/acl"
+	"github.com/hashicorp/consul/agent/connect"
+	"github.com/hashicorp/consul/agent/structs"
+	"github.com/hashicorp/consul/proto-public/pbconnectca"
+)
+
+func TestSign_ConnectDisabled(t *testing.T) {
+	server := NewServer(Config{ConnectEnabled: false})
+
+	_, err := server.Sign(context.Background(), &pbconnectca.SignRequest{})
+	require.Error(t, err)
+	require.Equal(t, codes.FailedPrecondition.String(), status.Code(err).String())
+	require.Contains(t, status.Convert(err).Message(), "Connect")
+}
+
+func TestSign_Validation(t *testing.T) {
+	aclResolver := &MockACLResolver{}
+	aclResolver.On("ResolveTokenAndDefaultMeta", mock.Anything, mock.Anything, mock.Anything).
+		Return(acl.AllowAll(), nil)
+
+	server := NewServer(Config{
+		Logger:         hclog.NewNullLogger(),
+		ACLResolver:    aclResolver,
+		ForwardRPC:     noopForwardRPC,
+		ConnectEnabled: true,
+	})
+
+	testCases := map[string]struct {
+		csr, err string
+	}{
+		"no csr": {
+			csr: "",
+			err: "CSR is required",
+		},
+		"invalid csr": {
+			csr: "bogus",
+			err: "no PEM-encoded data found",
+		},
+	}
+	for desc, tc := range testCases {
+		t.Run(desc, func(t *testing.T) {
+			_, err := server.Sign(context.Background(), &pbconnectca.SignRequest{
+				Csr: tc.csr,
+			})
+			require.Error(t, err)
+			require.Equal(t, codes.InvalidArgument.String(), status.Code(err).String())
+			require.Equal(t, tc.err, status.Convert(err).Message())
+		})
+	}
+}
+
+func TestSign_Unauthenticated(t *testing.T) {
+	aclResolver := &MockACLResolver{}
+	aclResolver.On("ResolveTokenAndDefaultMeta", mock.Anything, mock.Anything, mock.Anything).
+		Return(nil, acl.ErrNotFound)
+
+	server := NewServer(Config{
+		Logger:         hclog.NewNullLogger(),
+		ACLResolver:    aclResolver,
+		ForwardRPC:     noopForwardRPC,
+		ConnectEnabled: true,
+	})
+
+	csr, _ := connect.TestCSR(t, connect.TestSpiffeIDService(t, "web"))
+
+	_, err := server.Sign(context.Background(), &pbconnectca.SignRequest{
+		Csr: csr,
+	})
+	require.Error(t, err)
+	require.Equal(t, codes.Unauthenticated.String(), status.Code(err).String())
+}
+
+func TestSign_PermissionDenied(t *testing.T) {
+	aclResolver := &MockACLResolver{}
+	aclResolver.On("ResolveTokenAndDefaultMeta", mock.Anything, mock.Anything, mock.Anything).
+		Return(acl.AllowAll(), nil)
+
+	caManager := &MockCAManager{}
+	caManager.On("AuthorizeAndSignCertificate", mock.Anything, mock.Anything).
+		Return(nil, acl.ErrPermissionDenied)
+
+	server := NewServer(Config{
+		Logger:         hclog.NewNullLogger(),
+		ACLResolver:    aclResolver,
+		CAManager:      caManager,
+		ForwardRPC:     noopForwardRPC,
+		ConnectEnabled: true,
+	})
+
+	csr, _ := connect.TestCSR(t, connect.TestSpiffeIDService(t, "web"))
+
+	_, err := server.Sign(context.Background(), &pbconnectca.SignRequest{
+		Csr: csr,
+	})
+	require.Error(t, err)
+	require.Equal(t, codes.PermissionDenied.String(), status.Code(err).String())
+}
+
+func TestSign_InvalidCSR(t *testing.T) {
+	aclResolver := &MockACLResolver{}
+	aclResolver.On("ResolveTokenAndDefaultMeta", mock.Anything, mock.Anything, mock.Anything).
+		Return(acl.AllowAll(), nil)
+
+	caManager := &MockCAManager{}
+	caManager.On("AuthorizeAndSignCertificate", mock.Anything, mock.Anything).
+		Return(nil, connect.InvalidCSRError("nope"))
+
+	server := NewServer(Config{
+		Logger:         hclog.NewNullLogger(),
+		ACLResolver:    aclResolver,
+		CAManager:      caManager,
+		ForwardRPC:     noopForwardRPC,
+		ConnectEnabled: true,
+	})
+
+	csr, _ := connect.TestCSR(t, connect.TestSpiffeIDService(t, "web"))
+
+	_, err := server.Sign(context.Background(), &pbconnectca.SignRequest{
+		Csr: csr,
+	})
+	require.Error(t, err)
+	require.Equal(t, codes.InvalidArgument.String(), status.Code(err).String())
+}
+
+func TestSign_RateLimited(t *testing.T) {
+	aclResolver := &MockACLResolver{}
+	aclResolver.On("ResolveTokenAndDefaultMeta", mock.Anything, mock.Anything, mock.Anything).
+		Return(acl.AllowAll(), nil)
+
+	caManager := &MockCAManager{}
+	caManager.On("AuthorizeAndSignCertificate", mock.Anything, mock.Anything).
+		Return(nil, errors.New("Rate limit reached, try again later"))
+
+	server := NewServer(Config{
+		Logger:         hclog.NewNullLogger(),
+		ACLResolver:    aclResolver,
+		CAManager:      caManager,
+		ForwardRPC:     noopForwardRPC,
+		ConnectEnabled: true,
+	})
+
+	csr, _ := connect.TestCSR(t, connect.TestSpiffeIDService(t, "web"))
+
+	_, err := server.Sign(context.Background(), &pbconnectca.SignRequest{
+		Csr: csr,
+	})
+	require.Error(t, err)
+	require.Equal(t, codes.ResourceExhausted.String(), status.Code(err).String())
+}
+
+func TestSign_InternalError(t *testing.T) {
+	aclResolver := &MockACLResolver{}
+	aclResolver.On("ResolveTokenAndDefaultMeta", mock.Anything, mock.Anything, mock.Anything).
+		Return(acl.AllowAll(), nil)
+
+	caManager := &MockCAManager{}
+	caManager.On("AuthorizeAndSignCertificate", mock.Anything, mock.Anything).
+		Return(nil, errors.New("something went very wrong"))
+
+	server := NewServer(Config{
+		Logger:         hclog.NewNullLogger(),
+		ACLResolver:    aclResolver,
+		CAManager:      caManager,
+		ForwardRPC:     noopForwardRPC,
+		ConnectEnabled: true,
+	})
+
+	csr, _ := connect.TestCSR(t, connect.TestSpiffeIDService(t, "web"))
+
+	_, err := server.Sign(context.Background(), &pbconnectca.SignRequest{
+		Csr: csr,
+	})
+	require.Error(t, err)
+	require.Equal(t, codes.Internal.String(), status.Code(err).String())
+}
+
+func TestSign_Success(t *testing.T) {
+	aclResolver := &MockACLResolver{}
+	aclResolver.On("ResolveTokenAndDefaultMeta", mock.Anything, mock.Anything, mock.Anything).
+		Return(acl.AllowAll(), nil)
+
+	caManager := &MockCAManager{}
+	caManager.On("AuthorizeAndSignCertificate", mock.Anything, mock.Anything).
+		Return(&structs.IssuedCert{CertPEM: "this is the PEM"}, nil)
+
+	server := NewServer(Config{
+		Logger:         hclog.NewNullLogger(),
+		ACLResolver:    aclResolver,
+		CAManager:      caManager,
+		ForwardRPC:     noopForwardRPC,
+		ConnectEnabled: true,
+	})
+
+	csr, _ := connect.TestCSR(t, connect.TestSpiffeIDService(t, "web"))
+
+	rsp, err := server.Sign(context.Background(), &pbconnectca.SignRequest{
+		Csr: csr,
+	})
+	require.NoError(t, err)
+	require.Equal(t, "this is the PEM", rsp.CertPem)
+}
+
+func TestSign_RPCForwarding(t *testing.T) {
+	aclResolver := &MockACLResolver{}
+	aclResolver.On("ResolveTokenAndDefaultMeta", mock.Anything, mock.Anything, mock.Anything).
+		Return(acl.AllowAll(), nil)
+
+	caManager := &MockCAManager{}
+	caManager.On("AuthorizeAndSignCertificate", mock.Anything, mock.Anything).
+		Return(&structs.IssuedCert{CertPEM: "leader response"}, nil)
+
+	leader := NewServer(Config{
+		Logger:         hclog.NewNullLogger(),
+		ACLResolver:    aclResolver,
+		CAManager:      caManager,
+		ForwardRPC:     noopForwardRPC,
+		ConnectEnabled: true,
+	})
+	leaderConn, err := grpc.Dial(runTestServer(t, leader).String(), grpc.WithInsecure())
+	require.NoError(t, err)
+
+	follower := NewServer(Config{
+		Logger: hclog.NewNullLogger(),
+		ForwardRPC: func(_ structs.RPCInfo, fn func(*grpc.ClientConn) error) (bool, error) {
+			return true, fn(leaderConn)
+		},
+		ConnectEnabled: true,
+	})
+
+	csr, _ := connect.TestCSR(t, connect.TestSpiffeIDService(t, "web"))
+
+	rsp, err := follower.Sign(context.Background(), &pbconnectca.SignRequest{
+		Csr: csr,
+	})
+	require.NoError(t, err)
+	require.Equal(t, "leader response", rsp.CertPem)
+}
diff --git a/agent/grpc/public/services/connectca/watch_roots.go b/agent/grpc/public/services/connectca/watch_roots.go
index 7a7430783..1d458b558 100644
--- a/agent/grpc/public/services/connectca/watch_roots.go
+++ b/agent/grpc/public/services/connectca/watch_roots.go
@@ -26,8 +26,11 @@ import (
 // Connect CA roots. Current roots are sent immediately at the start of the
 // stream, and new lists will be sent whenever the roots are rotated.
 func (s *Server) WatchRoots(_ *emptypb.Empty, serverStream pbconnectca.ConnectCAService_WatchRootsServer) error {
-	logger := s.Logger.Named("watch-roots").With("stream_id", streamID())
+	if err := s.requireConnect(); err != nil {
+		return err
+	}
 
+	logger := s.Logger.Named("watch-roots").With("stream_id", traceID())
 	logger.Trace("starting stream")
 	defer logger.Trace("stream closed")
 
@@ -179,8 +182,8 @@ func (s *Server) authorize(token string) error {
 }
 
 // We tag logs with a unique identifier to ease debugging. In the future this
-// should probably be an Open Telemetry trace ID.
-func streamID() string {
+// should probably be a real Open Telemetry trace ID.
+func traceID() string {
 	id, err := uuid.GenerateUUID()
 	if err != nil {
 		return ""
diff --git a/agent/grpc/public/services/connectca/watch_roots_test.go b/agent/grpc/public/services/connectca/watch_roots_test.go
index 7bce07e1a..1106aa35d 100644
--- a/agent/grpc/public/services/connectca/watch_roots_test.go
+++ b/agent/grpc/public/services/connectca/watch_roots_test.go
@@ -26,6 +26,20 @@ import (
 
 const testACLToken = "acl-token"
 
+func TestWatchRoots_ConnectDisabled(t *testing.T) {
+	server := NewServer(Config{ConnectEnabled: false})
+
+	// Begin the stream.
+	client := testClient(t, server)
+	stream, err := client.WatchRoots(context.Background(), &emptypb.Empty{})
+	require.NoError(t, err)
+	rspCh := handleRootsStream(t, stream)
+
+	err = mustGetError(t, rspCh)
+	require.Equal(t, codes.FailedPrecondition.String(), status.Code(err).String())
+	require.Contains(t, status.Convert(err).Message(), "Connect")
+}
+
 func TestWatchRoots_Success(t *testing.T) {
 	fsm, publisher := setupFSMAndPublisher(t)
 
@@ -45,10 +59,11 @@ func TestWatchRoots_Success(t *testing.T) {
 	ctx := public.ContextWithToken(context.Background(), testACLToken)
 
 	server := NewServer(Config{
-		Publisher:   publisher,
-		GetStore:    func() StateStore { return fsm.GetStore() },
-		Logger:      testutil.Logger(t),
-		ACLResolver: aclResolver,
+		Publisher:      publisher,
+		GetStore:       func() StateStore { return fsm.GetStore() },
+		Logger:         testutil.Logger(t),
+		ACLResolver:    aclResolver,
+		ConnectEnabled: true,
 	})
 
 	// Begin the stream.
@@ -92,10 +107,11 @@ func TestWatchRoots_InvalidACLToken(t *testing.T) {
 	ctx := public.ContextWithToken(context.Background(), testACLToken)
 
 	server := NewServer(Config{
-		Publisher:   publisher,
-		GetStore:    func() StateStore { return fsm.GetStore() },
-		Logger:      testutil.Logger(t),
-		ACLResolver: aclResolver,
+		Publisher:      publisher,
+		GetStore:       func() StateStore { return fsm.GetStore() },
+		Logger:         testutil.Logger(t),
+		ACLResolver:    aclResolver,
+		ConnectEnabled: true,
 	})
 
 	// Start the stream.
@@ -129,10 +145,11 @@ func TestWatchRoots_ACLTokenInvalidated(t *testing.T) {
 	ctx := public.ContextWithToken(context.Background(), testACLToken)
 
 	server := NewServer(Config{
-		Publisher:   publisher,
-		GetStore:    func() StateStore { return fsm.GetStore() },
-		Logger:      testutil.Logger(t),
-		ACLResolver: aclResolver,
+		Publisher:      publisher,
+		GetStore:       func() StateStore { return fsm.GetStore() },
+		Logger:         testutil.Logger(t),
+		ACLResolver:    aclResolver,
+		ConnectEnabled: true,
 	})
 
 	// Start the stream.
@@ -196,10 +213,11 @@ func TestWatchRoots_StateStoreAbandoned(t *testing.T) {
 	ctx := public.ContextWithToken(context.Background(), testACLToken)
 
 	server := NewServer(Config{
-		Publisher:   publisher,
-		GetStore:    func() StateStore { return fsm.GetStore() },
-		Logger:      testutil.Logger(t),
-		ACLResolver: aclResolver,
+		Publisher:      publisher,
+		GetStore:       func() StateStore { return fsm.GetStore() },
+		Logger:         testutil.Logger(t),
+		ACLResolver:    aclResolver,
+		ConnectEnabled: true,
 	})
 
 	// Begin the stream.
diff --git a/proto-public/pbconnectca/ca.pb.binary.go b/proto-public/pbconnectca/ca.pb.binary.go
index e373db9b5..3db6ad209 100644
--- a/proto-public/pbconnectca/ca.pb.binary.go
+++ b/proto-public/pbconnectca/ca.pb.binary.go
@@ -26,3 +26,23 @@ func (msg *CARoot) MarshalBinary() ([]byte, error) {
 func (msg *CARoot) UnmarshalBinary(b []byte) error {
 	return proto.Unmarshal(b, msg)
 }
+
+// MarshalBinary implements encoding.BinaryMarshaler
+func (msg *SignRequest) MarshalBinary() ([]byte, error) {
+	return proto.Marshal(msg)
+}
+
+// UnmarshalBinary implements encoding.BinaryUnmarshaler
+func (msg *SignRequest) UnmarshalBinary(b []byte) error {
+	return proto.Unmarshal(b, msg)
+}
+
+// MarshalBinary implements encoding.BinaryMarshaler
+func (msg *SignResponse) MarshalBinary() ([]byte, error) {
+	return proto.Marshal(msg)
+}
+
+// UnmarshalBinary implements encoding.BinaryUnmarshaler
+func (msg *SignResponse) UnmarshalBinary(b []byte) error {
+	return proto.Unmarshal(b, msg)
+}
diff --git a/proto-public/pbconnectca/ca.pb.go b/proto-public/pbconnectca/ca.pb.go
index bb966a4de..a3f1d8777 100644
--- a/proto-public/pbconnectca/ca.pb.go
+++ b/proto-public/pbconnectca/ca.pb.go
@@ -228,6 +228,106 @@ func (x *CARoot) GetRotatedOutAt() *timestamppb.Timestamp {
 	return nil
 }
 
+type SignRequest struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	// csr is the PEM-encoded Certificate Signing Request (CSR).
+	//
+	// The CSR's SAN must include a SPIFFE ID that identifies a service or agent
+	// to which the ACL token provided in the `x-consul-token` metadata has write
+	// access.
+	Csr string `protobuf:"bytes,1,opt,name=csr,proto3" json:"csr,omitempty"`
+}
+
+func (x *SignRequest) Reset() {
+	*x = SignRequest{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_proto_public_pbconnectca_ca_proto_msgTypes[2]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *SignRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*SignRequest) ProtoMessage() {}
+
+func (x *SignRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_proto_public_pbconnectca_ca_proto_msgTypes[2]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use SignRequest.ProtoReflect.Descriptor instead.
+func (*SignRequest) Descriptor() ([]byte, []int) {
+	return file_proto_public_pbconnectca_ca_proto_rawDescGZIP(), []int{2}
+}
+
+func (x *SignRequest) GetCsr() string {
+	if x != nil {
+		return x.Csr
+	}
+	return ""
+}
+
+type SignResponse struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	// cert_pem is the PEM-encoded leaf certificate.
+	CertPem string `protobuf:"bytes,2,opt,name=cert_pem,json=certPem,proto3" json:"cert_pem,omitempty"`
+}
+
+func (x *SignResponse) Reset() {
+	*x = SignResponse{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_proto_public_pbconnectca_ca_proto_msgTypes[3]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *SignResponse) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*SignResponse) ProtoMessage() {}
+
+func (x *SignResponse) ProtoReflect() protoreflect.Message {
+	mi := &file_proto_public_pbconnectca_ca_proto_msgTypes[3]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use SignResponse.ProtoReflect.Descriptor instead.
+func (*SignResponse) Descriptor() ([]byte, []int) {
+	return file_proto_public_pbconnectca_ca_proto_rawDescGZIP(), []int{3}
+}
+
+func (x *SignResponse) GetCertPem() string {
+	if x != nil {
+		return x.CertPem
+	}
+	return ""
+}
+
 var File_proto_public_pbconnectca_ca_proto protoreflect.FileDescriptor
 
 var file_proto_public_pbconnectca_ca_proto_rawDesc = []byte{
@@ -264,17 +364,25 @@ var file_proto_public_pbconnectca_ca_proto_rawDesc = []byte{
 	0x75, 0x74, 0x5f, 0x61, 0x74, 0x18, 0x08, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f,
 	0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69,
 	0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0c, 0x72, 0x6f, 0x74, 0x61, 0x74, 0x65, 0x64,
-	0x4f, 0x75, 0x74, 0x41, 0x74, 0x32, 0x5b, 0x0a, 0x10, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74,
-	0x43, 0x41, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x47, 0x0a, 0x0a, 0x57, 0x61, 0x74,
-	0x63, 0x68, 0x52, 0x6f, 0x6f, 0x74, 0x73, 0x12, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65,
-	0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x1a,
-	0x1d, 0x2e, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x63, 0x61, 0x2e, 0x57, 0x61, 0x74, 0x63,
-	0x68, 0x52, 0x6f, 0x6f, 0x74, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00,
-	0x30, 0x01, 0x42, 0x36, 0x5a, 0x34, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d,
-	0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x63, 0x6f, 0x6e, 0x73, 0x75,
-	0x6c, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2d, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x2f, 0x70,
-	0x62, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x63, 0x61, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74,
-	0x6f, 0x33,
+	0x4f, 0x75, 0x74, 0x41, 0x74, 0x22, 0x1f, 0x0a, 0x0b, 0x53, 0x69, 0x67, 0x6e, 0x52, 0x65, 0x71,
+	0x75, 0x65, 0x73, 0x74, 0x12, 0x10, 0x0a, 0x03, 0x63, 0x73, 0x72, 0x18, 0x01, 0x20, 0x01, 0x28,
+	0x09, 0x52, 0x03, 0x63, 0x73, 0x72, 0x22, 0x29, 0x0a, 0x0c, 0x53, 0x69, 0x67, 0x6e, 0x52, 0x65,
+	0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x19, 0x0a, 0x08, 0x63, 0x65, 0x72, 0x74, 0x5f, 0x70,
+	0x65, 0x6d, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x63, 0x65, 0x72, 0x74, 0x50, 0x65,
+	0x6d, 0x32, 0x96, 0x01, 0x0a, 0x10, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x43, 0x41, 0x53,
+	0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x47, 0x0a, 0x0a, 0x57, 0x61, 0x74, 0x63, 0x68, 0x52,
+	0x6f, 0x6f, 0x74, 0x73, 0x12, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72,
+	0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x1a, 0x1d, 0x2e, 0x63,
+	0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x63, 0x61, 0x2e, 0x57, 0x61, 0x74, 0x63, 0x68, 0x52, 0x6f,
+	0x6f, 0x74, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x30, 0x01, 0x12,
+	0x39, 0x0a, 0x04, 0x53, 0x69, 0x67, 0x6e, 0x12, 0x16, 0x2e, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63,
+	0x74, 0x63, 0x61, 0x2e, 0x53, 0x69, 0x67, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a,
+	0x17, 0x2e, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x63, 0x61, 0x2e, 0x53, 0x69, 0x67, 0x6e,
+	0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x42, 0x36, 0x5a, 0x34, 0x67, 0x69,
+	0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f,
+	0x72, 0x70, 0x2f, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2d,
+	0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x2f, 0x70, 0x62, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74,
+	0x63, 0x61, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -289,20 +397,24 @@ func file_proto_public_pbconnectca_ca_proto_rawDescGZIP() []byte {
 	return file_proto_public_pbconnectca_ca_proto_rawDescData
 }
 
-var file_proto_public_pbconnectca_ca_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
+var file_proto_public_pbconnectca_ca_proto_msgTypes = make([]protoimpl.MessageInfo, 4)
 var file_proto_public_pbconnectca_ca_proto_goTypes = []interface{}{
 	(*WatchRootsResponse)(nil),    // 0: connectca.WatchRootsResponse
 	(*CARoot)(nil),                // 1: connectca.CARoot
-	(*timestamppb.Timestamp)(nil), // 2: google.protobuf.Timestamp
-	(*emptypb.Empty)(nil),         // 3: google.protobuf.Empty
+	(*SignRequest)(nil),           // 2: connectca.SignRequest
+	(*SignResponse)(nil),          // 3: connectca.SignResponse
+	(*timestamppb.Timestamp)(nil), // 4: google.protobuf.Timestamp
+	(*emptypb.Empty)(nil),         // 5: google.protobuf.Empty
 }
 var file_proto_public_pbconnectca_ca_proto_depIdxs = []int32{
 	1, // 0: connectca.WatchRootsResponse.roots:type_name -> connectca.CARoot
-	2, // 1: connectca.CARoot.rotated_out_at:type_name -> google.protobuf.Timestamp
-	3, // 2: connectca.ConnectCAService.WatchRoots:input_type -> google.protobuf.Empty
-	0, // 3: connectca.ConnectCAService.WatchRoots:output_type -> connectca.WatchRootsResponse
-	3, // [3:4] is the sub-list for method output_type
-	2, // [2:3] is the sub-list for method input_type
+	4, // 1: connectca.CARoot.rotated_out_at:type_name -> google.protobuf.Timestamp
+	5, // 2: connectca.ConnectCAService.WatchRoots:input_type -> google.protobuf.Empty
+	2, // 3: connectca.ConnectCAService.Sign:input_type -> connectca.SignRequest
+	0, // 4: connectca.ConnectCAService.WatchRoots:output_type -> connectca.WatchRootsResponse
+	3, // 5: connectca.ConnectCAService.Sign:output_type -> connectca.SignResponse
+	4, // [4:6] is the sub-list for method output_type
+	2, // [2:4] is the sub-list for method input_type
 	2, // [2:2] is the sub-list for extension type_name
 	2, // [2:2] is the sub-list for extension extendee
 	0, // [0:2] is the sub-list for field type_name
@@ -338,6 +450,30 @@ func file_proto_public_pbconnectca_ca_proto_init() {
 				return nil
 			}
 		}
+		file_proto_public_pbconnectca_ca_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*SignRequest); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_proto_public_pbconnectca_ca_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*SignResponse); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
 	}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
@@ -345,7 +481,7 @@ func file_proto_public_pbconnectca_ca_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_proto_public_pbconnectca_ca_proto_rawDesc,
 			NumEnums:      0,
-			NumMessages:   2,
+			NumMessages:   4,
 			NumExtensions: 0,
 			NumServices:   1,
 		},
@@ -375,6 +511,9 @@ type ConnectCAServiceClient interface {
 	// Connect CA roots. Current roots are sent immediately at the start of the
 	// stream, and new lists will be sent whenever the roots are rotated.
 	WatchRoots(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (ConnectCAService_WatchRootsClient, error)
+	// Sign a leaf certificate for the service or agent identified by the SPIFFE
+	// ID in the given CSR's SAN.
+	Sign(ctx context.Context, in *SignRequest, opts ...grpc.CallOption) (*SignResponse, error)
 }
 
 type connectCAServiceClient struct {
@@ -417,12 +556,24 @@ func (x *connectCAServiceWatchRootsClient) Recv() (*WatchRootsResponse, error) {
 	return m, nil
 }
 
+func (c *connectCAServiceClient) Sign(ctx context.Context, in *SignRequest, opts ...grpc.CallOption) (*SignResponse, error) {
+	out := new(SignResponse)
+	err := c.cc.Invoke(ctx, "/connectca.ConnectCAService/Sign", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 // ConnectCAServiceServer is the server API for ConnectCAService service.
 type ConnectCAServiceServer interface {
 	// WatchRoots provides a stream on which you can receive the list of active
 	// Connect CA roots. Current roots are sent immediately at the start of the
 	// stream, and new lists will be sent whenever the roots are rotated.
 	WatchRoots(*emptypb.Empty, ConnectCAService_WatchRootsServer) error
+	// Sign a leaf certificate for the service or agent identified by the SPIFFE
+	// ID in the given CSR's SAN.
+	Sign(context.Context, *SignRequest) (*SignResponse, error)
 }
 
 // UnimplementedConnectCAServiceServer can be embedded to have forward compatible implementations.
@@ -432,6 +583,9 @@ type UnimplementedConnectCAServiceServer struct {
 func (*UnimplementedConnectCAServiceServer) WatchRoots(*emptypb.Empty, ConnectCAService_WatchRootsServer) error {
 	return status.Errorf(codes.Unimplemented, "method WatchRoots not implemented")
 }
+func (*UnimplementedConnectCAServiceServer) Sign(context.Context, *SignRequest) (*SignResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method Sign not implemented")
+}
 
 func RegisterConnectCAServiceServer(s *grpc.Server, srv ConnectCAServiceServer) {
 	s.RegisterService(&_ConnectCAService_serviceDesc, srv)
@@ -458,10 +612,33 @@ func (x *connectCAServiceWatchRootsServer) Send(m *WatchRootsResponse) error {
 	return x.ServerStream.SendMsg(m)
 }
 
+func _ConnectCAService_Sign_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(SignRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(ConnectCAServiceServer).Sign(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/connectca.ConnectCAService/Sign",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(ConnectCAServiceServer).Sign(ctx, req.(*SignRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 var _ConnectCAService_serviceDesc = grpc.ServiceDesc{
 	ServiceName: "connectca.ConnectCAService",
 	HandlerType: (*ConnectCAServiceServer)(nil),
-	Methods:     []grpc.MethodDesc{},
+	Methods: []grpc.MethodDesc{
+		{
+			MethodName: "Sign",
+			Handler:    _ConnectCAService_Sign_Handler,
+		},
+	},
 	Streams: []grpc.StreamDesc{
 		{
 			StreamName:    "WatchRoots",
diff --git a/proto-public/pbconnectca/ca.proto b/proto-public/pbconnectca/ca.proto
index fef15fbc1..216a6e43c 100644
--- a/proto-public/pbconnectca/ca.proto
+++ b/proto-public/pbconnectca/ca.proto
@@ -12,6 +12,10 @@ service ConnectCAService {
   // Connect CA roots. Current roots are sent immediately at the start of the
   // stream, and new lists will be sent whenever the roots are rotated.
   rpc WatchRoots(google.protobuf.Empty) returns (stream WatchRootsResponse) {};
+
+  // Sign a leaf certificate for the service or agent identified by the SPIFFE
+  // ID in the given CSR's SAN.
+  rpc Sign(SignRequest) returns (SignResponse) {};
 }
 
 message WatchRootsResponse {
@@ -70,3 +74,17 @@ message CARoot {
    // active root.
    google.protobuf.Timestamp rotated_out_at = 8;
 }
+
+message SignRequest {
+  // csr is the PEM-encoded Certificate Signing Request (CSR).
+  //
+  // The CSR's SAN must include a SPIFFE ID that identifies a service or agent
+  // to which the ACL token provided in the `x-consul-token` metadata has write
+  // access.
+  string csr = 1;
+}
+
+message SignResponse {
+  // cert_pem is the PEM-encoded leaf certificate.
+  string cert_pem = 2;
+}