[DPE-4682] Add fix for user management (#374)

# Issue

Addressing #331 , namely that currently it's impossible to add users
manually to Opensearch, as they are cleaned up on the first
`update-status`.

# Solution

Saving every relation ID with the corresponding user's name in a
dictionary on the databag.
This allows us to keep track of relation users and the relations they
belong to.
This data structure is maintained such as
- `index-created` : adding new relation - usrer pair as craeting the
user in Opensearch
- `update-status`, `relation-broken`, `relation-departed`: removing
(stale or dedicated) relations - user pairs from the databag as removing
Opensearch users

## Developers' notes

In order to get involved scopes clear:
- Operations on the new databag structure are strongly coupled togeter.
Thus they MUST be in the same module.
  (This principle is also in attempt to decrease the "spagetti".)
- Since removing users is now clearly specific to client relations, this
functionality should reside in the Provider code.
Accompaigned with the suited user creation functionality encapsulating
all relation-specific details (role name same as username, databag
structures to be maintained, etc.)
- Note: since extra user roles may be shared across multiple users, and
re-used, we do NOT clean up those. But normal roles only.
- Note2: We are taking advantage of usernames being equal to their
related roles (and don't store roles specifically).

---------

Co-authored-by: Judit Novak <[email protected]>

lib/charms/opensearch/v0/constants_charm.py

@@ -109,6 +109,8 @@
 AdminUser = "admin"
 KibanaserverUser = "kibanaserver"
 KibanaserverRole = "kibana_server"
+ClientUsersDict = "client_relation_users"
+
 
 # Opensearch Snap revision
 OPENSEARCH_SNAP_REVISION = 58  # Keep in sync with `workload_version` file

lib/charms/opensearch/v0/opensearch_base_charm.py

@@ -617,13 +617,15 @@ def _on_update_status(self, event: UpdateStatusEvent):
 
         deployment_desc = self.opensearch_peer_cm.deployment_desc()
         if self.upgrade_in_progress:
-            logger.debug("Skipping `remove_users_and_roles()` because upgrade is in-progress")
+            logger.debug(
+                "Skipping `remove_lingering_users_and_roles()` because upgrade is in-progress"
+            )
         elif (
             self.unit.is_leader()
             and deployment_desc
             and deployment_desc.typ == DeploymentType.MAIN_ORCHESTRATOR
         ):
-            self.user_manager.remove_users_and_roles()
+            self.opensearch_provider.remove_lingering_relation_users_and_roles()
 
         # If relation not broken - leave
         if self.model.get_relation("certificates") is not None:

lib/charms/opensearch/v0/opensearch_plugins.py

@@ -261,7 +261,7 @@ def _on_update_status(self, event):
         for relation in self.model.relations.get(ClientRelationName, []):
             self.opensearch_provider.update_endpoints(relation)
 
-        self.user_manager.remove_users_and_roles()
+        self.opensearch_provider.remove_lingering_relation_users_and_roles()
         # If relation not broken - leave
         if self.model.get_relation("certificates") is not None:
             return

lib/charms/opensearch/v0/opensearch_relation_provider.py

@@ -28,6 +28,7 @@
 )
 from charms.opensearch.v0.constants_charm import (
     ClientRelationName,
+    ClientUsersDict,
     IndexCreationFailed,
     KibanaserverRole,
     KibanaserverUser,
@@ -198,7 +199,8 @@ def _on_index_requested(self, event: IndexRequestedEvent) -> None:  # noqa
         """
         if self.charm.upgrade_in_progress:
             logger.warning(
-                "Modifying relations during an upgrade is not supported. The charm may be in a broken, unrecoverable state"
+                "Modifying relations during an upgrade is not supported."
+                "The charm may be in a broken, unrecoverable state"
             )
             event.defer()
             return
@@ -233,7 +235,13 @@ def _on_index_requested(self, event: IndexRequestedEvent) -> None:  # noqa
             username = self._relation_username(event.relation)
             hashed_pwd, pwd = generate_hashed_password()
             try:
-                self.create_opensearch_users(username, hashed_pwd, event.index, extra_user_roles)
+                self.create_opensearch_users(
+                    username,
+                    hashed_pwd,
+                    event.index,
+                    extra_user_roles,
+                    relation_id=event.relation.id,
+                )
             except OpenSearchUserMgmtError as err:
                 logger.error(err)
                 self.charm.status.set(
@@ -258,9 +266,7 @@ def _on_index_requested(self, event: IndexRequestedEvent) -> None:  # noqa
         # Clear old statuses set by this hook
         self.charm.status.clear(NewIndexRequested.format(index=event.index))
         self.charm.status.clear(IndexCreationFailed.format(index=event.index))
-        self.charm.status.clear(
-            UserCreationFailed.format(rel_name=ClientRelationName, id=event.relation.id)
-        )
+        self.charm.status.clear(UserCreationFailed.format(rel_name=ClientRelationName, id=rel_id))
 
     def validate_index_name(self, index_name: str) -> bool:
         """Validates that the index name provided in the relation is acceptable."""
@@ -285,11 +291,7 @@ def validate_index_name(self, index_name: str) -> bool:
         return True
 
     def create_opensearch_users(
-        self,
-        username: str,
-        hashed_pwd: str,
-        index: str,
-        extra_user_roles: str,
+        self, username: str, hashed_pwd: str, index: str, extra_user_roles: str, relation_id: int
     ):
         """Creates necessary opensearch users and permissions for this relation.
 
@@ -299,6 +301,7 @@ def create_opensearch_users(
             index: the index to which the users must be granted access
             extra_user_roles: the level of permissions that the user should be given. Can be a
                 comma-separated list of roles, which should result in a merged list of permissions.
+            relation_id: the relation id for this relation, if it exists
 
         Raises:
             OpenSearchUserMgmtError if user creation fails
@@ -307,15 +310,11 @@ def create_opensearch_users(
             # Create a new role for this relation, encapsulating the permissions we care about. We
             # can't create a "default" and an "admin" role once because the permissions need to be
             # set to this relation's specific index.
-            self.user_manager.create_role(
-                role_name=username,
-                permissions=self.get_extra_user_role_permissions(extra_user_roles, index),
-            )
-            roles = [username]
-            self.user_manager.create_user(username, roles, hashed_pwd)
+            permissions = self.get_extra_user_role_permissions(extra_user_roles, index)
+            self._put_relation_user(username, permissions, hashed_pwd, relation_id)
             self.user_manager.patch_user(
                 username,
-                [{"op": "replace", "path": "/opendistro_security_roles", "value": roles}],
+                [{"op": "replace", "path": "/opendistro_security_roles", "value": [username]}],
             )
         except OpenSearchUserMgmtError as err:
             logger.error(err)
@@ -391,6 +390,8 @@ def _on_relation_departed(self, event: RelationDepartedEvent) -> None:
         if event.departing_unit == self.charm.unit:
             self.charm.peers_data.put(Scope.UNIT, self._depart_flag(event.relation), True)
 
+        self.remove_lingering_relation_users_and_roles(event.relation.id)
+
     def _on_relation_broken(self, event: RelationBrokenEvent) -> None:
         """Handle client relation-broken event."""
         if not self.unit.is_leader():
@@ -401,9 +402,10 @@ def _on_relation_broken(self, event: RelationBrokenEvent) -> None:
             return
         if self.charm.upgrade_in_progress:
             logger.warning(
-                "Modifying relations during an upgrade is not supported. The charm may be in a broken, unrecoverable state"
+                "Modifying relations during an upgrade is not supported."
+                "The charm may be in a broken, unrecoverable state"
             )
-        self.user_manager.remove_users_and_roles(event.relation.id)
+        self.remove_lingering_relation_users_and_roles(event.relation.id)
 
     def update_endpoints(self, relation: Relation, omit_endpoints: Optional[Set[str]] = None):
         """Updates endpoints in the databag for the given relation."""
@@ -438,3 +440,70 @@ def update_dashboards_password(self):
         pwd = self.secrets.get(Scope.APP, self.secrets.password_key(KibanaserverUser))
         for relation in self.dashboards_relations:
             self.opensearch_provides.set_credentials(relation.id, KibanaserverUser, pwd)
+
+    def _put_relation_user(
+        self, user: str, permissions: dict[str], hashed_pwd: str, relation_id: int
+    ):
+        """Create a relation user.
+
+        Relation users are registered with a dedicated role which maps to the username,
+        and their name is saved in the databag for later reference.
+        """
+        self.user_manager.create_role(role_name=user, permissions=permissions)
+        users = self.charm.peers_data.get_object(Scope.APP, ClientUsersDict) or {}
+
+        if users.get(relation_id):
+            logger.warning(
+                "User %s is already registered in Peer Relation data for relation %d.",
+                user,
+                relation_id,
+            )
+
+        self.user_manager.create_user(user, [user], hashed_pwd)
+        users[str(relation_id)] = user
+        self.charm.peers_data.put_object(Scope.APP, ClientUsersDict, users)
+
+    def remove_lingering_relation_users_and_roles(  # noqa: C901
+        self, departed_relation_id: int | None = None
+    ):
+        """Removes lingering relation users and roles from opensearch.
+
+        Args:
+            departed_relation_id: if a relation is departing, pass in the ID and its user will be
+                deleted.
+        """
+        if not self.opensearch.is_node_up() or not self.unit.is_leader():
+            return
+
+        relation_users = self.charm.peers_data.get_object(Scope.APP, ClientUsersDict) or {}
+
+        if departed_relation_id and (
+            not relation_users or departed_relation_id not in relation_users
+        ):
+            logging.warning(
+                "User for relation %d wasn't registered in internal cham workflows.",
+                departed_relation_id,
+            )
+
+        cleanup_rel_ids = []
+        if departed_relation_id:
+            cleanup_rel_ids = [str(departed_relation_id)]
+
+        rel_ids = [str(relation.id) for relation in self.opensearch_provides.relations]
+        cleanup_rel_ids += list(set(relation_users.keys()) - set(rel_ids))
+
+        for rel_id in cleanup_rel_ids:
+            if username := relation_users.get(rel_id):
+                try:
+                    self.user_manager.remove_user(username)
+                except OpenSearchUserMgmtError:
+                    logger.error(f"failed to remove user {username}")
+
+                try:
+                    self.user_manager.remove_role(username)
+                except OpenSearchUserMgmtError:
+                    logger.error(f"failed to remove role {username}")
+
+                del relation_users[rel_id]
+
+        self.charm.peers_data.put_object(Scope.APP, ClientUsersDict, relation_users)

lib/charms/opensearch/v0/opensearch_users.py

@@ -7,11 +7,10 @@
 """
 
 import logging
-from typing import Dict, List, Optional, Set
+from typing import Dict, List, Optional
 
 from charms.opensearch.v0.constants_charm import (
     AdminUser,
-    ClientRelationName,
     COSRole,
     COSUser,
     KibanaserverUser,
@@ -242,27 +241,6 @@ def patch_user(self, user_name: str, patches: List[Dict[str, any]]) -> Dict[str,
 
         return resp
 
-    def remove_users_and_roles(self, departed_relation_id: Optional[int] = None):
-        """Removes lingering relation users and roles from opensearch.
-
-        Args:
-            departed_relation_id: if a relation is departing, pass in the ID and its user will be
-                deleted.
-        """
-        if not self.opensearch.is_node_up() or not self.unit.is_leader():
-            return
-
-        relations = self.model.relations.get(ClientRelationName, [])
-        relation_users = set(
-            [
-                f"{ClientRelationName}_{relation.id}"
-                for relation in relations
-                if relation.id != departed_relation_id
-            ]
-        )
-        self._remove_lingering_users(relation_users)
-        self._remove_lingering_roles(relation_users)
-
     def update_user_password(self, username: str, hashed_pwd: str = None):
         """Change user hashed password."""
         resp = self.opensearch.request(
@@ -273,6 +251,10 @@ def update_user_password(self, username: str, hashed_pwd: str = None):
         if resp.get("status") != "OK":
             raise OpenSearchError(f"{resp}")
 
+    ##########################################################################
+    # Dedicated functionalities
+    ##########################################################################
+
     def put_internal_user(self, user: str, hashed_pwd: str):
         """User creation for specific system users."""
         if user not in OpenSearchUsers:
@@ -314,33 +296,3 @@ def put_internal_user(self, user: str, hashed_pwd: str):
                 COSUser,
                 [{"op": "replace", "path": "/opendistro_security_roles", "value": roles}],
             )
-
-    def _remove_lingering_users(self, relation_users: Set[str]):
-        app_users = relation_users | OpenSearchUsers
-        try:
-            database_users = set(self.get_users().keys())
-        except OpenSearchUserMgmtError:
-            logger.error("failed to get users")
-            return
-
-        for username in database_users - app_users:
-            try:
-                self.remove_user(username)
-            except OpenSearchUserMgmtError:
-                logger.error(f"failed to remove user {username}")
-
-    def _remove_lingering_roles(self, roles: Set[str]):
-        try:
-            database_roles = set(self.get_roles().keys())
-        except (OpenSearchUserMgmtError, OpenSearchHttpError):
-            logger.error("failed to get roles")
-            return
-
-        for role in database_roles - roles:
-            if not role.startswith(f"{ClientRelationName}_"):
-                # This role was not created by this charm, so leave it alone
-                continue
-            try:
-                self.remove_role(role)
-            except OpenSearchUserMgmtError:
-                logger.error(f"failed to remove role {role}")

tests/integration/test_charm.py

@@ -2,7 +2,9 @@
 # Copyright 2024 Canonical Ltd.
 # See LICENSE file for licensing details.
 
+import asyncio
 import logging
+import shlex
 import subprocess
 
 import pytest
@@ -70,6 +72,9 @@ async def test_deploy_and_remove_single_unit(ops_test: OpsTest) -> None:
 async def test_build_and_deploy(ops_test: OpsTest) -> None:
     """Build and deploy a couple of OpenSearch units."""
     my_charm = await ops_test.build_charm(".")
+    model_config = MODEL_CONFIG
+    model_config["update-status-hook-interval"] = "1m"
+
     await ops_test.model.set_config(MODEL_CONFIG)
 
     await ops_test.model.deploy(
@@ -324,6 +329,34 @@ async def test_all_units_have_internal_users_synced(ops_test: OpsTest) -> None:
 
 @pytest.mark.group(1)
 @pytest.mark.abort_on_fail
-async def test_remove_application(ops_test: OpsTest) -> None:
-    """Removes the application with two units."""
-    await ops_test.model.remove_application(APP_NAME, block_until_done=True)
+async def test_add_users_and_calling_update_status(ops_test: OpsTest) -> None:
+    """Add users and call update status."""
+    leader_id = await get_leader_unit_id(ops_test)
+    leader_ip = await get_leader_unit_ip(ops_test)
+    test_url = f"https://{leader_ip}:9200/_plugins/_security/api/internalusers/my_user"
+
+    http_resp_code = await http_request(
+        ops_test,
+        "PUT",
+        test_url,
+        resp_status_code=True,
+        payload={"hash": "1234"},
+    )
+    assert http_resp_code >= 200 and http_resp_code < 300
+
+    cmd = '"export JUJU_DISPATCH_PATH=hooks/update-status; ./dispatch"'
+    exec_cmd = f"juju exec -u opensearch/{leader_id} -m {ops_test.model.name} -- {cmd}"
+    try:
+        # The "normal" subprocess.run with "export ...; ..." cmd was failing
+        # Noticed that, for this case, canonical/jhack uses shlex instead to split.
+        # Adding it fixed the issue.
+        subprocess.run(shlex.split(exec_cmd))
+    except Exception as e:
+        logger.error(
+            f"Failed to apply state: process exited with {e.returncode}; "
+            f"stdout = {e.stdout}; "
+            f"stderr = {e.stderr}.",
+        )
+    await asyncio.sleep(300)
+    http_resp_code = await http_request(ops_test, "GET", test_url, resp_status_code=True)
+    assert http_resp_code >= 200 and http_resp_code < 300

tests/unit/lib/test_opensearch_base_charm.py

@@ -325,7 +325,9 @@ def test_on_start(
     @patch(f"{BASE_LIB_PATH}.opensearch_backups.OpenSearchBackup._is_restore_complete")
     @patch(f"{BASE_CHARM_CLASS}._stop_opensearch")
     @patch(f"{BASE_LIB_PATH}.opensearch_base_charm.cert_expiration_remaining_hours")
-    @patch(f"{BASE_LIB_PATH}.opensearch_users.OpenSearchUserManager.remove_users_and_roles")
+    @patch(
+        f"{BASE_LIB_PATH}.opensearch_relation_provider.OpenSearchProvider.remove_lingering_relation_users_and_roles"
+    )
     def test_on_update_status(self, _, cert_expiration_remaining_hours, _stop_opensearch, __, ___):
         """Test on update status."""
         with patch(