connect/ca: prevent blank CA config in snapshot

This PR both prevents a blank CA config from being written out to
a snapshot and allows Consul to gracefully recover from a snapshot
with an invalid CA config.

Fixes #4954.

agent/consul/fsm/snapshot_oss.go

@@ -267,18 +267,19 @@ func (s *snapshot) persistPreparedQueries(sink raft.SnapshotSink,
 
 func (s *snapshot) persistAutopilot(sink raft.SnapshotSink,
 	encoder *codec.Encoder) error {
-	autopilot, err := s.state.Autopilot()
+	config, err := s.state.Autopilot()
 	if err != nil {
 		return err
 	}
-	if autopilot == nil {
+	// Make sure we don't write a nil config out to a snapshot.
+	if config == nil {
 		return nil
 	}
 
 	if _, err := sink.Write([]byte{byte(structs.AutopilotRequestType)}); err != nil {
 		return err
 	}
-	if err := encoder.Encode(autopilot); err != nil {
+	if err := encoder.Encode(config); err != nil {
 		return err
 	}
 	return nil
@@ -309,6 +310,10 @@ func (s *snapshot) persistConnectCAConfig(sink raft.SnapshotSink,
 	if err != nil {
 		return err
 	}
+	// Make sure we don't write a nil config out to a snapshot.
+	if config == nil {
+		return nil
+	}
 
 	if _, err := sink.Write([]byte{byte(structs.ConnectCAConfigType)}); err != nil {
 		return err

agent/consul/fsm/snapshot_oss_test.go

@@ -453,3 +453,49 @@ func TestFSM_BadRestore_OSS(t *testing.T) {
 	default:
 	}
 }
+
+func TestFSM_BadSnapshot_NilCAConfig(t *testing.T) {
+	t.Parallel()
+
+	require := require.New(t)
+
+	// Create an FSM with no config entry.
+	fsm, err := New(nil, os.Stderr)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	// Snapshot
+	snap, err := fsm.Snapshot()
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	defer snap.Release()
+
+	// Persist
+	buf := bytes.NewBuffer(nil)
+	sink := &MockSink{buf, false}
+	if err := snap.Persist(sink); err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	// Try to restore on a new FSM
+	fsm2, err := New(nil, os.Stderr)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	// Do a restore
+	if err := fsm2.Restore(sink); err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	// Make sure there's no entry in the CA config table.
+	state := fsm2.State()
+	idx, config, err := state.CAConfig()
+	require.NoError(err)
+	require.Equal(uint64(0), idx)
+	if config != nil {
+		t.Fatalf("config should be nil")
+	}
+}

agent/consul/state/connect_ca.go

@@ -93,6 +93,12 @@ func (s *Snapshot) CAConfig() (*structs.CAConfiguration, error) {
 
 // CAConfig is used when restoring from a snapshot.
 func (s *Restore) CAConfig(config *structs.CAConfiguration) error {
+	// Don't restore a blank CA config
+	// https://github.com/hashicorp/consul/issues/4954
+	if config.Provider == "" {
+		return nil
+	}
+
 	if err := s.tx.Insert(caConfigTableName, config); err != nil {
 		return fmt.Errorf("failed restoring CA config: %s", err)
 	}

agent/consul/state/connect_ca_test.go

@@ -146,6 +146,43 @@ func TestStore_CAConfig_Snapshot_Restore(t *testing.T) {
 	verify.Values(t, "", before, res)
 }
 
+// Make sure we handle the case of a leftover blank CA config that
+// got stuck in a snapshot, as in https://github.com/hashicorp/consul/issues/4954
+func TestStore_CAConfig_Snapshot_Restore_BlankConfig(t *testing.T) {
+	s := testStateStore(t)
+	before := &structs.CAConfiguration{}
+	if err := s.CASetConfig(99, before); err != nil {
+		t.Fatal(err)
+	}
+
+	snap := s.Snapshot()
+	defer snap.Close()
+
+	snapped, err := snap.CAConfig()
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+	verify.Values(t, "", before, snapped)
+
+	s2 := testStateStore(t)
+	restore := s2.Restore()
+	if err := restore.CAConfig(snapped); err != nil {
+		t.Fatalf("err: %s", err)
+	}
+	restore.Commit()
+
+	idx, result, err := s2.CAConfig()
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+	if idx != 0 {
+		t.Fatalf("bad index: %d", idx)
+	}
+	if result != nil {
+		t.Fatalf("should be nil: %v", result)
+	}
+}
+
 func TestStore_CARootSetList(t *testing.T) {
 	assert := assert.New(t)
 	s := testStateStore(t)