diff --git a/agent/connect/ca/mock_Provider.go b/agent/connect/ca/mock_Provider.go
index e8b9e58929..ba42e0452c 100644
--- a/agent/connect/ca/mock_Provider.go
+++ b/agent/connect/ca/mock_Provider.go
@@ -2,8 +2,11 @@
 
 package ca
 
-import mock "github.com/stretchr/testify/mock"
-import x509 "crypto/x509"
+import (
+	x509 "crypto/x509"
+
+	mock "github.com/stretchr/testify/mock"
+)
 
 // MockProvider is an autogenerated mock type for the Provider type
 type MockProvider struct {
@@ -66,13 +69,13 @@ func (_m *MockProvider) Cleanup() error {
 	return r0
 }
 
-// Configure provides a mock function with given fields: clusterId, isRoot, rawConfig
-func (_m *MockProvider) Configure(clusterId string, isRoot bool, rawConfig map[string]interface{}) error {
-	ret := _m.Called(clusterId, isRoot, rawConfig)
+// Configure provides a mock function with given fields: clusterID, isRoot, rawConfig, state
+func (_m *MockProvider) Configure(clusterID string, isRoot bool, rawConfig map[string]interface{}, state map[string]string) error {
+	ret := _m.Called(clusterID, isRoot, rawConfig, state)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(string, bool, map[string]interface{}) error); ok {
-		r0 = rf(clusterId, isRoot, rawConfig)
+	if rf, ok := ret.Get(0).(func(string, bool, map[string]interface{}, map[string]string) error); ok {
+		r0 = rf(clusterID, isRoot, rawConfig, state)
 	} else {
 		r0 = ret.Error(0)
 	}
@@ -212,3 +215,26 @@ func (_m *MockProvider) SignIntermediate(_a0 *x509.CertificateRequest) (string,
 
 	return r0, r1
 }
+
+// State provides a mock function with given fields:
+func (_m *MockProvider) State() (map[string]string, error) {
+	ret := _m.Called()
+
+	var r0 map[string]string
+	if rf, ok := ret.Get(0).(func() map[string]string); ok {
+		r0 = rf()
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(map[string]string)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func() error); ok {
+		r1 = rf()
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
diff --git a/agent/connect/ca/plugin/plugin_test.go b/agent/connect/ca/plugin/plugin_test.go
index f4c100c288..18c4d9463b 100644
--- a/agent/connect/ca/plugin/plugin_test.go
+++ b/agent/connect/ca/plugin/plugin_test.go
@@ -21,17 +21,22 @@ func TestProvider_Configure(t *testing.T) {
 		m.On("Configure", "foo", false, map[string]interface{}{
 			"string": "bar",
 			"number": float64(42), // because json
+		}, map[string]string{
+			"foo": "bar",
 		}).Once().Return(nil)
 		require.NoError(p.Configure("foo", false, map[string]interface{}{
 			"string": "bar",
 			"number": float64(42),
+		}, map[string]string{
+			"foo": "bar",
 		}))
 		m.AssertExpectations(t)
 
 		// Try with an error
 		m.Mock = mock.Mock{}
-		m.On("Configure", "foo", false, map[string]interface{}{}).Once().Return(errors.New("hello world"))
-		err := p.Configure("foo", false, map[string]interface{}{})
+		m.On("Configure", "foo", false, map[string]interface{}{}, map[string]string{}).
+			Once().Return(errors.New("hello world"))
+		err := p.Configure("foo", false, map[string]interface{}{}, map[string]string{})
 		require.Error(err)
 		require.Contains(err.Error(), "hello")
 		m.AssertExpectations(t)
diff --git a/agent/connect/ca/plugin/provider.pb.binary.go b/agent/connect/ca/plugin/provider.pb.binary.go
index 57ebd890c9..20e951bd8d 100644
--- a/agent/connect/ca/plugin/provider.pb.binary.go
+++ b/agent/connect/ca/plugin/provider.pb.binary.go
@@ -57,6 +57,16 @@ func (msg *CrossSignCARequest) UnmarshalBinary(b []byte) error {
 	return proto.Unmarshal(b, msg)
 }
 
+// MarshalBinary implements encoding.BinaryMarshaler
+func (msg *StateResponse) MarshalBinary() ([]byte, error) {
+	return proto.Marshal(msg)
+}
+
+// UnmarshalBinary implements encoding.BinaryUnmarshaler
+func (msg *StateResponse) UnmarshalBinary(b []byte) error {
+	return proto.Unmarshal(b, msg)
+}
+
 // MarshalBinary implements encoding.BinaryMarshaler
 func (msg *ActiveRootResponse) MarshalBinary() ([]byte, error) {
 	return proto.Marshal(msg)
diff --git a/agent/connect/ca/plugin/provider.pb.go b/agent/connect/ca/plugin/provider.pb.go
index fcafb0b155..100b035023 100644
--- a/agent/connect/ca/plugin/provider.pb.go
+++ b/agent/connect/ca/plugin/provider.pb.go
@@ -27,6 +27,7 @@ type ConfigureRequest struct {
 	ClusterId            string   `protobuf:"bytes,1,opt,name=cluster_id,json=clusterId,proto3" json:"cluster_id,omitempty"`
 	IsRoot               bool     `protobuf:"varint,2,opt,name=is_root,json=isRoot,proto3" json:"is_root,omitempty"`
 	Config               []byte   `protobuf:"bytes,3,opt,name=config,proto3" json:"config,omitempty"`
+	State                []byte   `protobuf:"bytes,4,opt,name=state,proto3" json:"state,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
 	XXX_unrecognized     []byte   `json:"-"`
 	XXX_sizecache        int32    `json:"-"`
@@ -86,6 +87,13 @@ func (m *ConfigureRequest) GetConfig() []byte {
 	return nil
 }
 
+func (m *ConfigureRequest) GetState() []byte {
+	if m != nil {
+		return m.State
+	}
+	return nil
+}
+
 type SetIntermediateRequest struct {
 	IntermediatePem      string   `protobuf:"bytes,1,opt,name=intermediate_pem,json=intermediatePem,proto3" json:"intermediate_pem,omitempty"`
 	RootPem              string   `protobuf:"bytes,2,opt,name=root_pem,json=rootPem,proto3" json:"root_pem,omitempty"`
@@ -282,6 +290,53 @@ func (m *CrossSignCARequest) GetCrt() []byte {
 	return nil
 }
 
+type StateResponse struct {
+	State                []byte   `protobuf:"bytes,1,opt,name=state,proto3" json:"state,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *StateResponse) Reset()         { *m = StateResponse{} }
+func (m *StateResponse) String() string { return proto.CompactTextString(m) }
+func (*StateResponse) ProtoMessage()    {}
+func (*StateResponse) Descriptor() ([]byte, []int) {
+	return fileDescriptor_c6a9f3c02af3d1c8, []int{5}
+}
+func (m *StateResponse) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *StateResponse) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_StateResponse.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalTo(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *StateResponse) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_StateResponse.Merge(m, src)
+}
+func (m *StateResponse) XXX_Size() int {
+	return m.Size()
+}
+func (m *StateResponse) XXX_DiscardUnknown() {
+	xxx_messageInfo_StateResponse.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_StateResponse proto.InternalMessageInfo
+
+func (m *StateResponse) GetState() []byte {
+	if m != nil {
+		return m.State
+	}
+	return nil
+}
+
 type ActiveRootResponse struct {
 	CrtPem               string   `protobuf:"bytes,1,opt,name=crt_pem,json=crtPem,proto3" json:"crt_pem,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
@@ -293,7 +348,7 @@ func (m *ActiveRootResponse) Reset()         { *m = ActiveRootResponse{} }
 func (m *ActiveRootResponse) String() string { return proto.CompactTextString(m) }
 func (*ActiveRootResponse) ProtoMessage()    {}
 func (*ActiveRootResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_c6a9f3c02af3d1c8, []int{5}
+	return fileDescriptor_c6a9f3c02af3d1c8, []int{6}
 }
 func (m *ActiveRootResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -340,7 +395,7 @@ func (m *GenerateIntermediateCSRResponse) Reset()         { *m = GenerateInterme
 func (m *GenerateIntermediateCSRResponse) String() string { return proto.CompactTextString(m) }
 func (*GenerateIntermediateCSRResponse) ProtoMessage()    {}
 func (*GenerateIntermediateCSRResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_c6a9f3c02af3d1c8, []int{6}
+	return fileDescriptor_c6a9f3c02af3d1c8, []int{7}
 }
 func (m *GenerateIntermediateCSRResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -387,7 +442,7 @@ func (m *ActiveIntermediateResponse) Reset()         { *m = ActiveIntermediateRe
 func (m *ActiveIntermediateResponse) String() string { return proto.CompactTextString(m) }
 func (*ActiveIntermediateResponse) ProtoMessage()    {}
 func (*ActiveIntermediateResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_c6a9f3c02af3d1c8, []int{7}
+	return fileDescriptor_c6a9f3c02af3d1c8, []int{8}
 }
 func (m *ActiveIntermediateResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -434,7 +489,7 @@ func (m *GenerateIntermediateResponse) Reset()         { *m = GenerateIntermedia
 func (m *GenerateIntermediateResponse) String() string { return proto.CompactTextString(m) }
 func (*GenerateIntermediateResponse) ProtoMessage()    {}
 func (*GenerateIntermediateResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_c6a9f3c02af3d1c8, []int{8}
+	return fileDescriptor_c6a9f3c02af3d1c8, []int{9}
 }
 func (m *GenerateIntermediateResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -481,7 +536,7 @@ func (m *SignResponse) Reset()         { *m = SignResponse{} }
 func (m *SignResponse) String() string { return proto.CompactTextString(m) }
 func (*SignResponse) ProtoMessage()    {}
 func (*SignResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_c6a9f3c02af3d1c8, []int{9}
+	return fileDescriptor_c6a9f3c02af3d1c8, []int{10}
 }
 func (m *SignResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -528,7 +583,7 @@ func (m *SignIntermediateResponse) Reset()         { *m = SignIntermediateRespon
 func (m *SignIntermediateResponse) String() string { return proto.CompactTextString(m) }
 func (*SignIntermediateResponse) ProtoMessage()    {}
 func (*SignIntermediateResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_c6a9f3c02af3d1c8, []int{10}
+	return fileDescriptor_c6a9f3c02af3d1c8, []int{11}
 }
 func (m *SignIntermediateResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -575,7 +630,7 @@ func (m *CrossSignCAResponse) Reset()         { *m = CrossSignCAResponse{} }
 func (m *CrossSignCAResponse) String() string { return proto.CompactTextString(m) }
 func (*CrossSignCAResponse) ProtoMessage()    {}
 func (*CrossSignCAResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_c6a9f3c02af3d1c8, []int{11}
+	return fileDescriptor_c6a9f3c02af3d1c8, []int{12}
 }
 func (m *CrossSignCAResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -623,7 +678,7 @@ func (m *Empty) Reset()         { *m = Empty{} }
 func (m *Empty) String() string { return proto.CompactTextString(m) }
 func (*Empty) ProtoMessage()    {}
 func (*Empty) Descriptor() ([]byte, []int) {
-	return fileDescriptor_c6a9f3c02af3d1c8, []int{12}
+	return fileDescriptor_c6a9f3c02af3d1c8, []int{13}
 }
 func (m *Empty) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -658,6 +713,7 @@ func init() {
 	proto.RegisterType((*SignRequest)(nil), "plugin.SignRequest")
 	proto.RegisterType((*SignIntermediateRequest)(nil), "plugin.SignIntermediateRequest")
 	proto.RegisterType((*CrossSignCARequest)(nil), "plugin.CrossSignCARequest")
+	proto.RegisterType((*StateResponse)(nil), "plugin.StateResponse")
 	proto.RegisterType((*ActiveRootResponse)(nil), "plugin.ActiveRootResponse")
 	proto.RegisterType((*GenerateIntermediateCSRResponse)(nil), "plugin.GenerateIntermediateCSRResponse")
 	proto.RegisterType((*ActiveIntermediateResponse)(nil), "plugin.ActiveIntermediateResponse")
@@ -671,40 +727,42 @@ func init() {
 func init() { proto.RegisterFile("provider.proto", fileDescriptor_c6a9f3c02af3d1c8) }
 
 var fileDescriptor_c6a9f3c02af3d1c8 = []byte{
-	// 523 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x84, 0x54, 0xcd, 0x6e, 0xd3, 0x4c,
-	0x14, 0xfd, 0x9c, 0x7e, 0x75, 0x9a, 0xdb, 0x40, 0xad, 0x69, 0xd5, 0x18, 0x03, 0x4e, 0x64, 0x21,
-	0x12, 0x04, 0x44, 0x82, 0x82, 0x2a, 0xb1, 0x22, 0x58, 0x50, 0x55, 0x6c, 0x8a, 0x23, 0xb6, 0x44,
-	0xa9, 0x73, 0x89, 0x46, 0x8a, 0x3d, 0x66, 0x66, 0x5c, 0x89, 0x37, 0xe1, 0x8d, 0x60, 0xc9, 0x23,
-	0xa0, 0xf0, 0x22, 0xc8, 0x8e, 0xed, 0xda, 0x13, 0xb7, 0xde, 0x79, 0xee, 0x9c, 0x73, 0xee, 0xdf,
-	0x19, 0xc3, 0xdd, 0x88, 0xb3, 0x2b, 0xba, 0x40, 0x3e, 0x8e, 0x38, 0x93, 0x8c, 0xe8, 0xd1, 0x2a,
-	0x5e, 0xd2, 0xd0, 0xb9, 0x04, 0xc3, 0x65, 0xe1, 0x57, 0xba, 0x8c, 0x39, 0x7a, 0xf8, 0x2d, 0x46,
-	0x21, 0xc9, 0x43, 0x00, 0x7f, 0x15, 0x0b, 0x89, 0x7c, 0x46, 0x17, 0xa6, 0x36, 0xd0, 0x46, 0x1d,
-	0xaf, 0x93, 0x45, 0xce, 0x17, 0xa4, 0x07, 0x6d, 0x2a, 0x66, 0x9c, 0x31, 0x69, 0xb6, 0x06, 0xda,
-	0x68, 0xcf, 0xd3, 0xa9, 0xf0, 0x18, 0x93, 0xe4, 0x18, 0x74, 0x3f, 0xd5, 0x32, 0x77, 0x06, 0xda,
-	0xa8, 0xeb, 0x65, 0x27, 0xe7, 0x0b, 0x1c, 0x4f, 0x51, 0x9e, 0x87, 0x12, 0x79, 0x80, 0x0b, 0x3a,
-	0x97, 0x45, 0xa6, 0x27, 0x60, 0xd0, 0x52, 0x78, 0x16, 0x61, 0x90, 0xe5, 0x3b, 0x28, 0xc7, 0x2f,
-	0x30, 0x20, 0xf7, 0x60, 0x2f, 0x49, 0x99, 0x42, 0x5a, 0x29, 0xa4, 0x9d, 0x9c, 0x2f, 0x30, 0x70,
-	0xfa, 0xb0, 0x3f, 0xa5, 0xcb, 0x30, 0x17, 0x35, 0x60, 0xc7, 0x17, 0x3c, 0xd5, 0xe9, 0x7a, 0xc9,
-	0xa7, 0xf3, 0x14, 0x7a, 0x09, 0xa0, 0xae, 0x82, 0x6d, 0xf0, 0x63, 0x20, 0x2e, 0x67, 0x42, 0x24,
-	0x0c, 0x77, 0x52, 0xc6, 0x71, 0x59, 0xe0, 0xb8, 0x74, 0x9e, 0x03, 0x99, 0xf8, 0x92, 0x5e, 0x61,
-	0xd2, 0xbb, 0x87, 0x22, 0x62, 0xa1, 0xc0, 0x64, 0x38, 0x3e, 0x97, 0xa5, 0x46, 0x74, 0x9f, 0xa7,
-	0x45, 0xbe, 0x81, 0xfe, 0x19, 0x86, 0xc8, 0xe7, 0x12, 0xcb, 0x75, 0xb8, 0x53, 0xaf, 0xc2, 0x15,
-	0xbc, 0xc2, 0x15, 0x3c, 0xe1, 0xbe, 0x06, 0x6b, 0x93, 0xaa, 0xda, 0x41, 0x53, 0xca, 0x53, 0x78,
-	0x50, 0x97, 0xb2, 0x99, 0x38, 0x84, 0xee, 0x66, 0xa0, 0x4d, 0xc0, 0x13, 0x30, 0xb7, 0x07, 0xdb,
-	0x44, 0x1a, 0xc3, 0x61, 0x65, 0xc0, 0x4d, 0xf8, 0x36, 0xec, 0xbe, 0x0f, 0x22, 0xf9, 0xfd, 0xe5,
-	0xcf, 0x5d, 0x68, 0xb9, 0x13, 0xf2, 0x0a, 0x3a, 0x85, 0x65, 0x89, 0x39, 0xde, 0x18, 0x79, 0xac,
-	0xba, 0xd8, 0xba, 0x93, 0xdf, 0xa4, 0x64, 0xf2, 0x0c, 0xba, 0xf9, 0x30, 0x52, 0xb3, 0x56, 0xaf,
-	0x55, 0xf4, 0x29, 0xc0, 0xf5, 0x72, 0x55, 0xac, 0x95, 0x1f, 0x6b, 0xf6, 0xff, 0x09, 0x7a, 0x37,
-	0xac, 0x59, 0x55, 0x19, 0xe6, 0xc7, 0x26, 0x5b, 0xbc, 0x85, 0x03, 0xe5, 0xf9, 0x10, 0x3b, 0xe7,
-	0xd6, 0xbf, 0x2b, 0xb5, 0x9b, 0xb3, 0xdc, 0xaa, 0x15, 0x11, 0xa5, 0x1e, 0xa7, 0xda, 0x55, 0xed,
-	0x4e, 0x3f, 0xc2, 0x51, 0x5d, 0xb5, 0xaa, 0xd4, 0xa3, 0xdb, 0x5a, 0x2b, 0xc4, 0x5e, 0xc0, 0xff,
-	0x89, 0x05, 0xc8, 0x61, 0xd1, 0xcc, 0xf5, 0x23, 0xb6, 0x8e, 0xaa, 0xc1, 0x8c, 0xf2, 0x19, 0x0c,
-	0xd5, 0x6f, 0xa4, 0x5f, 0x46, 0xd6, 0x0d, 0x63, 0x70, 0x33, 0x20, 0x93, 0xfd, 0x00, 0xfb, 0x25,
-	0x47, 0x92, 0x62, 0xbf, 0xdb, 0xff, 0x01, 0xeb, 0x7e, 0xed, 0x5d, 0xa6, 0x33, 0x84, 0xb6, 0xbb,
-	0xc2, 0x79, 0x18, 0x47, 0xb7, 0xdb, 0xeb, 0x9d, 0xf1, 0x6b, 0x6d, 0x6b, 0xbf, 0xd7, 0xb6, 0xf6,
-	0x67, 0x6d, 0x6b, 0x3f, 0xfe, 0xda, 0xff, 0x5d, 0xea, 0xe9, 0x6f, 0xf9, 0xe4, 0x5f, 0x00, 0x00,
-	0x00, 0xff, 0xff, 0x75, 0xd0, 0x68, 0xcb, 0xa8, 0x05, 0x00, 0x00,
+	// 560 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x84, 0x55, 0xc1, 0x6e, 0xd3, 0x40,
+	0x10, 0xc5, 0x69, 0x93, 0x34, 0xd3, 0x94, 0x5a, 0xdb, 0xd0, 0x18, 0x03, 0x4e, 0x64, 0x01, 0x09,
+	0x82, 0x46, 0x82, 0x82, 0x2a, 0x71, 0x22, 0x58, 0x50, 0x55, 0x5c, 0x8a, 0x23, 0xae, 0x44, 0xc1,
+	0x59, 0xa2, 0x95, 0x62, 0xaf, 0xd9, 0x5d, 0x57, 0xf0, 0x27, 0x7c, 0x12, 0x47, 0x3e, 0x01, 0x85,
+	0x7f, 0xe0, 0x8c, 0xbc, 0xb1, 0x1d, 0x7b, 0xe3, 0xd6, 0xb7, 0xcc, 0xf8, 0xcd, 0xdb, 0x79, 0xb3,
+	0x6f, 0x36, 0x70, 0x3b, 0x64, 0xf4, 0x8a, 0xcc, 0x31, 0x1b, 0x85, 0x8c, 0x0a, 0x8a, 0x1a, 0xe1,
+	0x32, 0x5a, 0x90, 0xc0, 0xfe, 0x0e, 0xba, 0x43, 0x83, 0xaf, 0x64, 0x11, 0x31, 0xec, 0xe2, 0x6f,
+	0x11, 0xe6, 0x02, 0x3d, 0x00, 0xf0, 0x96, 0x11, 0x17, 0x98, 0x4d, 0xc9, 0xdc, 0xd0, 0xfa, 0xda,
+	0xb0, 0xe5, 0xb6, 0x92, 0xcc, 0xc5, 0x1c, 0x75, 0xa1, 0x49, 0xf8, 0x94, 0x51, 0x2a, 0x8c, 0x5a,
+	0x5f, 0x1b, 0xee, 0xb9, 0x0d, 0xc2, 0x5d, 0x4a, 0x05, 0x3a, 0x86, 0x86, 0x27, 0xb9, 0x8c, 0x9d,
+	0xbe, 0x36, 0x6c, 0xbb, 0x49, 0x84, 0x3a, 0x50, 0xe7, 0x62, 0x26, 0xb0, 0xb1, 0x2b, 0xd3, 0xeb,
+	0xc0, 0xfe, 0x0c, 0xc7, 0x13, 0x2c, 0x2e, 0x02, 0x81, 0x99, 0x8f, 0xe7, 0x64, 0x26, 0xb2, 0xf3,
+	0x9f, 0x80, 0x4e, 0x72, 0xe9, 0x69, 0x88, 0xfd, 0xa4, 0x8b, 0xc3, 0x7c, 0xfe, 0x12, 0xfb, 0xe8,
+	0x2e, 0xec, 0xc5, 0x8d, 0x48, 0x48, 0x4d, 0x42, 0x9a, 0x71, 0x7c, 0x89, 0x7d, 0xbb, 0x07, 0xfb,
+	0x13, 0xb2, 0x08, 0x52, 0x52, 0x1d, 0x76, 0x3c, 0xce, 0x24, 0x4f, 0xdb, 0x8d, 0x7f, 0xda, 0x4f,
+	0xa1, 0x1b, 0x03, 0xca, 0x3a, 0xd8, 0x06, 0x3f, 0x06, 0xe4, 0x30, 0xca, 0x79, 0x5c, 0xe1, 0x8c,
+	0xf3, 0x38, 0x26, 0x32, 0x1c, 0x13, 0xf6, 0x23, 0x38, 0x98, 0x08, 0xc9, 0xc4, 0x43, 0x1a, 0x70,
+	0xbc, 0x11, 0xaf, 0xe5, 0xc5, 0x9f, 0x00, 0x1a, 0x7b, 0x82, 0x5c, 0xe1, 0x78, 0x70, 0x19, 0xb6,
+	0x0b, 0x4d, 0x8f, 0x89, 0x9c, 0xde, 0x86, 0xc7, 0xa4, 0x96, 0xd7, 0xd0, 0x3b, 0xc7, 0x01, 0x66,
+	0x33, 0x81, 0xf3, 0xed, 0x3a, 0x13, 0xb7, 0x50, 0xcb, 0x59, 0xa1, 0x96, 0xb3, 0xb8, 0xf6, 0x15,
+	0x98, 0xeb, 0xa3, 0x8a, 0x42, 0xab, 0x8e, 0x3c, 0x83, 0xfb, 0x65, 0x47, 0x56, 0x17, 0x0e, 0xa0,
+	0xbd, 0x9e, 0x7b, 0x15, 0xf0, 0x14, 0x8c, 0xed, 0xf9, 0x57, 0x15, 0x8d, 0xe0, 0xa8, 0x70, 0x0f,
+	0x55, 0xf8, 0x26, 0xd4, 0xdf, 0xf9, 0xa1, 0xf8, 0xf1, 0xe2, 0x5f, 0x1d, 0x6a, 0xce, 0x18, 0xbd,
+	0x84, 0x56, 0xe6, 0x77, 0x64, 0x8c, 0xd6, 0x5b, 0x30, 0x52, 0x57, 0xc0, 0x3c, 0x48, 0xbf, 0xc8,
+	0x62, 0x74, 0x02, 0x75, 0x79, 0xab, 0xa8, 0x98, 0x37, 0xef, 0xa4, 0x61, 0xf1, 0xce, 0x9f, 0x41,
+	0x3b, 0x9d, 0x9d, 0x5c, 0x0c, 0xa5, 0x4a, 0x21, 0x3f, 0x03, 0xd8, 0x78, 0x41, 0xc5, 0x9a, 0x69,
+	0x58, 0x62, 0x97, 0x8f, 0xd0, 0xbd, 0xc6, 0x15, 0x2a, 0xcb, 0x20, 0x0d, 0xab, 0x5c, 0xf4, 0x06,
+	0x0e, 0x95, 0xa5, 0x44, 0x56, 0xa6, 0xb1, 0x74, 0x5b, 0x55, 0x35, 0xe7, 0xa9, 0xb3, 0x0b, 0x24,
+	0x4a, 0x3f, 0x76, 0x51, 0x55, 0xa9, 0x05, 0x3e, 0x40, 0xa7, 0xac, 0x5b, 0x95, 0xea, 0xe1, 0x4d,
+	0xd2, 0x32, 0xb2, 0xe7, 0xb0, 0x1b, 0x3b, 0x06, 0x1d, 0x65, 0x62, 0x36, 0x4f, 0x83, 0xd9, 0x29,
+	0x26, 0x93, 0x92, 0x4f, 0xa0, 0xab, 0xf6, 0x44, 0xbd, 0x3c, 0xb2, 0x6c, 0x18, 0xfd, 0xeb, 0x01,
+	0x09, 0xed, 0x7b, 0xd8, 0xcf, 0x19, 0x18, 0x65, 0xf7, 0xbb, 0xfd, 0xba, 0x98, 0xf7, 0x4a, 0xbf,
+	0x25, 0x3c, 0x03, 0x68, 0x3a, 0x4b, 0x3c, 0x0b, 0xa2, 0xf0, 0x66, 0x7b, 0xbd, 0xd5, 0x7f, 0xad,
+	0x2c, 0xed, 0xf7, 0xca, 0xd2, 0xfe, 0xac, 0x2c, 0xed, 0xe7, 0x5f, 0xeb, 0xd6, 0x97, 0x86, 0xfc,
+	0x0b, 0x38, 0xfd, 0x1f, 0x00, 0x00, 0xff, 0xff, 0xd1, 0xcf, 0x1a, 0x9a, 0x14, 0x06, 0x00, 0x00,
 }
 
 // Reference imports to suppress errors if they are not otherwise used.
@@ -720,6 +778,7 @@ const _ = grpc.SupportPackageIsVersion4
 // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream.
 type CAClient interface {
 	Configure(ctx context.Context, in *ConfigureRequest, opts ...grpc.CallOption) (*Empty, error)
+	State(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*StateResponse, error)
 	GenerateRoot(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*Empty, error)
 	ActiveRoot(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*ActiveRootResponse, error)
 	GenerateIntermediateCSR(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*GenerateIntermediateCSRResponse, error)
@@ -749,6 +808,15 @@ func (c *cAClient) Configure(ctx context.Context, in *ConfigureRequest, opts ...
 	return out, nil
 }
 
+func (c *cAClient) State(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*StateResponse, error) {
+	out := new(StateResponse)
+	err := c.cc.Invoke(ctx, "/plugin.CA/State", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 func (c *cAClient) GenerateRoot(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*Empty, error) {
 	out := new(Empty)
 	err := c.cc.Invoke(ctx, "/plugin.CA/GenerateRoot", in, out, opts...)
@@ -842,6 +910,7 @@ func (c *cAClient) Cleanup(ctx context.Context, in *Empty, opts ...grpc.CallOpti
 // CAServer is the server API for CA service.
 type CAServer interface {
 	Configure(context.Context, *ConfigureRequest) (*Empty, error)
+	State(context.Context, *Empty) (*StateResponse, error)
 	GenerateRoot(context.Context, *Empty) (*Empty, error)
 	ActiveRoot(context.Context, *Empty) (*ActiveRootResponse, error)
 	GenerateIntermediateCSR(context.Context, *Empty) (*GenerateIntermediateCSRResponse, error)
@@ -876,6 +945,24 @@ func _CA_Configure_Handler(srv interface{}, ctx context.Context, dec func(interf
 	return interceptor(ctx, in, info, handler)
 }
 
+func _CA_State_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(Empty)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(CAServer).State(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/plugin.CA/State",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(CAServer).State(ctx, req.(*Empty))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 func _CA_GenerateRoot_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(Empty)
 	if err := dec(in); err != nil {
@@ -1064,6 +1151,10 @@ var _CA_serviceDesc = grpc.ServiceDesc{
 			MethodName: "Configure",
 			Handler:    _CA_Configure_Handler,
 		},
+		{
+			MethodName: "State",
+			Handler:    _CA_State_Handler,
+		},
 		{
 			MethodName: "GenerateRoot",
 			Handler:    _CA_GenerateRoot_Handler,
@@ -1146,6 +1237,12 @@ func (m *ConfigureRequest) MarshalTo(dAtA []byte) (int, error) {
 		i = encodeVarintProvider(dAtA, i, uint64(len(m.Config)))
 		i += copy(dAtA[i:], m.Config)
 	}
+	if len(m.State) > 0 {
+		dAtA[i] = 0x22
+		i++
+		i = encodeVarintProvider(dAtA, i, uint64(len(m.State)))
+		i += copy(dAtA[i:], m.State)
+	}
 	if m.XXX_unrecognized != nil {
 		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
@@ -1266,6 +1363,33 @@ func (m *CrossSignCARequest) MarshalTo(dAtA []byte) (int, error) {
 	return i, nil
 }
 
+func (m *StateResponse) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalTo(dAtA)
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *StateResponse) MarshalTo(dAtA []byte) (int, error) {
+	var i int
+	_ = i
+	var l int
+	_ = l
+	if len(m.State) > 0 {
+		dAtA[i] = 0xa
+		i++
+		i = encodeVarintProvider(dAtA, i, uint64(len(m.State)))
+		i += copy(dAtA[i:], m.State)
+	}
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	return i, nil
+}
+
 func (m *ActiveRootResponse) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
@@ -1502,6 +1626,10 @@ func (m *ConfigureRequest) Size() (n int) {
 	if l > 0 {
 		n += 1 + l + sovProvider(uint64(l))
 	}
+	l = len(m.State)
+	if l > 0 {
+		n += 1 + l + sovProvider(uint64(l))
+	}
 	if m.XXX_unrecognized != nil {
 		n += len(m.XXX_unrecognized)
 	}
@@ -1576,6 +1704,22 @@ func (m *CrossSignCARequest) Size() (n int) {
 	return n
 }
 
+func (m *StateResponse) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	l = len(m.State)
+	if l > 0 {
+		n += 1 + l + sovProvider(uint64(l))
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
 func (m *ActiveRootResponse) Size() (n int) {
 	if m == nil {
 		return 0
@@ -1828,6 +1972,40 @@ func (m *ConfigureRequest) Unmarshal(dAtA []byte) error {
 				m.Config = []byte{}
 			}
 			iNdEx = postIndex
+		case 4:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field State", wireType)
+			}
+			var byteLen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowProvider
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				byteLen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if byteLen < 0 {
+				return ErrInvalidLengthProvider
+			}
+			postIndex := iNdEx + byteLen
+			if postIndex < 0 {
+				return ErrInvalidLengthProvider
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.State = append(m.State[:0], dAtA[iNdEx:postIndex]...)
+			if m.State == nil {
+				m.State = []byte{}
+			}
+			iNdEx = postIndex
 		default:
 			iNdEx = preIndex
 			skippy, err := skipProvider(dAtA[iNdEx:])
@@ -2235,6 +2413,94 @@ func (m *CrossSignCARequest) Unmarshal(dAtA []byte) error {
 	}
 	return nil
 }
+func (m *StateResponse) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowProvider
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: StateResponse: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: StateResponse: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field State", wireType)
+			}
+			var byteLen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowProvider
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				byteLen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if byteLen < 0 {
+				return ErrInvalidLengthProvider
+			}
+			postIndex := iNdEx + byteLen
+			if postIndex < 0 {
+				return ErrInvalidLengthProvider
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.State = append(m.State[:0], dAtA[iNdEx:postIndex]...)
+			if m.State == nil {
+				m.State = []byte{}
+			}
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipProvider(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if skippy < 0 {
+				return ErrInvalidLengthProvider
+			}
+			if (iNdEx + skippy) < 0 {
+				return ErrInvalidLengthProvider
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
 func (m *ActiveRootResponse) Unmarshal(dAtA []byte) error {
 	l := len(dAtA)
 	iNdEx := 0
diff --git a/agent/connect/ca/plugin/provider.proto b/agent/connect/ca/plugin/provider.proto
index bc0d9aa785..de15778fe4 100644
--- a/agent/connect/ca/plugin/provider.proto
+++ b/agent/connect/ca/plugin/provider.proto
@@ -14,6 +14,7 @@ package plugin;
 
 service CA {
     rpc Configure(ConfigureRequest) returns (Empty);
+    rpc State(Empty) returns (StateResponse);
     rpc GenerateRoot(Empty) returns (Empty);
     rpc ActiveRoot(Empty) returns (ActiveRootResponse);
     rpc GenerateIntermediateCSR(Empty) returns (GenerateIntermediateCSRResponse);
@@ -30,6 +31,7 @@ message ConfigureRequest {
     string cluster_id = 1;
     bool is_root = 2;
     bytes config = 3; // JSON-encoded structure
+    bytes state = 4; // JSON-encoded structure
 }
 
 message SetIntermediateRequest {
@@ -49,6 +51,10 @@ message CrossSignCARequest {
     bytes crt = 1;
 }
 
+message StateResponse {
+    bytes state = 1; // JSON-encoded map[string]string
+}
+
 message ActiveRootResponse {
     string crt_pem = 1;
 }
diff --git a/agent/connect/ca/plugin/transport_grpc.go b/agent/connect/ca/plugin/transport_grpc.go
index cb21e559cd..09ff0957c3 100644
--- a/agent/connect/ca/plugin/transport_grpc.go
+++ b/agent/connect/ca/plugin/transport_grpc.go
@@ -19,8 +19,24 @@ func (p *providerPluginGRPCServer) Configure(_ context.Context, req *ConfigureRe
 	if err := json.Unmarshal(req.Config, &rawConfig); err != nil {
 		return nil, err
 	}
+	var state map[string]string
+	if err := json.Unmarshal(req.State, &state); err != nil {
+		return nil, err
+	}
 
-	return &Empty{}, p.impl.Configure(req.ClusterId, req.IsRoot, rawConfig)
+	return &Empty{}, p.impl.Configure(req.ClusterId, req.IsRoot, rawConfig, state)
+}
+
+func (p *providerPluginGRPCServer) State(context.Context, *Empty) (*StateResponse, error) {
+	got, err := p.impl.State()
+	var jsonBs []byte
+	if err == nil {
+		jsonBs, err = json.Marshal(got)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return &StateResponse{State: jsonBs}, nil
 }
 
 func (p *providerPluginGRPCServer) GenerateRoot(context.Context, *Empty) (*Empty, error) {
@@ -96,20 +112,42 @@ type providerPluginGRPCClient struct {
 func (p *providerPluginGRPCClient) Configure(
 	clusterId string,
 	isRoot bool,
-	rawConfig map[string]interface{}) error {
+	rawConfig map[string]interface{},
+	state map[string]string) error {
+
 	config, err := json.Marshal(rawConfig)
 	if err != nil {
 		return err
 	}
 
+	stateJSON, err := json.Marshal(state)
+	if err != nil {
+		return err
+	}
+
 	_, err = p.client.Configure(p.doneCtx, &ConfigureRequest{
 		ClusterId: clusterId,
 		IsRoot:    isRoot,
 		Config:    config,
+		State:     stateJSON,
 	})
 	return p.err(err)
 }
 
+func (p *providerPluginGRPCClient) State() (map[string]string, error) {
+	stateResp, err := p.client.State(p.doneCtx, &Empty{})
+	if err != nil {
+		return nil, p.err(err)
+	}
+
+	var state map[string]string
+	err = json.Unmarshal(stateResp.State, &state)
+	if err != nil {
+		return nil, err
+	}
+	return state, nil
+}
+
 func (p *providerPluginGRPCClient) GenerateRoot() error {
 	_, err := p.client.GenerateRoot(p.doneCtx, &Empty{})
 	return p.err(err)
diff --git a/agent/connect/ca/plugin/transport_netrpc.go b/agent/connect/ca/plugin/transport_netrpc.go
index 8186e1d00c..335facdd99 100644
--- a/agent/connect/ca/plugin/transport_netrpc.go
+++ b/agent/connect/ca/plugin/transport_netrpc.go
@@ -2,6 +2,7 @@ package plugin
 
 import (
 	"crypto/x509"
+	"encoding/json"
 	"net/rpc"
 
 	"github.com/hashicorp/consul/agent/connect/ca"
@@ -15,7 +16,19 @@ type providerPluginRPCServer struct {
 }
 
 func (p *providerPluginRPCServer) Configure(args *ConfigureRPCRequest, _ *struct{}) error {
-	return p.impl.Configure(args.ClusterId, args.IsRoot, args.RawConfig)
+	return p.impl.Configure(args.ClusterId, args.IsRoot, args.RawConfig, args.State)
+}
+
+func (p *providerPluginRPCServer) State(_ struct{}, resp *StateResponse) error {
+	state, err := p.impl.State()
+	if err != nil {
+		return err
+	}
+	resp.State, err = json.Marshal(state)
+	if err != nil {
+		return err
+	}
+	return nil
 }
 
 func (p *providerPluginRPCServer) GenerateRoot(struct{}, *struct{}) error {
@@ -96,14 +109,30 @@ type providerPluginRPCClient struct {
 func (p *providerPluginRPCClient) Configure(
 	clusterId string,
 	isRoot bool,
-	rawConfig map[string]interface{}) error {
+	rawConfig map[string]interface{},
+	state map[string]string) error {
 	return p.client.Call("Plugin.Configure", &ConfigureRPCRequest{
 		ClusterId: clusterId,
 		IsRoot:    isRoot,
 		RawConfig: rawConfig,
+		State:     state,
 	}, &struct{}{})
 }
 
+func (p *providerPluginRPCClient) State() (map[string]string, error) {
+	var resp StateResponse
+	err := p.client.Call("Plugin.State", struct{}{}, &resp)
+	if err != nil {
+		return nil, err
+	}
+	var state map[string]string
+	err = json.Unmarshal(resp.State, &state)
+	if err != nil {
+		return nil, err
+	}
+	return state, nil
+}
+
 func (p *providerPluginRPCClient) GenerateRoot() error {
 	return p.client.Call("Plugin.GenerateRoot", struct{}{}, &struct{}{})
 }
@@ -177,6 +206,7 @@ type ConfigureRPCRequest struct {
 	ClusterId string
 	IsRoot    bool
 	RawConfig map[string]interface{}
+	State     map[string]string
 }
 
 type SetIntermediateRPCRequest struct {
diff --git a/agent/connect/ca/provider.go b/agent/connect/ca/provider.go
index 386211d38d..ff14635991 100644
--- a/agent/connect/ca/provider.go
+++ b/agent/connect/ca/provider.go
@@ -11,9 +11,33 @@ import (
 // an external CA that provides leaf certificate signing for
 // given SpiffeIDServices.
 type Provider interface {
-	// Configure initializes the provider based on the given cluster ID, root status
-	// and configuration values.
-	Configure(clusterId string, isRoot bool, rawConfig map[string]interface{}) error
+	// Configure initializes the provider based on the given cluster ID, root
+	// status and configuration values. rawConfig contains the user-provided
+	// Config. State contains a the State the same provider last persisted on a
+	// restart or reconfiguration. The provider must not modify `rawConfig` or
+	// `state` maps directly as it may be being read from other goroutines.
+	Configure(clusterID string, isRoot bool, rawConfig map[string]interface{}, state map[string]string) error
+
+	// State returns the current provider state. If the provider doesn't need to
+	// store anything other than what the user configured this can return nil. It
+	// is called after any config change before the new active config is stored in
+	// the state store and the most recent value returned by the provider is given
+	// in subsequent `Configure` calls provided that the current provider is the
+	// same type as the new provider instance being configured. This provides a
+	// simple way for providers to persist information like UUIDs of resources
+	// they manage. This state is visible to anyone with operator:read via the API
+	// so it's not intended for storing secrets like root private keys. Only
+	// strings are permitted since this has to pass through msgpack and so
+	// interface values will end up mangled in many cases which is ugly for all
+	// provider code to have to remember to reason about.
+	//
+	// Note that the map returned will be accessed (read-only) in other goroutines
+	// - for example passed to Configure in the Connect CA Config RPC endpoint -
+	// so it must not just be a pointer to a map that may internally be modified.
+	// If the Provider only writes to it during Configure it's safe to return
+	// as-is, but otherwise it's assumed the map returned is a copy of the state
+	// in the Provider struct so it won't change after being returned.
+	State() (map[string]string, error)
 
 	// GenerateRoot causes the creation of a new root certificate for this provider.
 	// This can also be a no-op if a root certificate already exists for the given
diff --git a/agent/connect/ca/provider_consul.go b/agent/connect/ca/provider_consul.go
index b246704136..8ebaaae4ee 100644
--- a/agent/connect/ca/provider_consul.go
+++ b/agent/connect/ca/provider_consul.go
@@ -32,6 +32,14 @@ type ConsulProvider struct {
 	spiffeID  *connect.SpiffeIDSigning
 	logger    *log.Logger
 
+	// testState is only used to test Consul leader's handling of providers that
+	// need to persist state. Consul provider actually manages it's state directly
+	// in the FSM since it is highly sensitive not (root private keys) not just
+	// metadata for lookups. We could make a whole mock provider to keep this out
+	// of Consul but that would still need to be configurable through real config
+	// and is a lot more boilerplate to test this for equivalent functionality.
+	testState map[string]string
+
 	sync.RWMutex
 }
 
@@ -41,7 +49,7 @@ type ConsulProviderStateDelegate interface {
 }
 
 // Configure sets up the provider using the given configuration.
-func (c *ConsulProvider) Configure(clusterID string, isRoot bool, rawConfig map[string]interface{}) error {
+func (c *ConsulProvider) Configure(clusterID string, isRoot bool, rawConfig map[string]interface{}, state map[string]string) error {
 	// Parse the raw config and update our ID.
 	config, err := ParseConsulCAConfig(rawConfig)
 	if err != nil {
@@ -54,6 +62,9 @@ func (c *ConsulProvider) Configure(clusterID string, isRoot bool, rawConfig map[
 	c.isRoot = isRoot
 	c.spiffeID = connect.SpiffeIDSigningForCluster(&structs.CAConfiguration{ClusterID: clusterID})
 
+	// Passthrough test state for state handling tests. See testState doc.
+	c.parseTestState(rawConfig)
+
 	// Exit early if the state store has an entry for this provider's config.
 	_, providerState, err := c.Delegate.State().CAProviderState(c.id)
 	if err != nil {
@@ -114,6 +125,15 @@ func (c *ConsulProvider) Configure(clusterID string, isRoot bool, rawConfig map[
 	return nil
 }
 
+// State implements Provider. Consul actually does store all it's state in raft
+// but it manages it independently through a separate table already so this is a
+// no-op. This method just passes through testState which allows tests to verify
+// state handling behavior without needing to plumb a full test mock provider
+// right through Consul server code.
+func (c *ConsulProvider) State() (map[string]string, error) {
+	return c.testState, nil
+}
+
 // ActiveRoot returns the active root CA certificate.
 func (c *ConsulProvider) ActiveRoot() (string, error) {
 	_, providerState, err := c.getState()
@@ -641,3 +661,28 @@ func (c *ConsulProvider) generateCA(privateKey string, sn uint64) (string, error
 func (c *ConsulProvider) SetLogger(logger *log.Logger) {
 	c.logger = logger
 }
+
+func (c *ConsulProvider) parseTestState(rawConfig map[string]interface{}) {
+	c.testState = nil
+	if rawTestState, ok := rawConfig["test_state"]; ok {
+		if ts, ok := rawTestState.(map[string]string); ok {
+			c.testState = ts
+			return
+		}
+
+		// Secondary's config takes a trip through the state store before Configure
+		// is called unlike primaries. This means we end up with map[string]string
+		// encoded as map[string]interface{}. Just handle that case and ignore the
+		// rest as this is test-only code and we'd rather not leave a way to error
+		// CA setup and leave cluster unavailable in prod by accidentally setting a
+		// bad test_state config.
+		if ts, ok := rawTestState.(map[string]interface{}); ok {
+			c.testState = make(map[string]string)
+			for k, v := range ts {
+				if s, ok := v.(string); ok {
+					c.testState[k] = s
+				}
+			}
+		}
+	}
+}
diff --git a/agent/connect/ca/provider_consul_test.go b/agent/connect/ca/provider_consul_test.go
index 1829d0b0b6..e01a580d19 100644
--- a/agent/connect/ca/provider_consul_test.go
+++ b/agent/connect/ca/provider_consul_test.go
@@ -84,7 +84,7 @@ func TestConsulCAProvider_Bootstrap(t *testing.T) {
 	delegate := newMockDelegate(t, conf)
 
 	provider := TestConsulProvider(t, delegate)
-	require.NoError(provider.Configure(conf.ClusterID, true, conf.Config))
+	require.NoError(provider.Configure(conf.ClusterID, true, conf.Config, nil))
 	require.NoError(provider.GenerateRoot())
 
 	root, err := provider.ActiveRoot()
@@ -117,7 +117,7 @@ func TestConsulCAProvider_Bootstrap_WithCert(t *testing.T) {
 	delegate := newMockDelegate(t, conf)
 
 	provider := TestConsulProvider(t, delegate)
-	require.NoError(provider.Configure(conf.ClusterID, true, conf.Config))
+	require.NoError(provider.Configure(conf.ClusterID, true, conf.Config, nil))
 	require.NoError(provider.GenerateRoot())
 
 	root, err := provider.ActiveRoot()
@@ -139,7 +139,7 @@ func TestConsulCAProvider_SignLeaf(t *testing.T) {
 			delegate := newMockDelegate(t, conf)
 
 			provider := TestConsulProvider(t, delegate)
-			require.NoError(provider.Configure(conf.ClusterID, true, conf.Config))
+			require.NoError(provider.Configure(conf.ClusterID, true, conf.Config, nil))
 			require.NoError(provider.GenerateRoot())
 
 			spiffeService := &connect.SpiffeIDService{
@@ -245,7 +245,7 @@ func TestConsulCAProvider_CrossSignCA(t *testing.T) {
 			provider1 := TestConsulProvider(t, delegate1)
 			conf1.Config["PrivateKeyType"] = tc.SigningKeyType
 			conf1.Config["PrivateKeyBits"] = tc.SigningKeyBits
-			require.NoError(provider1.Configure(conf1.ClusterID, true, conf1.Config))
+			require.NoError(provider1.Configure(conf1.ClusterID, true, conf1.Config, nil))
 			require.NoError(provider1.GenerateRoot())
 
 			conf2 := testConsulCAConfig()
@@ -254,7 +254,7 @@ func TestConsulCAProvider_CrossSignCA(t *testing.T) {
 			provider2 := TestConsulProvider(t, delegate2)
 			conf2.Config["PrivateKeyType"] = tc.CSRKeyType
 			conf2.Config["PrivateKeyBits"] = tc.CSRKeyBits
-			require.NoError(provider2.Configure(conf2.ClusterID, true, conf2.Config))
+			require.NoError(provider2.Configure(conf2.ClusterID, true, conf2.Config, nil))
 			require.NoError(provider2.GenerateRoot())
 
 			testCrossSignProviders(t, provider1, provider2)
@@ -363,7 +363,7 @@ func TestConsulProvider_SignIntermediate(t *testing.T) {
 			provider1 := TestConsulProvider(t, delegate1)
 			conf1.Config["PrivateKeyType"] = tc.SigningKeyType
 			conf1.Config["PrivateKeyBits"] = tc.SigningKeyBits
-			require.NoError(provider1.Configure(conf1.ClusterID, true, conf1.Config))
+			require.NoError(provider1.Configure(conf1.ClusterID, true, conf1.Config, nil))
 			require.NoError(provider1.GenerateRoot())
 
 			conf2 := testConsulCAConfig()
@@ -372,7 +372,7 @@ func TestConsulProvider_SignIntermediate(t *testing.T) {
 			provider2 := TestConsulProvider(t, delegate2)
 			conf2.Config["PrivateKeyType"] = tc.CSRKeyType
 			conf2.Config["PrivateKeyBits"] = tc.CSRKeyBits
-			require.NoError(provider2.Configure(conf2.ClusterID, false, conf2.Config))
+			require.NoError(provider2.Configure(conf2.ClusterID, false, conf2.Config, nil))
 
 			testSignIntermediateCrossDC(t, provider1, provider2)
 		})
@@ -452,7 +452,7 @@ func TestConsulCAProvider_MigrateOldID(t *testing.T) {
 	require.NotNil(providerState)
 
 	provider := TestConsulProvider(t, delegate)
-	require.NoError(provider.Configure(conf.ClusterID, true, conf.Config))
+	require.NoError(provider.Configure(conf.ClusterID, true, conf.Config, nil))
 	require.NoError(provider.GenerateRoot())
 
 	// After running Configure, the old ID entry should be gone.
diff --git a/agent/connect/ca/provider_vault.go b/agent/connect/ca/provider_vault.go
index f522d7a637..a193109377 100644
--- a/agent/connect/ca/provider_vault.go
+++ b/agent/connect/ca/provider_vault.go
@@ -40,7 +40,7 @@ func vaultTLSConfig(config *structs.VaultCAProviderConfig) *vaultapi.TLSConfig {
 }
 
 // Configure sets up the provider using the given configuration.
-func (v *VaultProvider) Configure(clusterID string, isRoot bool, rawConfig map[string]interface{}) error {
+func (v *VaultProvider) Configure(clusterID string, isRoot bool, rawConfig map[string]interface{}, state map[string]string) error {
 	config, err := ParseVaultCAConfig(rawConfig)
 	if err != nil {
 		return err
@@ -68,6 +68,12 @@ func (v *VaultProvider) Configure(clusterID string, isRoot bool, rawConfig map[s
 	return nil
 }
 
+// State implements Provider. Vault provider needs no state other than the
+// user-provided config currently.
+func (v *VaultProvider) State() (map[string]string, error) {
+	return nil, nil
+}
+
 // ActiveRoot returns the active root CA certificate.
 func (v *VaultProvider) ActiveRoot() (string, error) {
 	return v.getCA(v.config.RootPKIPath)
diff --git a/agent/connect/ca/provider_vault_test.go b/agent/connect/ca/provider_vault_test.go
index 37bc813467..a0580ff88d 100644
--- a/agent/connect/ca/provider_vault_test.go
+++ b/agent/connect/ca/provider_vault_test.go
@@ -289,7 +289,7 @@ func TestVaultProvider_SignIntermediateConsul(t *testing.T) {
 		conf := testConsulCAConfig()
 		delegate := newMockDelegate(t, conf)
 		provider2 := TestConsulProvider(t, delegate)
-		require.NoError(t, provider2.Configure(conf.ClusterID, false, conf.Config))
+		require.NoError(t, provider2.Configure(conf.ClusterID, false, conf.Config, nil))
 
 		testSignIntermediateCrossDC(t, provider1, provider2)
 	})
@@ -299,7 +299,7 @@ func TestVaultProvider_SignIntermediateConsul(t *testing.T) {
 		conf := testConsulCAConfig()
 		delegate := newMockDelegate(t, conf)
 		provider1 := TestConsulProvider(t, delegate)
-		require.NoError(t, provider1.Configure(conf.ClusterID, true, conf.Config))
+		require.NoError(t, provider1.Configure(conf.ClusterID, true, conf.Config, nil))
 		require.NoError(t, provider1.GenerateRoot())
 
 		provider2, testVault2 := testVaultProviderWithConfig(t, false, nil)
@@ -335,7 +335,7 @@ func testVaultProviderWithConfig(t *testing.T, isRoot bool, rawConf map[string]i
 
 	provider := &VaultProvider{}
 
-	if err := provider.Configure(connect.TestClusterID, isRoot, conf); err != nil {
+	if err := provider.Configure(connect.TestClusterID, isRoot, conf, nil); err != nil {
 		testVault.Stop()
 		t.Fatalf("err: %v", err)
 	}
diff --git a/agent/connect_ca_endpoint.go b/agent/connect_ca_endpoint.go
index 3052f3a23e..7f245f1713 100644
--- a/agent/connect_ca_endpoint.go
+++ b/agent/connect_ca_endpoint.go
@@ -51,7 +51,6 @@ func (s *HTTPServer) ConnectCAConfigurationGet(resp http.ResponseWriter, req *ht
 		return nil, err
 	}
 
-	fixupConfig(&reply)
 	return reply, nil
 }
 
@@ -72,26 +71,3 @@ func (s *HTTPServer) ConnectCAConfigurationSet(resp http.ResponseWriter, req *ht
 	err := s.agent.RPC("ConnectCA.ConfigurationSet", &args, &reply)
 	return nil, err
 }
-
-// A hack to fix up the config types inside of the map[string]interface{}
-// so that they get formatted correctly during json.Marshal. Without this,
-// string values that get converted to []uint8 end up getting output back
-// to the user in base64-encoded form.
-func fixupConfig(conf *structs.CAConfiguration) {
-	for k, v := range conf.Config {
-		if raw, ok := v.([]uint8); ok {
-			strVal := structs.Uint8ToString(raw)
-			conf.Config[k] = strVal
-			switch conf.Provider {
-			case structs.ConsulCAProvider:
-				if k == "PrivateKey" && strVal != "" {
-					conf.Config["PrivateKey"] = "hidden"
-				}
-			case structs.VaultCAProvider:
-				if k == "Token" && strVal != "" {
-					conf.Config["Token"] = "hidden"
-				}
-			}
-		}
-	}
-}
diff --git a/agent/consul/connect_ca_endpoint.go b/agent/consul/connect_ca_endpoint.go
index 51d7fdbcd0..9ee2a86773 100644
--- a/agent/consul/connect_ca_endpoint.go
+++ b/agent/consul/connect_ca_endpoint.go
@@ -168,6 +168,16 @@ func (s *ConnectCA) ConfigurationSet(
 		return nil
 	}
 
+	// If the provider hasn't changed, we need to load the current Provider state
+	// so it can decide if it needs to change resources or not based on the config
+	// change.
+	if args.Config.Provider == config.Provider {
+		// Note this is a shallow copy since the State method doc requires the
+		// provider return a map that will not be further modified and should not
+		// modify the one we pass to Configure.
+		args.Config.State = config.State
+	}
+
 	// Create a new instance of the provider described by the config
 	// and get the current active root CA. This acts as a good validation
 	// of the config and makes sure the provider is functioning correctly
@@ -176,7 +186,8 @@ func (s *ConnectCA) ConfigurationSet(
 	if err != nil {
 		return fmt.Errorf("could not initialize provider: %v", err)
 	}
-	if err := newProvider.Configure(args.Config.ClusterID, true, args.Config.Config); err != nil {
+	if err := newProvider.Configure(args.Config.ClusterID, true,
+		args.Config.Config, args.Config.State); err != nil {
 		return fmt.Errorf("error configuring provider: %v", err)
 	}
 	if err := newProvider.GenerateRoot(); err != nil {
@@ -193,6 +204,13 @@ func (s *ConnectCA) ConfigurationSet(
 		return err
 	}
 
+	// See if the provider needs to persist any state along with the config
+	pState, err := newProvider.State()
+	if err != nil {
+		return fmt.Errorf("error getting provider state: %v", err)
+	}
+	args.Config.State = pState
+
 	// Compare the new provider's root CA ID to the current one. If they
 	// match, just update the existing provider with the new config.
 	// If they don't match, begin the root rotation process.
diff --git a/agent/consul/connect_ca_endpoint_test.go b/agent/consul/connect_ca_endpoint_test.go
index db48086def..6693703987 100644
--- a/agent/consul/connect_ca_endpoint_test.go
+++ b/agent/consul/connect_ca_endpoint_test.go
@@ -101,6 +101,8 @@ func TestConnectCAConfig_GetSet(t *testing.T) {
 		assert.Equal(actual, expected)
 	}
 
+	testState := map[string]string{"foo": "bar"}
+
 	// Update a config value
 	newConfig := &structs.CAConfiguration{
 		Provider: "consul",
@@ -108,6 +110,9 @@ func TestConnectCAConfig_GetSet(t *testing.T) {
 			"PrivateKey":     "",
 			"RootCert":       "",
 			"RotationPeriod": 180 * 24 * time.Hour,
+			// This verifies the state persistence for providers although Consul
+			// provider doesn't actually use that mechanism outside of tests.
+			"test_state": testState,
 		},
 	}
 	{
@@ -135,6 +140,7 @@ func TestConnectCAConfig_GetSet(t *testing.T) {
 		assert.NoError(err)
 		assert.Equal(reply.Provider, newConfig.Provider)
 		assert.Equal(actual, expected)
+		assert.Equal(testState, reply.State)
 	}
 }
 
diff --git a/agent/consul/leader_connect.go b/agent/consul/leader_connect.go
index 2f142b63f8..da76c1aaab 100644
--- a/agent/consul/leader_connect.go
+++ b/agent/consul/leader_connect.go
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"reflect"
 	"strings"
 	"time"
 
@@ -211,7 +212,7 @@ func (s *Server) initializeCA() error {
 
 // initializeRootCA runs the initialization logic for a root CA.
 func (s *Server) initializeRootCA(provider ca.Provider, conf *structs.CAConfiguration) error {
-	if err := provider.Configure(conf.ClusterID, true, conf.Config); err != nil {
+	if err := provider.Configure(conf.ClusterID, true, conf.Config, conf.State); err != nil {
 		return fmt.Errorf("error configuring provider: %v", err)
 	}
 	if err := provider.GenerateRoot(); err != nil {
@@ -238,6 +239,24 @@ func (s *Server) initializeRootCA(provider ca.Provider, conf *structs.CAConfigur
 		return fmt.Errorf("error getting intermediate cert: %v", err)
 	}
 
+	// If the provider has state to persist and it's changed or new then update
+	// CAConfig.
+	pState, err := provider.State()
+	if err != nil {
+		return fmt.Errorf("error getting provider state: %v", err)
+	}
+	if !reflect.DeepEqual(conf.State, pState) {
+		// Update the CAConfig in raft to persist the provider state
+		conf.State = pState
+		req := structs.CARequest{
+			Op:     structs.CAOpSetConfig,
+			Config: conf,
+		}
+		if _, err = s.raftApply(structs.ConnectCARequestType, req); err != nil {
+			return fmt.Errorf("error persisting provider state: %v", err)
+		}
+	}
+
 	// Check if the CA root is already initialized and exit if it is,
 	// adding on any existing intermediate certs since they aren't directly
 	// tied to the provider.
@@ -396,6 +415,12 @@ func (s *Server) initializeSecondaryCA(provider ca.Provider, roots structs.Index
 		newConf := *config
 		newConf.ClusterID = newActiveRoot.ExternalTrustDomain
 
+		// Persist any state the provider needs us to
+		newConf.State, err = provider.State()
+		if err != nil {
+			return fmt.Errorf("error getting provider state: %v", err)
+		}
+
 		// Copy the root list and append the new active root, updating the old root
 		// with the time it was rotated out.
 		var newRoots structs.CARoots
@@ -766,7 +791,7 @@ func (s *Server) initializeSecondaryProvider(provider ca.Provider, roots structs
 		return err
 	}
 
-	if err := provider.Configure(clusterID, false, conf.Config); err != nil {
+	if err := provider.Configure(clusterID, false, conf.Config, conf.State); err != nil {
 		return fmt.Errorf("error configuring provider: %v", err)
 	}
 
diff --git a/agent/consul/leader_connect_test.go b/agent/consul/leader_connect_test.go
index 0b68e6f9b4..0c161fd9bf 100644
--- a/agent/consul/leader_connect_test.go
+++ b/agent/consul/leader_connect_test.go
@@ -35,6 +35,9 @@ func TestLeader_SecondaryCA_Initialize(t *testing.T) {
 		{"rsa", 2048},
 	}
 
+	dc1State := map[string]string{"foo": "dc1-value"}
+	dc2State := map[string]string{"foo": "dc2-value"}
+
 	for _, tc := range tests {
 		tc := tc
 		t.Run(fmt.Sprintf("%s-%d", tc.keyType, tc.keyBits), func(t *testing.T) {
@@ -50,6 +53,7 @@ func TestLeader_SecondaryCA_Initialize(t *testing.T) {
 				c.ACLDefaultPolicy = "deny"
 				c.CAConfig.Config["PrivateKeyType"] = tc.keyType
 				c.CAConfig.Config["PrivateKeyBits"] = tc.keyBits
+				c.CAConfig.Config["test_state"] = dc1State
 			})
 			defer os.RemoveAll(dir1)
 			defer s1.Shutdown()
@@ -68,6 +72,7 @@ func TestLeader_SecondaryCA_Initialize(t *testing.T) {
 				c.ACLTokenReplication = true
 				c.CAConfig.Config["PrivateKeyType"] = tc.keyType
 				c.CAConfig.Config["PrivateKeyBits"] = tc.keyBits
+				c.CAConfig.Config["test_state"] = dc2State
 			})
 			defer os.RemoveAll(dir2)
 			defer s2.Shutdown()
@@ -137,6 +142,22 @@ func TestLeader_SecondaryCA_Initialize(t *testing.T) {
 			// Check that the leaf signed by the new cert can be verified using the
 			// returned cert chain (signed intermediate + remote root).
 			require.NoError(t, connect.ValidateLeaf(caRoot.RootCert, leafPEM, []string{intermediatePEM}))
+
+			// Verify that both primary and secondary persisted state as expected -
+			// pass through from the config.
+			{
+				state := s1.fsm.State()
+				_, caConfig, err := state.CAConfig(nil)
+				require.NoError(t, err)
+				require.Equal(t, dc1State, caConfig.State)
+			}
+			{
+				state := s2.fsm.State()
+				_, caConfig, err := state.CAConfig(nil)
+				require.NoError(t, err)
+				require.Equal(t, dc2State, caConfig.State)
+			}
+
 		})
 	}
 }
@@ -183,8 +204,8 @@ func TestLeader_SecondaryCA_IntermediateRefresh(t *testing.T) {
 	}
 
 	// Wait for current state to be reflected in both datacenters.
-	waitForActiveCARoot(t, s1, "dc1", originalRoot)
-	waitForActiveCARoot(t, s2, "dc2", originalRoot)
+	testrpc.WaitForActiveCARoot(t, s1.RPC, "dc1", originalRoot)
+	testrpc.WaitForActiveCARoot(t, s2.RPC, "dc2", originalRoot)
 
 	// Update the provider config to use a new private key, which should
 	// cause a rotation.
@@ -227,8 +248,8 @@ func TestLeader_SecondaryCA_IntermediateRefresh(t *testing.T) {
 	})
 	require.NoError(err)
 
-	waitForActiveCARoot(t, s1, "dc1", updatedRoot)
-	waitForActiveCARoot(t, s2, "dc2", updatedRoot)
+	testrpc.WaitForActiveCARoot(t, s1.RPC, "dc1", updatedRoot)
+	testrpc.WaitForActiveCARoot(t, s2.RPC, "dc2", updatedRoot)
 
 	// Verify the root lists have been rotated in each DC's state store.
 	state1 := s1.fsm.State()
@@ -602,32 +623,6 @@ func TestLeader_SecondaryCA_UpgradeBeforePrimary(t *testing.T) {
 	require.NoError(t, err)
 }
 
-func waitForActiveCARoot(t *testing.T, s *Server, datacenter string, expect *structs.CARoot) {
-	retry.Run(t, func(r *retry.R) {
-		args := &structs.DCSpecificRequest{
-			Datacenter: datacenter,
-		}
-		var reply structs.IndexedCARoots
-		if err := s.RPC("ConnectCA.Roots", args, &reply); err != nil {
-			r.Fatalf("err: %v", err)
-		}
-
-		var root *structs.CARoot
-		for _, r := range reply.Roots {
-			if r.ID == reply.ActiveRootID {
-				root = r
-				break
-			}
-		}
-		if root == nil {
-			r.Fatal("no active root")
-		}
-		if root.ID != expect.ID {
-			r.Fatalf("current active root is %s; waiting for %s", root.ID, expect.ID)
-		}
-	})
-}
-
 func getTestRoots(s *Server, datacenter string) (*structs.IndexedCARoots, *structs.CARoot, error) {
 	rootReq := &structs.DCSpecificRequest{
 		Datacenter: datacenter,
diff --git a/agent/structs/connect_ca.go b/agent/structs/connect_ca.go
index f458ef310e..5fc7c96f10 100644
--- a/agent/structs/connect_ca.go
+++ b/agent/structs/connect_ca.go
@@ -5,6 +5,8 @@ import (
 	"reflect"
 	"time"
 
+	"github.com/hashicorp/consul/lib"
+	"github.com/hashicorp/go-msgpack/codec"
 	"github.com/mitchellh/mapstructure"
 )
 
@@ -228,9 +230,67 @@ type CAConfiguration struct {
 	// and maps).
 	Config map[string]interface{}
 
+	// State is optionally used by the provider to persist information it needs
+	// between reloads like UUIDs of resources it manages. It only supports string
+	// values to avoid gotchas with interface{} since this is encoded through
+	// msgpack when it's written through raft. For example if providers used a
+	// custom struct or even a simple `int` type, msgpack with loose type
+	// information during encode/decode and providers will end up getting back
+	// different types have have to remember to test multiple variants of state
+	// handling to account for cases where it's been through msgpack or not.
+	// Keeping this as strings only forces compatibility and leaves the input
+	// Providers have to work with unambiguous - they can parse ints or other
+	// types as they need. We expect this only to be used to store a handful of
+	// identifiers anyway so this is simpler.
+	State map[string]string
+
 	RaftIndex
 }
 
+// MarshalBinary writes CAConfiguration as msgpack encoded. It's only here
+// because we need custom decoding of the raw interface{} values and this
+// completes the interface.
+func (c *CAConfiguration) MarshalBinary() (data []byte, err error) {
+	// bs will grow if needed but allocate enough to avoid reallocation in common
+	// case.
+	bs := make([]byte, 128)
+	enc := codec.NewEncoderBytes(&bs, msgpackHandle)
+
+	type Alias CAConfiguration
+
+	if err := enc.Encode((*Alias)(c)); err != nil {
+		return nil, err
+	}
+
+	return bs, nil
+}
+
+// UnmarshalBinary decodes msgpack encoded CAConfiguration. It used
+// default msgpack encoding but fixes up the uint8 strings and other problems we
+// have with encoding map[string]interface{}.
+func (c *CAConfiguration) UnmarshalBinary(data []byte) error {
+	dec := codec.NewDecoderBytes(data, msgpackHandle)
+
+	type Alias CAConfiguration
+	var a Alias
+
+	if err := dec.Decode(&a); err != nil {
+		return err
+	}
+
+	*c = CAConfiguration(a)
+
+	var err error
+
+	// Fix strings and maps in the returned maps
+	c.Config, err = lib.MapWalk(c.Config)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
 func (c *CAConfiguration) GetCommonConfig() (*CommonCAProviderConfig, error) {
 	if c == nil {
 		return nil, fmt.Errorf("config map was nil")
diff --git a/testrpc/wait.go b/testrpc/wait.go
index 496fc43523..8e3e616122 100644
--- a/testrpc/wait.go
+++ b/testrpc/wait.go
@@ -111,3 +111,34 @@ func WaitForTestAgent(t *testing.T, rpc rpcFn, dc string, options ...waitOption)
 		}
 	})
 }
+
+// WaitForActiveCARoot polls until the server returns an active Connect root CA
+// with the same ID field as expect. If expect is nil, it just waits until _any_
+// active root is returned. This is useful because initializing CA happens after
+// raft leadership is gained so WaitForLeader isn't sufficient to be sure that
+// the CA is fully initialized.
+func WaitForActiveCARoot(t *testing.T, rpc rpcFn, dc string, expect *structs.CARoot) {
+	retry.Run(t, func(r *retry.R) {
+		args := &structs.DCSpecificRequest{
+			Datacenter: dc,
+		}
+		var reply structs.IndexedCARoots
+		if err := rpc("ConnectCA.Roots", args, &reply); err != nil {
+			r.Fatalf("err: %v", err)
+		}
+
+		var root *structs.CARoot
+		for _, r := range reply.Roots {
+			if r.ID == reply.ActiveRootID {
+				root = r
+				break
+			}
+		}
+		if root == nil {
+			r.Fatal("no active root")
+		}
+		if expect != nil && root.ID != expect.ID {
+			r.Fatalf("current active root is %s; waiting for %s", root.ID, expect.ID)
+		}
+	})
+}