Add anonymize user action

actions.yaml

@@ -1,6 +1,16 @@
 # Copyright 2023 Canonical Ltd.
 # See LICENSE file for licensing details.
 
+anonymize-user:
+  description: |
+    Anonymizes an user in order to make it GDPR compliant.
+  properties:
+    username:
+      description: |
+        User name to be anonymized.
+      type: string
+  required:
+    - username
 reset-instance:
   description: |
     Set a new server_name before running this action.

docs/tutorial/getting-started.md

@@ -31,7 +31,7 @@ Synapse requires connections to PostgreSQL. Deploy both charm applications.
 ### Deploy the charms:
 ```
 juju deploy postgresql-k8s
-juju deploy synapse-k8s
+juju deploy synapse
 ```
 
 Run `juju status` to see the current status of the deployment. Synapse

src/charm.py

@@ -87,6 +87,7 @@ def __init__(self, *args: typing.Any) -> None:
         self.framework.observe(
             self.on.promote_user_admin_action, self._on_promote_user_admin_action
         )
+        self.framework.observe(self.on.anonymize_user_action, self._on_anonymize_user_action)
 
     def replan_nginx(self) -> None:
         """Replan NGINX."""
@@ -286,6 +287,36 @@ def _on_promote_user_admin_action(self, event: ActionEvent) -> None:
             return
         event.set_results(results)
 
+    def _on_anonymize_user_action(self, event: ActionEvent) -> None:
+        """Anonymize user and report action result.
+
+        Args:
+            event: Event triggering the anonymize user action.
+        """
+        results = {
+            "anonymize-user": False,
+        }
+        container = self.unit.get_container(synapse.SYNAPSE_CONTAINER_NAME)
+        if not container.can_connect():
+            event.fail("Container not yet ready. Try again later")
+            return
+        try:
+            admin_access_token = self.get_admin_access_token()
+            if not admin_access_token:
+                event.fail("Failed to get admin access token")
+                return
+            username = event.params["username"]
+            server = self._charm_state.synapse_config.server_name
+            user = User(username=username, admin=False)
+            synapse.deactivate_user(
+                user=user, server=server, admin_access_token=admin_access_token
+            )
+            results["anonymize-user"] = True
+        except synapse.APIError:
+            event.fail("Failed to anonymize the user. Check if the user is created and active.")
+            return
+        event.set_results(results)
+
 
 if __name__ == "__main__":  # pragma: nocover
     main(SynapseCharm)

tests/integration/test_charm.py

@@ -13,6 +13,7 @@
 from juju.action import Action
 from juju.application import Application
 from juju.model import Model
+from juju.unit import Unit
 from ops.model import ActiveStatus
 from pytest_operator.plugin import OpsTest
 from saml_test_helper import SamlK8sTestHelper
@@ -426,3 +427,55 @@ async def test_promote_user_admin(
         timeout=5,
     )
     assert res.status_code == 200
+
+
+async def test_anonymize_user(
+    synapse_app: Application,
+    get_unit_ips: typing.Callable[[str], typing.Awaitable[tuple[str, ...]]],
+) -> None:
+    """
+    arrange: build and deploy the Synapse charm, create an user, get the access token and assert
+        that the user is not an admin.
+    act:  run action to anonymize user.
+    assert: the Synapse application is active and the API request returns as expected.
+    """
+    operator_username = "operator-new"
+    synapse_unit: Unit = next(iter(synapse_app.units))
+    action_register_user: Action = await synapse_unit.run_action(
+        "register-user", username=operator_username, admin=False
+    )
+    await action_register_user.wait()
+    assert action_register_user.status == "completed"
+    password = action_register_user.results["user-password"]
+    synapse_ip = (await get_unit_ips(synapse_app.name))[0]
+    with requests.session() as sess:
+        res = sess.post(
+            f"http://{synapse_ip}:8080/_matrix/client/r0/login",
+            # same thing is done on fixture but we are creating a non-admin user here.
+            json={  # pylint: disable=duplicate-code
+                "identifier": {"type": "m.id.user", "user": operator_username},
+                "password": password,
+                "type": "m.login.password",
+            },
+            timeout=5,
+        )
+        res.raise_for_status()
+
+    action_anonymize: Action = await synapse_unit.run_action(
+        "anonymize-user", username=operator_username
+    )
+    await action_anonymize.wait()
+    assert action_anonymize.status == "completed"
+
+    with requests.session() as sess:
+        res = sess.post(
+            f"http://{synapse_ip}:8080/_matrix/client/r0/login",
+            # same thing is done on fixture but we are creating a non-admin user here.
+            json={  # pylint: disable=duplicate-code
+                "identifier": {"type": "m.id.user", "user": operator_username},
+                "password": password,
+                "type": "m.login.password",
+            },
+            timeout=5,
+        )
+    assert res.status_code == 403

tests/unit/test_anonymize_user_action.py

@@ -0,0 +1,133 @@
+# Copyright 2023 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+"""Register user action unit tests."""
+
+# Disabled to access _on_register_user_action
+# pylint: disable=protected-access
+
+import unittest.mock
+from secrets import token_hex
+
+import pytest
+from ops.charm import ActionEvent
+from ops.testing import Harness
+
+import synapse
+from charm import SynapseCharm
+
+
+def test_anonymize_user_action(harness: Harness, monkeypatch: pytest.MonkeyPatch) -> None:
+    """
+    arrange: start the Synapse charm, set Synapse container to be ready and set server_name.
+    act: run anonymize-user action.
+    assert: event results are returned as expected.
+    """
+    harness.begin_with_initial_hooks()
+    admin_access_token = token_hex(16)
+    anonymize_user_mock = unittest.mock.Mock()
+    monkeypatch.setattr(
+        SynapseCharm,
+        "get_admin_access_token",
+        unittest.mock.MagicMock(return_value=admin_access_token),
+    )
+    monkeypatch.setattr("synapse.deactivate_user", anonymize_user_mock)
+    user = "username"
+    admin = True
+    event = unittest.mock.MagicMock(spec=ActionEvent)
+    event.params = {
+        "username": user,
+        "admin": admin,
+    }
+
+    harness.charm._on_anonymize_user_action(event)
+
+    assert event.set_results.call_count == 1
+    event.set_results.assert_called_with({"anonymize-user": True})
+    anonymize_user_mock.assert_called_with(
+        user=unittest.mock.ANY, server=unittest.mock.ANY, admin_access_token=admin_access_token
+    )
+
+
+def test_anonymize_user_api_error(harness: Harness, monkeypatch: pytest.MonkeyPatch) -> None:
+    """
+    arrange: start the Synapse charm, set Synapse container to be ready and set server_name.
+    act: run anonymize-user action.
+    assert: event fails as expected.
+    """
+    harness.begin_with_initial_hooks()
+    fail_message = "Failed to anonymize the user. Check if the user is created and active."
+    synapse_api_error = synapse.APIError(fail_message)
+    anonymize_user_mock = unittest.mock.MagicMock(side_effect=synapse_api_error)
+    monkeypatch.setattr("synapse.deactivate_user", anonymize_user_mock)
+    admin_access_token = token_hex(16)
+    user = "username"
+    admin = True
+    monkeypatch.setattr(
+        SynapseCharm,
+        "get_admin_access_token",
+        unittest.mock.MagicMock(return_value=admin_access_token),
+    )
+    event = unittest.mock.MagicMock(spec=ActionEvent)
+    event.params = {
+        "username": user,
+        "admin": admin,
+    }
+
+    def event_store_failure(failure_message: str) -> None:
+        """Define a failure message for the event.
+
+        Args:
+            failure_message: failure message content to be defined.
+        """
+        event.fail_message = failure_message
+
+    event.fail = event_store_failure
+    event.params = {
+        "username": user,
+        "admin": admin,
+    }
+
+    harness.charm._on_anonymize_user_action(event)
+
+    assert fail_message in event.fail_message
+
+
+def test_anonymize_user_container_down(harness: Harness) -> None:
+    """
+    arrange: start the Synapse charm, set Synapse container to be off.
+    act: run anonymize-user action.
+    assert: event fails as expected.
+    """
+    harness.begin_with_initial_hooks()
+    harness.set_can_connect(harness.model.unit.containers[synapse.SYNAPSE_CONTAINER_NAME], False)
+    event = unittest.mock.Mock()
+
+    harness.charm._on_anonymize_user_action(event)
+
+    assert event.set_results.call_count == 0
+    assert event.fail.call_count == 1
+    assert "Container not yet ready. Try again later" == event.fail.call_args[0][0]
+
+
+def test_anonymize_user_action_no_token(harness: Harness, monkeypatch: pytest.MonkeyPatch) -> None:
+    """
+    arrange: start the Synapse charm, set Synapse container to be ready and set server_name.
+    act: run anonymize-user action.
+    assert: event fails as expected.
+    """
+    harness.begin_with_initial_hooks()
+    anonymize_user_mock = unittest.mock.Mock()
+    monkeypatch.setattr(
+        SynapseCharm,
+        "get_admin_access_token",
+        unittest.mock.MagicMock(return_value=None),
+    )
+    monkeypatch.setattr("synapse.deactivate_user", anonymize_user_mock)
+    event = unittest.mock.Mock()
+
+    harness.charm._on_anonymize_user_action(event)
+
+    assert event.set_results.call_count == 0
+    assert event.fail.call_count == 1
+    assert "Failed to get admin access token" == event.fail.call_args[0][0]