[DPE-7500] Create catalog/database level roles

src/charm.py

@@ -1712,7 +1712,7 @@ def _start_primary(self, event: StartEvent) -> None:  # noqa: C901
             return
 
         try:
-            self.postgresql.create_predefined_roles()
+            self.postgresql.create_predefined_instance_roles()
         except PostgreSQLCreatePredefinedRolesError as e:
             logger.exception(e)
             self.unit.status = BlockedStatus("Failed to create pre-defined roles")
@@ -1724,8 +1724,6 @@ def _start_primary(self, event: StartEvent) -> None:  # noqa: C901
             # This event can be run on a replica if the machines are restarted.
             # For that case, check whether the postgres user already exits.
             users = self.postgresql.list_users()
-            if "postgres" not in users:
-                self.postgresql.create_user("postgres", new_password(), admin=True)
             # Create the backup user.
             if BACKUP_USER not in users:
                 self.postgresql.create_user(

src/relations/postgresql_provider.py

@@ -110,12 +110,13 @@ def _on_database_requested(self, event: DatabaseRequestedEvent) -> None:
         try:
             # Creates the user and the database for this specific relation.
             user = f"relation-{event.relation.id}"
-            password = new_password()
-            self.charm.postgresql.create_user(user, password, extra_user_roles=extra_user_roles)
             plugins = self.charm.get_plugins()
 
-            self.charm.postgresql.create_database(
-                database, user, plugins=plugins, client_relations=self.charm.client_relations
+            self.charm.postgresql.create_database(database, plugins=plugins)
+
+            password = new_password()
+            self.charm.postgresql.create_user(
+                user, password, extra_user_roles=extra_user_roles, in_role=f"{database}_admin"
             )
 
             # Share the credentials with the application.

templates/patroni.yml.j2

@@ -96,7 +96,11 @@ bootstrap:
         log_truncate_on_rotation: 'on'
         logging_collector: 'on'
         wal_level: logical
-        shared_preload_libraries: 'timescaledb,pgaudit'
+        shared_preload_libraries: 'timescaledb,pgaudit,set_user'
+        session_preload_libraries: 'login_hook'
+        set_user.block_log_statement: 'on'
+        set_user.exit_on_error: 'on'
+        set_user.superuser_allowlist: '+charmed_dba'
 
   {%- if restoring_backup %}
   method: pgbackrest
@@ -132,7 +136,11 @@ postgresql:
   bin_dir: /snap/charmed-postgresql/current/usr/lib/postgresql/{{ version }}/bin
   data_dir: {{ data_path }}
   parameters:
-    shared_preload_libraries: 'timescaledb,pgaudit'
+    shared_preload_libraries: 'timescaledb,pgaudit,set_user'
+    session_preload_libraries: 'login_hook'
+    set_user.block_log_statement: 'on'
+    set_user.exit_on_error: 'on'
+    set_user.superuser_allowlist: '+charmed_dba'
     {%- if enable_pgbackrest_archiving %}
     archive_command: 'pgbackrest {{ pgbackrest_configuration_file }} --stanza={{ stanza }} archive-push %p'
     {% else %}

tests/integration/helpers.py

@@ -14,6 +14,7 @@
 
 import botocore
 import psycopg2
+import pytest
 import requests
 import yaml
 from juju.model import Model
@@ -37,6 +38,7 @@
 DATABASE_APP_NAME = METADATA["name"]
 STORAGE_PATH = METADATA["storage"]["data"]["location"]
 APPLICATION_NAME = "postgresql-test-app"
+DATA_INTEGRATOR_APP_NAME = "data-integrator"
 
 
 class SecretNotFoundError(Exception):
@@ -737,6 +739,80 @@ def get_unit_address(ops_test: OpsTest, unit_name: str, model: Model = None) ->
     return model.units.get(unit_name).public_address
 
 
+def check_connected_user(
+    cursor, session_user: str, current_user: str, primary: bool = True
+) -> None:
+    cursor.execute("SELECT session_user,current_user;")
+    result = cursor.fetchone()
+    if result is not None:
+        instance = "primary" if primary else "replica"
+        assert result[0] == session_user, (
+            f"The session user should be the {session_user} user in the {instance}"
+        )
+        assert result[1] == current_user, (
+            f"The current user should be the {current_user} user in the {instance}"
+        )
+    else:
+        assert False, "No result returned from the query"
+
+
+async def check_roles_and_their_permissions(
+    ops_test: OpsTest, relation_endpoint: str, database_name: str
+) -> None:
+    action = await ops_test.model.units[f"{DATA_INTEGRATOR_APP_NAME}/0"].run_action(
+        action_name="get-credentials"
+    )
+    result = await action.wait()
+    data_integrator_credentials = result.results
+    username = data_integrator_credentials[relation_endpoint]["username"]
+    uris = data_integrator_credentials[relation_endpoint]["uris"]
+    connection = None
+    try:
+        connection = psycopg2.connect(uris)
+        connection.autocommit = True
+        with connection.cursor() as cursor:
+            logger.info(
+                "Checking that the relation user is automatically escalated to the database owner user"
+            )
+            check_connected_user(cursor, username, f"{database_name}_owner")
+            logger.info("Creating a test table and inserting data")
+            cursor.execute("CREATE TABLE test_table (id INTEGER);")
+            logger.info("Inserting data into the test table")
+            cursor.execute("INSERT INTO test_table(id) VALUES(1);")
+            logger.info("Reading data from the test table")
+            cursor.execute("SELECT * FROM test_table;")
+            result = cursor.fetchall()
+            assert len(result) == 1, "The database owner user should be able to read the data"
+
+            logger.info("Checking that the database owner user can't create a database")
+            with pytest.raises(psycopg2.errors.InsufficientPrivilege):
+                cursor.execute(f"CREATE DATABASE {database_name}_2;")
+
+            logger.info("Checking that the relation user can't create a table")
+            cursor.execute("RESET ROLE;")
+            check_connected_user(cursor, username, username)
+            with pytest.raises(psycopg2.errors.InsufficientPrivilege):
+                cursor.execute("CREATE TABLE test_table_2 (id INTEGER);")
+    finally:
+        if connection is not None:
+            connection.close()
+
+    connection_string = f"host={data_integrator_credentials[relation_endpoint]['read-only-endpoints'].split(':')[0]} dbname={data_integrator_credentials[relation_endpoint]['database']} user={username} password={data_integrator_credentials[relation_endpoint]['password']}"
+    connection = None
+    try:
+        connection = psycopg2.connect(connection_string)
+        with connection.cursor() as cursor:
+            logger.info("Checking that the relation user can read data from the database")
+            check_connected_user(cursor, username, username, primary=False)
+            logger.info("Reading data from the test table")
+            cursor.execute("SELECT * FROM test_table;")
+            result = cursor.fetchall()
+            assert len(result) == 1, "The relation user should be able to read the data"
+    finally:
+        if connection is not None:
+            connection.close()
+
+
 async def check_tls(ops_test: OpsTest, unit_name: str, enabled: bool) -> bool:
     """Returns whether TLS is enabled on the specific PostgreSQL instance.
 
@@ -900,6 +976,14 @@ async def primary_changed(ops_test: OpsTest, old_primary: str) -> bool:
     return primary != old_primary
 
 
+def relations(ops_test: OpsTest, provider_app: str, requirer_app: str) -> list:
+    return [
+        relation
+        for relation in ops_test.model.applications[provider_app].relations
+        if not relation.is_peer and relation.requires.application_name == requirer_app
+    ]
+
+
 async def restart_machine(ops_test: OpsTest, unit_name: str) -> None:
     """Restart the machine where a unit run on.
 

tests/integration/new_relations/test_new_relations_1.py

@@ -2,8 +2,6 @@
 # See LICENSE file for licensing details.
 import asyncio
 import logging
-import secrets
-import string
 from pathlib import Path
 
 import psycopg2
@@ -484,97 +482,10 @@ async def test_relation_with_no_database_name(ops_test: OpsTest):
         await ops_test.model.wait_for_idle(apps=APP_NAMES, status="active", raise_on_blocked=True)
 
 
-async def test_admin_role(ops_test: OpsTest):
-    """Test that the admin role gives access to all the databases."""
-    all_app_names = [DATA_INTEGRATOR_APP_NAME]
-    all_app_names.extend(APP_NAMES)
+async def test_invalid_extra_user_roles(ops_test: OpsTest):
     async with ops_test.fast_forward():
         await ops_test.model.deploy(DATA_INTEGRATOR_APP_NAME, base=CHARM_BASE)
         await ops_test.model.wait_for_idle(apps=[DATA_INTEGRATOR_APP_NAME], status="blocked")
-        await ops_test.model.applications[DATA_INTEGRATOR_APP_NAME].set_config({
-            "database-name": DATA_INTEGRATOR_APP_NAME.replace("-", "_"),
-            "extra-user-roles": "admin",
-        })
-        await ops_test.model.wait_for_idle(apps=[DATA_INTEGRATOR_APP_NAME], status="blocked")
-        await ops_test.model.add_relation(DATA_INTEGRATOR_APP_NAME, DATABASE_APP_NAME)
-        await ops_test.model.wait_for_idle(apps=all_app_names, status="active")
-
-    # Check that the user can access all the databases.
-    for database in [
-        DATABASE_DEFAULT_NAME,
-        f"{APPLICATION_APP_NAME.replace('-', '_')}_database",
-        "another_application_database",
-    ]:
-        logger.info(f"connecting to the following database: {database}")
-        connection_string = await build_connection_string(
-            ops_test, DATA_INTEGRATOR_APP_NAME, "postgresql", database=database
-        )
-        connection = None
-        should_fail = False
-        try:
-            with psycopg2.connect(connection_string) as connection, connection.cursor() as cursor:
-                # Check the version that the application received is the same on the
-                # database server.
-                cursor.execute("SELECT version();")
-                data = cursor.fetchone()[0].split(" ")[1]
-
-                # Get the version of the database and compare with the information that
-                # was retrieved directly from the database.
-                version = await get_application_relation_data(
-                    ops_test, DATA_INTEGRATOR_APP_NAME, "postgresql", "version"
-                )
-                assert version == data
-
-                # Write some data (it should fail in the default database name).
-                random_name = (
-                    f"test_{''.join(secrets.choice(string.ascii_lowercase) for _ in range(10))}"
-                )
-                should_fail = database == DATABASE_DEFAULT_NAME
-                cursor.execute(f"CREATE SCHEMA test; CREATE TABLE test.{random_name}(data TEXT);")
-                if should_fail:
-                    assert False, (
-                        f"failed to run a statement in the following database: {database}"
-                    )
-        except psycopg2.errors.InsufficientPrivilege as e:
-            if not should_fail:
-                logger.exception(e)
-                assert False, (
-                    f"failed to connect to or run a statement in the following database: {database}"
-                )
-        finally:
-            if connection is not None:
-                connection.close()
-
-    # Test the creation and deletion of databases.
-    connection_string = await build_connection_string(
-        ops_test, DATA_INTEGRATOR_APP_NAME, "postgresql", database=DATABASE_DEFAULT_NAME
-    )
-    connection = psycopg2.connect(connection_string)
-    connection.autocommit = True
-    cursor = connection.cursor()
-    random_name = f"test_{''.join(secrets.choice(string.ascii_lowercase) for _ in range(10))}"
-    cursor.execute(f"CREATE DATABASE {random_name};")
-    cursor.execute(f"DROP DATABASE {random_name};")
-    try:
-        cursor.execute(f"DROP DATABASE {DATABASE_DEFAULT_NAME};")
-        assert False, (
-            f"the admin extra user role was able to drop the `{DATABASE_DEFAULT_NAME}` system database"
-        )
-    except psycopg2.errors.InsufficientPrivilege:
-        # Ignore the error, as the admin extra user role mustn't be able to drop
-        # the "postgres" system database.
-        pass
-    finally:
-        connection.close()
-
-
-async def test_invalid_extra_user_roles(ops_test: OpsTest):
-    async with ops_test.fast_forward():
-        # Remove the relation between the database and the first data integrator.
-        await ops_test.model.applications[DATABASE_APP_NAME].remove_relation(
-            DATABASE_APP_NAME, DATA_INTEGRATOR_APP_NAME
-        )
-        await ops_test.model.wait_for_idle(apps=APP_NAMES, status="active", raise_on_blocked=True)
 
         another_data_integrator_app_name = f"another-{DATA_INTEGRATOR_APP_NAME}"
         data_integrator_apps_names = [DATA_INTEGRATOR_APP_NAME, another_data_integrator_app_name]

tests/integration/new_relations/test_relations_coherence.py

@@ -82,40 +82,6 @@ async def test_relations(ops_test: OpsTest, charm):
             f"{DATABASE_APP_NAME}:database", f"{DATA_INTEGRATOR_APP_NAME}:postgresql"
         )
 
-        # Re-relation with admin role with checking permission
-        await ops_test.model.applications[DATA_INTEGRATOR_APP_NAME].set_config({
-            "database-name": DATA_INTEGRATOR_APP_NAME.replace("-", "_"),
-            "extra-user-roles": "admin",
-        })
-        await ops_test.model.wait_for_idle(apps=[DATA_INTEGRATOR_APP_NAME], status="blocked")
-        await ops_test.model.add_relation(DATA_INTEGRATOR_APP_NAME, DATABASE_APP_NAME)
-        await ops_test.model.wait_for_idle(apps=APP_NAMES, status="active")
-
-        connection_string = await build_connection_string(
-            ops_test,
-            DATA_INTEGRATOR_APP_NAME,
-            "postgresql",
-            database=DATA_INTEGRATOR_APP_NAME.replace("-", "_"),
-        )
-        try:
-            connection = psycopg2.connect(connection_string)
-            connection.autocommit = True
-            cursor = connection.cursor()
-            random_name = (
-                f"test_{''.join(secrets.choice(string.ascii_lowercase) for _ in range(10))}"
-            )
-            cursor.execute(f"CREATE DATABASE {random_name};")
-        except psycopg2.errors.InsufficientPrivilege:
-            assert False, (
-                f"failed connect to {random_name} or run a statement in the following database"
-            )
-        finally:
-            connection.close()
-
-        await ops_test.model.applications[DATABASE_APP_NAME].remove_relation(
-            f"{DATABASE_APP_NAME}:database", f"{DATA_INTEGRATOR_APP_NAME}:postgresql"
-        )
-
         # Re-relation again with user role and checking write data
         await ops_test.model.applications[DATA_INTEGRATOR_APP_NAME].set_config({
             "database-name": DATA_INTEGRATOR_APP_NAME.replace("-", "_"),