kaslo/synapse
1
0
Fork 0
Code Wiki Activity

RoomAccessRules cleanup (#62)

Browse Source 
Various cleanups of the DINUM-specific `RoomAccessRules` module, including:

* Type hints
* Docstring cleanups
* Some code cleanups
This commit is contained in:
Andrew Morgan
2020-09-10 19:04:34 +01:00
committed by GitHub
parent ff91a451b1
commit 3d1c941a6e
3 changed files with 214 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog.d/62.misc Normal file
View File

@@ -0,0 +1 @@
Type hinting and other cleanups for `synapse.third_party_rules.access_rules`.

305
synapse/third_party_rules/access_rules.py
View File

@@ -12,25 +12,31 @@
# 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 email.utils
from typing import Dict, List, Optional, Tuple
from twisted.internet import defer
from synapse.api.constants import EventTypes, JoinRules, Membership, RoomCreationPreset
from synapse.api.errors import SynapseError
from synapse.config._base import ConfigError
from synapse.types import get_domain_from_id
from synapse.events import EventBase
from synapse.http.client import SimpleHttpClient
from synapse.types import Requester, StateMap, get_domain_from_id
ACCESS_RULES_TYPE = "im.vector.room.access_rules"
ACCESS_RULE_RESTRICTED = "restricted"
ACCESS_RULE_UNRESTRICTED = "unrestricted"
ACCESS_RULE_DIRECT = "direct"
class AccessRules:
DIRECT = "direct"
RESTRICTED = "restricted"
UNRESTRICTED = "unrestricted"
VALID_ACCESS_RULES = (
ACCESS_RULE_DIRECT,
ACCESS_RULE_RESTRICTED,
ACCESS_RULE_UNRESTRICTED,
AccessRules.DIRECT,
AccessRules.RESTRICTED,
AccessRules.UNRESTRICTED,
)
# Rules to which we need to apply the power levels restrictions.
@@ -44,7 +50,7 @@ VALID_ACCESS_RULES = (
# * the default power level for users (users_default) being set to anything other than 0.
# * a non-default power level being assigned to any user which would be forbidden from
# joining a restricted room.
RULES_WITH_RESTRICTED_POWER_LEVELS = (ACCESS_RULE_UNRESTRICTED,)
RULES_WITH_RESTRICTED_POWER_LEVELS = (AccessRules.UNRESTRICTED,)
class RoomAccessRules(object):
@@ -68,7 +74,7 @@ class RoomAccessRules(object):
Don't forget to consider if you can invite users from your own domain.
"""
def __init__(self, config, http_client):
def __init__(self, config: Dict, http_client: SimpleHttpClient):
self.http_client = http_client
self.id_server = config["id_server"]
@@ -78,18 +84,42 @@ class RoomAccessRules(object):
)
@staticmethod
def parse_config(config):
if "id_server" in config:
return config
else:
def parse_config(config: Dict) -> Dict:
"""Parses and validates the options specified in the homeserver config.
Args:
config: The config dict.
Returns:
The config dict.
Raises:
ConfigError: If there was an issue with the provided module configuration.
"""
if "id_server" not in config:
raise ConfigError("No IS for event rules TchapEventRules")
def on_create_room(self, requester, config, is_requester_admin) -> bool:
"""Implements synapse.events.ThirdPartyEventRules.on_create_room
return config
def on_create_room(
self, requester: Requester, config: Dict, is_requester_admin: bool,
) -> bool:
"""Implements synapse.events.ThirdPartyEventRules.on_create_room.
Checks if a im.vector.room.access_rules event is being set during room creation.
If yes, make sure the event is correct. Otherwise, append an event with the
default rule to the initial state.
Args:
requester: The user who is making the createRoom request.
config: The createRoom config dict provided by the user.
is_requester_admin: Whether the requester is a Synapse admin.
Returns:
Whether the request is allowed.
Raises:
SynapseError: If the createRoom config dict is invalid or its contents blocked.
"""
is_direct = config.get("is_direct")
preset = config.get("preset")
@@ -110,9 +140,8 @@ class RoomAccessRules(object):
if access_rule not in VALID_ACCESS_RULES:
raise SynapseError(400, "Invalid access rule")
# Make sure the rule is "direct" if the room is a direct chat.
if (is_direct and access_rule != ACCESS_RULE_DIRECT) or (
access_rule == ACCESS_RULE_DIRECT and not is_direct
if (is_direct and access_rule != AccessRules.DIRECT) or (
access_rule == AccessRules.DIRECT and not is_direct
):
raise SynapseError(400, "Invalid access rule")
@@ -123,12 +152,12 @@ class RoomAccessRules(object):
# If there's no access rules event in the initial state, create one with the
# default setting.
if is_direct:
default_rule = ACCESS_RULE_DIRECT
default_rule = AccessRules.DIRECT
else:
# If the default value for non-direct chat changes, we should make another
# case here for rooms created with either a "public" join_rule or the
# "public_chat" preset to make sure those keep defaulting to "restricted"
default_rule = ACCESS_RULE_RESTRICTED
default_rule = AccessRules.RESTRICTED
if not config.get("initial_state"):
config["initial_state"] = []
@@ -148,7 +177,7 @@ class RoomAccessRules(object):
# a "public" join rule, the access rule must be "restricted").
if (
join_rule == JoinRules.PUBLIC or preset == RoomCreationPreset.PUBLIC_CHAT
) and access_rule != ACCESS_RULE_RESTRICTED:
) and access_rule != AccessRules.RESTRICTED:
raise SynapseError(400, "Invalid access rule")
# Check if the creator can override values for the power levels.
@@ -170,30 +199,41 @@ class RoomAccessRules(object):
return True
@defer.inlineCallbacks
def check_threepid_can_be_invited(self, medium, address, state_events):
"""Implements synapse.events.ThirdPartyEventRules.check_threepid_can_be_invited
def check_threepid_can_be_invited(
self, medium: str, address: str, state_events: StateMap[EventBase],
) -> bool:
"""Implements synapse.events.ThirdPartyEventRules.check_threepid_can_be_invited.
Check if a threepid can be invited to the room via a 3PID invite given the current
rules and the threepid's address, by retrieving the HS it's mapped to from the
configured identity server, and checking if we can invite users from it.
Args:
medium: The medium of the threepid.
address: The address of the threepid.
state_events: A dict mapping (event type, state key) to state event.
State events in the room the threepid is being invited to.
Returns:
Whether the threepid invite is allowed.
"""
rule = self._get_rule_from_state(state_events)
if medium != "email":
defer.returnValue(False)
return False
if rule != ACCESS_RULE_RESTRICTED:
if rule != AccessRules.RESTRICTED:
# Only "restricted" requires filtering 3PID invites. We don't need to do
# anything for "direct" here, because only "restricted" requires filtering
# based on the HS the address is mapped to.
defer.returnValue(True)
return True
parsed_address = email.utils.parseaddr(address)[1]
if parsed_address != address:
# Avoid reproducing the security issue described here:
# https://matrix.org/blog/2019/04/18/security-update-sydent-1-0-2
# It's probably not worth it but let's just be overly safe here.
defer.returnValue(False)
return False
# Get the HS this address belongs to from the identity server.
res = yield self.http_client.get_json(
@@ -203,17 +243,27 @@ class RoomAccessRules(object):
# Look for a domain that's not forbidden from being invited.
if not res.get("hs"):
defer.returnValue(False)
return False
if res.get("hs") in self.domains_forbidden_when_restricted:
defer.returnValue(False)
return False
defer.returnValue(True)
return True
def check_event_allowed(self, event, state_events):
"""Implements synapse.events.ThirdPartyEventRules.check_event_allowed
def check_event_allowed(
self, event: EventBase, state_events: StateMap[EventBase],
) -> bool:
"""Implements synapse.events.ThirdPartyEventRules.check_event_allowed.
Checks the event's type and the current rule and calls the right function to
determine whether the event can be allowed.
Args:
event: The event to check.
state_events: A dict mapping (event type, state key) to state event.
State events in the room the event originated from.
Returns:
True if the event can be allowed, False otherwise.
"""
if event.type == ACCESS_RULES_TYPE:
return self._on_rules_change(event, state_events)
@@ -241,16 +291,19 @@ class RoomAccessRules(object):
return True
def _on_rules_change(self, event, state_events):
def _on_rules_change(
self, event: EventBase, state_events: StateMap[EventBase],
) -> bool:
"""Implement the checks and behaviour specified on allowing or forbidding a new
im.vector.room.access_rules event.
Args:
event (synapse.events.EventBase): The event to check.
state_events (dict[tuple[event type, state key], EventBase]): The state of the
room before the event was sent.
event: The event to check.
state_events: A dict mapping (event type, state key) to state event.
State events in the room before the event was sent.
Returns:
bool, True if the event can be allowed, False otherwise.
True if the event can be allowed, False otherwise.
"""
new_rule = event.content.get("rule")
@@ -261,11 +314,11 @@ class RoomAccessRules(object):
# We must not allow rooms with the "public" join rule to be given any other access
# rule than "restricted".
join_rule = self._get_join_rule_from_state(state_events)
if join_rule == JoinRules.PUBLIC and new_rule != ACCESS_RULE_RESTRICTED:
if join_rule == JoinRules.PUBLIC and new_rule != AccessRules.RESTRICTED:
return False
# Make sure we don't apply "direct" if the room has more than two members.
if new_rule == ACCESS_RULE_DIRECT:
if new_rule == AccessRules.DIRECT:
existing_members, threepid_tokens = self._get_members_and_tokens_from_state(
state_events
)
@@ -283,28 +336,30 @@ class RoomAccessRules(object):
prev_rule = prev_rules_event.content.get("rule")
# Currently, we can only go from "restricted" to "unrestricted".
if prev_rule == ACCESS_RULE_RESTRICTED and new_rule == ACCESS_RULE_UNRESTRICTED:
return True
return (
prev_rule == AccessRules.RESTRICTED and new_rule == AccessRules.UNRESTRICTED
)
return False
def _on_membership_or_invite(self, event, rule, state_events):
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
m.room.third_party_invite events.
Args:
event (synapse.events.EventBase): The event to check.
rule (str): The name of the rule to apply.
state_events (dict[tuple[event type, state key], EventBase]): The state of the
room before the event was sent.
event: The event to check.
rule: The name of the rule to apply.
state_events: A dict mapping (event type, state key) to state event.
The state of the room before the event was sent.
Returns:
bool, True if the event can be allowed, False otherwise.
True if the event can be allowed, False otherwise.
"""
if rule == ACCESS_RULE_RESTRICTED:
if rule == AccessRules.RESTRICTED:
ret = self._on_membership_or_invite_restricted(event)
elif rule == ACCESS_RULE_UNRESTRICTED:
elif rule == AccessRules.UNRESTRICTED:
ret = self._on_membership_or_invite_unrestricted()
elif rule == ACCESS_RULE_DIRECT:
elif rule == AccessRules.DIRECT:
ret = self._on_membership_or_invite_direct(event, state_events)
else:
# We currently apply the default (restricted) if we don't know the rule, we
@@ -313,16 +368,17 @@ class RoomAccessRules(object):
return ret
def _on_membership_or_invite_restricted(self, event):
def _on_membership_or_invite_restricted(self, event: EventBase) -> bool:
"""Implements the checks and behaviour specified for the "restricted" rule.
"restricted" currently means that users can only invite users if their server is
included in a limited list of domains.
Args:
event (synapse.events.EventBase): The event to check.
event: The event to check.
Returns:
bool, True if the event can be allowed, False otherwise.
True if the event can be allowed, False otherwise.
"""
# We're not applying the rules on m.room.third_party_member events here because
# the filtering on threepids is done in check_threepid_can_be_invited, which is
@@ -340,29 +396,31 @@ class RoomAccessRules(object):
invitee_domain = get_domain_from_id(event.state_key)
return invitee_domain not in self.domains_forbidden_when_restricted
def _on_membership_or_invite_unrestricted(self):
def _on_membership_or_invite_unrestricted(self) -> bool:
"""Implements the checks and behaviour specified for the "unrestricted" rule.
"unrestricted" currently means that every event is allowed.
Returns:
bool, True if the event can be allowed, False otherwise.
True if the event can be allowed, False otherwise.
"""
return True
def _on_membership_or_invite_direct(self, event, state_events):
def _on_membership_or_invite_direct(
self, event: EventBase, state_events: StateMap[EventBase],
) -> bool:
"""Implements the checks and behaviour specified for the "direct" rule.
"direct" currently means that no member is allowed apart from the two initial
members the room was created for (i.e. the room's creator and their first
invitee).
members the room was created for (i.e. the room's creator and their first invitee).
Args:
event (synapse.events.EventBase): The event to check.
state_events (dict[tuple[event type, state key], EventBase]): The state of the
room before the event was sent.
event: The event to check.
state_events: A dict mapping (event type, state key) to state event.
The state of the room before the event was sent.
Returns:
bool, True if the event can be allowed, False otherwise.
True if the event can be allowed, False otherwise.
"""
# Get the room memberships and 3PID invite tokens from the room's state.
existing_members, threepid_tokens = self._get_members_and_tokens_from_state(
@@ -406,14 +464,11 @@ class RoomAccessRules(object):
is_from_threepid_invite = self._is_invite_from_threepid(
event, threepid_tokens[0]
)
if is_from_threepid_invite or target == existing_members[0]:
return True
return False
return is_from_threepid_invite or target == existing_members[0]
return True
def _is_power_level_content_allowed(self, content, access_rule):
def _is_power_level_content_allowed(self, content: Dict, access_rule: str) -> bool:
"""Check if a given power levels event is permitted under the given access rule.
It shouldn't be allowed if it either changes the default PL to a non-0 value or
@@ -421,10 +476,11 @@ class RoomAccessRules(object):
under a more restrictive access rule.
Args:
content (dict[]): The content of the m.room.power_levels event to check.
access_rule (str): The access rule in place in this room.
content: The content of the m.room.power_levels event to check.
access_rule: The access rule in place in this room.
Returns:
bool, True if the event can be allowed, False otherwise.
Whether the content of the power levels event is valid.
"""
# Check if we need to apply the restrictions with the current rule.
if access_rule not in RULES_WITH_RESTRICTED_POWER_LEVELS:
@@ -448,10 +504,11 @@ class RoomAccessRules(object):
return True
def _on_join_rule_change(self, event, rule):
def _on_join_rule_change(self, event: EventBase, rule: str) -> bool:
"""Check whether a join rule change is allowed. A join rule change is always
allowed unless the new join rule is "public" and the current access rule isn't
"restricted".
The rationale is that external users (those whose server would be denied access
to rooms enforcing the "restricted" access rule) should always rely on non-
external users for access to rooms, therefore they shouldn't be able to access
@@ -465,100 +522,107 @@ class RoomAccessRules(object):
that in mind if we need to change the default access rule in the future.
Args:
event (synapse.events.EventBase): The event to check.
rule (str): The name of the rule to apply.
event: The event to check.
rule: The name of the rule to apply.
Returns:
bool, True if the event can be allowed, False otherwise.
Whether the change is allowed.
"""
if event.content.get("join_rule") == JoinRules.PUBLIC:
return rule == ACCESS_RULE_RESTRICTED
return rule == AccessRules.RESTRICTED
return True
def _on_room_avatar_change(self, event, rule):
def _on_room_avatar_change(self, event: EventBase, rule: str) -> bool:
"""Check whether a change of room avatar is allowed.
The current rule is to forbid such a change in direct chats but allow it
everywhere else.
Args:
event (synapse.events.EventBase): The event to check.
rule (str): The name of the rule to apply.
Returns:
bool, True if the event can be allowed, False otherwise.
"""
return rule != ACCESS_RULE_DIRECT
event: The event to check.
rule: The name of the rule to apply.
def _on_room_name_change(self, event, rule):
Returns:
True if the event can be allowed, False otherwise.
"""
return rule != AccessRules.DIRECT
def _on_room_name_change(self, event: EventBase, rule: str) -> bool:
"""Check whether a change of room name is allowed.
The current rule is to forbid such a change in direct chats but allow it
everywhere else.
Args:
event (synapse.events.EventBase): The event to check.
rule (str): The name of the rule to apply.
Returns:
bool, True if the event can be allowed, False otherwise.
"""
return rule != ACCESS_RULE_DIRECT
event: The event to check.
rule: The name of the rule to apply.
def _on_room_topic_change(self, event, rule):
Returns:
True if the event can be allowed, False otherwise.
"""
return rule != AccessRules.DIRECT
def _on_room_topic_change(self, event: EventBase, rule: str) -> bool:
"""Check whether a change of room topic is allowed.
The current rule is to forbid such a change in direct chats but allow it
everywhere else.
Args:
event (synapse.events.EventBase): The event to check.
rule (str): The name of the rule to apply.
event: The event to check.
rule: The name of the rule to apply.
Returns:
bool, True if the event can be allowed, False otherwise.
True if the event can be allowed, False otherwise.
"""
return rule != ACCESS_RULE_DIRECT
return rule != AccessRules.DIRECT
@staticmethod
def _get_rule_from_state(state_events):
def _get_rule_from_state(state_events: StateMap[EventBase]) -> Optional[str]:
"""Extract the rule to be applied from the given set of state events.
Args:
state_events (dict[tuple[event type, state key], EventBase]): The set of state
events.
state_events: A dict mapping (event type, state key) to state event.
Returns:
str, the name of the rule (either "direct", "restricted" or "unrestricted")
The name of the rule (either "direct", "restricted" or "unrestricted") if found,
else None.
"""
access_rules = state_events.get((ACCESS_RULES_TYPE, ""))
if access_rules is None:
rule = ACCESS_RULE_RESTRICTED
else:
rule = access_rules.content.get("rule")
return rule
return AccessRules.RESTRICTED
return access_rules.content.get("rule")
@staticmethod
def _get_join_rule_from_state(state_events):
def _get_join_rule_from_state(state_events: StateMap[EventBase]) -> Optional[str]:
"""Extract the room's join rule from the given set of state events.
Args:
state_events (dict[tuple[event type, state key], EventBase]): The set of state
events.
Returns:
str, the name of the join rule (either "public", or "invite")
The name of the join rule (either "public", or "invite") if found, else None.
"""
join_rule_event = state_events.get((EventTypes.JoinRules, ""))
if join_rule_event is None:
return None
return join_rule_event.content.get("join_rule")
@staticmethod
def _get_members_and_tokens_from_state(state_events):
"""Retrieves from a list of state events the list of users that have a
m.room.member event in the room, and the tokens of 3PID invites in the room.
def _get_members_and_tokens_from_state(
state_events: StateMap[EventBase],
) -> Tuple[List[str], List[str]]:
"""Retrieves the list of users that have a m.room.member event in the room,
as well as 3PID invites tokens in the room.
Args:
state_events (dict[tuple[event type, state key], EventBase]): The set of state
events.
state_events: A dict mapping (event type, state key) to state event.
Returns:
existing_members (list[str]): List of targets of the m.room.member events in
the state.
threepid_invite_tokens (list[str]): List of tokens of the 3PID invites in the
state.
A tuple containing the:
* targets of the m.room.member events in the state.
* 3PID invite tokens in the state.
"""
existing_members = []
threepid_invite_tokens = []
@@ -572,12 +636,15 @@ class RoomAccessRules(object):
return existing_members, threepid_invite_tokens
@staticmethod
def _is_invite_from_threepid(invite, threepid_invite_token):
def _is_invite_from_threepid(invite: EventBase, threepid_invite_token: str) -> bool:
"""Checks whether the given invite follows the given 3PID invite.
Args:
invite (EventBase): The m.room.member event with "invite" membership.
threepid_invite_token (str): The state key from the 3PID invite.
invite: The m.room.member event with "invite" membership.
threepid_invite_token: The state key from the 3PID invite.
Returns:
Whether the invite is due to the given 3PID invite.
"""
token = (
invite.content.get("third_party_invite", {})

61
tests/rest/client/test_room_access_rules.py
View File

@@ -12,8 +12,6 @@
# 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
@@ -25,12 +23,7 @@ from twisted.internet import defer
from synapse.api.constants import EventTypes, JoinRules, RoomCreationPreset
from synapse.rest import admin
from synapse.rest.client.v1 import login, room
from synapse.third_party_rules.access_rules import (
ACCESS_RULE_DIRECT,
ACCESS_RULE_RESTRICTED,
ACCESS_RULE_UNRESTRICTED,
ACCESS_RULES_TYPE,
)
from synapse.third_party_rules.access_rules import ACCESS_RULES_TYPE, AccessRules
from tests import unittest
@@ -109,7 +102,7 @@ class RoomAccessTestCase(unittest.HomeserverTestCase):
self.tok = self.login("kermit", "monkey")
self.restricted_room = self.create_room()
self.unrestricted_room = self.create_room(rule=ACCESS_RULE_UNRESTRICTED)
self.unrestricted_room = self.create_room(rule=AccessRules.UNRESTRICTED)
self.direct_rooms = [
self.create_room(direct=True),
self.create_room(direct=True),
@@ -127,34 +120,34 @@ class RoomAccessTestCase(unittest.HomeserverTestCase):
)
def test_create_room_no_rule(self):
"""Tests that creating a room with no rule will set the default value."""
"""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, ACCESS_RULE_RESTRICTED)
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 value."""
"""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, ACCESS_RULE_DIRECT)
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 value."""
room_id = self.create_room(rule=ACCESS_RULE_UNRESTRICTED)
"""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, ACCESS_RULE_UNRESTRICTED)
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=ACCESS_RULE_DIRECT, expected_code=400)
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=ACCESS_RULE_RESTRICTED, expected_code=400)
self.create_room(direct=True, rule=AccessRules.RESTRICTED, expected_code=400)
def test_public_room(self):
"""Tests that it's not possible to have a room with the public join rule and an
@@ -164,7 +157,7 @@ class RoomAccessTestCase(unittest.HomeserverTestCase):
# rule to restricted.
preset_room_id = self.create_room(preset=RoomCreationPreset.PUBLIC_CHAT)
self.assertEqual(
self.current_rule_in_room(preset_room_id), ACCESS_RULE_RESTRICTED
self.current_rule_in_room(preset_room_id), AccessRules.RESTRICTED
)
# Creating a room with the public join rule in its initial state should succeed
@@ -178,21 +171,21 @@ class RoomAccessTestCase(unittest.HomeserverTestCase):
]
)
self.assertEqual(
self.current_rule_in_room(init_state_room_id), ACCESS_RULE_RESTRICTED
self.current_rule_in_room(init_state_room_id), AccessRules.RESTRICTED
)
# Changing access rule to unrestricted should fail.
self.change_rule_in_room(
preset_room_id, ACCESS_RULE_UNRESTRICTED, expected_code=403
preset_room_id, AccessRules.UNRESTRICTED, expected_code=403
)
self.change_rule_in_room(
init_state_room_id, ACCESS_RULE_UNRESTRICTED, expected_code=403
init_state_room_id, AccessRules.UNRESTRICTED, expected_code=403
)
# Changing access rule to direct should fail.
self.change_rule_in_room(preset_room_id, ACCESS_RULE_DIRECT, expected_code=403)
self.change_rule_in_room(preset_room_id, AccessRules.DIRECT, expected_code=403)
self.change_rule_in_room(
init_state_room_id, ACCESS_RULE_DIRECT, expected_code=403
init_state_room_id, AccessRules.DIRECT, expected_code=403
)
# Changing join rule to public in an unrestricted room should fail.
@@ -208,12 +201,12 @@ class RoomAccessTestCase(unittest.HomeserverTestCase):
# restricted should fail.
self.create_room(
preset=RoomCreationPreset.PUBLIC_CHAT,
rule=ACCESS_RULE_UNRESTRICTED,
rule=AccessRules.UNRESTRICTED,
expected_code=400,
)
self.create_room(
preset=RoomCreationPreset.PUBLIC_CHAT,
rule=ACCESS_RULE_DIRECT,
rule=AccessRules.DIRECT,
expected_code=400,
)
@@ -226,7 +219,7 @@ class RoomAccessTestCase(unittest.HomeserverTestCase):
"content": {"join_rule": JoinRules.PUBLIC},
}
],
rule=ACCESS_RULE_UNRESTRICTED,
rule=AccessRules.UNRESTRICTED,
expected_code=400,
)
self.create_room(
@@ -236,7 +229,7 @@ class RoomAccessTestCase(unittest.HomeserverTestCase):
"content": {"join_rule": JoinRules.PUBLIC},
}
],
rule=ACCESS_RULE_DIRECT,
rule=AccessRules.DIRECT,
expected_code=400,
)
@@ -446,40 +439,40 @@ class RoomAccessTestCase(unittest.HomeserverTestCase):
# We can change the rule from restricted to unrestricted.
self.change_rule_in_room(
room_id=self.restricted_room,
new_rule=ACCESS_RULE_UNRESTRICTED,
new_rule=AccessRules.UNRESTRICTED,
expected_code=200,
)
# We can't change the rule from restricted to direct.
self.change_rule_in_room(
room_id=self.restricted_room, new_rule=ACCESS_RULE_DIRECT, expected_code=403
room_id=self.restricted_room, new_rule=AccessRules.DIRECT, expected_code=403
)
# We can't change the rule from unrestricted to restricted.
self.change_rule_in_room(
room_id=self.unrestricted_room,
new_rule=ACCESS_RULE_RESTRICTED,
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=ACCESS_RULE_DIRECT,
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=ACCESS_RULE_RESTRICTED,
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=ACCESS_RULE_UNRESTRICTED,
new_rule=AccessRules.UNRESTRICTED,
expected_code=403,
)