test_appservice.py 38 KB

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