test_room_access_rules.py

# -*- coding: utf-8 -*-
# Copyright 2019 New Vector Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import json
import random
import string
from typing import Optional

from mock import Mock

from twisted.internet import defer

from synapse.api.constants import EventTypes, JoinRules, Membership, RoomCreationPreset
from synapse.api.errors import SynapseError
from synapse.rest import admin
from synapse.rest.client.v1 import directory, login, room
from synapse.third_party_rules.access_rules import (
    ACCESS_RULES_TYPE,
    AccessRules,
    RoomAccessRules,
)
from synapse.types import JsonDict, create_requester

from tests import unittest


class RoomAccessTestCase(unittest.HomeserverTestCase):

    servlets = [
        admin.register_servlets,
        login.register_servlets,
        room.register_servlets,
        directory.register_servlets,
    ]

    def make_homeserver(self, reactor, clock):
        config = self.default_config()

        config["third_party_event_rules"] = {
            "module": "synapse.third_party_rules.access_rules.RoomAccessRules",
            "config": {
                "domains_forbidden_when_restricted": ["forbidden_domain"],
                "id_server": "testis",
                "freeze_room_with_no_admin": "true",
            },
        }
        config["trusted_third_party_id_servers"] = ["testis"]

        def send_invite(destination, room_id, event_id, pdu):
            return defer.succeed(pdu)

        def get_json(uri, args={}, headers=None):
            address_domain = args["address"].split("@")[1]
            return defer.succeed({"hs": address_domain})

        def post_json_get_json(uri, post_json, args={}, headers=None):
            token = "".join(random.choice(string.ascii_letters) for _ in range(10))
            return defer.succeed(
                {
                    "token": token,
                    "public_keys": [
                        {
                            "public_key": "serverpublickey",
                            "key_validity_url": "https://testis/pubkey/isvalid",
                        },
                        {
                            "public_key": "phemeralpublickey",
                            "key_validity_url": "https://testis/pubkey/ephemeral/isvalid",
                        },
                    ],
                    "display_name": "f...@b...",
                }
            )

        mock_federation_client = Mock(spec=["send_invite"])
        mock_federation_client.send_invite.side_effect = send_invite

        mock_http_client = Mock(spec=["get_json", "post_json_get_json"],)
        # Mocking the response for /info on the IS API.
        mock_http_client.get_json.side_effect = get_json
        # Mocking the response for /store-invite on the IS API.
        mock_http_client.post_json_get_json.side_effect = post_json_get_json
        self.hs = self.setup_test_homeserver(
            config=config,
            federation_client=mock_federation_client,
            simple_http_client=mock_http_client,
        )

        # TODO: This class does not use a singleton to get it's http client
        # This should be fixed for easier testing
        # https://github.com/matrix-org/synapse-dinsic/issues/26
        self.hs.get_identity_handler().blacklisting_http_client = mock_http_client

        self.third_party_event_rules = self.hs.get_third_party_event_rules()

        return self.hs

    def prepare(self, reactor, clock, homeserver):
        self.user_id = self.register_user("kermit", "monkey")
        self.tok = self.login("kermit", "monkey")

        self.restricted_room = self.create_room()
        self.unrestricted_room = self.create_room(rule=AccessRules.UNRESTRICTED)
        self.direct_rooms = [
            self.create_room(direct=True),
            self.create_room(direct=True),
            self.create_room(direct=True),
        ]

        self.invitee_id = self.register_user("invitee", "test")
        self.invitee_tok = self.login("invitee", "test")

        self.helper.invite(
            room=self.direct_rooms[0],
            src=self.user_id,
            targ=self.invitee_id,
            tok=self.tok,
        )

    def test_create_room_no_rule(self):
        """Tests that creating a room with no rule will set the default."""
        room_id = self.create_room()
        rule = self.current_rule_in_room(room_id)

        self.assertEqual(rule, AccessRules.RESTRICTED)

    def test_create_room_direct_no_rule(self):
        """Tests that creating a direct room with no rule will set the default."""
        room_id = self.create_room(direct=True)
        rule = self.current_rule_in_room(room_id)

        self.assertEqual(rule, AccessRules.DIRECT)

    def test_create_room_valid_rule(self):
        """Tests that creating a room with a valid rule will set the right."""
        room_id = self.create_room(rule=AccessRules.UNRESTRICTED)
        rule = self.current_rule_in_room(room_id)

        self.assertEqual(rule, AccessRules.UNRESTRICTED)

    def test_create_room_invalid_rule(self):
        """Tests that creating a room with an invalid rule will set fail."""
        self.create_room(rule=AccessRules.DIRECT, expected_code=400)

    def test_create_room_direct_invalid_rule(self):
        """Tests that creating a direct room with an invalid rule will fail.
        """
        self.create_room(direct=True, rule=AccessRules.RESTRICTED, expected_code=400)

    def test_create_room_default_power_level_rules(self):
        """Tests that a room created with no power level overrides instead uses the dinum
        defaults
        """
        room_id = self.create_room(direct=True, rule=AccessRules.DIRECT)
        power_levels = self.helper.get_state(room_id, "m.room.power_levels", self.tok)

        # Inviting another user should require PL50, even in private rooms
        self.assertEqual(power_levels["invite"], 50)
        # Sending arbitrary state events should require PL100
        self.assertEqual(power_levels["state_default"], 100)

    def test_create_room_fails_on_incorrect_power_level_rules(self):
        """Tests that a room created with power levels lower than that required are rejected"""
        modified_power_levels = RoomAccessRules._get_default_power_levels(self.user_id)
        modified_power_levels["invite"] = 0
        modified_power_levels["state_default"] = 50

        self.create_room(
            direct=True,
            rule=AccessRules.DIRECT,
            initial_state=[
                {"type": "m.room.power_levels", "content": modified_power_levels}
            ],
            expected_code=400,
        )

    def test_create_room_with_missing_power_levels_use_default_values(self):
        """
        Tests that a room created with custom power levels, but without defining invite or state_default
        succeeds, but the missing values are replaced with the defaults.
        """

        # Attempt to create a room without defining "invite" or "state_default"
        modified_power_levels = RoomAccessRules._get_default_power_levels(self.user_id)
        del modified_power_levels["invite"]
        del modified_power_levels["state_default"]
        room_id = self.create_room(
            direct=True,
            rule=AccessRules.DIRECT,
            initial_state=[
                {"type": "m.room.power_levels", "content": modified_power_levels}
            ],
        )

        # This should succeed, but the defaults should be put in place instead
        room_power_levels = self.helper.get_state(
            room_id, "m.room.power_levels", self.tok
        )
        self.assertEqual(room_power_levels["invite"], 50)
        self.assertEqual(room_power_levels["state_default"], 100)

        # And now the same test, but using power_levels_content_override instead
        # of initial_state (which takes a slightly different codepath)
        modified_power_levels = RoomAccessRules._get_default_power_levels(self.user_id)
        del modified_power_levels["invite"]
        del modified_power_levels["state_default"]
        room_id = self.create_room(
            direct=True,
            rule=AccessRules.DIRECT,
            power_levels_content_override=modified_power_levels,
        )

        # This should succeed, but the defaults should be put in place instead
        room_power_levels = self.helper.get_state(
            room_id, "m.room.power_levels", self.tok
        )
        self.assertEqual(room_power_levels["invite"], 50)
        self.assertEqual(room_power_levels["state_default"], 100)

    def test_existing_room_can_change_power_levels(self):
        """Tests that a room created with default power levels can have their power levels
        dropped after room creation
        """
        # Creates a room with the default power levels
        room_id = self.create_room(
            direct=True, rule=AccessRules.DIRECT, expected_code=200,
        )

        # Attempt to drop invite and state_default power levels after the fact
        room_power_levels = self.helper.get_state(
            room_id, "m.room.power_levels", self.tok
        )
        room_power_levels["invite"] = 0
        room_power_levels["state_default"] = 50
        self.helper.send_state(
            room_id, "m.room.power_levels", room_power_levels, self.tok
        )

    def test_public_room(self):
        """Tests that it's only possible to have a room listed in the public room list
        if the access rule is restricted.
        """
        # Creating a room with the public_chat preset should succeed and set the access
        # rule to restricted.
        preset_room_id = self.create_room(preset=RoomCreationPreset.PUBLIC_CHAT)
        self.assertEqual(
            self.current_rule_in_room(preset_room_id), AccessRules.RESTRICTED
        )

        # Creating a room with the public join rule in its initial state should succeed
        # and set the access rule to restricted.
        init_state_room_id = self.create_room(
            initial_state=[
                {
                    "type": "m.room.join_rules",
                    "content": {"join_rule": JoinRules.PUBLIC},
                }
            ]
        )
        self.assertEqual(
            self.current_rule_in_room(init_state_room_id), AccessRules.RESTRICTED
        )

        # List preset_room_id in the public room list
        request, channel = self.make_request(
            "PUT",
            "/_matrix/client/r0/directory/list/room/%s" % (preset_room_id,),
            {"visibility": "public"},
            access_token=self.tok,
        )
        self.assertEqual(channel.code, 200, channel.result)

        # List init_state_room_id in the public room list
        request, channel = self.make_request(
            "PUT",
            "/_matrix/client/r0/directory/list/room/%s" % (init_state_room_id,),
            {"visibility": "public"},
            access_token=self.tok,
        )
        self.assertEqual(channel.code, 200, channel.result)

        # Changing access rule to unrestricted should fail.
        self.change_rule_in_room(
            preset_room_id, AccessRules.UNRESTRICTED, expected_code=403
        )
        self.change_rule_in_room(
            init_state_room_id, AccessRules.UNRESTRICTED, expected_code=403
        )

        # Changing access rule to direct should fail.
        self.change_rule_in_room(preset_room_id, AccessRules.DIRECT, expected_code=403)
        self.change_rule_in_room(
            init_state_room_id, AccessRules.DIRECT, expected_code=403
        )

        # Creating a new room with the public_chat preset and an access rule of direct
        # should fail.
        self.create_room(
            preset=RoomCreationPreset.PUBLIC_CHAT,
            rule=AccessRules.DIRECT,
            expected_code=400,
        )

        # Changing join rule to public in an direct room should fail.
        self.change_join_rule_in_room(
            self.direct_rooms[0], JoinRules.PUBLIC, expected_code=403
        )

    def test_restricted(self):
        """Tests that in restricted mode we're unable to invite users from blacklisted
        servers but can invite other users.

        Also tests that the room can be published to, and removed from, the public room
        list.
        """
        # We can't invite a user from a forbidden HS.
        self.helper.invite(
            room=self.restricted_room,
            src=self.user_id,
            targ="@test:forbidden_domain",
            tok=self.tok,
            expect_code=403,
        )

        # We can invite a user which HS isn't forbidden.
        self.helper.invite(
            room=self.restricted_room,
            src=self.user_id,
            targ="@test:allowed_domain",
            tok=self.tok,
            expect_code=200,
        )

        # We can't send a 3PID invite to an address that is mapped to a forbidden HS.
        self.send_threepid_invite(
            address="test@forbidden_domain",
            room_id=self.restricted_room,
            expected_code=403,
        )

        # We can send a 3PID invite to an address that is mapped to an HS that's not
        # forbidden.
        self.send_threepid_invite(
            address="test@allowed_domain",
            room_id=self.restricted_room,
            expected_code=200,
        )

        # We are allowed to publish the room to the public room list
        url = "/_matrix/client/r0/directory/list/room/%s" % self.restricted_room
        data = {"visibility": "public"}

        request, channel = self.make_request("PUT", url, data, access_token=self.tok)
        self.assertEqual(channel.code, 200, channel.result)

        # We are allowed to remove the room from the public room list
        url = "/_matrix/client/r0/directory/list/room/%s" % self.restricted_room
        data = {"visibility": "private"}

        request, channel = self.make_request("PUT", url, data, access_token=self.tok)
        self.assertEqual(channel.code, 200, channel.result)

    def test_direct(self):
        """Tests that, in direct mode, other users than the initial two can't be invited,
        but the following scenario works:
          * invited user joins the room
          * invited user leaves the room
          * room creator re-invites invited user

        Tests that a user from a HS that's in the list of forbidden domains (to use
        in restricted mode) can be invited.

        Tests that the room cannot be published to the public room list.
        """
        not_invited_user = "@not_invited:forbidden_domain"

        # We can't invite a new user to the room.
        self.helper.invite(
            room=self.direct_rooms[0],
            src=self.user_id,
            targ=not_invited_user,
            tok=self.tok,
            expect_code=403,
        )

        # The invited user can join the room.
        self.helper.join(
            room=self.direct_rooms[0],
            user=self.invitee_id,
            tok=self.invitee_tok,
            expect_code=200,
        )

        # The invited user can leave the room.
        self.helper.leave(
            room=self.direct_rooms[0],
            user=self.invitee_id,
            tok=self.invitee_tok,
            expect_code=200,
        )

        # The invited user can be re-invited to the room.
        self.helper.invite(
            room=self.direct_rooms[0],
            src=self.user_id,
            targ=self.invitee_id,
            tok=self.tok,
            expect_code=200,
        )

        # If we're alone in the room and have always been the only member, we can invite
        # someone.
        self.helper.invite(
            room=self.direct_rooms[1],
            src=self.user_id,
            targ=not_invited_user,
            tok=self.tok,
            expect_code=200,
        )

        # Disable the 3pid invite ratelimiter
        burst = self.hs.config.rc_third_party_invite.burst_count
        per_second = self.hs.config.rc_third_party_invite.per_second
        self.hs.config.rc_third_party_invite.burst_count = 10
        self.hs.config.rc_third_party_invite.per_second = 0.1

        # We can't send a 3PID invite to a room that already has two members.
        self.send_threepid_invite(
            address="test@allowed_domain",
            room_id=self.direct_rooms[0],
            expected_code=403,
        )

        # We can't send a 3PID invite to a room that already has a pending invite.
        self.send_threepid_invite(
            address="test@allowed_domain",
            room_id=self.direct_rooms[1],
            expected_code=403,
        )

        # We can send a 3PID invite to a room in which we've always been the only member.
        self.send_threepid_invite(
            address="test@forbidden_domain",
            room_id=self.direct_rooms[2],
            expected_code=200,
        )

        # We can send a 3PID invite to a room in which there's a 3PID invite.
        self.send_threepid_invite(
            address="test@forbidden_domain",
            room_id=self.direct_rooms[2],
            expected_code=403,
        )

        self.hs.config.rc_third_party_invite.burst_count = burst
        self.hs.config.rc_third_party_invite.per_second = per_second

        # We can't publish the room to the public room list
        url = "/_matrix/client/r0/directory/list/room/%s" % self.direct_rooms[0]
        data = {"visibility": "public"}

        request, channel = self.make_request("PUT", url, data, access_token=self.tok)
        self.assertEqual(channel.code, 403, channel.result)

    def test_unrestricted(self):
        """Tests that, in unrestricted mode, we can invite whoever we want, but we can
        only change the power level of users that wouldn't be forbidden in restricted
        mode.

        Tests that the room cannot be published to the public room list.
        """
        # We can invite
        self.helper.invite(
            room=self.unrestricted_room,
            src=self.user_id,
            targ="@test:forbidden_domain",
            tok=self.tok,
            expect_code=200,
        )

        self.helper.invite(
            room=self.unrestricted_room,
            src=self.user_id,
            targ="@test:not_forbidden_domain",
            tok=self.tok,
            expect_code=200,
        )

        # We can send a 3PID invite to an address that is mapped to a forbidden HS.
        self.send_threepid_invite(
            address="test@forbidden_domain",
            room_id=self.unrestricted_room,
            expected_code=200,
        )

        # We can send a 3PID invite to an address that is mapped to an HS that's not
        # forbidden.
        self.send_threepid_invite(
            address="test@allowed_domain",
            room_id=self.unrestricted_room,
            expected_code=200,
        )

        # We can send a power level event that doesn't redefine the default PL or set a
        # non-default PL for a user that would be forbidden in restricted mode.
        self.helper.send_state(
            room_id=self.unrestricted_room,
            event_type=EventTypes.PowerLevels,
            body={"users": {self.user_id: 100, "@test:not_forbidden_domain": 10}},
            tok=self.tok,
            expect_code=200,
        )

        # We can't send a power level event that redefines the default PL and doesn't set
        # a non-default PL for a user that would be forbidden in restricted mode.
        self.helper.send_state(
            room_id=self.unrestricted_room,
            event_type=EventTypes.PowerLevels,
            body={
                "users": {self.user_id: 100, "@test:not_forbidden_domain": 10},
                "users_default": 10,
            },
            tok=self.tok,
            expect_code=403,
        )

        # We can't send a power level event that doesn't redefines the default PL but sets
        # a non-default PL for a user that would be forbidden in restricted mode.
        self.helper.send_state(
            room_id=self.unrestricted_room,
            event_type=EventTypes.PowerLevels,
            body={"users": {self.user_id: 100, "@test:forbidden_domain": 10}},
            tok=self.tok,
            expect_code=403,
        )

        # We can't publish the room to the public room list
        url = "/_matrix/client/r0/directory/list/room/%s" % self.unrestricted_room
        data = {"visibility": "public"}

        request, channel = self.make_request("PUT", url, data, access_token=self.tok)
        self.assertEqual(channel.code, 403, channel.result)

    def test_change_rules(self):
        """Tests that we can only change the current rule from restricted to
        unrestricted.
        """
        # We can't change the rule from restricted to direct.
        self.change_rule_in_room(
            room_id=self.restricted_room, new_rule=AccessRules.DIRECT, expected_code=403
        )

        # We can change the rule from restricted to unrestricted.
        # Note that this changes self.restricted_room to an unrestricted room
        self.change_rule_in_room(
            room_id=self.restricted_room,
            new_rule=AccessRules.UNRESTRICTED,
            expected_code=200,
        )

        # We can't change the rule from unrestricted to restricted.
        self.change_rule_in_room(
            room_id=self.unrestricted_room,
            new_rule=AccessRules.RESTRICTED,
            expected_code=403,
        )

        # We can't change the rule from unrestricted to direct.
        self.change_rule_in_room(
            room_id=self.unrestricted_room,
            new_rule=AccessRules.DIRECT,
            expected_code=403,
        )

        # We can't change the rule from direct to restricted.
        self.change_rule_in_room(
            room_id=self.direct_rooms[0],
            new_rule=AccessRules.RESTRICTED,
            expected_code=403,
        )

        # We can't change the rule from direct to unrestricted.
        self.change_rule_in_room(
            room_id=self.direct_rooms[0],
            new_rule=AccessRules.UNRESTRICTED,
            expected_code=403,
        )

        # We can't publish a room to the public room list and then change its rule to
        # unrestricted

        # Create a restricted room
        test_room_id = self.create_room(rule=AccessRules.RESTRICTED)

        # Publish the room to the public room list
        url = "/_matrix/client/r0/directory/list/room/%s" % test_room_id
        data = {"visibility": "public"}

        request, channel = self.make_request("PUT", url, data, access_token=self.tok)
        self.assertEqual(channel.code, 200, channel.result)

        # Attempt to switch the room to "unrestricted"
        self.change_rule_in_room(
            room_id=test_room_id, new_rule=AccessRules.UNRESTRICTED, expected_code=403
        )

        # Attempt to switch the room to "direct"
        self.change_rule_in_room(
            room_id=test_room_id, new_rule=AccessRules.DIRECT, expected_code=403
        )

    def test_change_room_avatar(self):
        """Tests that changing the room avatar is always allowed unless the room is a
        direct chat, in which case it's forbidden.
        """

        avatar_content = {
            "info": {"h": 398, "mimetype": "image/jpeg", "size": 31037, "w": 394},
            "url": "mxc://example.org/JWEIFJgwEIhweiWJE",
        }

        self.helper.send_state(
            room_id=self.restricted_room,
            event_type=EventTypes.RoomAvatar,
            body=avatar_content,
            tok=self.tok,
            expect_code=200,
        )

        self.helper.send_state(
            room_id=self.unrestricted_room,
            event_type=EventTypes.RoomAvatar,
            body=avatar_content,
            tok=self.tok,
            expect_code=200,
        )

        self.helper.send_state(
            room_id=self.direct_rooms[0],
            event_type=EventTypes.RoomAvatar,
            body=avatar_content,
            tok=self.tok,
            expect_code=403,
        )

    def test_change_room_name(self):
        """Tests that changing the room name is always allowed unless the room is a direct
        chat, in which case it's forbidden.
        """

        name_content = {"name": "My super room"}

        self.helper.send_state(
            room_id=self.restricted_room,
            event_type=EventTypes.Name,
            body=name_content,
            tok=self.tok,
            expect_code=200,
        )

        self.helper.send_state(
            room_id=self.unrestricted_room,
            event_type=EventTypes.Name,
            body=name_content,
            tok=self.tok,
            expect_code=200,
        )

        self.helper.send_state(
            room_id=self.direct_rooms[0],
            event_type=EventTypes.Name,
            body=name_content,
            tok=self.tok,
            expect_code=403,
        )

    def test_change_room_topic(self):
        """Tests that changing the room topic is always allowed unless the room is a
        direct chat, in which case it's forbidden.
        """

        topic_content = {"topic": "Welcome to this room"}

        self.helper.send_state(
            room_id=self.restricted_room,
            event_type=EventTypes.Topic,
            body=topic_content,
            tok=self.tok,
            expect_code=200,
        )

        self.helper.send_state(
            room_id=self.unrestricted_room,
            event_type=EventTypes.Topic,
            body=topic_content,
            tok=self.tok,
            expect_code=200,
        )

        self.helper.send_state(
            room_id=self.direct_rooms[0],
            event_type=EventTypes.Topic,
            body=topic_content,
            tok=self.tok,
            expect_code=403,
        )

    def test_revoke_3pid_invite_direct(self):
        """Tests that revoking a 3PID invite doesn't cause the room access rules module to
        confuse the revokation as a new 3PID invite.
        """
        invite_token = "sometoken"

        invite_body = {
            "display_name": "ker...@exa...",
            "public_keys": [
                {
                    "key_validity_url": "https://validity_url",
                    "public_key": "ta8IQ0u1sp44HVpxYi7dFOdS/bfwDjcy4xLFlfY5KOA",
                },
                {
                    "key_validity_url": "https://validity_url",
                    "public_key": "4_9nzEeDwR5N9s51jPodBiLnqH43A2_g2InVT137t9I",
                },
            ],
            "key_validity_url": "https://validity_url",
            "public_key": "ta8IQ0u1sp44HVpxYi7dFOdS/bfwDjcy4xLFlfY5KOA",
        }

        self.send_state_with_state_key(
            room_id=self.direct_rooms[1],
            event_type=EventTypes.ThirdPartyInvite,
            state_key=invite_token,
            body=invite_body,
            tok=self.tok,
        )

        self.send_state_with_state_key(
            room_id=self.direct_rooms[1],
            event_type=EventTypes.ThirdPartyInvite,
            state_key=invite_token,
            body={},
            tok=self.tok,
        )

        invite_token = "someothertoken"

        self.send_state_with_state_key(
            room_id=self.direct_rooms[1],
            event_type=EventTypes.ThirdPartyInvite,
            state_key=invite_token,
            body=invite_body,
            tok=self.tok,
        )

    def test_check_event_allowed(self):
        """Tests that RoomAccessRules.check_event_allowed behaves accordingly.

        It tests that:
            * forbidden users cannot join restricted rooms.
            * forbidden users can only join unrestricted rooms if they have an invite.
        """
        event_creator = self.hs.get_event_creation_handler()

        # Test that forbidden users cannot join restricted rooms
        requester = create_requester(self.user_id)
        allowed_requester = create_requester("@user:allowed_domain")
        forbidden_requester = create_requester("@user:forbidden_domain")

        # Assert a join event from a forbidden user to a restricted room is rejected
        self.get_failure(
            event_creator.create_event(
                forbidden_requester,
                {
                    "type": EventTypes.Member,
                    "room_id": self.restricted_room,
                    "sender": forbidden_requester.user.to_string(),
                    "content": {"membership": Membership.JOIN},
                    "state_key": forbidden_requester.user.to_string(),
                },
            ),
            SynapseError,
        )

        # A join event from an non-forbidden user to a restricted room is allowed
        self.get_success(
            event_creator.create_event(
                allowed_requester,
                {
                    "type": EventTypes.Member,
                    "room_id": self.restricted_room,
                    "sender": allowed_requester.user.to_string(),
                    "content": {"membership": Membership.JOIN},
                    "state_key": allowed_requester.user.to_string(),
                },
            )
        )

        # Test that forbidden users can only join unrestricted rooms if they have an invite

        # A forbidden user without an invite should not be able to join an unrestricted room
        self.get_failure(
            event_creator.create_event(
                forbidden_requester,
                {
                    "type": EventTypes.Member,
                    "room_id": self.unrestricted_room,
                    "sender": forbidden_requester.user.to_string(),
                    "content": {"membership": Membership.JOIN},
                    "state_key": forbidden_requester.user.to_string(),
                },
            ),
            SynapseError,
        )

        # However, if we then invite this user...
        self.helper.invite(
            room=self.unrestricted_room,
            src=requester.user.to_string(),
            targ=forbidden_requester.user.to_string(),
            tok=self.tok,
        )

        # And create another join event, making sure that its context states it's coming
        # in after the above invite was made...
        # Then the forbidden user should be able to join!
        self.get_success(
            event_creator.create_event(
                forbidden_requester,
                {
                    "type": EventTypes.Member,
                    "room_id": self.unrestricted_room,
                    "sender": forbidden_requester.user.to_string(),
                    "content": {"membership": Membership.JOIN},
                    "state_key": forbidden_requester.user.to_string(),
                },
            )
        )

    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.
        """

        def freeze_room_with_id_and_power_levels(
            room_id: str, custom_power_levels_content: Optional[JsonDict] = None,
        ):
            # Invite a user to the room, they join with PL 0
            self.helper.invite(
                room=room_id, src=self.user_id, targ=self.invitee_id, tok=self.tok,
            )

            # Invitee joins the room
            self.helper.join(
                room=room_id, user=self.invitee_id, tok=self.invitee_tok,
            )

            if not custom_power_levels_content:
                # Retrieve the room's current power levels event content
                power_levels = self.helper.get_state(
                    room_id=room_id, event_type="m.room.power_levels", tok=self.tok,
                )
            else:
                power_levels = custom_power_levels_content

                # Override the room's power levels with the given power levels content
                self.helper.send_state(
                    room_id=room_id,
                    event_type="m.room.power_levels",
                    body=custom_power_levels_content,
                    tok=self.tok,
                )

            # Ensure that the invitee leaving the room does not change the power levels
            self.helper.leave(
                room=room_id, 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=room_id, 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=room_id, src=self.user_id, targ=self.invitee_id, tok=self.tok,
            )

            # Invitee joins the room
            self.helper.join(
                room=room_id, user=self.invitee_id, tok=self.invitee_tok,
            )

            # Now the admin leaves the room
            self.helper.leave(
                room=room_id, user=self.user_id, tok=self.tok,
            )

            # Check the power levels again
            new_power_levels = self.helper.get_state(
                room_id=room_id, 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)

        # Test that freezing a room with the default power level state event content works
        room1 = self.create_room()
        freeze_room_with_id_and_power_levels(room1)

        # Test that freezing a room with a power level state event that is missing
        # `state_default` and `event_default` keys behaves as expected
        room2 = self.create_room()
        freeze_room_with_id_and_power_levels(
            room2,
            {
                "ban": 50,
                "events": {
                    "m.room.avatar": 50,
                    "m.room.canonical_alias": 50,
                    "m.room.history_visibility": 100,
                    "m.room.name": 50,
                    "m.room.power_levels": 100,
                },
                "invite": 0,
                "kick": 50,
                "redact": 50,
                "users": {self.user_id: 100},
                "users_default": 0,
                # Explicitly remove `state_default` and `event_default` keys
            },
        )

        # Test that freezing a room with a power level state event that is *additionally*
        # missing `ban`, `invite`, `kick` and `redact` keys behaves as expected
        room3 = self.create_room()
        freeze_room_with_id_and_power_levels(
            room3,
            {
                "events": {
                    "m.room.avatar": 50,
                    "m.room.canonical_alias": 50,
                    "m.room.history_visibility": 100,
                    "m.room.name": 50,
                    "m.room.power_levels": 100,
                },
                "users": {self.user_id: 100},
                "users_default": 0,
                # Explicitly remove `state_default` and `event_default` keys
                # Explicitly remove `ban`, `invite`, `kick` and `redact` keys
            },
        )

    def create_room(
        self,
        direct=False,
        rule=None,
        preset=RoomCreationPreset.TRUSTED_PRIVATE_CHAT,
        initial_state=None,
        power_levels_content_override=None,
        expected_code=200,
    ):
        content = {"is_direct": direct, "preset": preset}

        if rule:
            content["initial_state"] = [
                {"type": ACCESS_RULES_TYPE, "state_key": "", "content": {"rule": rule}}
            ]

        if initial_state:
            if "initial_state" not in content:
                content["initial_state"] = []

            content["initial_state"] += initial_state

        if power_levels_content_override:
            content["power_levels_content_override"] = power_levels_content_override

        request, channel = self.make_request(
            "POST", "/_matrix/client/r0/createRoom", content, access_token=self.tok,
        )

        self.assertEqual(channel.code, expected_code, channel.result)

        if expected_code == 200:
            return channel.json_body["room_id"]

    def current_rule_in_room(self, room_id):
        request, channel = self.make_request(
            "GET",
            "/_matrix/client/r0/rooms/%s/state/%s" % (room_id, ACCESS_RULES_TYPE),
            access_token=self.tok,
        )

        self.assertEqual(channel.code, 200, channel.result)
        return channel.json_body["rule"]

    def change_rule_in_room(self, room_id, new_rule, expected_code=200):
        data = {"rule": new_rule}
        request, channel = self.make_request(
            "PUT",
            "/_matrix/client/r0/rooms/%s/state/%s" % (room_id, ACCESS_RULES_TYPE),
            json.dumps(data),
            access_token=self.tok,
        )

        self.assertEqual(channel.code, expected_code, channel.result)

    def change_join_rule_in_room(self, room_id, new_join_rule, expected_code=200):
        data = {"join_rule": new_join_rule}
        request, channel = self.make_request(
            "PUT",
            "/_matrix/client/r0/rooms/%s/state/%s" % (room_id, EventTypes.JoinRules),
            json.dumps(data),
            access_token=self.tok,
        )

        self.assertEqual(channel.code, expected_code, channel.result)

    def send_threepid_invite(self, address, room_id, expected_code=200):
        params = {"id_server": "testis", "medium": "email", "address": address}

        request, channel = self.make_request(
            "POST",
            "/_matrix/client/r0/rooms/%s/invite" % room_id,
            json.dumps(params),
            access_token=self.tok,
        )
        self.assertEqual(channel.code, expected_code, channel.result)

    def send_state_with_state_key(
        self, room_id, event_type, state_key, body, tok, expect_code=200
    ):
        path = "/_matrix/client/r0/rooms/%s/state/%s/%s" % (
            room_id,
            event_type,
            state_key,
        )

        request, channel = self.make_request(
            "PUT", path, json.dumps(body), access_token=tok
        )

        self.assertEqual(channel.code, expect_code, channel.result)

        return channel.json_body