postgresql_database: Reassign objects owners if database owner changes (#458)

Hi,

this is my first Pull Request, so if something is missing or wrong, let
me know.

To resolve
[439](#439)
I suggest to add an additional parameter to the database resource called
`alter_object_ownership`. If set to true, it will issue a REASSIGN OWNED
BY query on the database when the owner changes, transfer the whole
ownership in the database from the previous owner to the new one.

As far as I can tell this is only possible if the user in the provider
configuration is a superuser or a user which has grants for both the
previous owner role and the new one. So I skipped the test in the
rds-like test suite.

---------

Co-authored-by: Lukas Albani <lukas.albani@deutschebahn.com>
Co-authored-by: Cyril Gaudin <cyril.gaudin@gmail.com>

Author: 3 people

postgresql/resource_postgresql_database.go

@@ -14,16 +14,17 @@ import (
 )
 
 const (
-	dbAllowConnsAttr = "allow_connections"
-	dbCTypeAttr      = "lc_ctype"
-	dbCollationAttr  = "lc_collate"
-	dbConnLimitAttr  = "connection_limit"
-	dbEncodingAttr   = "encoding"
-	dbIsTemplateAttr = "is_template"
-	dbNameAttr       = "name"
-	dbOwnerAttr      = "owner"
-	dbTablespaceAttr = "tablespace_name"
-	dbTemplateAttr   = "template"
+	dbAllowConnsAttr       = "allow_connections"
+	dbCTypeAttr            = "lc_ctype"
+	dbCollationAttr        = "lc_collate"
+	dbConnLimitAttr        = "connection_limit"
+	dbEncodingAttr         = "encoding"
+	dbIsTemplateAttr       = "is_template"
+	dbNameAttr             = "name"
+	dbOwnerAttr            = "owner"
+	dbTablespaceAttr       = "tablespace_name"
+	dbTemplateAttr         = "template"
+	dbAlterObjectOwnership = "alter_object_ownership"
 )
 
 func resourcePostgreSQLDatabase() *schema.Resource {
@@ -102,6 +103,12 @@ func resourcePostgreSQLDatabase() *schema.Resource {
 				Computed:    true,
 				Description: "If true, then this database can be cloned by any user with CREATEDB privileges",
 			},
+			dbAlterObjectOwnership: {
+				Type:        schema.TypeBool,
+				Optional:    true,
+				Default:     false,
+				Description: "If true, the owner of already existing objects will change if the owner changes",
+			},
 		},
 	}
 }
@@ -393,6 +400,10 @@ func resourcePostgreSQLDatabaseUpdate(db *DBConnection, d *schema.ResourceData)
 		return err
 	}
 
+	if err := setAlterOwnership(db, d); err != nil {
+		return err
+	}
+
 	if err := setDBOwner(db, d); err != nil {
 		return err
 	}
@@ -468,6 +479,7 @@ func setDBOwner(db *DBConnection, d *schema.ResourceData) error {
 	}
 
 	dbName := d.Get(dbNameAttr).(string)
+
 	sql := fmt.Sprintf("ALTER DATABASE %s OWNER TO %s", pq.QuoteIdentifier(dbName), pq.QuoteIdentifier(owner))
 	if _, err := db.Exec(sql); err != nil {
 		return fmt.Errorf("Error updating database OWNER: %w", err)
@@ -476,6 +488,60 @@ func setDBOwner(db *DBConnection, d *schema.ResourceData) error {
 	return err
 }
 
+func setAlterOwnership(db *DBConnection, d *schema.ResourceData) error {
+	if !d.HasChange(dbOwnerAttr) && !d.HasChange(dbAlterObjectOwnership) {
+		return nil
+	}
+	owner := d.Get(dbOwnerAttr).(string)
+	if owner == "" {
+		return nil
+	}
+
+	alterOwnership := d.Get(dbAlterObjectOwnership).(bool)
+	if !alterOwnership {
+		return nil
+	}
+	currentUser := db.client.config.getDatabaseUsername()
+
+	dbName := d.Get(dbNameAttr).(string)
+
+	lockTxn, err := startTransaction(db.client, dbName)
+	if err := pgLockRole(lockTxn, currentUser); err != nil {
+		return err
+	}
+	defer deferredRollback(lockTxn)
+
+	currentOwner, err := getDatabaseOwner(db, dbName)
+	if err != nil {
+		return fmt.Errorf("Error getting current database OWNER: %w", err)
+	}
+
+	newOwner := d.Get(dbOwnerAttr).(string)
+
+	if currentOwner == newOwner {
+		return nil
+	}
+
+	currentOwnerGranted, err := grantRoleMembership(db, currentOwner, currentUser)
+	if err != nil {
+		return err
+	}
+	if currentOwnerGranted {
+		defer func() {
+			_, err = revokeRoleMembership(db, currentOwner, currentUser)
+		}()
+	}
+	sql := fmt.Sprintf("REASSIGN OWNED BY %s TO %s", pq.QuoteIdentifier(currentOwner), pq.QuoteIdentifier(newOwner))
+	if _, err := lockTxn.Exec(sql); err != nil {
+		return fmt.Errorf("Error reassigning objects owned by '%s': %w", currentOwner, err)
+	}
+
+	if err := lockTxn.Commit(); err != nil {
+		return fmt.Errorf("error committing reassign: %w", err)
+	}
+	return nil
+}
+
 func setDBTablespace(db QueryAble, d *schema.ResourceData) error {
 	if !d.HasChange(dbTablespaceAttr) {
 		return nil

postgresql/resource_postgresql_database_test.go

@@ -43,6 +43,8 @@ func TestAccPostgresqlDatabase_Basic(t *testing.T) {
 						"postgresql_database.default_opts", "connection_limit", "-1"),
 					resource.TestCheckResourceAttr(
 						"postgresql_database.default_opts", "is_template", "false"),
+					resource.TestCheckResourceAttr(
+						"postgresql_database.default_opts", "alter_object_ownership", "false"),
 
 					resource.TestCheckResourceAttr(
 						"postgresql_database.modified_opts", "owner", "myrole"),
@@ -62,6 +64,8 @@ func TestAccPostgresqlDatabase_Basic(t *testing.T) {
 						"postgresql_database.modified_opts", "connection_limit", "10"),
 					resource.TestCheckResourceAttr(
 						"postgresql_database.modified_opts", "is_template", "true"),
+					resource.TestCheckResourceAttr(
+						"postgresql_database.modified_opts", "alter_object_ownership", "true"),
 
 					resource.TestCheckResourceAttr(
 						"postgresql_database.pathological_opts", "owner", "myrole"),
@@ -266,6 +270,78 @@ resource postgresql_database "test_db" {
 	})
 }
 
+// Test the case where the owned objects by the previous database owner are altered.
+func TestAccPostgresqlDatabase_AlterObjectOwnership(t *testing.T) {
+	skipIfNotAcc(t)
+
+	const (
+		databaseSuffix = "ownership"
+		tableName      = "testtable1"
+		previous_owner = "previous_owner"
+		new_owner      = "new_owner"
+	)
+
+	databaseName := fmt.Sprintf("%s_%s", dbNamePrefix, databaseSuffix)
+
+	config := getTestConfig(t)
+	dsn := config.connStr("postgres")
+
+	for _, role := range []string{previous_owner, new_owner} {
+		dbExecute(
+			t, dsn,
+			fmt.Sprintf("CREATE ROLE %s;", role),
+		)
+		defer func(role string) {
+			dbExecute(t, dsn, fmt.Sprintf("DROP ROLE %s", role))
+		}(role)
+
+	}
+
+	resource.Test(t, resource.TestCase{
+		PreCheck: func() {
+			testAccPreCheck(t)
+			testSuperuserPreCheck(t)
+		},
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckPostgresqlDatabaseDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: `
+resource postgresql_database "test_db" {
+       name  = "tf_tests_db_ownership"
+       owner = "previous_owner"
+	   alter_object_ownership = true
+}
+`,
+				Check: func(*terraform.State) error {
+					// To test default privileges, we need to create a table
+					// after having apply the state.
+					_ = createTestTables(t, databaseSuffix, []string{tableName}, previous_owner)
+					return nil
+				},
+			},
+			{
+				Config: `
+resource postgresql_database "test_db" {
+       name  = "tf_tests_db_ownership"
+       owner = "new_owner"
+	   alter_object_ownership = true
+}
+`,
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckPostgresqlDatabaseExists("postgresql_database.test_db"),
+					resource.TestCheckResourceAttr("postgresql_database.test_db", "name", databaseName),
+					resource.TestCheckResourceAttr("postgresql_database.test_db", "owner", new_owner),
+					resource.TestCheckResourceAttr("postgresql_database.test_db", "alter_object_ownership", "true"),
+
+					checkTableOwnership(t, config.connStr(databaseName), new_owner, tableName),
+				),
+			},
+		},
+	})
+
+}
+
 func checkUserMembership(
 	t *testing.T, dsn, member, role string, shouldHaveRole bool,
 ) resource.TestCheckFunc {
@@ -306,6 +382,38 @@ func checkUserMembership(
 	}
 }
 
+func checkTableOwnership(
+	t *testing.T, dsn, owner, tableName string,
+) resource.TestCheckFunc {
+	return func(s *terraform.State) error {
+		db, err := sql.Open("postgres", dsn)
+		if err != nil {
+			t.Fatalf("could not create connection pool: %v", err)
+		}
+		defer db.Close()
+
+		var _rez int
+
+		err = db.QueryRow(`
+			SELECT 1 FROM pg_tables
+			WHERE tablename = $1 AND tableowner = $2
+		`, tableName, owner).Scan(&_rez)
+
+		switch {
+		case err == sql.ErrNoRows:
+			return fmt.Errorf(
+				"User %s should be owner of %s but is not", owner, tableName,
+			)
+		case err != nil:
+			t.Fatalf("Error checking table ownership. %v", err)
+
+		}
+
+		return nil
+
+	}
+}
+
 func testAccCheckPostgresqlDatabaseDestroy(s *terraform.State) error {
 	client := testAccProvider.Meta().(*Client)
 
@@ -396,6 +504,7 @@ resource "postgresql_database" "default_opts" {
    lc_ctype = "C"
    connection_limit = -1
    is_template = false
+   alter_object_ownership = false
 }
 
 resource "postgresql_database" "modified_opts" {
@@ -407,6 +516,7 @@ resource "postgresql_database" "modified_opts" {
    lc_ctype = "en_US.UTF-8"
    connection_limit = 10
    is_template = true
+   alter_object_ownership = true
 }
 
 resource "postgresql_database" "pathological_opts" {

website/docs/r/postgresql_database.html.markdown

@@ -17,12 +17,13 @@ within a PostgreSQL server instance.
 
 ```hcl
 resource "postgresql_database" "my_db" {
-  name              = "my_db"
-  owner             = "my_role"
-  template          = "template0"
-  lc_collate        = "C"
-  connection_limit  = -1
-  allow_connections = true
+  name                   = "my_db"
+  owner                  = "my_role"
+  template               = "template0"
+  lc_collate             = "C"
+  connection_limit       = -1
+  allow_connections      = true
+  alter_object_ownership = true
 }
 ```
 
@@ -82,6 +83,14 @@ resource "postgresql_database" "my_db" {
   force the creation of a new resource as this value can only be changed when a
   database is created.
 
+* `alter_object_ownership` - (Optional) If `true`, the change of the database
+  `owner` will also include a reassignment of the ownership of preexisting
+  objects like tables or sequences from the previous owner to the new one.
+  If set to `false` (the default), then the previous database `owner` will still
+  hold the ownership of the objects in that database. To alter existing objects in
+  the database, you must be a direct or indirect member of the specified role, or
+  the username in the provider must be superuser.
+
 ## Import Example
 
 `postgresql_database` supports importing resources.  Supposing the following