mirrors/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2024-11-22 09:35:45 +03:00
Code Issues Projects Releases Packages Wiki Activity

Merge pull request from GHSA-3x4c-pq33-4w3q

Browse source 
* Add some tests to characterise the problem

Some failing. Current states:

  RoomsMemberListTestCase
test_get_member_list ...
[OK]
test_get_member_list_mixed_memberships ...
[OK]
test_get_member_list_no_permission ...
[OK]
test_get_member_list_no_permission_former_member ...
[OK]
test_get_member_list_no_permission_former_member_with_at_token ...
[FAIL]
test_get_member_list_no_room ...
[OK]
test_get_member_list_no_permission_with_at_token ...
[FAIL]

* Correct the tests

* Check user is/was member before divulging room membership

* Pull out only the 1 membership event we want.

* Update tests/rest/client/v1/test_rooms.py

Co-authored-by: Erik Johnston <erik@matrix.org>

* Fixup tests (following apply review suggestion)

Co-authored-by: Erik Johnston <erik@matrix.org>
This commit is contained in:
reivilibre 2021-08-31 10:09:58 +01:00 committed by GitHub
parent 8f98260552
commit 52c7a51cfc
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 103 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
synapse/handlers/message.py
View file

@ -183,20 +183,37 @@ class MessageHandler:
if not last_events:
raise NotFoundError("Can't find event for token %s" % (at_token,))
last_event = last_events[0]
# check whether the user is in the room at that time to determine
# whether they should be treated as peeking.
state_map = await self.state_store.get_state_for_event(
last_event.event_id,
StateFilter.from_types([(EventTypes.Member, user_id)]),
)
joined = False
membership_event = state_map.get((EventTypes.Member, user_id))
if membership_event:
joined = membership_event.membership == Membership.JOIN
is_peeking = not joined
visible_events = await filter_events_for_client(
self.storage,
user_id,
last_events,
filter_send_to_client=False,
is_peeking=is_peeking,
)
event = last_events[0]
if visible_events:
room_state_events = await self.state_store.get_state_for_events(
[event.event_id], state_filter=state_filter
[last_event.event_id], state_filter=state_filter
)
room_state: Mapping[Any, EventBase] = room_state_events[event.event_id]
room_state: Mapping[Any, EventBase] = room_state_events[
last_event.event_id
]
else:
raise AuthError(
403,

84
tests/rest/client/v1/test_rooms.py
View file

@ -29,7 +29,7 @@ from synapse.api.constants import EventContentFields, EventTypes, Membership
from synapse.api.errors import HttpResponseException
from synapse.handlers.pagination import PurgeStatus
from synapse.rest import admin
from synapse.rest.client import account, directory, login, profile, room
from synapse.rest.client import account, directory, login, profile, room, sync
from synapse.types import JsonDict, RoomAlias, UserID, create_requester
from synapse.util.stringutils import random_string
@ -381,6 +381,8 @@ class RoomPermissionsTestCase(RoomBase):
class RoomsMemberListTestCase(RoomBase):
"""Tests /rooms/$room_id/members/list REST events."""
servlets = RoomBase.servlets + [sync.register_servlets]
user_id = "@sid1:red"
def test_get_member_list(self):
@ -397,6 +399,86 @@ class RoomsMemberListTestCase(RoomBase):
channel = self.make_request("GET", "/rooms/%s/members" % room_id)
self.assertEquals(403, channel.code, msg=channel.result["body"])
def test_get_member_list_no_permission_with_at_token(self):
"""
Tests that a stranger to the room cannot get the member list
(in the case that they use an at token).
"""
room_id = self.helper.create_room_as("@someone.else:red")
# first sync to get an at token
channel = self.make_request("GET", "/sync")
self.assertEquals(200, channel.code)
sync_token = channel.json_body["next_batch"]
# check that permission is denied for @sid1:red to get the
# memberships of @someone.else:red's room.
channel = self.make_request(
"GET",
f"/rooms/{room_id}/members?at={sync_token}",
)
self.assertEquals(403, channel.code, msg=channel.result["body"])
def test_get_member_list_no_permission_former_member(self):
"""
Tests that a former member of the room can not get the member list.
"""
# create a room, invite the user and the user joins
room_id = self.helper.create_room_as("@alice:red")
self.helper.invite(room_id, "@alice:red", self.user_id)
self.helper.join(room_id, self.user_id)
# check that the user can see the member list to start with
channel = self.make_request("GET", "/rooms/%s/members" % room_id)
self.assertEquals(200, channel.code, msg=channel.result["body"])
# ban the user
self.helper.change_membership(room_id, "@alice:red", self.user_id, "ban")
# check the user can no longer see the member list
channel = self.make_request("GET", "/rooms/%s/members" % room_id)
self.assertEquals(403, channel.code, msg=channel.result["body"])
def test_get_member_list_no_permission_former_member_with_at_token(self):
"""
Tests that a former member of the room can not get the member list
(in the case that they use an at token).
"""
# create a room, invite the user and the user joins
room_id = self.helper.create_room_as("@alice:red")
self.helper.invite(room_id, "@alice:red", self.user_id)
self.helper.join(room_id, self.user_id)
# sync to get an at token
channel = self.make_request("GET", "/sync")
self.assertEquals(200, channel.code)
sync_token = channel.json_body["next_batch"]
# check that the user can see the member list to start with
channel = self.make_request(
"GET", "/rooms/%s/members?at=%s" % (room_id, sync_token)
)
self.assertEquals(200, channel.code, msg=channel.result["body"])
# ban the user (Note: the user is actually allowed to see this event and
# state so that they know they're banned!)
self.helper.change_membership(room_id, "@alice:red", self.user_id, "ban")
# invite a third user and let them join
self.helper.invite(room_id, "@alice:red", "@bob:red")
self.helper.join(room_id, "@bob:red")
# now, with the original user, sync again to get a new at token
channel = self.make_request("GET", "/sync")
self.assertEquals(200, channel.code)
sync_token = channel.json_body["next_batch"]
# check the user can no longer see the updated member list
channel = self.make_request(
"GET", "/rooms/%s/members?at=%s" % (room_id, sync_token)
)
self.assertEquals(403, channel.code, msg=channel.result["body"])
def test_get_member_list_mixed_memberships(self):
room_creator = "@some_other_guy:red"
room_id = self.helper.create_room_as(room_creator)