Move connect CA provider to separate package

agent/connect/ca/ca_provider.go

@@ -4,10 +4,10 @@ import (
 	"crypto/x509"
 )
 
-// CAProvider is the interface for Consul to interact with
+// Provider is the interface for Consul to interact with
 // an external CA that provides leaf certificate signing for
 // given SpiffeIDServices.
-type CAProvider interface {
+type Provider interface {
 	// Active root returns the currently active root CA for this
 	// provider. This should be a parent of the certificate returned by
 	// ActiveIntermediate()

agent/connect/ca/ca_provider_consul.go

@@ -1,4 +1,4 @@
-package consul
+package connect
 
 import (
 	"bytes"
@@ -15,34 +15,39 @@ import (
 	"time"
 
 	"github.com/hashicorp/consul/agent/connect"
+	"github.com/hashicorp/consul/agent/consul/state"
 	"github.com/hashicorp/consul/agent/structs"
 	"github.com/mitchellh/mapstructure"
 )
 
 type ConsulCAProvider struct {
 	config   *structs.ConsulCAProviderConfig
-
 	id       string
-	srv *Server
+	delegate ConsulCAStateDelegate
 	sync.RWMutex
 }
 
+type ConsulCAStateDelegate interface {
+	State() *state.Store
+	ApplyCARequest(*structs.CARequest) error
+}
+
 // NewConsulCAProvider returns a new instance of the Consul CA provider,
 // bootstrapping its state in the state store necessary
-func NewConsulCAProvider(rawConfig map[string]interface{}, srv *Server) (*ConsulCAProvider, error) {
+func NewConsulCAProvider(rawConfig map[string]interface{}, delegate ConsulCAStateDelegate) (*ConsulCAProvider, error) {
 	conf, err := ParseConsulCAConfig(rawConfig)
 	if err != nil {
 		return nil, err
 	}
 	provider := &ConsulCAProvider{
 		config:   conf,
-		srv:    srv,
+		delegate: delegate,
 		id:       fmt.Sprintf("%s,%s", conf.PrivateKey, conf.RootCert),
 	}
 
 	// Check if this configuration of the provider has already been
 	// initialized in the state store.
-	state := srv.fsm.State()
+	state := delegate.State()
 	_, providerState, err := state.CAProviderState(provider.id)
 	if err != nil {
 		return nil, err
@@ -64,13 +69,9 @@ func NewConsulCAProvider(rawConfig map[string]interface{}, srv *Server) (*Consul
 			Op:            structs.CAOpSetProviderState,
 			ProviderState: &newState,
 		}
-		resp, err := srv.raftApply(structs.ConnectCARequestType, args)
+		if err := delegate.ApplyCARequest(args); err != nil {
-		if err != nil {
 			return nil, err
 		}
-		if respErr, ok := resp.(error); ok {
-			return nil, respErr
-		}
 	}
 
 	idx, _, err := state.CAProviderState(provider.id)
@@ -80,7 +81,7 @@ func NewConsulCAProvider(rawConfig map[string]interface{}, srv *Server) (*Consul
 
 	// Generate a private key if needed
 	if conf.PrivateKey == "" {
-		pk, err := generatePrivateKey()
+		pk, err := GeneratePrivateKey()
 		if err != nil {
 			return nil, err
 		}
@@ -105,13 +106,9 @@ func NewConsulCAProvider(rawConfig map[string]interface{}, srv *Server) (*Consul
 		Op:            structs.CAOpSetProviderState,
 		ProviderState: &newState,
 	}
-	resp, err := srv.raftApply(structs.ConnectCARequestType, args)
+	if err := delegate.ApplyCARequest(args); err != nil {
-	if err != nil {
 		return nil, err
 	}
-	if respErr, ok := resp.(error); ok {
-		return nil, respErr
-	}
 
 	return provider, nil
 }
@@ -131,7 +128,7 @@ func ParseConsulCAConfig(raw map[string]interface{}) (*structs.ConsulCAProviderC
 
 // Return the active root CA and generate a new one if needed
 func (c *ConsulCAProvider) ActiveRoot() (string, error) {
-	state := c.srv.fsm.State()
+	state := c.delegate.State()
 	_, providerState, err := state.CAProviderState(c.id)
 	if err != nil {
 		return "", err
@@ -165,13 +162,9 @@ func (c *ConsulCAProvider) Cleanup() error {
 		Op:            structs.CAOpDeleteProviderState,
 		ProviderState: &structs.CAConsulProviderState{ID: c.id},
 	}
-	resp, err := c.srv.raftApply(structs.ConnectCARequestType, args)
+	if err := c.delegate.ApplyCARequest(args); err != nil {
-	if err != nil {
 		return err
 	}
-	if respErr, ok := resp.(error); ok {
-		return respErr
-	}
 
 	return nil
 }
@@ -185,7 +178,7 @@ func (c *ConsulCAProvider) Sign(csr *x509.CertificateRequest) (string, error) {
 	defer c.Unlock()
 
 	// Get the provider state
-	state := c.srv.fsm.State()
+	state := c.delegate.State()
 	_, providerState, err := state.CAProviderState(c.id)
 	if err != nil {
 		return "", err
@@ -274,7 +267,7 @@ func (c *ConsulCAProvider) CrossSignCA(cert *x509.Certificate) (string, error) {
 	defer c.Unlock()
 
 	// Get the provider state
-	state := c.srv.fsm.State()
+	state := c.delegate.State()
 	_, providerState, err := state.CAProviderState(c.id)
 	if err != nil {
 		return "", err
@@ -333,19 +326,15 @@ func (c *ConsulCAProvider) incrementSerialIndex(providerState *structs.CAConsulP
 		Op:            structs.CAOpSetProviderState,
 		ProviderState: &newState,
 	}
-	resp, err := c.srv.raftApply(structs.ConnectCARequestType, args)
+	if err := c.delegate.ApplyCARequest(args); err != nil {
-	if err != nil {
 		return err
 	}
-	if respErr, ok := resp.(error); ok {
-		return respErr
-	}
 
 	return nil
 }
 
-// generatePrivateKey returns a new private key
+// GeneratePrivateKey returns a new private key
-func generatePrivateKey() (string, error) {
+func GeneratePrivateKey() (string, error) {
 	var pk *ecdsa.PrivateKey
 
 	pk, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
@@ -369,7 +358,7 @@ func generatePrivateKey() (string, error) {
 
 // generateCA makes a new root CA using the current private key
 func (c *ConsulCAProvider) generateCA(privateKey string, sn uint64) (string, error) {
-	state := c.srv.fsm.State()
+	state := c.delegate.State()
 	_, config, err := state.CAConfig()
 	if err != nil {
 		return "", err

agent/connect/ca/ca_provider_consul_test.go

@@ -1,28 +1,81 @@
-package consul
+package connect
 
 import (
-	"os"
+	"fmt"
 	"testing"
 	"time"
 
 	"github.com/hashicorp/consul/agent/connect"
-	"github.com/hashicorp/consul/testrpc"
+	"github.com/hashicorp/consul/agent/consul/state"
+	"github.com/hashicorp/consul/agent/structs"
 	"github.com/stretchr/testify/assert"
 )
 
+type consulCAMockDelegate struct {
+	state *state.Store
+}
+
+func (c *consulCAMockDelegate) State() *state.Store {
+	return c.state
+}
+
+func (c *consulCAMockDelegate) ApplyCARequest(req *structs.CARequest) error {
+	idx, _, err := c.state.CAConfig()
+	if err != nil {
+		return err
+	}
+
+	switch req.Op {
+	case structs.CAOpSetProviderState:
+		_, err := c.state.CASetProviderState(idx+1, req.ProviderState)
+		if err != nil {
+			return err
+		}
+
+		return nil
+	case structs.CAOpDeleteProviderState:
+		if err := c.state.CADeleteProviderState(req.ProviderState.ID); err != nil {
+			return err
+		}
+
+		return nil
+	default:
+		return fmt.Errorf("Invalid CA operation '%s'", req.Op)
+	}
+}
+
+func newMockDelegate(t *testing.T, conf *structs.CAConfiguration) *consulCAMockDelegate {
+	s, err := state.NewStateStore(nil)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+	if s == nil {
+		t.Fatalf("missing state store")
+	}
+	if err := s.CASetConfig(0, conf); err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	return &consulCAMockDelegate{s}
+}
+
+func testConsulCAConfig() *structs.CAConfiguration {
+	return &structs.CAConfiguration{
+		ClusterID: "asdf",
+		Provider:  "consul",
+		Config:    map[string]interface{}{},
+	}
+}
+
 func TestCAProvider_Bootstrap(t *testing.T) {
 	t.Parallel()
 
 	assert := assert.New(t)
-	dir1, s1 := testServer(t)
+	conf := testConsulCAConfig()
-	defer os.RemoveAll(dir1)
+	delegate := newMockDelegate(t, conf)
-	defer s1.Shutdown()
-	codec := rpcClient(t, s1)
-	defer codec.Close()
 
-	testrpc.WaitForLeader(t, s1.RPC, "dc1")
+	provider, err := NewConsulCAProvider(conf.Config, delegate)
-
+	assert.NoError(err)
-	provider := s1.getCAProvider()
 
 	root, err := provider.ActiveRoot()
 	assert.NoError(err)
@@ -30,14 +83,12 @@ func TestCAProvider_Bootstrap(t *testing.T) {
 	// Intermediate should be the same cert.
 	inter, err := provider.ActiveIntermediate()
 	assert.NoError(err)
+	assert.Equal(root, inter)
 
-	// Make sure we initialize without errors and that the
+	// Should be a valid cert
-	// root cert gets set to the active cert.
+	parsed, err := connect.ParseCert(root)
-	state := s1.fsm.State()
-	_, activeRoot, err := state.CARootActive(nil)
 	assert.NoError(err)
-	assert.Equal(root, activeRoot.RootCert)
+	assert.Equal(parsed.URIs[0].String(), fmt.Sprintf("spiffe://%s.consul", conf.ClusterID))
-	assert.Equal(inter, activeRoot.RootCert)
 }
 
 func TestCAProvider_Bootstrap_WithCert(t *testing.T) {
@@ -46,49 +97,35 @@ func TestCAProvider_Bootstrap_WithCert(t *testing.T) {
 	// Make sure setting a custom private key/root cert works.
 	assert := assert.New(t)
 	rootCA := connect.TestCA(t, nil)
-	dir1, s1 := testServerWithConfig(t, func(c *Config) {
+	conf := testConsulCAConfig()
-		c.CAConfig.Config["PrivateKey"] = rootCA.SigningKey
+	conf.Config = map[string]interface{}{
-		c.CAConfig.Config["RootCert"] = rootCA.RootCert
+		"PrivateKey": rootCA.SigningKey,
-	})
+		"RootCert":   rootCA.RootCert,
-	defer os.RemoveAll(dir1)
+	}
-	defer s1.Shutdown()
+	delegate := newMockDelegate(t, conf)
-	codec := rpcClient(t, s1)
-	defer codec.Close()
 
-	testrpc.WaitForLeader(t, s1.RPC, "dc1")
+	provider, err := NewConsulCAProvider(conf.Config, delegate)
-
+	assert.NoError(err)
-	provider := s1.getCAProvider()
 
 	root, err := provider.ActiveRoot()
 	assert.NoError(err)
-
+	assert.Equal(root, rootCA.RootCert)
-	// Make sure we initialize without errors and that the
-	// root cert we provided gets set to the active cert.
-	state := s1.fsm.State()
-	_, activeRoot, err := state.CARootActive(nil)
-	assert.NoError(err)
-	assert.Equal(root, activeRoot.RootCert)
-	assert.Equal(rootCA.RootCert, activeRoot.RootCert)
 }
 
 func TestCAProvider_SignLeaf(t *testing.T) {
 	t.Parallel()
 
 	assert := assert.New(t)
-	dir1, s1 := testServer(t)
+	conf := testConsulCAConfig()
-	defer os.RemoveAll(dir1)
+	delegate := newMockDelegate(t, conf)
-	defer s1.Shutdown()
-	codec := rpcClient(t, s1)
-	defer codec.Close()
 
-	testrpc.WaitForLeader(t, s1.RPC, "dc1")
+	provider, err := NewConsulCAProvider(conf.Config, delegate)
-
+	assert.NoError(err)
-	provider := s1.getCAProvider()
 
 	spiffeService := &connect.SpiffeIDService{
-		Host:       s1.config.NodeName,
+		Host:       "node1",
 		Namespace:  "default",
-		Datacenter: s1.config.Datacenter,
+		Datacenter: "dc1",
 		Service:    "foo",
 	}
 
@@ -141,18 +178,12 @@ func TestCAProvider_CrossSignCA(t *testing.T) {
 	t.Parallel()
 
 	assert := assert.New(t)
+	conf := testConsulCAConfig()
+	delegate := newMockDelegate(t, conf)
+	provider, err := NewConsulCAProvider(conf.Config, delegate)
+	assert.NoError(err)
 
-	// Make sure setting a custom private key/root cert works.
+	// Make a new CA cert to get cross-signed.
-	dir1, s1 := testServer(t)
-	defer os.RemoveAll(dir1)
-	defer s1.Shutdown()
-	codec := rpcClient(t, s1)
-	defer codec.Close()
-
-	testrpc.WaitForLeader(t, s1.RPC, "dc1")
-
-	provider := s1.getCAProvider()
-
 	rootCA := connect.TestCA(t, nil)
 	rootPEM, err := provider.ActiveRoot()
 	assert.NoError(err)

agent/consul/connect_ca_endpoint_test.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/hashicorp/consul/agent/connect"
+	connect_ca "github.com/hashicorp/consul/agent/connect/ca"
 	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/testrpc"
 	"github.com/hashicorp/net-rpc-msgpackrpc"
@@ -82,9 +83,9 @@ func TestConnectCAConfig_GetSet(t *testing.T) {
 		var reply structs.CAConfiguration
 		assert.NoError(msgpackrpc.CallWithCodec(codec, "ConnectCA.ConfigurationGet", args, &reply))
 
-		actual, err := ParseConsulCAConfig(reply.Config)
+		actual, err := connect_ca.ParseConsulCAConfig(reply.Config)
 		assert.NoError(err)
-		expected, err := ParseConsulCAConfig(s1.config.CAConfig.Config)
+		expected, err := connect_ca.ParseConsulCAConfig(s1.config.CAConfig.Config)
 		assert.NoError(err)
 		assert.Equal(reply.Provider, s1.config.CAConfig.Provider)
 		assert.Equal(actual, expected)
@@ -117,9 +118,9 @@ func TestConnectCAConfig_GetSet(t *testing.T) {
 		var reply structs.CAConfiguration
 		assert.NoError(msgpackrpc.CallWithCodec(codec, "ConnectCA.ConfigurationGet", args, &reply))
 
-		actual, err := ParseConsulCAConfig(reply.Config)
+		actual, err := connect_ca.ParseConsulCAConfig(reply.Config)
 		assert.NoError(err)
-		expected, err := ParseConsulCAConfig(newConfig.Config)
+		expected, err := connect_ca.ParseConsulCAConfig(newConfig.Config)
 		assert.NoError(err)
 		assert.Equal(reply.Provider, newConfig.Provider)
 		assert.Equal(actual, expected)
@@ -149,7 +150,7 @@ func TestConnectCAConfig_TriggerRotation(t *testing.T) {
 
 	// Update the provider config to use a new private key, which should
 	// cause a rotation.
-	newKey, err := generatePrivateKey()
+	newKey, err := connect_ca.GeneratePrivateKey()
 	assert.NoError(err)
 	newConfig := &structs.CAConfiguration{
 		Provider: "consul",
@@ -219,9 +220,9 @@ func TestConnectCAConfig_TriggerRotation(t *testing.T) {
 		var reply structs.CAConfiguration
 		assert.NoError(msgpackrpc.CallWithCodec(codec, "ConnectCA.ConfigurationGet", args, &reply))
 
-		actual, err := ParseConsulCAConfig(reply.Config)
+		actual, err := connect_ca.ParseConsulCAConfig(reply.Config)
 		assert.NoError(err)
-		expected, err := ParseConsulCAConfig(newConfig.Config)
+		expected, err := connect_ca.ParseConsulCAConfig(newConfig.Config)
 		assert.NoError(err)
 		assert.Equal(reply.Provider, newConfig.Provider)
 		assert.Equal(actual, expected)

agent/consul/consul_ca_delegate.go

@@ -0,0 +1,28 @@
+package consul
+
+import (
+	"github.com/hashicorp/consul/agent/consul/state"
+	"github.com/hashicorp/consul/agent/structs"
+)
+
+// consulCADelegate providers callbacks for the Consul CA provider
+// to use the state store for its operations.
+type consulCADelegate struct {
+	srv *Server
+}
+
+func (c *consulCADelegate) State() *state.Store {
+	return c.srv.fsm.State()
+}
+
+func (c *consulCADelegate) ApplyCARequest(req *structs.CARequest) error {
+	resp, err := c.srv.raftApply(structs.ConnectCARequestType, req)
+	if err != nil {
+		return err
+	}
+	if respErr, ok := resp.(error); ok {
+		return respErr
+	}
+
+	return nil
+}

agent/consul/leader.go

@@ -10,6 +10,7 @@ import (
 	"github.com/armon/go-metrics"
 	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/agent/connect"
+	connect_ca "github.com/hashicorp/consul/agent/connect/ca"
 	"github.com/hashicorp/consul/agent/consul/autopilot"
 	"github.com/hashicorp/consul/agent/metadata"
 	"github.com/hashicorp/consul/agent/structs"
@@ -478,10 +479,10 @@ func (s *Server) initializeCA() error {
 }
 
 // createProvider returns a connect CA provider from the given config.
-func (s *Server) createCAProvider(conf *structs.CAConfiguration) (connect.CAProvider, error) {
+func (s *Server) createCAProvider(conf *structs.CAConfiguration) (connect_ca.Provider, error) {
 	switch conf.Provider {
 	case structs.ConsulCAProvider:
-		return NewConsulCAProvider(conf.Config, s)
+		return connect_ca.NewConsulCAProvider(conf.Config, &consulCADelegate{s})
 	default:
 		return nil, fmt.Errorf("unknown CA provider %q", conf.Provider)
 	}
@@ -510,7 +511,7 @@ func (s *Server) getCAProvider() connect.CAProvider {
 	return result
 }
 
-func (s *Server) setCAProvider(newProvider connect.CAProvider) {
+func (s *Server) setCAProvider(newProvider connect_ca.Provider) {
 	s.caProviderLock.Lock()
 	defer s.caProviderLock.Unlock()
 	s.caProvider = newProvider

agent/consul/server.go

@@ -18,7 +18,7 @@ import (
 	"time"
 
 	"github.com/hashicorp/consul/acl"
-	"github.com/hashicorp/consul/agent/connect"
+	connect_ca "github.com/hashicorp/consul/agent/connect/ca"
 	"github.com/hashicorp/consul/agent/consul/autopilot"
 	"github.com/hashicorp/consul/agent/consul/fsm"
 	"github.com/hashicorp/consul/agent/consul/state"
@@ -99,7 +99,7 @@ type Server struct {
 
 	// caProvider is the current CA provider in use for Connect. This is
 	// only non-nil when we are the leader.
-	caProvider     connect.CAProvider
+	caProvider     connect_ca.Provider
 	caProviderLock sync.RWMutex
 
 	// Consul configuration