Forbid encryption of public rooms (#1)

* Forbid encryption of public rooms

* Also forbid change of join_rule for existing public rooms

---------

Co-authored-by: Mathieu Velten <mathieuv@matrix.org>

room_access_rules/__init__.py

@@ -350,7 +350,7 @@ async def check_event_allowed(
                 )
 
             if event.type == EventTypes.JoinRules:
-                return self._on_join_rule_change(event, rule), None
+                return self._on_join_rule_change(event, rule, state_events), None
 
             if event.type == EventTypes.RoomAvatar:
                 return self._on_room_avatar_change(event, rule), None
@@ -361,6 +361,9 @@ async def check_event_allowed(
             if event.type == EventTypes.Topic:
                 return self._on_room_topic_change(event, rule), None
 
+            if event.type == EventTypes.RoomEncryption:
+                return self._on_room_encryption_change(event, state_events), None
+
         return True, None
 
     async def check_visibility_can_be_modified(
@@ -654,7 +657,9 @@ def _is_power_level_content_allowed(
 
         return True
 
-    def _on_join_rule_change(self, event: EventBase, rule: str) -> bool:
+    def _on_join_rule_change(
+        self, event: EventBase, rule: str, state_events: StateMap[EventBase]
+    ) -> bool:
         """Check whether a join rule change is allowed.
 
         A join rule change is always allowed unless the new join rule is "public" and
@@ -670,6 +675,9 @@ def _on_join_rule_change(self, event: EventBase, rule: str) -> bool:
         if event.content.get("join_rule") == JoinRules.PUBLIC:
             return rule != AccessRules.DIRECT
 
+        if self._get_join_rule_from_state(state_events) == JoinRules.PUBLIC:
+            return False
+
         return True
 
     def _on_room_avatar_change(self, event: EventBase, rule: str) -> bool:
@@ -714,6 +722,23 @@ def _on_room_topic_change(self, event: EventBase, rule: str) -> bool:
         """
         return rule != AccessRules.DIRECT
 
+    def _on_room_encryption_change(
+        self, event: EventBase, state_events: StateMap[EventBase]
+    ) -> bool:
+        """Check whether a room can have its encryption enabled.
+        The current rule is to forbid such a change in public rooms.
+
+        Args:
+            event: The event to check.
+            state_events: A dict mapping (event type, state key) to state event.
+
+        Returns:
+            True if the event can be allowed, False otherwise.
+        """
+        if self._get_join_rule_from_state(state_events) == JoinRules.PUBLIC:
+            return False
+        return True
+
     @staticmethod
     def _get_rule_from_state(state_events: StateMap[EventBase]) -> str:
         """Extract the rule to be applied from the given set of state events.

tests/test_event_allowed.py

@@ -611,6 +611,50 @@ async def test_revoke_3pid_invite_direct(self):
 
         self.assertTrue(allowed)
 
+    async def test_forbid_join_rule_change_public_room(self):
+        """Tests that the join_rule of a public room can't be changed."""
+        state_events = self.restricted_room_state.copy()
+        state_events[(EventTypes.JoinRules, "")] = MockEvent(
+            sender=self.room_creator,
+            type=EventTypes.JoinRules,
+            content={"join_rule": JoinRules.PUBLIC},
+            state_key="",
+        )
+
+        allowed, _ = await self.module.check_event_allowed(
+            event=MockEvent(
+                sender=self.room_creator,
+                type=EventTypes.JoinRules,
+                content={"join_rule": JoinRules.PRIVATE},
+                state_key="",
+            ),
+            state_events=state_events,
+        )
+
+        self.assertFalse(allowed)
+
+    async def test_forbid_encryption_public_room(self):
+        """Tests that a public room can't have its encryption enabled."""
+        state_events = self.restricted_room_state.copy()
+        state_events[(EventTypes.JoinRules, "")] = MockEvent(
+            sender=self.room_creator,
+            type=EventTypes.JoinRules,
+            content={"join_rule": JoinRules.PUBLIC},
+            state_key="",
+        )
+
+        allowed, _ = await self.module.check_event_allowed(
+            event=MockEvent(
+                sender=self.room_creator,
+                type=EventTypes.RoomEncryption,
+                content={},
+                state_key="",
+            ),
+            state_events=state_events,
+        )
+
+        self.assertFalse(allowed)
+
     async def test_forbidden_users_join(self):
         """Tests that RoomAccessRules.check_event_allowed behaves accordingly.