Fix deadlock in root credential rotation (#4309)

* fix deadlock in root credential rotation

* add more logging of errors

* adding cassandra test

builtin/logical/database/backend.go

@@ -249,7 +249,7 @@ func (b *databaseBackend) CloseIfShutdown(db *dbPluginInstance, err error) {
 	case rpc.ErrShutdown, dbplugin.ErrPluginShutdown:
 		// Put this in a goroutine so that requests can run with the read or write lock
 		// and simply defer the unlock.  Since we are attaching the instance and matching
-		// the id in the conneciton map, we can safely do this.
+		// the id in the connection map, we can safely do this.
 		go func() {
 			b.Lock()
 			defer b.Unlock()

builtin/logical/database/backend_test.go

@@ -927,6 +927,107 @@ func TestBackend_allowedRoles(t *testing.T) {
 	}
 }
 
+func TestBackend_RotateRootCredentials(t *testing.T) {
+	cluster, sys := getCluster(t)
+	defer cluster.Cleanup()
+
+	config := logical.TestBackendConfig()
+	config.StorageView = &logical.InmemStorage{}
+	config.System = sys
+
+	b, err := Factory(context.Background(), config)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer b.Cleanup(context.Background())
+
+	cleanup, connURL := preparePostgresTestContainer(t, config.StorageView, b)
+	defer cleanup()
+
+	connURL = strings.Replace(connURL, "postgres:secret", "{{username}}:{{password}}", -1)
+
+	// Configure a connection
+	data := map[string]interface{}{
+		"connection_url": connURL,
+		"plugin_name":    "postgresql-database-plugin",
+		"allowed_roles":  []string{"plugin-role-test"},
+		"username":       "postgres",
+		"password":       "secret",
+	}
+	req := &logical.Request{
+		Operation: logical.UpdateOperation,
+		Path:      "config/plugin-test",
+		Storage:   config.StorageView,
+		Data:      data,
+	}
+	resp, err := b.HandleRequest(context.Background(), req)
+	if err != nil || (resp != nil && resp.IsError()) {
+		t.Fatalf("err:%s resp:%#v\n", err, resp)
+	}
+
+	// Create a role
+	data = map[string]interface{}{
+		"db_name":             "plugin-test",
+		"creation_statements": testRole,
+		"max_ttl":             "10m",
+	}
+	req = &logical.Request{
+		Operation: logical.UpdateOperation,
+		Path:      "roles/plugin-role-test",
+		Storage:   config.StorageView,
+		Data:      data,
+	}
+	resp, err = b.HandleRequest(context.Background(), req)
+	if err != nil || (resp != nil && resp.IsError()) {
+		t.Fatalf("err:%s resp:%#v\n", err, resp)
+	}
+	// Get creds
+	data = map[string]interface{}{}
+	req = &logical.Request{
+		Operation: logical.ReadOperation,
+		Path:      "creds/plugin-role-test",
+		Storage:   config.StorageView,
+		Data:      data,
+	}
+	credsResp, err := b.HandleRequest(context.Background(), req)
+	if err != nil || (credsResp != nil && credsResp.IsError()) {
+		t.Fatalf("err:%s resp:%#v\n", err, credsResp)
+	}
+
+	data = map[string]interface{}{}
+	req = &logical.Request{
+		Operation: logical.UpdateOperation,
+		Path:      "rotate-root/plugin-test",
+		Storage:   config.StorageView,
+		Data:      data,
+	}
+	resp, err = b.HandleRequest(context.Background(), req)
+	if err != nil || (credsResp != nil && credsResp.IsError()) {
+		t.Fatalf("err:%s resp:%#v\n", err, credsResp)
+	}
+
+	dbConfig, err := b.(*databaseBackend).DatabaseConfig(context.Background(), config.StorageView, "plugin-test")
+	if err != nil {
+		t.Fatalf("err: %#v", err)
+	}
+	if dbConfig.ConnectionDetails["password"].(string) == "secret" {
+		t.Fatal("root credentials not rotated")
+	}
+
+	// Get creds to make sure it still works
+	data = map[string]interface{}{}
+	req = &logical.Request{
+		Operation: logical.ReadOperation,
+		Path:      "creds/plugin-role-test",
+		Storage:   config.StorageView,
+		Data:      data,
+	}
+	credsResp, err = b.HandleRequest(context.Background(), req)
+	if err != nil || (credsResp != nil && credsResp.IsError()) {
+		t.Fatalf("err:%s resp:%#v\n", err, credsResp)
+	}
+}
+
 func testCredsExist(t *testing.T, resp *logical.Response, connURL string) bool {
 	t.Helper()
 	var d struct {

builtin/logical/database/path_rotate_credentials.go

@@ -19,7 +19,7 @@ func pathRotateCredentials(b *databaseBackend) *framework.Path {
 		},
 
 		Callbacks: map[logical.Operation]framework.OperationFunc{
-			logical.ReadOperation: b.pathRotateCredentialsUpdate(),
+			logical.UpdateOperation: b.pathRotateCredentialsUpdate(),
 		},
 
 		HelpSynopsis:    pathCredsCreateReadHelpSyn,
@@ -44,8 +44,11 @@ func (b *databaseBackend) pathRotateCredentialsUpdate() framework.OperationFunc
 			return nil, err
 		}
 
-		// Take the write lock instead of read since we are updating the
-		// connection
+		// Take out the backend lock since we are swapping out the connection
+		b.Lock()
+		defer b.Unlock()
+
+		// Take the write lock on the instance
 		db.Lock()
 		defer db.Unlock()
 
@@ -63,9 +66,13 @@ func (b *databaseBackend) pathRotateCredentialsUpdate() framework.OperationFunc
 			return nil, err
 		}
 
-		if err := b.ClearConnection(name); err != nil {
-			return nil, err
+		// Close the plugin
+		db.closed = true
+		if err := db.Database.Close(); err != nil {
+			b.Logger().Error("error closing the database plugin connection", "err", err)
 		}
+		// Even on error, still remove the connection
+		delete(b.connections, name)
 
 		return nil, nil
 	}

plugins/database/cassandra/cassandra_test.go

@@ -252,6 +252,45 @@ func TestCassandra_RevokeUser(t *testing.T) {
 	}
 }
 
+func TestCassandra_RotateRootCredentials(t *testing.T) {
+	cleanup, address, port := prepareCassandraTestContainer(t)
+	defer cleanup()
+
+	connectionDetails := map[string]interface{}{
+		"hosts":            address,
+		"port":             port,
+		"username":         "cassandra",
+		"password":         "cassandra",
+		"protocol_version": 4,
+	}
+
+	db := new()
+
+	connProducer := db.cassandraConnectionProducer
+
+	_, err := db.Init(context.Background(), connectionDetails, true)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	if !connProducer.Initialized {
+		t.Fatal("Database should be initialized")
+	}
+
+	newConf, err := db.RotateRootCredentials(context.Background(), nil)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	if newConf["password"] == "cassandra" {
+		t.Fatal("password was not updated")
+	}
+
+	err = db.Close()
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+}
+
 func testCredsExist(t testing.TB, address string, port int, username, password string) error {
 	clusterConfig := gocql.NewCluster(address)
 	clusterConfig.Authenticator = gocql.PasswordAuthenticator{

plugins/database/cassandra/test-fixtures/cassandra.yaml

@@ -572,7 +572,7 @@ ssl_storage_port: 7001
 #
 # Setting listen_address to 0.0.0.0 is always wrong.
 #
-listen_address: 172.17.0.2
+listen_address: 172.17.0.6
 
 # Set listen_address OR listen_interface, not both. Interfaces must correspond
 # to a single address, IP aliasing is not supported.