MirrorHub/synapse
0
0
Fork 1
mirror of https://mau.dev/maunium/synapse.git synced 2024-12-13 11:23:25 +01:00
Code Issues Projects Releases Wiki Activity

EventAuthTestCase: build events for the right room version

Browse source 
In practice, when we run the auth rules, all of the events have the right room
version. Let's stop building Room V1 events for these tests and use the right
version.
This commit is contained in:
Richard van der Hoff 2022-06-10 10:44:19 +01:00
parent 8a499d7a60
commit 2959184a42
1 changed files with 205 additions and 119 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

324
tests/test_event_auth.py
View file

@ -15,10 +15,12 @@
import unittest import unittest
from typing import Optional from typing import Optional
from parameterized import parameterized
from synapse import event_auth from synapse import event_auth
from synapse.api.constants import EventContentFields from synapse.api.constants import EventContentFields
from synapse.api.errors import AuthError from synapse.api.errors import AuthError
from synapse.api.room_versions import RoomVersions from synapse.api.room_versions import EventFormatVersions, RoomVersion, RoomVersions
from synapse.events import EventBase, make_event_from_dict from synapse.events import EventBase, make_event_from_dict
from synapse.types import JsonDict, get_domain_from_id from synapse.types import JsonDict, get_domain_from_id
@ -30,19 +32,23 @@ class EventAuthTestCase(unittest.TestCase):
""" """
creator = "@creator:example.com" creator = "@creator:example.com"
auth_events = [ auth_events = [
_create_event(creator), _create_event(RoomVersions.V9, creator),
_join_event(creator), _join_event(RoomVersions.V9, creator),
] ]
# creator should be able to send state # creator should be able to send state
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V9, RoomVersions.V9,
_random_state_event(creator), _random_state_event(RoomVersions.V9, creator),
auth_events, auth_events,
) )
# ... but a rejected join_rules event should cause it to be rejected # ... but a rejected join_rules event should cause it to be rejected
rejected_join_rules = _join_rules_event(creator, "public") rejected_join_rules = _join_rules_event(
RoomVersions.V9,
creator,
"public",
)
rejected_join_rules.rejected_reason = "stinky" rejected_join_rules.rejected_reason = "stinky"
auth_events.append(rejected_join_rules) auth_events.append(rejected_join_rules)
@ -50,18 +56,18 @@ class EventAuthTestCase(unittest.TestCase):
AuthError, AuthError,
event_auth.check_auth_rules_for_event, event_auth.check_auth_rules_for_event,
RoomVersions.V9, RoomVersions.V9,
_random_state_event(creator), _random_state_event(RoomVersions.V9, creator),
auth_events, auth_events,
) )
# ... even if there is *also* a good join rules # ... even if there is *also* a good join rules
auth_events.append(_join_rules_event(creator, "public")) auth_events.append(_join_rules_event(RoomVersions.V9, creator, "public"))
self.assertRaises( self.assertRaises(
AuthError, AuthError,
event_auth.check_auth_rules_for_event, event_auth.check_auth_rules_for_event,
RoomVersions.V9, RoomVersions.V9,
_random_state_event(creator), _random_state_event(RoomVersions.V9, creator),
auth_events, auth_events,
) )
@ -73,15 +79,15 @@ class EventAuthTestCase(unittest.TestCase):
creator = "@creator:example.com" creator = "@creator:example.com"
joiner = "@joiner:example.com" joiner = "@joiner:example.com"
auth_events = [ auth_events = [
_create_event(creator), _create_event(RoomVersions.V1, creator),
_join_event(creator), _join_event(RoomVersions.V1, creator),
_join_event(joiner), _join_event(RoomVersions.V1, joiner),
] ]
# creator should be able to send state # creator should be able to send state
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V1, RoomVersions.V1,
_random_state_event(creator), _random_state_event(RoomVersions.V1, creator),
auth_events, auth_events,
) )
@ -90,7 +96,7 @@ class EventAuthTestCase(unittest.TestCase):
AuthError, AuthError,
event_auth.check_auth_rules_for_event, event_auth.check_auth_rules_for_event,
RoomVersions.V1, RoomVersions.V1,
_random_state_event(joiner), _random_state_event(RoomVersions.V1, joiner),
auth_events, auth_events,
) )
@ -104,13 +110,15 @@ class EventAuthTestCase(unittest.TestCase):
king = "@joiner2:example.com" king = "@joiner2:example.com"
auth_events = [ auth_events = [
_create_event(creator), _create_event(RoomVersions.V1, creator),
_join_event(creator), _join_event(RoomVersions.V1, creator),
_power_levels_event( _power_levels_event(
creator, {"state_default": "30", "users": {pleb: "29", king: "30"}} RoomVersions.V1,
creator,
{"state_default": "30", "users": {pleb: "29", king: "30"}},
), ),
_join_event(pleb), _join_event(RoomVersions.V1, pleb),
_join_event(king), _join_event(RoomVersions.V1, king),
] ]
# pleb should not be able to send state # pleb should not be able to send state
@ -118,14 +126,14 @@ class EventAuthTestCase(unittest.TestCase):
AuthError, AuthError,
event_auth.check_auth_rules_for_event, event_auth.check_auth_rules_for_event,
RoomVersions.V1, RoomVersions.V1,
_random_state_event(pleb), _random_state_event(RoomVersions.V1, pleb),
auth_events, auth_events,
), ),
# king should be able to send state # king should be able to send state
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V1, RoomVersions.V1,
_random_state_event(king), _random_state_event(RoomVersions.V1, king),
auth_events, auth_events,
) )
@ -134,14 +142,14 @@ class EventAuthTestCase(unittest.TestCase):
creator = "@creator:example.com" creator = "@creator:example.com"
other = "@other:example.com" other = "@other:example.com"
auth_events = [ auth_events = [
_create_event(creator), _create_event(RoomVersions.V1, creator),
_join_event(creator), _join_event(RoomVersions.V1, creator),
] ]
# creator should be able to send aliases # creator should be able to send aliases
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V1, RoomVersions.V1,
_alias_event(creator), _alias_event(RoomVersions.V1, creator),
auth_events, auth_events,
) )
@ -149,7 +157,7 @@ class EventAuthTestCase(unittest.TestCase):
with self.assertRaises(AuthError): with self.assertRaises(AuthError):
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V1, RoomVersions.V1,
_alias_event(creator, state_key=""), _alias_event(RoomVersions.V1, creator, state_key=""),
auth_events, auth_events,
) )
@ -157,14 +165,14 @@ class EventAuthTestCase(unittest.TestCase):
with self.assertRaises(AuthError): with self.assertRaises(AuthError):
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V1, RoomVersions.V1,
_alias_event(creator, state_key="test.com"), _alias_event(RoomVersions.V1, creator, state_key="test.com"),
auth_events, auth_events,
) )
# Note that the member does *not* need to be in the room. # Note that the member does *not* need to be in the room.
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V1, RoomVersions.V1,
_alias_event(other), _alias_event(RoomVersions.V1, other),
auth_events, auth_events,
) )
@ -173,26 +181,26 @@ class EventAuthTestCase(unittest.TestCase):
creator = "@creator:example.com" creator = "@creator:example.com"
other = "@other:example.com" other = "@other:example.com"
auth_events = [ auth_events = [
_create_event(creator), _create_event(RoomVersions.V6, creator),
_join_event(creator), _join_event(RoomVersions.V6, creator),
] ]
# creator should be able to send aliases # creator should be able to send aliases
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V6, RoomVersions.V6,
_alias_event(creator), _alias_event(RoomVersions.V6, creator),
auth_events, auth_events,
) )
# No particular checks are done on the state key. # No particular checks are done on the state key.
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V6, RoomVersions.V6,
_alias_event(creator, state_key=""), _alias_event(RoomVersions.V6, creator, state_key=""),
auth_events, auth_events,
) )
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V6, RoomVersions.V6,
_alias_event(creator, state_key="test.com"), _alias_event(RoomVersions.V6, creator, state_key="test.com"),
auth_events, auth_events,
) )
@ -200,11 +208,12 @@ class EventAuthTestCase(unittest.TestCase):
with self.assertRaises(AuthError): with self.assertRaises(AuthError):
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V6, RoomVersions.V6,
_alias_event(other), _alias_event(RoomVersions.V6, other),
auth_events, auth_events,
) )
def test_msc2209(self): @parameterized.expand([(RoomVersions.V1, True), (RoomVersions.V6, False)])
def test_notifications(self, room_version: RoomVersion, allow_modification: bool):
""" """
Notifications power levels get checked due to MSC2209. Notifications power levels get checked due to MSC2209.
""" """
@ -212,28 +221,28 @@ class EventAuthTestCase(unittest.TestCase):
pleb = "@joiner:example.com" pleb = "@joiner:example.com"
auth_events = [ auth_events = [
_create_event(creator), _create_event(room_version, creator),
_join_event(creator), _join_event(room_version, creator),
_power_levels_event( _power_levels_event(
creator, {"state_default": "30", "users": {pleb: "30"}} room_version, creator, {"state_default": "30", "users": {pleb: "30"}}
), ),
_join_event(pleb), _join_event(room_version, pleb),
] ]
# pleb should be able to modify the notifications power level. pl_event = _power_levels_event(
event_auth.check_auth_rules_for_event( room_version, pleb, {"notifications": {"room": 100}}
RoomVersions.V1,
_power_levels_event(pleb, {"notifications": {"room": 100}}),
auth_events,
) )
# But an MSC2209 room rejects this change. # on room V1, pleb should be able to modify the notifications power level.
with self.assertRaises(AuthError): if allow_modification:
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(room_version, pl_event, auth_events)
RoomVersions.V6,
_power_levels_event(pleb, {"notifications": {"room": 100}}), else:
auth_events, # But an MSC2209 room rejects this change.
) with self.assertRaises(AuthError):
event_auth.check_auth_rules_for_event(
room_version, pl_event, auth_events
)
def test_join_rules_public(self): def test_join_rules_public(self):
""" """
@ -243,15 +252,17 @@ class EventAuthTestCase(unittest.TestCase):
pleb = "@joiner:example.com" pleb = "@joiner:example.com"
auth_events = { auth_events = {
("m.room.create", ""): _create_event(creator), ("m.room.create", ""): _create_event(RoomVersions.V6, creator),
("m.room.member", creator): _join_event(creator), ("m.room.member", creator): _join_event(RoomVersions.V6, creator),
("m.room.join_rules", ""): _join_rules_event(creator, "public"), ("m.room.join_rules", ""): _join_rules_event(
RoomVersions.V6, creator, "public"
),
} }
# Check join. # Check join.
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V6, RoomVersions.V6,
_join_event(pleb), _join_event(RoomVersions.V6, pleb),
auth_events.values(), auth_events.values(),
) )
@ -259,42 +270,48 @@ class EventAuthTestCase(unittest.TestCase):
with self.assertRaises(AuthError): with self.assertRaises(AuthError):
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V6, RoomVersions.V6,
_member_event(pleb, "join", sender=creator), _member_event(RoomVersions.V6, pleb, "join", sender=creator),
auth_events.values(), auth_events.values(),
) )
# Banned should be rejected. # Banned should be rejected.
auth_events[("m.room.member", pleb)] = _member_event(pleb, "ban") auth_events[("m.room.member", pleb)] = _member_event(
RoomVersions.V6, pleb, "ban"
)
with self.assertRaises(AuthError): with self.assertRaises(AuthError):
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V6, RoomVersions.V6,
_join_event(pleb), _join_event(RoomVersions.V6, pleb),
auth_events.values(), auth_events.values(),
) )
# A user who left can re-join. # A user who left can re-join.
auth_events[("m.room.member", pleb)] = _member_event(pleb, "leave") auth_events[("m.room.member", pleb)] = _member_event(
RoomVersions.V6, pleb, "leave"
)
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V6, RoomVersions.V6,
_join_event(pleb), _join_event(RoomVersions.V6, pleb),
auth_events.values(), auth_events.values(),
) )
# A user can send a join if they're in the room. # A user can send a join if they're in the room.
auth_events[("m.room.member", pleb)] = _member_event(pleb, "join") auth_events[("m.room.member", pleb)] = _member_event(
RoomVersions.V6, pleb, "join"
)
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V6, RoomVersions.V6,
_join_event(pleb), _join_event(RoomVersions.V6, pleb),
auth_events.values(), auth_events.values(),
) )
# A user can accept an invite. # A user can accept an invite.
auth_events[("m.room.member", pleb)] = _member_event( auth_events[("m.room.member", pleb)] = _member_event(
pleb, "invite", sender=creator RoomVersions.V6, pleb, "invite", sender=creator
) )
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V6, RoomVersions.V6,
_join_event(pleb), _join_event(RoomVersions.V6, pleb),
auth_events.values(), auth_events.values(),
) )
@ -306,16 +323,18 @@ class EventAuthTestCase(unittest.TestCase):
pleb = "@joiner:example.com" pleb = "@joiner:example.com"
auth_events = { auth_events = {
("m.room.create", ""): _create_event(creator), ("m.room.create", ""): _create_event(RoomVersions.V6, creator),
("m.room.member", creator): _join_event(creator), ("m.room.member", creator): _join_event(RoomVersions.V6, creator),
("m.room.join_rules", ""): _join_rules_event(creator, "invite"), ("m.room.join_rules", ""): _join_rules_event(
RoomVersions.V6, creator, "invite"
),
} }
# A join without an invite is rejected. # A join without an invite is rejected.
with self.assertRaises(AuthError): with self.assertRaises(AuthError):
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V6, RoomVersions.V6,
_join_event(pleb), _join_event(RoomVersions.V6, pleb),
auth_events.values(), auth_events.values(),
) )
@ -323,47 +342,76 @@ class EventAuthTestCase(unittest.TestCase):
with self.assertRaises(AuthError): with self.assertRaises(AuthError):
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V6, RoomVersions.V6,
_member_event(pleb, "join", sender=creator), _member_event(RoomVersions.V6, pleb, "join", sender=creator),
auth_events.values(), auth_events.values(),
) )
# Banned should be rejected. # Banned should be rejected.
auth_events[("m.room.member", pleb)] = _member_event(pleb, "ban") auth_events[("m.room.member", pleb)] = _member_event(
RoomVersions.V6, pleb, "ban"
)
with self.assertRaises(AuthError): with self.assertRaises(AuthError):
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V6, RoomVersions.V6,
_join_event(pleb), _join_event(RoomVersions.V6, pleb),
auth_events.values(), auth_events.values(),
) )
# A user who left cannot re-join. # A user who left cannot re-join.
auth_events[("m.room.member", pleb)] = _member_event(pleb, "leave") auth_events[("m.room.member", pleb)] = _member_event(
RoomVersions.V6, pleb, "leave"
)
with self.assertRaises(AuthError): with self.assertRaises(AuthError):
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V6, RoomVersions.V6,
_join_event(pleb), _join_event(RoomVersions.V6, pleb),
auth_events.values(), auth_events.values(),
) )
# A user can send a join if they're in the room. # A user can send a join if they're in the room.
auth_events[("m.room.member", pleb)] = _member_event(pleb, "join") auth_events[("m.room.member", pleb)] = _member_event(
RoomVersions.V6, pleb, "join"
)
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V6, RoomVersions.V6,
_join_event(pleb), _join_event(RoomVersions.V6, pleb),
auth_events.values(), auth_events.values(),
) )
# A user can accept an invite. # A user can accept an invite.
auth_events[("m.room.member", pleb)] = _member_event( auth_events[("m.room.member", pleb)] = _member_event(
pleb, "invite", sender=creator RoomVersions.V6, pleb, "invite", sender=creator
) )
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V6, RoomVersions.V6,
_join_event(pleb), _join_event(RoomVersions.V6, pleb),
auth_events.values(), auth_events.values(),
) )
def test_join_rules_msc3083_restricted(self): def test_join_rules_restricted_old_room(self) -> None:
"""Old room versions should reject joins to restricted rooms"""
creator = "@creator:example.com"
pleb = "@joiner:example.com"
auth_events = {
("m.room.create", ""): _create_event(RoomVersions.V6, creator),
("m.room.member", creator): _join_event(RoomVersions.V6, creator),
("m.room.power_levels", ""): _power_levels_event(
RoomVersions.V6, creator, {"invite": 0}
),
("m.room.join_rules", ""): _join_rules_event(
RoomVersions.V6, creator, "restricted"
),
}
with self.assertRaises(AuthError):
event_auth.check_auth_rules_for_event(
RoomVersions.V6,
_join_event(RoomVersions.V6, pleb),
auth_events.values(),
)
def test_join_rules_msc3083_restricted(self) -> None:
""" """
Test joining a restricted room from MSC3083. Test joining a restricted room from MSC3083.
@ -377,22 +425,19 @@ class EventAuthTestCase(unittest.TestCase):
pleb = "@joiner:example.com" pleb = "@joiner:example.com"
auth_events = { auth_events = {
("m.room.create", ""): _create_event(creator), ("m.room.create", ""): _create_event(RoomVersions.V8, creator),
("m.room.member", creator): _join_event(creator), ("m.room.member", creator): _join_event(RoomVersions.V8, creator),
("m.room.power_levels", ""): _power_levels_event(creator, {"invite": 0}), ("m.room.power_levels", ""): _power_levels_event(
("m.room.join_rules", ""): _join_rules_event(creator, "restricted"), RoomVersions.V8, creator, {"invite": 0}
),
("m.room.join_rules", ""): _join_rules_event(
RoomVersions.V8, creator, "restricted"
),
} }
# Older room versions don't understand this join rule
with self.assertRaises(AuthError):
event_auth.check_auth_rules_for_event(
RoomVersions.V6,
_join_event(pleb),
auth_events.values(),
)
# A properly formatted join event should work. # A properly formatted join event should work.
authorised_join_event = _join_event( authorised_join_event = _join_event(
RoomVersions.V8,
pleb, pleb,
additional_content={ additional_content={
EventContentFields.AUTHORISING_USER: "@creator:example.com" EventContentFields.AUTHORISING_USER: "@creator:example.com"
@ -408,14 +453,17 @@ class EventAuthTestCase(unittest.TestCase):
# are done properly). # are done properly).
pl_auth_events = auth_events.copy() pl_auth_events = auth_events.copy()
pl_auth_events[("m.room.power_levels", "")] = _power_levels_event( pl_auth_events[("m.room.power_levels", "")] = _power_levels_event(
creator, {"invite": 100, "users": {"@inviter:foo.test": 150}} RoomVersions.V8,
creator,
{"invite": 100, "users": {"@inviter:foo.test": 150}},
) )
pl_auth_events[("m.room.member", "@inviter:foo.test")] = _join_event( pl_auth_events[("m.room.member", "@inviter:foo.test")] = _join_event(
"@inviter:foo.test" RoomVersions.V8, "@inviter:foo.test"
) )
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V8, RoomVersions.V8,
_join_event( _join_event(
RoomVersions.V8,
pleb, pleb,
additional_content={ additional_content={
EventContentFields.AUTHORISING_USER: "@inviter:foo.test" EventContentFields.AUTHORISING_USER: "@inviter:foo.test"
@ -428,19 +476,22 @@ class EventAuthTestCase(unittest.TestCase):
with self.assertRaises(AuthError): with self.assertRaises(AuthError):
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V8, RoomVersions.V8,
_join_event(pleb), _join_event(RoomVersions.V8, pleb),
auth_events.values(), auth_events.values(),
) )
# An join authorised by a user who is not in the room is rejected. # An join authorised by a user who is not in the room is rejected.
pl_auth_events = auth_events.copy() pl_auth_events = auth_events.copy()
pl_auth_events[("m.room.power_levels", "")] = _power_levels_event( pl_auth_events[("m.room.power_levels", "")] = _power_levels_event(
creator, {"invite": 100, "users": {"@other:example.com": 150}} RoomVersions.V8,
creator,
{"invite": 100, "users": {"@other:example.com": 150}},
) )
with self.assertRaises(AuthError): with self.assertRaises(AuthError):
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V8, RoomVersions.V8,
_join_event( _join_event(
RoomVersions.V8,
pleb, pleb,
additional_content={ additional_content={
EventContentFields.AUTHORISING_USER: "@other:example.com" EventContentFields.AUTHORISING_USER: "@other:example.com"
@ -455,6 +506,7 @@ class EventAuthTestCase(unittest.TestCase):
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V8, RoomVersions.V8,
_member_event( _member_event(
RoomVersions.V8,
pleb, pleb,
"join", "join",
sender=creator, sender=creator,
@ -466,7 +518,9 @@ class EventAuthTestCase(unittest.TestCase):
) )
# Banned should be rejected. # Banned should be rejected.
auth_events[("m.room.member", pleb)] = _member_event(pleb, "ban") auth_events[("m.room.member", pleb)] = _member_event(
RoomVersions.V8, pleb, "ban"
)
with self.assertRaises(AuthError): with self.assertRaises(AuthError):
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V8, RoomVersions.V8,
@ -475,7 +529,9 @@ class EventAuthTestCase(unittest.TestCase):
) )
# A user who left can re-join. # A user who left can re-join.
auth_events[("m.room.member", pleb)] = _member_event(pleb, "leave") auth_events[("m.room.member", pleb)] = _member_event(
RoomVersions.V8, pleb, "leave"
)
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V8, RoomVersions.V8,
authorised_join_event, authorised_join_event,
@ -484,21 +540,23 @@ class EventAuthTestCase(unittest.TestCase):
# A user can send a join if they're in the room. (This doesn't need to # A user can send a join if they're in the room. (This doesn't need to
# be authorised since the user is already joined.) # be authorised since the user is already joined.)
auth_events[("m.room.member", pleb)] = _member_event(pleb, "join") auth_events[("m.room.member", pleb)] = _member_event(
RoomVersions.V8, pleb, "join"
)
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V8, RoomVersions.V8,
_join_event(pleb), _join_event(RoomVersions.V8, pleb),
auth_events.values(), auth_events.values(),
) )
# A user can accept an invite. (This doesn't need to be authorised since # A user can accept an invite. (This doesn't need to be authorised since
# the user was invited.) # the user was invited.)
auth_events[("m.room.member", pleb)] = _member_event( auth_events[("m.room.member", pleb)] = _member_event(
pleb, "invite", sender=creator RoomVersions.V8, pleb, "invite", sender=creator
) )
event_auth.check_auth_rules_for_event( event_auth.check_auth_rules_for_event(
RoomVersions.V8, RoomVersions.V8,
_join_event(pleb), _join_event(RoomVersions.V8, pleb),
auth_events.values(), auth_events.values(),
) )
@ -508,20 +566,25 @@ class EventAuthTestCase(unittest.TestCase):
TEST_ROOM_ID = "!test:room" TEST_ROOM_ID = "!test:room"
def _create_event(user_id: str) -> EventBase: def _create_event(
room_version: RoomVersion,
user_id: str,
) -> EventBase:
return make_event_from_dict( return make_event_from_dict(
{ {
"room_id": TEST_ROOM_ID, "room_id": TEST_ROOM_ID,
"event_id": _get_event_id(), **_maybe_get_event_id_dict_for_room_version(room_version),
"type": "m.room.create", "type": "m.room.create",
"state_key": "", "state_key": "",
"sender": user_id, "sender": user_id,
"content": {"creator": user_id}, "content": {"creator": user_id},
} },
room_version=room_version,
) )
def _member_event( def _member_event(
room_version: RoomVersion,
user_id: str, user_id: str,
membership: str, membership: str,
sender: Optional[str] = None, sender: Optional[str] = None,
@ -530,79 +593,102 @@ def _member_event(
return make_event_from_dict( return make_event_from_dict(
{ {
"room_id": TEST_ROOM_ID, "room_id": TEST_ROOM_ID,
"event_id": _get_event_id(), **_maybe_get_event_id_dict_for_room_version(room_version),
"type": "m.room.member", "type": "m.room.member",
"sender": sender or user_id, "sender": sender or user_id,
"state_key": user_id, "state_key": user_id,
"content": {"membership": membership, **(additional_content or {})}, "content": {"membership": membership, **(additional_content or {})},
"prev_events": [], "prev_events": [],
} },
room_version=room_version,
) )
def _join_event(user_id: str, additional_content: Optional[dict] = None) -> EventBase: def _join_event(
return _member_event(user_id, "join", additional_content=additional_content) room_version: RoomVersion,
user_id: str,
additional_content: Optional[dict] = None,
) -> EventBase:
return _member_event(
room_version,
user_id,
"join",
additional_content=additional_content,
)
def _power_levels_event(sender: str, content: JsonDict) -> EventBase: def _power_levels_event(
room_version: RoomVersion,
sender: str,
content: JsonDict,
) -> EventBase:
return make_event_from_dict( return make_event_from_dict(
{ {
"room_id": TEST_ROOM_ID, "room_id": TEST_ROOM_ID,
"event_id": _get_event_id(), **_maybe_get_event_id_dict_for_room_version(room_version),
"type": "m.room.power_levels", "type": "m.room.power_levels",
"sender": sender, "sender": sender,
"state_key": "", "state_key": "",
"content": content, "content": content,
} },
room_version=room_version,
) )
def _alias_event(sender: str, **kwargs) -> EventBase: def _alias_event(room_version: RoomVersion, sender: str, **kwargs) -> EventBase:
data = { data = {
"room_id": TEST_ROOM_ID, "room_id": TEST_ROOM_ID,
"event_id": _get_event_id(), **_maybe_get_event_id_dict_for_room_version(room_version),
"type": "m.room.aliases", "type": "m.room.aliases",
"sender": sender, "sender": sender,
"state_key": get_domain_from_id(sender), "state_key": get_domain_from_id(sender),
"content": {"aliases": []}, "content": {"aliases": []},
} }
data.update(**kwargs) data.update(**kwargs)
return make_event_from_dict(data) return make_event_from_dict(data, room_version=room_version)
def _random_state_event(sender: str) -> EventBase: def _random_state_event(room_version: RoomVersion, sender: str) -> EventBase:
return make_event_from_dict( return make_event_from_dict(
{ {
"room_id": TEST_ROOM_ID, "room_id": TEST_ROOM_ID,
"event_id": _get_event_id(), **_maybe_get_event_id_dict_for_room_version(room_version),
"type": "test.state", "type": "test.state",
"sender": sender, "sender": sender,
"state_key": "", "state_key": "",
"content": {"membership": "join"}, "content": {"membership": "join"},
} },
room_version=room_version,
) )
def _join_rules_event(sender: str, join_rule: str) -> EventBase: def _join_rules_event(
room_version: RoomVersion, sender: str, join_rule: str
) -> EventBase:
return make_event_from_dict( return make_event_from_dict(
{ {
"room_id": TEST_ROOM_ID, "room_id": TEST_ROOM_ID,
"event_id": _get_event_id(), **_maybe_get_event_id_dict_for_room_version(room_version),
"type": "m.room.join_rules", "type": "m.room.join_rules",
"sender": sender, "sender": sender,
"state_key": "", "state_key": "",
"content": { "content": {
"join_rule": join_rule, "join_rule": join_rule,
}, },
} },
room_version=room_version,
) )
event_count = 0 event_count = 0
def _get_event_id() -> str: def _maybe_get_event_id_dict_for_room_version(room_version: RoomVersion) -> dict:
"""If this room version needs it, generate an event id"""
if room_version.event_format != EventFormatVersions.V1:
return {}
global event_count global event_count
c = event_count c = event_count
event_count += 1 event_count += 1
return "!%i:example.com" % (c,) return {"event_id": "!%i:example.com" % (c,)}