Fix LDAP component to use dedicated user instead of operator user

docs/explanation/users.md

@@ -17,6 +17,7 @@ The operator uses the following internal DB users:
 * `rewind` - the internal user for synchronizing a PostgreSQL cluster with another copy of the same cluster.
 * `monitoring` - the user for [COS integration](/how-to/monitoring-cos/enable-monitoring).
 * `backups` - the user to [perform/list/restore backups](/how-to/back-up-and-restore/create-a-backup).
+* `ldap` - the user for [LDAP synchronization](/how-to/enable-ldap) when LDAP integration is enabled. This user has the `CREATEROLE` privilege to manage PostgreSQL roles synchronized from LDAP.
 
 The full list of internal users is available in charm [source code](https://github.com/canonical/postgresql-operator/blob/main/src/constants.py). The full dump of internal users (on the newly installed charm):
 
@@ -26,6 +27,7 @@ postgres=# \du
   Role name  |                         Attributes                         |  Member of   
 -------------+------------------------------------------------------------+--------------
  backup      | Superuser                                                  | {}
+ ldap        | Create role                                                | {}
  monitoring  |                                                            | {pg_monitor}
  operator    | Superuser, Create role, Create DB, Replication, Bypass RLS | {}
  postgres    | Superuser                                                  | {}
@@ -51,7 +53,7 @@ password:
   description: The password will be auto-generated if this option is not specified.
 username:
   type: string
-  description: The username, the default value 'operator'. Possible values - operator, replication, rewind.
+  description: The username, the default value 'operator'. Possible values - operator, replication, rewind, monitoring, backup, ldap.
 ```
 
 For example, to generate a new random password for *internal* user:

docs/how-to/manage-passwords.md

@@ -30,3 +30,5 @@ To set a manual password for another user:
 juju run postgresql/leader set-password username=<username> password=<password>
 ```
 
+Where `<username>` can be any of the system users: `operator`, `replication`, `rewind`, `monitoring`, `backup`, or `ldap`.
+

src/charm.py

@@ -76,6 +76,8 @@
     BACKUP_USER,
     DATABASE_DEFAULT_NAME,
     DATABASE_PORT,
+    LDAP_PASSWORD_KEY,
+    LDAP_USER,
     METRICS_PORT,
     MONITORING_PASSWORD_KEY,
     MONITORING_SNAP_SERVICE,
@@ -1106,6 +1108,7 @@ def _on_leader_elected(self, event: LeaderElectedEvent) -> None:
             REPLICATION_PASSWORD_KEY,
             REWIND_PASSWORD_KEY,
             MONITORING_PASSWORD_KEY,
+            LDAP_PASSWORD_KEY,
             RAFT_PASSWORD_KEY,
             PATRONI_PASSWORD_KEY,
         ):
@@ -1407,8 +1410,8 @@ def _setup_ldap_sync(self, postgres_snap: snap.Snap | None = None) -> None:
             "ldap-sync.postgres_host": "127.0.0.1",
             "ldap-sync.postgres_port": DATABASE_PORT,
             "ldap-sync.postgres_database": DATABASE_DEFAULT_NAME,
-            "ldap-sync.postgres_username": USER,
-            "ldap-sync.postgres_password": self._get_password(),
+            "ldap-sync.postgres_username": LDAP_USER,
+            "ldap-sync.postgres_password": self.get_secret(APP_SCOPE, LDAP_PASSWORD_KEY),
         })
 
         logger.debug("Starting LDAP sync service")
@@ -1446,6 +1449,13 @@ def _start_primary(self, event: StartEvent) -> None:
                     self.get_secret(APP_SCOPE, MONITORING_PASSWORD_KEY),
                     extra_user_roles=["pg_monitor"],
                 )
+            if LDAP_USER not in users:
+                # Create the LDAP user with minimal privileges to create roles.
+                self.postgresql.create_user(
+                    LDAP_USER,
+                    self.get_secret(APP_SCOPE, LDAP_PASSWORD_KEY),
+                    extra_user_roles=["CREATEROLE"],
+                )
         except PostgreSQLCreateUserError as e:
             logger.exception(e)
             self.unit.status = BlockedStatus("Failed to create postgres user")

src/constants.py

@@ -18,6 +18,7 @@
 BACKUP_USER = "backup"
 REPLICATION_USER = "replication"
 REWIND_USER = "rewind"
+LDAP_USER = "ldap"
 TLS_KEY_FILE = "key.pem"
 TLS_CA_FILE = "ca.pem"
 TLS_CERT_FILE = "cert.pem"
@@ -27,7 +28,7 @@
 PATRONI_SERVICE_NAME = "snap.charmed-postgresql.patroni.service"
 PATRONI_SERVICE_DEFAULT_PATH = f"/etc/systemd/system/{PATRONI_SERVICE_NAME}"
 # List of system usernames needed for correct work of the charm/workload.
-SYSTEM_USERS = [BACKUP_USER, REPLICATION_USER, REWIND_USER, USER, MONITORING_USER]
+SYSTEM_USERS = [BACKUP_USER, REPLICATION_USER, REWIND_USER, USER, MONITORING_USER, LDAP_USER]
 
 # Snap constants.
 PGBACKREST_EXECUTABLE = "charmed-postgresql.pgbackrest"
@@ -67,6 +68,7 @@
 REWIND_PASSWORD_KEY = "rewind-password"  # noqa: S105
 USER_PASSWORD_KEY = "operator-password"  # noqa: S105
 MONITORING_PASSWORD_KEY = "monitoring-password"  # noqa: S105
+LDAP_PASSWORD_KEY = "ldap-password"  # noqa: S105
 RAFT_PASSWORD_KEY = "raft-password"  # noqa: S105
 PATRONI_PASSWORD_KEY = "patroni-password"  # noqa: S105
 SECRET_INTERNAL_LABEL = "internal-secret"  # noqa: S105

tests/integration/test_password_rotation.py

@@ -51,6 +51,7 @@ async def test_password_rotation(ops_test: OpsTest):
     monitoring_password = await get_password(ops_test, any_unit_name, "monitoring")
     backup_password = await get_password(ops_test, any_unit_name, "backup")
     rewind_password = await get_password(ops_test, any_unit_name, "rewind")
+    ldap_password = await get_password(ops_test, any_unit_name, "ldap")
     patroni_password = await get_password(ops_test, any_unit_name, "patroni")
 
     # Get the leader unit name (because passwords can only be set through it).
@@ -97,6 +98,14 @@ async def test_password_rotation(ops_test: OpsTest):
     assert "password" in result
     await ops_test.model.wait_for_idle(apps=[APP_NAME], status="active", timeout=1000)
 
+    # For ldap, generate a specific password and pass it to the action.
+    new_ldap_password = "test-password"
+    result = await set_password(
+        ops_test, unit_name=leader, username="ldap", password=new_ldap_password
+    )
+    assert "password" in result
+    await ops_test.model.wait_for_idle(apps=[APP_NAME], status="active", timeout=1000)
+
     new_superuser_password = await get_password(ops_test, any_unit_name)
     assert superuser_password != new_superuser_password
     assert new_replication_password == await get_password(ops_test, any_unit_name, "replication")
@@ -107,6 +116,8 @@ async def test_password_rotation(ops_test: OpsTest):
     assert backup_password != new_backup_password
     assert new_rewind_password == await get_password(ops_test, any_unit_name, "rewind")
     assert rewind_password != new_rewind_password
+    assert new_ldap_password == await get_password(ops_test, any_unit_name, "ldap")
+    assert ldap_password != new_ldap_password
 
     # Restart Patroni on any non-leader unit and check that
     # Patroni and PostgreSQL continue to work.

tests/unit/test_charm.py

@@ -825,6 +825,7 @@ def test_on_get_password(harness):
             {
                 "operator-password": "test-password",
                 "replication-password": "replication-test-password",
+                "ldap-password": "ldap-test-password",
             },
         )
 
@@ -840,12 +841,18 @@ def test_on_get_password(harness):
         harness.charm._on_get_password(mock_event)
         mock_event.set_results.assert_called_once_with({"password": "test-password"})
 
-        # Also test providing the username option.
+        # Test providing the replication username option.
         mock_event.reset_mock()
         mock_event.params["username"] = "replication"
         harness.charm._on_get_password(mock_event)
         mock_event.set_results.assert_called_once_with({"password": "replication-test-password"})
 
+        # Test providing the ldap username option.
+        mock_event.reset_mock()
+        mock_event.params["username"] = "ldap"
+        harness.charm._on_get_password(mock_event)
+        mock_event.set_results.assert_called_once_with({"password": "ldap-test-password"})
+
 
 def test_on_set_password(harness):
     with (
@@ -900,14 +907,21 @@ def test_on_set_password(harness):
         mock_event.fail.assert_called_once()
         _set_secret.assert_not_called()
 
-        # Also test providing the username option.
+        # Test providing the replication username option.
         _set_secret.reset_mock()
         mock_event.params["username"] = "replication"
         harness.charm._on_set_password(mock_event)
         _set_secret.assert_called_once_with("app", "replication-password", "newpass")
 
+        # Test providing the ldap username option.
+        _set_secret.reset_mock()
+        mock_event.params["username"] = "ldap"
+        harness.charm._on_set_password(mock_event)
+        _set_secret.assert_called_once_with("app", "ldap-password", "newpass")
+
         # And test providing both the username and password options.
         _set_secret.reset_mock()
+        mock_event.params["username"] = "replication"
         mock_event.params["password"] = "replication-test-password"
         harness.charm._on_set_password(mock_event)
         _set_secret.assert_called_once_with(