test_visibility.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327
  1. # -*- coding: utf-8 -*-
  2. # Copyright 2018 New Vector Ltd
  3. #
  4. # Licensed under the Apache License, Version 2.0 (the "License");
  5. # you may not use this file except in compliance with the License.
  6. # You may obtain a copy of the License at
  7. #
  8. # http://www.apache.org/licenses/LICENSE-2.0
  9. #
  10. # Unless required by applicable law or agreed to in writing, software
  11. # distributed under the License is distributed on an "AS IS" BASIS,
  12. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. # See the License for the specific language governing permissions and
  14. # limitations under the License.
  15. import logging
  16. from twisted.internet import defer
  17. from twisted.internet.defer import succeed
  18. from synapse.events import FrozenEvent
  19. from synapse.visibility import filter_events_for_server
  20. import tests.unittest
  21. from tests.utils import setup_test_homeserver
  22. logger = logging.getLogger(__name__)
  23. TEST_ROOM_ID = "!TEST:ROOM"
  24. class FilterEventsForServerTestCase(tests.unittest.TestCase):
  25. @defer.inlineCallbacks
  26. def setUp(self):
  27. self.hs = yield setup_test_homeserver(self.addCleanup)
  28. self.event_creation_handler = self.hs.get_event_creation_handler()
  29. self.event_builder_factory = self.hs.get_event_builder_factory()
  30. self.store = self.hs.get_datastore()
  31. @defer.inlineCallbacks
  32. def test_filtering(self):
  33. #
  34. # The events to be filtered consist of 10 membership events (it doesn't
  35. # really matter if they are joins or leaves, so let's make them joins).
  36. # One of those membership events is going to be for a user on the
  37. # server we are filtering for (so we can check the filtering is doing
  38. # the right thing).
  39. #
  40. # before we do that, we persist some other events to act as state.
  41. self.inject_visibility("@admin:hs", "joined")
  42. for i in range(0, 10):
  43. yield self.inject_room_member("@resident%i:hs" % i)
  44. events_to_filter = []
  45. for i in range(0, 10):
  46. user = "@user%i:%s" % (i, "test_server" if i == 5 else "other_server")
  47. evt = yield self.inject_room_member(user, extra_content={"a": "b"})
  48. events_to_filter.append(evt)
  49. filtered = yield filter_events_for_server(
  50. self.store, "test_server", events_to_filter
  51. )
  52. # the result should be 5 redacted events, and 5 unredacted events.
  53. for i in range(0, 5):
  54. self.assertEqual(events_to_filter[i].event_id, filtered[i].event_id)
  55. self.assertNotIn("a", filtered[i].content)
  56. for i in range(5, 10):
  57. self.assertEqual(events_to_filter[i].event_id, filtered[i].event_id)
  58. self.assertEqual(filtered[i].content["a"], "b")
  59. @tests.unittest.DEBUG
  60. @defer.inlineCallbacks
  61. def test_erased_user(self):
  62. # 4 message events, from erased and unerased users, with a membership
  63. # change in the middle of them.
  64. events_to_filter = []
  65. evt = yield self.inject_message("@unerased:local_hs")
  66. events_to_filter.append(evt)
  67. evt = yield self.inject_message("@erased:local_hs")
  68. events_to_filter.append(evt)
  69. evt = yield self.inject_room_member("@joiner:remote_hs")
  70. events_to_filter.append(evt)
  71. evt = yield self.inject_message("@unerased:local_hs")
  72. events_to_filter.append(evt)
  73. evt = yield self.inject_message("@erased:local_hs")
  74. events_to_filter.append(evt)
  75. # the erasey user gets erased
  76. self.hs.get_datastore().mark_user_erased("@erased:local_hs")
  77. # ... and the filtering happens.
  78. filtered = yield filter_events_for_server(
  79. self.store, "test_server", events_to_filter
  80. )
  81. for i in range(0, len(events_to_filter)):
  82. self.assertEqual(
  83. events_to_filter[i].event_id,
  84. filtered[i].event_id,
  85. "Unexpected event at result position %i" % (i,),
  86. )
  87. for i in (0, 3):
  88. self.assertEqual(
  89. events_to_filter[i].content["body"],
  90. filtered[i].content["body"],
  91. "Unexpected event content at result position %i" % (i,),
  92. )
  93. for i in (1, 4):
  94. self.assertNotIn("body", filtered[i].content)
  95. @defer.inlineCallbacks
  96. def inject_visibility(self, user_id, visibility):
  97. content = {"history_visibility": visibility}
  98. builder = self.event_builder_factory.new(
  99. {
  100. "type": "m.room.history_visibility",
  101. "sender": user_id,
  102. "state_key": "",
  103. "room_id": TEST_ROOM_ID,
  104. "content": content,
  105. }
  106. )
  107. event, context = yield self.event_creation_handler.create_new_client_event(
  108. builder
  109. )
  110. yield self.hs.get_datastore().persist_event(event, context)
  111. defer.returnValue(event)
  112. @defer.inlineCallbacks
  113. def inject_room_member(self, user_id, membership="join", extra_content={}):
  114. content = {"membership": membership}
  115. content.update(extra_content)
  116. builder = self.event_builder_factory.new(
  117. {
  118. "type": "m.room.member",
  119. "sender": user_id,
  120. "state_key": user_id,
  121. "room_id": TEST_ROOM_ID,
  122. "content": content,
  123. }
  124. )
  125. event, context = yield self.event_creation_handler.create_new_client_event(
  126. builder
  127. )
  128. yield self.hs.get_datastore().persist_event(event, context)
  129. defer.returnValue(event)
  130. @defer.inlineCallbacks
  131. def inject_message(self, user_id, content=None):
  132. if content is None:
  133. content = {"body": "testytest"}
  134. builder = self.event_builder_factory.new(
  135. {
  136. "type": "m.room.message",
  137. "sender": user_id,
  138. "room_id": TEST_ROOM_ID,
  139. "content": content,
  140. }
  141. )
  142. event, context = yield self.event_creation_handler.create_new_client_event(
  143. builder
  144. )
  145. yield self.hs.get_datastore().persist_event(event, context)
  146. defer.returnValue(event)
  147. @defer.inlineCallbacks
  148. def test_large_room(self):
  149. # see what happens when we have a large room with hundreds of thousands
  150. # of membership events
  151. # As above, the events to be filtered consist of 10 membership events,
  152. # where one of them is for a user on the server we are filtering for.
  153. import cProfile
  154. import pstats
  155. import time
  156. # we stub out the store, because building up all that state the normal
  157. # way is very slow.
  158. test_store = _TestStore()
  159. # our initial state is 100000 membership events and one
  160. # history_visibility event.
  161. room_state = []
  162. history_visibility_evt = FrozenEvent(
  163. {
  164. "event_id": "$history_vis",
  165. "type": "m.room.history_visibility",
  166. "sender": "@resident_user_0:test.com",
  167. "state_key": "",
  168. "room_id": TEST_ROOM_ID,
  169. "content": {"history_visibility": "joined"},
  170. }
  171. )
  172. room_state.append(history_visibility_evt)
  173. test_store.add_event(history_visibility_evt)
  174. for i in range(0, 100000):
  175. user = "@resident_user_%i:test.com" % (i,)
  176. evt = FrozenEvent(
  177. {
  178. "event_id": "$res_event_%i" % (i,),
  179. "type": "m.room.member",
  180. "state_key": user,
  181. "sender": user,
  182. "room_id": TEST_ROOM_ID,
  183. "content": {"membership": "join", "extra": "zzz,"},
  184. }
  185. )
  186. room_state.append(evt)
  187. test_store.add_event(evt)
  188. events_to_filter = []
  189. for i in range(0, 10):
  190. user = "@user%i:%s" % (i, "test_server" if i == 5 else "other_server")
  191. evt = FrozenEvent(
  192. {
  193. "event_id": "$evt%i" % (i,),
  194. "type": "m.room.member",
  195. "state_key": user,
  196. "sender": user,
  197. "room_id": TEST_ROOM_ID,
  198. "content": {"membership": "join", "extra": "zzz"},
  199. }
  200. )
  201. events_to_filter.append(evt)
  202. room_state.append(evt)
  203. test_store.add_event(evt)
  204. test_store.set_state_ids_for_event(
  205. evt, {(e.type, e.state_key): e.event_id for e in room_state}
  206. )
  207. pr = cProfile.Profile()
  208. pr.enable()
  209. logger.info("Starting filtering")
  210. start = time.time()
  211. filtered = yield filter_events_for_server(
  212. test_store, "test_server", events_to_filter
  213. )
  214. logger.info("Filtering took %f seconds", time.time() - start)
  215. pr.disable()
  216. with open("filter_events_for_server.profile", "w+") as f:
  217. ps = pstats.Stats(pr, stream=f).sort_stats('cumulative')
  218. ps.print_stats()
  219. # the result should be 5 redacted events, and 5 unredacted events.
  220. for i in range(0, 5):
  221. self.assertEqual(events_to_filter[i].event_id, filtered[i].event_id)
  222. self.assertNotIn("extra", filtered[i].content)
  223. for i in range(5, 10):
  224. self.assertEqual(events_to_filter[i].event_id, filtered[i].event_id)
  225. self.assertEqual(filtered[i].content["extra"], "zzz")
  226. test_large_room.skip = "Disabled by default because it's slow"
  227. class _TestStore(object):
  228. """Implements a few methods of the DataStore, so that we can test
  229. filter_events_for_server
  230. """
  231. def __init__(self):
  232. # data for get_events: a map from event_id to event
  233. self.events = {}
  234. # data for get_state_ids_for_events mock: a map from event_id to
  235. # a map from (type_state_key) -> event_id for the state at that
  236. # event
  237. self.state_ids_for_events = {}
  238. def add_event(self, event):
  239. self.events[event.event_id] = event
  240. def set_state_ids_for_event(self, event, state):
  241. self.state_ids_for_events[event.event_id] = state
  242. def get_state_ids_for_events(self, events, types):
  243. res = {}
  244. include_memberships = False
  245. for (type, state_key) in types:
  246. if type == "m.room.history_visibility":
  247. continue
  248. if type != "m.room.member" or state_key is not None:
  249. raise RuntimeError(
  250. "Unimplemented: get_state_ids with type (%s, %s)"
  251. % (type, state_key)
  252. )
  253. include_memberships = True
  254. if include_memberships:
  255. for event_id in events:
  256. res[event_id] = self.state_ids_for_events[event_id]
  257. else:
  258. k = ("m.room.history_visibility", "")
  259. for event_id in events:
  260. hve = self.state_ids_for_events[event_id][k]
  261. res[event_id] = {k: hve}
  262. return succeed(res)
  263. def get_events(self, events):
  264. return succeed({event_id: self.events[event_id] for event_id in events})
  265. def are_users_erased(self, users):
  266. return succeed({u: False for u in users})