"Freeze" a room when the last admin of that room leaves

changelog.d/59.feature

@@ -0,0 +1 @@
+Freeze a room when the last administrator in the room leaves.

synapse/third_party_rules/access_rules.py

@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import email.utils
+import logging
 from typing import Dict, List, Optional, Tuple
 
 from twisted.internet import defer
@@ -22,7 +23,9 @@
 from synapse.config._base import ConfigError
 from synapse.events import EventBase
 from synapse.module_api import ModuleApi
-from synapse.types import Requester, StateMap, get_domain_from_id
+from synapse.types import Requester, StateMap, UserID, get_domain_from_id
+
+logger = logging.getLogger(__name__)
 
 ACCESS_RULES_TYPE = "im.vector.room.access_rules"
 
@@ -323,7 +326,7 @@ async def check_event_allowed(
             )
 
         if event.type == EventTypes.Member or event.type == EventTypes.ThirdPartyInvite:
-            return self._on_membership_or_invite(event, rule, state_events)
+            return await self._on_membership_or_invite(event, rule, state_events)
 
         if event.type == EventTypes.JoinRules:
             return self._on_join_rule_change(event, rule)
@@ -420,7 +423,7 @@ async def _on_rules_change(
             prev_rule == AccessRules.RESTRICTED and new_rule == AccessRules.UNRESTRICTED
         )
 
-    def _on_membership_or_invite(
+    async def _on_membership_or_invite(
         self, event: EventBase, rule: str, state_events: StateMap[EventBase],
     ) -> bool:
         """Applies the correct rule for incoming m.room.member and
@@ -446,8 +449,136 @@ def _on_membership_or_invite(
             # might want to change that in the future.
             ret = self._on_membership_or_invite_restricted(event)
 
+        if event.type == "m.room.member":
+            # If this is an admin leaving, and they are the last admin in the room,
+            # raise the power levels of the room so that the room is 'frozen'.
+            #
+            # We have to freeze the room by puppeting an admin user, which we can
+            # only do for local users
+            if (
+                self._is_local_user(event.sender)
+                and event.membership == Membership.LEAVE
+            ):
+                await self._freeze_room_if_last_admin_is_leaving(event, state_events)
+
         return ret
 
+    async def _freeze_room_if_last_admin_is_leaving(
+        self, event: EventBase, state_events: StateMap[EventBase]
+    ):
+        power_level_state_event = state_events.get(
+            (EventTypes.PowerLevels, "")
+        )  # type: EventBase
+        if not power_level_state_event:
+            return
+        power_level_content = power_level_state_event.content.copy()
+
+        # Do some validation checks on the power level state event
+        if (
+            not isinstance(power_level_content, dict)
+            or "users" not in power_level_content
+            or not isinstance(power_level_content["users"], dict)
+        ):
+            # We can't use this power level event to determine whether the room should be
+            # frozen. Bail out.
+            return
+
+        user_id = event.get("sender")
+        if not user_id:
+            return
+
+        # Get every admin user defined in the room's state
+        admin_users = {
+            user
+            for user, power_level in power_level_content["users"].items()
+            if power_level >= 100
+        }
+
+        if user_id not in admin_users:
+            # This user is not an admin, ignore them
+            return
+
+        # Filter these users to only those who are actually joined or invited to the room
+        joined_members = [
+            user_id
+            for (event_type, user_id), event in state_events.items()
+            if event_type == EventTypes.Member
+            and event.membership in [Membership.JOIN, Membership.INVITE]
+        ]
+        admin_users = {user for user in admin_users if user in joined_members}
+
+        if len(admin_users) > 1:
+            # There's another admin user in, or invited to, the room
+            return
+
+        logger.info("Freezing room '%s'", event.room_id)
+
+        # Modify the existing power levels to raise all required types to 100
+        #
+        # This changes a power level state event's content from something like:
+        # {
+        #     "redact": 50,
+        #     "state_default": 50,
+        #     "ban": 50,
+        #     "notifications": {
+        #         "room": 50
+        #     },
+        #     "events": {
+        #         "m.room.avatar": 50,
+        #         "m.room.encryption": 50,
+        #         "m.room.canonical_alias": 50,
+        #         "m.room.name": 50,
+        #         "im.vector.modular.widgets": 50,
+        #         "m.room.topic": 50,
+        #         "m.room.tombstone": 50,
+        #         "m.room.history_visibility": 100,
+        #         "m.room.power_levels": 100
+        #     },
+        #     "users_default": 0,
+        #     "events_default": 0,
+        #     "users": {
+        #         "@admin:example.com": 100,
+        #     },
+        #     "kick": 50,
+        #     "invite": 0
+        # }
+        #
+        # to
+        #
+        # {
+        #     "redact": 100,
+        #     "state_default": 100,
+        #     "ban": 100,
+        #     "notifications": {
+        #         "room": 50
+        #     },
+        #     "events": {}
+        #     "users_default": 0,
+        #     "events_default": 100,
+        #     "users": {
+        #         "@admin:example.com": 100,
+        #     },
+        #     "kick": 100,
+        #     "invite": 100
+        # }
+        for key, value in power_level_content.items():
+            # Do not change "users_default", as that key specifies the default power
+            # level of new users
+            if isinstance(value, int) and key != "users_default":
+                power_level_content[key] = 100
+        power_level_content["events"] = {}
+
+        # Freeze the room by raising the required power level to send events to 100
+        await self.module_api.create_and_send_event_into_room(
+            {
+                "room_id": event.room_id,
+                "sender": user_id,
+                "type": EventTypes.PowerLevels,
+                "content": power_level_content,
+                "state_key": "",
+            }
+        )
+
     def _on_membership_or_invite_restricted(self, event: EventBase) -> bool:
         """Implements the checks and behaviour specified for the "restricted" rule.
 
@@ -753,6 +884,25 @@ def _is_invite_from_threepid(invite: EventBase, threepid_invite_token: str) -> b
 
         return token == threepid_invite_token
 
+    def _is_local_user(self, user_id: str) -> bool:
+        """Checks whether a given user ID belongs to this homeserver, or a remote
+
+        Args:
+            user_id: A user ID to check.
+
+        Returns:
+            True if the user belongs to this homeserver, False otherwise.
+        """
+        user = UserID.from_string(user_id)
+
+        # Extract the localpart and ask the module API for a user ID from the localpart
+        # The module API will append the local homeserver's server_name
+        local_user_id = self.module_api.get_qualified_user_id(user.localpart)
+
+        # If the user ID we get based on the localpart is the same as the original user ID,
+        # then they were a local user
+        return user_id == local_user_id
+
     def _user_is_invited_to_room(
         self, user_id: str, state_events: StateMap[EventBase]
     ) -> bool:

tests/rest/client/test_room_access_rules.py

@@ -840,6 +840,84 @@ def test_check_event_allowed(self):
         )
         self.assertTrue(can_join)
 
+    def test_freezing_a_room(self):
+        """Tests that the power levels in a room change to prevent new events from
+        non-admin users when the last admin of a room leaves.
+        """
+        # Invite a user to the room, they join with PL 0
+        self.helper.invite(
+            room=self.restricted_room,
+            src=self.user_id,
+            targ=self.invitee_id,
+            tok=self.tok,
+        )
+
+        # Invitee joins the room
+        self.helper.join(
+            room=self.restricted_room, user=self.invitee_id, tok=self.invitee_tok,
+        )
+
+        # Retrieve the room's current power levels event content
+        power_levels = self.helper.get_state(
+            room_id=self.restricted_room,
+            event_type="m.room.power_levels",
+            tok=self.tok,
+        )
+
+        # Ensure that the invitee leaving the room does not change the power levels
+        self.helper.leave(
+            room=self.restricted_room, user=self.invitee_id, tok=self.invitee_tok,
+        )
+
+        # Retrieve the new power levels of the room
+        new_power_levels = self.helper.get_state(
+            room_id=self.restricted_room,
+            event_type="m.room.power_levels",
+            tok=self.tok,
+        )
+
+        # Ensure they have not changed
+        self.assertDictEqual(power_levels, new_power_levels)
+
+        # Invite the user back again
+        self.helper.invite(
+            room=self.restricted_room,
+            src=self.user_id,
+            targ=self.invitee_id,
+            tok=self.tok,
+        )
+
+        # Invitee joins the room
+        self.helper.join(
+            room=self.restricted_room, user=self.invitee_id, tok=self.invitee_tok,
+        )
+
+        # Now the admin leaves the room
+        self.helper.leave(
+            room=self.restricted_room, user=self.user_id, tok=self.tok,
+        )
+
+        # Check the power levels again
+        new_power_levels = self.helper.get_state(
+            room_id=self.restricted_room,
+            event_type="m.room.power_levels",
+            tok=self.invitee_tok,
+        )
+
+        # Ensure that the new power levels prevent anyone but admins from sending
+        # certain events
+        self.assertEquals(new_power_levels["state_default"], 100)
+        self.assertEquals(new_power_levels["events_default"], 100)
+        self.assertEquals(new_power_levels["kick"], 100)
+        self.assertEquals(new_power_levels["invite"], 100)
+        self.assertEquals(new_power_levels["ban"], 100)
+        self.assertEquals(new_power_levels["redact"], 100)
+        self.assertDictEqual(new_power_levels["events"], {})
+        self.assertDictEqual(new_power_levels["users"], {self.user_id: 100})
+
+        # Ensure new users entering the room aren't going to immediately become admins
+        self.assertEquals(new_power_levels["users_default"], 0)
+
     def create_room(
         self,
         direct=False,