1
0

test_appservice.py 44 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153
  1. # Copyright 2015-2021 The Matrix.org Foundation C.I.C.
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License");
  4. # you may not use this file except in compliance with the License.
  5. # You may obtain a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. # See the License for the specific language governing permissions and
  13. # limitations under the License.
  14. from typing import Dict, Iterable, List, Optional
  15. from unittest.mock import Mock
  16. from parameterized import parameterized
  17. from twisted.internet import defer
  18. from twisted.test.proto_helpers import MemoryReactor
  19. import synapse.rest.admin
  20. import synapse.storage
  21. from synapse.api.constants import EduTypes, EventTypes
  22. from synapse.appservice import (
  23. ApplicationService,
  24. TransactionOneTimeKeysCount,
  25. TransactionUnusedFallbackKeys,
  26. )
  27. from synapse.handlers.appservice import ApplicationServicesHandler
  28. from synapse.rest.client import login, receipts, register, room, sendtodevice
  29. from synapse.server import HomeServer
  30. from synapse.types import JsonDict, RoomStreamToken
  31. from synapse.util import Clock
  32. from synapse.util.stringutils import random_string
  33. from tests import unittest
  34. from tests.test_utils import event_injection, make_awaitable, simple_async_mock
  35. from tests.unittest import override_config
  36. from tests.utils import MockClock
  37. class AppServiceHandlerTestCase(unittest.TestCase):
  38. """Tests the ApplicationServicesHandler."""
  39. def setUp(self) -> None:
  40. self.mock_store = Mock()
  41. self.mock_as_api = Mock()
  42. self.mock_scheduler = Mock()
  43. hs = Mock()
  44. hs.get_datastores.return_value = Mock(main=self.mock_store)
  45. self.mock_store.get_appservice_last_pos.return_value = make_awaitable(None)
  46. self.mock_store.set_appservice_last_pos.return_value = make_awaitable(None)
  47. self.mock_store.set_appservice_stream_type_pos.return_value = make_awaitable(
  48. None
  49. )
  50. hs.get_application_service_api.return_value = self.mock_as_api
  51. hs.get_application_service_scheduler.return_value = self.mock_scheduler
  52. hs.get_clock.return_value = MockClock()
  53. self.handler = ApplicationServicesHandler(hs)
  54. self.event_source = hs.get_event_sources()
  55. def test_notify_interested_services(self) -> None:
  56. interested_service = self._mkservice(is_interested_in_event=True)
  57. services = [
  58. self._mkservice(is_interested_in_event=False),
  59. interested_service,
  60. self._mkservice(is_interested_in_event=False),
  61. ]
  62. self.mock_as_api.query_user.return_value = make_awaitable(True)
  63. self.mock_store.get_app_services.return_value = services
  64. self.mock_store.get_user_by_id.return_value = make_awaitable([])
  65. event = Mock(
  66. sender="@someone:anywhere", type="m.room.message", room_id="!foo:bar"
  67. )
  68. self.mock_store.get_all_new_event_ids_stream.side_effect = [
  69. make_awaitable((0, {})),
  70. make_awaitable((1, {event.event_id: 0})),
  71. ]
  72. self.mock_store.get_events_as_list.side_effect = [
  73. make_awaitable([]),
  74. make_awaitable([event]),
  75. ]
  76. self.handler.notify_interested_services(RoomStreamToken(None, 1))
  77. self.mock_scheduler.enqueue_for_appservice.assert_called_once_with(
  78. interested_service, events=[event]
  79. )
  80. def test_query_user_exists_unknown_user(self) -> None:
  81. user_id = "@someone:anywhere"
  82. services = [self._mkservice(is_interested_in_event=True)]
  83. services[0].is_interested_in_user.return_value = True
  84. self.mock_store.get_app_services.return_value = services
  85. self.mock_store.get_user_by_id.return_value = make_awaitable(None)
  86. event = Mock(sender=user_id, type="m.room.message", room_id="!foo:bar")
  87. self.mock_as_api.query_user.return_value = make_awaitable(True)
  88. self.mock_store.get_all_new_event_ids_stream.side_effect = [
  89. make_awaitable((0, {event.event_id: 0})),
  90. ]
  91. self.mock_store.get_events_as_list.side_effect = [make_awaitable([event])]
  92. self.handler.notify_interested_services(RoomStreamToken(None, 0))
  93. self.mock_as_api.query_user.assert_called_once_with(services[0], user_id)
  94. def test_query_user_exists_known_user(self) -> None:
  95. user_id = "@someone:anywhere"
  96. services = [self._mkservice(is_interested_in_event=True)]
  97. services[0].is_interested_in_user.return_value = True
  98. self.mock_store.get_app_services.return_value = services
  99. self.mock_store.get_user_by_id.return_value = make_awaitable({"name": user_id})
  100. event = Mock(sender=user_id, type="m.room.message", room_id="!foo:bar")
  101. self.mock_as_api.query_user.return_value = make_awaitable(True)
  102. self.mock_store.get_all_new_event_ids_stream.side_effect = [
  103. make_awaitable((0, [event], {event.event_id: 0})),
  104. ]
  105. self.handler.notify_interested_services(RoomStreamToken(None, 0))
  106. self.assertFalse(
  107. self.mock_as_api.query_user.called,
  108. "query_user called when it shouldn't have been.",
  109. )
  110. def test_query_room_alias_exists(self) -> None:
  111. room_alias_str = "#foo:bar"
  112. room_alias = Mock()
  113. room_alias.to_string.return_value = room_alias_str
  114. room_id = "!alpha:bet"
  115. servers = ["aperture"]
  116. interested_service = self._mkservice_alias(is_room_alias_in_namespace=True)
  117. services = [
  118. self._mkservice_alias(is_room_alias_in_namespace=False),
  119. interested_service,
  120. self._mkservice_alias(is_room_alias_in_namespace=False),
  121. ]
  122. self.mock_as_api.query_alias.return_value = make_awaitable(True)
  123. self.mock_store.get_app_services.return_value = services
  124. self.mock_store.get_association_from_room_alias.return_value = make_awaitable(
  125. Mock(room_id=room_id, servers=servers)
  126. )
  127. result = self.successResultOf(
  128. defer.ensureDeferred(self.handler.query_room_alias_exists(room_alias))
  129. )
  130. self.mock_as_api.query_alias.assert_called_once_with(
  131. interested_service, room_alias_str
  132. )
  133. self.assertEqual(result.room_id, room_id)
  134. self.assertEqual(result.servers, servers)
  135. def test_get_3pe_protocols_no_appservices(self) -> None:
  136. self.mock_store.get_app_services.return_value = []
  137. response = self.successResultOf(
  138. defer.ensureDeferred(self.handler.get_3pe_protocols("my-protocol"))
  139. )
  140. self.mock_as_api.get_3pe_protocol.assert_not_called()
  141. self.assertEqual(response, {})
  142. def test_get_3pe_protocols_no_protocols(self) -> None:
  143. service = self._mkservice(False, [])
  144. self.mock_store.get_app_services.return_value = [service]
  145. response = self.successResultOf(
  146. defer.ensureDeferred(self.handler.get_3pe_protocols())
  147. )
  148. self.mock_as_api.get_3pe_protocol.assert_not_called()
  149. self.assertEqual(response, {})
  150. def test_get_3pe_protocols_protocol_no_response(self) -> None:
  151. service = self._mkservice(False, ["my-protocol"])
  152. self.mock_store.get_app_services.return_value = [service]
  153. self.mock_as_api.get_3pe_protocol.return_value = make_awaitable(None)
  154. response = self.successResultOf(
  155. defer.ensureDeferred(self.handler.get_3pe_protocols())
  156. )
  157. self.mock_as_api.get_3pe_protocol.assert_called_once_with(
  158. service, "my-protocol"
  159. )
  160. self.assertEqual(response, {})
  161. def test_get_3pe_protocols_select_one_protocol(self) -> None:
  162. service = self._mkservice(False, ["my-protocol"])
  163. self.mock_store.get_app_services.return_value = [service]
  164. self.mock_as_api.get_3pe_protocol.return_value = make_awaitable(
  165. {"x-protocol-data": 42, "instances": []}
  166. )
  167. response = self.successResultOf(
  168. defer.ensureDeferred(self.handler.get_3pe_protocols("my-protocol"))
  169. )
  170. self.mock_as_api.get_3pe_protocol.assert_called_once_with(
  171. service, "my-protocol"
  172. )
  173. self.assertEqual(
  174. response, {"my-protocol": {"x-protocol-data": 42, "instances": []}}
  175. )
  176. def test_get_3pe_protocols_one_protocol(self) -> None:
  177. service = self._mkservice(False, ["my-protocol"])
  178. self.mock_store.get_app_services.return_value = [service]
  179. self.mock_as_api.get_3pe_protocol.return_value = make_awaitable(
  180. {"x-protocol-data": 42, "instances": []}
  181. )
  182. response = self.successResultOf(
  183. defer.ensureDeferred(self.handler.get_3pe_protocols())
  184. )
  185. self.mock_as_api.get_3pe_protocol.assert_called_once_with(
  186. service, "my-protocol"
  187. )
  188. self.assertEqual(
  189. response, {"my-protocol": {"x-protocol-data": 42, "instances": []}}
  190. )
  191. def test_get_3pe_protocols_multiple_protocol(self) -> None:
  192. service_one = self._mkservice(False, ["my-protocol"])
  193. service_two = self._mkservice(False, ["other-protocol"])
  194. self.mock_store.get_app_services.return_value = [service_one, service_two]
  195. self.mock_as_api.get_3pe_protocol.return_value = make_awaitable(
  196. {"x-protocol-data": 42, "instances": []}
  197. )
  198. response = self.successResultOf(
  199. defer.ensureDeferred(self.handler.get_3pe_protocols())
  200. )
  201. self.mock_as_api.get_3pe_protocol.assert_called()
  202. self.assertEqual(
  203. response,
  204. {
  205. "my-protocol": {"x-protocol-data": 42, "instances": []},
  206. "other-protocol": {"x-protocol-data": 42, "instances": []},
  207. },
  208. )
  209. def test_get_3pe_protocols_multiple_info(self) -> None:
  210. service_one = self._mkservice(False, ["my-protocol"])
  211. service_two = self._mkservice(False, ["my-protocol"])
  212. async def get_3pe_protocol(
  213. service: ApplicationService, protocol: str
  214. ) -> Optional[JsonDict]:
  215. if service == service_one:
  216. return {
  217. "x-protocol-data": 42,
  218. "instances": [{"desc": "Alice's service"}],
  219. }
  220. if service == service_two:
  221. return {
  222. "x-protocol-data": 36,
  223. "x-not-used": 45,
  224. "instances": [{"desc": "Bob's service"}],
  225. }
  226. raise Exception("Unexpected service")
  227. self.mock_store.get_app_services.return_value = [service_one, service_two]
  228. self.mock_as_api.get_3pe_protocol = get_3pe_protocol
  229. response = self.successResultOf(
  230. defer.ensureDeferred(self.handler.get_3pe_protocols())
  231. )
  232. # It's expected that the second service's data doesn't appear in the response
  233. self.assertEqual(
  234. response,
  235. {
  236. "my-protocol": {
  237. "x-protocol-data": 42,
  238. "instances": [
  239. {
  240. "desc": "Alice's service",
  241. },
  242. {"desc": "Bob's service"},
  243. ],
  244. },
  245. },
  246. )
  247. def test_notify_interested_services_ephemeral(self) -> None:
  248. """
  249. Test sending ephemeral events to the appservice handler are scheduled
  250. to be pushed out to interested appservices, and that the stream ID is
  251. updated accordingly.
  252. """
  253. interested_service = self._mkservice(is_interested_in_event=True)
  254. services = [interested_service]
  255. self.mock_store.get_app_services.return_value = services
  256. self.mock_store.get_type_stream_id_for_appservice.return_value = make_awaitable(
  257. 579
  258. )
  259. event = Mock(event_id="event_1")
  260. self.event_source.sources.receipt.get_new_events_as.return_value = (
  261. make_awaitable(([event], None))
  262. )
  263. self.handler.notify_interested_services_ephemeral(
  264. "receipt_key", 580, ["@fakerecipient:example.com"]
  265. )
  266. self.mock_scheduler.enqueue_for_appservice.assert_called_once_with(
  267. interested_service, ephemeral=[event]
  268. )
  269. self.mock_store.set_appservice_stream_type_pos.assert_called_once_with(
  270. interested_service,
  271. "read_receipt",
  272. 580,
  273. )
  274. def test_notify_interested_services_ephemeral_out_of_order(self) -> None:
  275. """
  276. Test sending out of order ephemeral events to the appservice handler
  277. are ignored.
  278. """
  279. interested_service = self._mkservice(is_interested_in_event=True)
  280. services = [interested_service]
  281. self.mock_store.get_app_services.return_value = services
  282. self.mock_store.get_type_stream_id_for_appservice.return_value = make_awaitable(
  283. 580
  284. )
  285. event = Mock(event_id="event_1")
  286. self.event_source.sources.receipt.get_new_events_as.return_value = (
  287. make_awaitable(([event], None))
  288. )
  289. self.handler.notify_interested_services_ephemeral(
  290. "receipt_key", 580, ["@fakerecipient:example.com"]
  291. )
  292. # This method will be called, but with an empty list of events
  293. self.mock_scheduler.enqueue_for_appservice.assert_called_once_with(
  294. interested_service, ephemeral=[]
  295. )
  296. def _mkservice(
  297. self, is_interested_in_event: bool, protocols: Optional[Iterable] = None
  298. ) -> Mock:
  299. """
  300. Create a new mock representing an ApplicationService.
  301. Args:
  302. is_interested_in_event: Whether this application service will be considered
  303. interested in all events.
  304. protocols: The third-party protocols that this application service claims to
  305. support.
  306. Returns:
  307. A mock representing the ApplicationService.
  308. """
  309. service = Mock()
  310. service.is_interested_in_event.return_value = make_awaitable(
  311. is_interested_in_event
  312. )
  313. service.token = "mock_service_token"
  314. service.url = "mock_service_url"
  315. service.protocols = protocols
  316. return service
  317. def _mkservice_alias(self, is_room_alias_in_namespace: bool) -> Mock:
  318. """
  319. Create a new mock representing an ApplicationService that is or is not interested
  320. any given room aliase.
  321. Args:
  322. is_room_alias_in_namespace: If true, the application service will be interested
  323. in all room aliases that are queried against it. If false, the application
  324. service will not be interested in any room aliases.
  325. Returns:
  326. A mock representing the ApplicationService.
  327. """
  328. service = Mock()
  329. service.is_room_alias_in_namespace.return_value = is_room_alias_in_namespace
  330. service.token = "mock_service_token"
  331. service.url = "mock_service_url"
  332. return service
  333. class ApplicationServicesHandlerSendEventsTestCase(unittest.HomeserverTestCase):
  334. """
  335. Tests that the ApplicationServicesHandler sends events to application
  336. services correctly.
  337. """
  338. servlets = [
  339. synapse.rest.admin.register_servlets_for_client_rest_resource,
  340. login.register_servlets,
  341. room.register_servlets,
  342. sendtodevice.register_servlets,
  343. receipts.register_servlets,
  344. ]
  345. def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
  346. self.hs = hs
  347. # Mock the ApplicationServiceScheduler's _TransactionController's send method so that
  348. # we can track any outgoing ephemeral events
  349. self.send_mock = simple_async_mock()
  350. hs.get_application_service_handler().scheduler.txn_ctrl.send = self.send_mock # type: ignore[assignment]
  351. # Mock out application services, and allow defining our own in tests
  352. self._services: List[ApplicationService] = []
  353. self.hs.get_datastores().main.get_app_services = Mock( # type: ignore[assignment]
  354. return_value=self._services
  355. )
  356. # A user on the homeserver.
  357. self.local_user_device_id = "local_device"
  358. self.local_user = self.register_user("local_user", "password")
  359. self.local_user_token = self.login(
  360. "local_user", "password", self.local_user_device_id
  361. )
  362. # A user on the homeserver which lies within an appservice's exclusive user namespace.
  363. self.exclusive_as_user_device_id = "exclusive_as_device"
  364. self.exclusive_as_user = self.register_user("exclusive_as_user", "password")
  365. self.exclusive_as_user_token = self.login(
  366. "exclusive_as_user", "password", self.exclusive_as_user_device_id
  367. )
  368. def _notify_interested_services(self) -> None:
  369. # This is normally set in `notify_interested_services` but we need to call the
  370. # internal async version so the reactor gets pushed to completion.
  371. self.hs.get_application_service_handler().current_max += 1
  372. self.get_success(
  373. self.hs.get_application_service_handler()._notify_interested_services(
  374. RoomStreamToken(
  375. None, self.hs.get_application_service_handler().current_max
  376. )
  377. )
  378. )
  379. @parameterized.expand(
  380. [
  381. ("@local_as_user:test", True),
  382. # Defining remote users in an application service user namespace regex is a
  383. # footgun since the appservice might assume that it'll receive all events
  384. # sent by that remote user, but it will only receive events in rooms that
  385. # are shared with a local user. So we just remove this footgun possibility
  386. # entirely and we won't notify the application service based on remote
  387. # users.
  388. ("@remote_as_user:remote", False),
  389. ]
  390. )
  391. def test_match_interesting_room_members(
  392. self, interesting_user: str, should_notify: bool
  393. ) -> None:
  394. """
  395. Test to make sure that a interesting user (local or remote) in the room is
  396. notified as expected when someone else in the room sends a message.
  397. """
  398. # Register an application service that's interested in the `interesting_user`
  399. interested_appservice = self._register_application_service(
  400. namespaces={
  401. ApplicationService.NS_USERS: [
  402. {
  403. "regex": interesting_user,
  404. "exclusive": False,
  405. },
  406. ],
  407. },
  408. )
  409. # Create a room
  410. alice = self.register_user("alice", "pass")
  411. alice_access_token = self.login("alice", "pass")
  412. room_id = self.helper.create_room_as(room_creator=alice, tok=alice_access_token)
  413. # Join the interesting user to the room
  414. self.get_success(
  415. event_injection.inject_member_event(
  416. self.hs, room_id, interesting_user, "join"
  417. )
  418. )
  419. # Kick the appservice into checking this membership event to get the event out
  420. # of the way
  421. self._notify_interested_services()
  422. # We don't care about the interesting user join event (this test is making sure
  423. # the next thing works)
  424. self.send_mock.reset_mock()
  425. # Send a message from an uninteresting user
  426. self.helper.send_event(
  427. room_id,
  428. type=EventTypes.Message,
  429. content={
  430. "msgtype": "m.text",
  431. "body": "message from uninteresting user",
  432. },
  433. tok=alice_access_token,
  434. )
  435. # Kick the appservice into checking this new event
  436. self._notify_interested_services()
  437. if should_notify:
  438. self.send_mock.assert_called_once()
  439. (
  440. service,
  441. events,
  442. _ephemeral,
  443. _to_device_messages,
  444. _otks,
  445. _fbks,
  446. _device_list_summary,
  447. ) = self.send_mock.call_args[0]
  448. # Even though the message came from an uninteresting user, it should still
  449. # notify us because the interesting user is joined to the room where the
  450. # message was sent.
  451. self.assertEqual(service, interested_appservice)
  452. self.assertEqual(events[0]["type"], "m.room.message")
  453. self.assertEqual(events[0]["sender"], alice)
  454. else:
  455. self.send_mock.assert_not_called()
  456. def test_application_services_receive_events_sent_by_interesting_local_user(
  457. self,
  458. ) -> None:
  459. """
  460. Test to make sure that a messages sent from a local user can be interesting and
  461. picked up by the appservice.
  462. """
  463. # Register an application service that's interested in all local users
  464. interested_appservice = self._register_application_service(
  465. namespaces={
  466. ApplicationService.NS_USERS: [
  467. {
  468. "regex": ".*",
  469. "exclusive": False,
  470. },
  471. ],
  472. },
  473. )
  474. # Create a room
  475. alice = self.register_user("alice", "pass")
  476. alice_access_token = self.login("alice", "pass")
  477. room_id = self.helper.create_room_as(room_creator=alice, tok=alice_access_token)
  478. # We don't care about interesting events before this (this test is making sure
  479. # the next thing works)
  480. self.send_mock.reset_mock()
  481. # Send a message from the interesting local user
  482. self.helper.send_event(
  483. room_id,
  484. type=EventTypes.Message,
  485. content={
  486. "msgtype": "m.text",
  487. "body": "message from interesting local user",
  488. },
  489. tok=alice_access_token,
  490. )
  491. # Kick the appservice into checking this new event
  492. self._notify_interested_services()
  493. self.send_mock.assert_called_once()
  494. (
  495. service,
  496. events,
  497. _ephemeral,
  498. _to_device_messages,
  499. _otks,
  500. _fbks,
  501. _device_list_summary,
  502. ) = self.send_mock.call_args[0]
  503. # Events sent from an interesting local user should also be picked up as
  504. # interesting to the appservice.
  505. self.assertEqual(service, interested_appservice)
  506. self.assertEqual(events[0]["type"], "m.room.message")
  507. self.assertEqual(events[0]["sender"], alice)
  508. def test_sending_read_receipt_batches_to_application_services(self) -> None:
  509. """Tests that a large batch of read receipts are sent correctly to
  510. interested application services.
  511. """
  512. # Register an application service that's interested in a certain user
  513. # and room prefix
  514. interested_appservice = self._register_application_service(
  515. namespaces={
  516. ApplicationService.NS_USERS: [
  517. {
  518. "regex": "@exclusive_as_user:.+",
  519. "exclusive": True,
  520. }
  521. ],
  522. ApplicationService.NS_ROOMS: [
  523. {
  524. "regex": "!fakeroom_.*",
  525. "exclusive": True,
  526. }
  527. ],
  528. },
  529. )
  530. # Now, pretend that we receive a large burst of read receipts (300 total) that
  531. # all come in at once.
  532. for i in range(300):
  533. self.get_success(
  534. # Insert a fake read receipt into the database
  535. self.hs.get_datastores().main.insert_receipt(
  536. # We have to use unique room ID + user ID combinations here, as the db query
  537. # is an upsert.
  538. room_id=f"!fakeroom_{i}:test",
  539. receipt_type="m.read",
  540. user_id=self.local_user,
  541. event_ids=[f"$eventid_{i}"],
  542. thread_id=None,
  543. data={},
  544. )
  545. )
  546. # Now notify the appservice handler that 300 read receipts have all arrived
  547. # at once. What will it do!
  548. # note: stream tokens start at 2
  549. for stream_token in range(2, 303):
  550. self.get_success(
  551. self.hs.get_application_service_handler()._notify_interested_services_ephemeral(
  552. services=[interested_appservice],
  553. stream_key="receipt_key",
  554. new_token=stream_token,
  555. users=[self.exclusive_as_user],
  556. )
  557. )
  558. # Using our txn send mock, we can see what the AS received. After iterating over every
  559. # transaction, we'd like to see all 300 read receipts accounted for.
  560. # No more, no less.
  561. all_ephemeral_events = []
  562. for call in self.send_mock.call_args_list:
  563. ephemeral_events = call[0][2]
  564. all_ephemeral_events += ephemeral_events
  565. # Ensure that no duplicate events were sent
  566. self.assertEqual(len(all_ephemeral_events), 300)
  567. # Check that the ephemeral event is a read receipt with the expected structure
  568. latest_read_receipt = all_ephemeral_events[-1]
  569. self.assertEqual(latest_read_receipt["type"], EduTypes.RECEIPT)
  570. event_id = list(latest_read_receipt["content"].keys())[0]
  571. self.assertEqual(
  572. latest_read_receipt["content"][event_id]["m.read"], {self.local_user: {}}
  573. )
  574. @unittest.override_config(
  575. {"experimental_features": {"msc2409_to_device_messages_enabled": True}}
  576. )
  577. def test_application_services_receive_local_to_device(self) -> None:
  578. """
  579. Test that when a user sends a to-device message to another user
  580. that is an application service's user namespace, the
  581. application service will receive it.
  582. """
  583. interested_appservice = self._register_application_service(
  584. namespaces={
  585. ApplicationService.NS_USERS: [
  586. {
  587. "regex": "@exclusive_as_user:.+",
  588. "exclusive": True,
  589. }
  590. ],
  591. },
  592. )
  593. # Have local_user send a to-device message to exclusive_as_user
  594. message_content = {"some_key": "some really interesting value"}
  595. chan = self.make_request(
  596. "PUT",
  597. "/_matrix/client/r0/sendToDevice/m.room_key_request/3",
  598. content={
  599. "messages": {
  600. self.exclusive_as_user: {
  601. self.exclusive_as_user_device_id: message_content
  602. }
  603. }
  604. },
  605. access_token=self.local_user_token,
  606. )
  607. self.assertEqual(chan.code, 200, chan.result)
  608. # Have exclusive_as_user send a to-device message to local_user
  609. chan = self.make_request(
  610. "PUT",
  611. "/_matrix/client/r0/sendToDevice/m.room_key_request/4",
  612. content={
  613. "messages": {
  614. self.local_user: {self.local_user_device_id: message_content}
  615. }
  616. },
  617. access_token=self.exclusive_as_user_token,
  618. )
  619. self.assertEqual(chan.code, 200, chan.result)
  620. # Check if our application service - that is interested in exclusive_as_user - received
  621. # the to-device message as part of an AS transaction.
  622. # Only the local_user -> exclusive_as_user to-device message should have been forwarded to the AS.
  623. #
  624. # The uninterested application service should not have been notified at all.
  625. self.send_mock.assert_called_once()
  626. (
  627. service,
  628. _events,
  629. _ephemeral,
  630. to_device_messages,
  631. _otks,
  632. _fbks,
  633. _device_list_summary,
  634. ) = self.send_mock.call_args[0]
  635. # Assert that this was the same to-device message that local_user sent
  636. self.assertEqual(service, interested_appservice)
  637. self.assertEqual(to_device_messages[0]["type"], "m.room_key_request")
  638. self.assertEqual(to_device_messages[0]["sender"], self.local_user)
  639. # Additional fields 'to_user_id' and 'to_device_id' specifically for
  640. # to-device messages via the AS API
  641. self.assertEqual(to_device_messages[0]["to_user_id"], self.exclusive_as_user)
  642. self.assertEqual(
  643. to_device_messages[0]["to_device_id"], self.exclusive_as_user_device_id
  644. )
  645. self.assertEqual(to_device_messages[0]["content"], message_content)
  646. @unittest.override_config(
  647. {"experimental_features": {"msc2409_to_device_messages_enabled": True}}
  648. )
  649. def test_application_services_receive_bursts_of_to_device(self) -> None:
  650. """
  651. Test that when a user sends >100 to-device messages at once, any
  652. interested AS's will receive them in separate transactions.
  653. Also tests that uninterested application services do not receive messages.
  654. """
  655. # Register two application services with exclusive interest in a user
  656. interested_appservices = []
  657. for _ in range(2):
  658. appservice = self._register_application_service(
  659. namespaces={
  660. ApplicationService.NS_USERS: [
  661. {
  662. "regex": "@exclusive_as_user:.+",
  663. "exclusive": True,
  664. }
  665. ],
  666. },
  667. )
  668. interested_appservices.append(appservice)
  669. # ...and an application service which does not have any user interest.
  670. self._register_application_service()
  671. to_device_message_content = {
  672. "some key": "some interesting value",
  673. }
  674. # We need to send a large burst of to-device messages. We also would like to
  675. # include them all in the same application service transaction so that we can
  676. # test large transactions.
  677. #
  678. # To do this, we can send a single to-device message to many user devices at
  679. # once.
  680. #
  681. # We insert number_of_messages - 1 messages into the database directly. We'll then
  682. # send a final to-device message to the real device, which will also kick off
  683. # an AS transaction (as just inserting messages into the DB won't).
  684. number_of_messages = 150
  685. fake_device_ids = [f"device_{num}" for num in range(number_of_messages - 1)]
  686. messages = {
  687. self.exclusive_as_user: {
  688. device_id: {
  689. "type": "test_to_device_message",
  690. "sender": "@some:sender",
  691. "content": to_device_message_content,
  692. }
  693. for device_id in fake_device_ids
  694. }
  695. }
  696. # Create a fake device per message. We can't send to-device messages to
  697. # a device that doesn't exist.
  698. self.get_success(
  699. self.hs.get_datastores().main.db_pool.simple_insert_many(
  700. desc="test_application_services_receive_burst_of_to_device",
  701. table="devices",
  702. keys=("user_id", "device_id"),
  703. values=[
  704. (
  705. self.exclusive_as_user,
  706. device_id,
  707. )
  708. for device_id in fake_device_ids
  709. ],
  710. )
  711. )
  712. # Seed the device_inbox table with our fake messages
  713. self.get_success(
  714. self.hs.get_datastores().main.add_messages_to_device_inbox(messages, {})
  715. )
  716. # Now have local_user send a final to-device message to exclusive_as_user. All unsent
  717. # to-device messages should be sent to any application services
  718. # interested in exclusive_as_user.
  719. chan = self.make_request(
  720. "PUT",
  721. "/_matrix/client/r0/sendToDevice/m.room_key_request/4",
  722. content={
  723. "messages": {
  724. self.exclusive_as_user: {
  725. self.exclusive_as_user_device_id: to_device_message_content
  726. }
  727. }
  728. },
  729. access_token=self.local_user_token,
  730. )
  731. self.assertEqual(chan.code, 200, chan.result)
  732. self.send_mock.assert_called()
  733. # Count the total number of to-device messages that were sent out per-service.
  734. # Ensure that we only sent to-device messages to interested services, and that
  735. # each interested service received the full count of to-device messages.
  736. service_id_to_message_count: Dict[str, int] = {}
  737. for call in self.send_mock.call_args_list:
  738. (
  739. service,
  740. _events,
  741. _ephemeral,
  742. to_device_messages,
  743. _otks,
  744. _fbks,
  745. _device_list_summary,
  746. ) = call[0]
  747. # Check that this was made to an interested service
  748. self.assertIn(service, interested_appservices)
  749. # Add to the count of messages for this application service
  750. service_id_to_message_count.setdefault(service.id, 0)
  751. service_id_to_message_count[service.id] += len(to_device_messages)
  752. # Assert that each interested service received the full count of messages
  753. for count in service_id_to_message_count.values():
  754. self.assertEqual(count, number_of_messages)
  755. def _register_application_service(
  756. self,
  757. namespaces: Optional[Dict[str, Iterable[Dict]]] = None,
  758. ) -> ApplicationService:
  759. """
  760. Register a new application service, with the given namespaces of interest.
  761. Args:
  762. namespaces: A dictionary containing any user, room or alias namespaces that
  763. the application service is interested in.
  764. Returns:
  765. The registered application service.
  766. """
  767. # Create an application service
  768. appservice = ApplicationService(
  769. token=random_string(10),
  770. id=random_string(10),
  771. sender="@as:example.com",
  772. rate_limited=False,
  773. namespaces=namespaces,
  774. supports_ephemeral=True,
  775. )
  776. # Register the application service
  777. self._services.append(appservice)
  778. return appservice
  779. class ApplicationServicesHandlerDeviceListsTestCase(unittest.HomeserverTestCase):
  780. """
  781. Tests that the ApplicationServicesHandler sends device list updates to application
  782. services correctly.
  783. """
  784. servlets = [
  785. synapse.rest.admin.register_servlets_for_client_rest_resource,
  786. login.register_servlets,
  787. room.register_servlets,
  788. ]
  789. def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
  790. # Allow us to modify cached feature flags mid-test
  791. self.as_handler = hs.get_application_service_handler()
  792. # Mock ApplicationServiceApi's put_json, so we can verify the raw JSON that
  793. # will be sent over the wire
  794. self.put_json = simple_async_mock()
  795. hs.get_application_service_api().put_json = self.put_json # type: ignore[assignment]
  796. # Mock out application services, and allow defining our own in tests
  797. self._services: List[ApplicationService] = []
  798. self.hs.get_datastores().main.get_app_services = Mock( # type: ignore[assignment]
  799. return_value=self._services
  800. )
  801. # Test across a variety of configuration values
  802. @parameterized.expand(
  803. [
  804. (True, True, True),
  805. (True, False, False),
  806. (False, True, False),
  807. (False, False, False),
  808. ]
  809. )
  810. def test_application_service_receives_device_list_updates(
  811. self,
  812. experimental_feature_enabled: bool,
  813. as_supports_txn_extensions: bool,
  814. as_should_receive_device_list_updates: bool,
  815. ) -> None:
  816. """
  817. Tests that an application service receives notice of changed device
  818. lists for a user, when a user changes their device lists.
  819. Arguments above are populated by parameterized.
  820. Args:
  821. as_should_receive_device_list_updates: Whether we expect the AS to receive the
  822. device list changes.
  823. experimental_feature_enabled: Whether the "msc3202_transaction_extensions" experimental
  824. feature is enabled. This feature must be enabled for device lists to ASs to work.
  825. as_supports_txn_extensions: Whether the application service has explicitly registered
  826. to receive information defined by MSC3202 - which includes device list changes.
  827. """
  828. # Change whether the experimental feature is enabled or disabled before making
  829. # device list changes
  830. self.as_handler._msc3202_transaction_extensions_enabled = (
  831. experimental_feature_enabled
  832. )
  833. # Create an appservice that is interested in "local_user"
  834. appservice = ApplicationService(
  835. token=random_string(10),
  836. id=random_string(10),
  837. sender="@as:example.com",
  838. rate_limited=False,
  839. namespaces={
  840. ApplicationService.NS_USERS: [
  841. {
  842. "regex": "@local_user:.+",
  843. "exclusive": False,
  844. }
  845. ],
  846. },
  847. supports_ephemeral=True,
  848. msc3202_transaction_extensions=as_supports_txn_extensions,
  849. # Must be set for Synapse to try pushing data to the AS
  850. hs_token="abcde",
  851. url="some_url",
  852. )
  853. # Register the application service
  854. self._services.append(appservice)
  855. # Register a user on the homeserver
  856. self.local_user = self.register_user("local_user", "password")
  857. self.local_user_token = self.login("local_user", "password")
  858. if as_should_receive_device_list_updates:
  859. # Ensure that the resulting JSON uses the unstable prefix and contains the
  860. # expected users
  861. self.put_json.assert_called_once()
  862. json_body = self.put_json.call_args[1]["json_body"]
  863. # Our application service should have received a device list update with
  864. # "local_user" in the "changed" list
  865. device_list_dict = json_body.get("org.matrix.msc3202.device_lists", {})
  866. self.assertEqual([], device_list_dict["left"])
  867. self.assertEqual([self.local_user], device_list_dict["changed"])
  868. else:
  869. # No device list changes should have been sent out
  870. self.put_json.assert_not_called()
  871. class ApplicationServicesHandlerOtkCountsTestCase(unittest.HomeserverTestCase):
  872. # Argument indices for pulling out arguments from a `send_mock`.
  873. ARG_OTK_COUNTS = 4
  874. ARG_FALLBACK_KEYS = 5
  875. servlets = [
  876. synapse.rest.admin.register_servlets_for_client_rest_resource,
  877. login.register_servlets,
  878. register.register_servlets,
  879. room.register_servlets,
  880. sendtodevice.register_servlets,
  881. receipts.register_servlets,
  882. ]
  883. def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
  884. # Mock the ApplicationServiceScheduler's _TransactionController's send method so that
  885. # we can track what's going out
  886. self.send_mock = simple_async_mock()
  887. hs.get_application_service_handler().scheduler.txn_ctrl.send = self.send_mock # type: ignore[assignment] # We assign to a method.
  888. # Define an application service for the tests
  889. self._service_token = "VERYSECRET"
  890. self._service = ApplicationService(
  891. self._service_token,
  892. "as1",
  893. "@as.sender:test",
  894. namespaces={
  895. "users": [
  896. {"regex": "@_as_.*:test", "exclusive": True},
  897. {"regex": "@as.sender:test", "exclusive": True},
  898. ]
  899. },
  900. msc3202_transaction_extensions=True,
  901. )
  902. self.hs.get_datastores().main.services_cache = [self._service]
  903. # Register some appservice users
  904. self._sender_user, self._sender_device = self.register_appservice_user(
  905. "as.sender", self._service_token
  906. )
  907. self._namespaced_user, self._namespaced_device = self.register_appservice_user(
  908. "_as_user1", self._service_token
  909. )
  910. # Register a real user as well.
  911. self._real_user = self.register_user("real.user", "meow")
  912. self._real_user_token = self.login("real.user", "meow")
  913. async def _add_otks_for_device(
  914. self, user_id: str, device_id: str, otk_count: int
  915. ) -> None:
  916. """
  917. Add some dummy keys. It doesn't matter if they're not a real algorithm;
  918. that should be opaque to the server anyway.
  919. """
  920. await self.hs.get_datastores().main.add_e2e_one_time_keys(
  921. user_id,
  922. device_id,
  923. self.clock.time_msec(),
  924. [("algo", f"k{i}", "{}") for i in range(otk_count)],
  925. )
  926. async def _add_fallback_key_for_device(
  927. self, user_id: str, device_id: str, used: bool
  928. ) -> None:
  929. """
  930. Adds a fake fallback key to a device, optionally marking it as used
  931. right away.
  932. """
  933. store = self.hs.get_datastores().main
  934. await store.set_e2e_fallback_keys(user_id, device_id, {"algo:fk": "fall back!"})
  935. if used is True:
  936. # Mark the key as used
  937. await store.db_pool.simple_update_one(
  938. table="e2e_fallback_keys_json",
  939. keyvalues={
  940. "user_id": user_id,
  941. "device_id": device_id,
  942. "algorithm": "algo",
  943. "key_id": "fk",
  944. },
  945. updatevalues={"used": True},
  946. desc="_get_fallback_key_set_used",
  947. )
  948. def _set_up_devices_and_a_room(self) -> str:
  949. """
  950. Helper to set up devices for all the users
  951. and a room for the users to talk in.
  952. """
  953. async def preparation() -> None:
  954. await self._add_otks_for_device(self._sender_user, self._sender_device, 42)
  955. await self._add_fallback_key_for_device(
  956. self._sender_user, self._sender_device, used=True
  957. )
  958. await self._add_otks_for_device(
  959. self._namespaced_user, self._namespaced_device, 36
  960. )
  961. await self._add_fallback_key_for_device(
  962. self._namespaced_user, self._namespaced_device, used=False
  963. )
  964. # Register a device for the real user, too, so that we can later ensure
  965. # that we don't leak information to the AS about the non-AS user.
  966. await self.hs.get_datastores().main.store_device(
  967. self._real_user, "REALDEV", "UltraMatrix 3000"
  968. )
  969. await self._add_otks_for_device(self._real_user, "REALDEV", 50)
  970. self.get_success(preparation())
  971. room_id = self.helper.create_room_as(
  972. self._real_user, is_public=True, tok=self._real_user_token
  973. )
  974. self.helper.join(
  975. room_id,
  976. self._namespaced_user,
  977. tok=self._service_token,
  978. appservice_user_id=self._namespaced_user,
  979. )
  980. # Check it was called for sanity. (This was to send the join event to the AS.)
  981. self.send_mock.assert_called()
  982. self.send_mock.reset_mock()
  983. return room_id
  984. @override_config(
  985. {"experimental_features": {"msc3202_transaction_extensions": True}}
  986. )
  987. def test_application_services_receive_otk_counts_and_fallback_key_usages_with_pdus(
  988. self,
  989. ) -> None:
  990. """
  991. Tests that:
  992. - the AS receives one-time key counts and unused fallback keys for:
  993. - the specified sender; and
  994. - any user who is in receipt of the PDUs
  995. """
  996. room_id = self._set_up_devices_and_a_room()
  997. # Send a message into the AS's room
  998. self.helper.send(room_id, "woof woof", tok=self._real_user_token)
  999. # Capture what was sent as an AS transaction.
  1000. self.send_mock.assert_called()
  1001. last_args, _last_kwargs = self.send_mock.call_args
  1002. otks: Optional[TransactionOneTimeKeysCount] = last_args[self.ARG_OTK_COUNTS]
  1003. unused_fallbacks: Optional[TransactionUnusedFallbackKeys] = last_args[
  1004. self.ARG_FALLBACK_KEYS
  1005. ]
  1006. self.assertEqual(
  1007. otks,
  1008. {
  1009. "@as.sender:test": {self._sender_device: {"algo": 42}},
  1010. "@_as_user1:test": {self._namespaced_device: {"algo": 36}},
  1011. },
  1012. )
  1013. self.assertEqual(
  1014. unused_fallbacks,
  1015. {
  1016. "@as.sender:test": {self._sender_device: []},
  1017. "@_as_user1:test": {self._namespaced_device: ["algo"]},
  1018. },
  1019. )