1
0

test_federation_sender.py 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544
  1. # -*- coding: utf-8 -*-
  2. # Copyright 2019 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. from typing import Optional
  16. from mock import Mock
  17. from signedjson import key, sign
  18. from signedjson.types import BaseKey, SigningKey
  19. from twisted.internet import defer
  20. from synapse.rest import admin
  21. from synapse.rest.client.v1 import login
  22. from synapse.types import JsonDict, ReadReceipt
  23. from tests.unittest import HomeserverTestCase, override_config
  24. class FederationSenderReceiptsTestCases(HomeserverTestCase):
  25. def make_homeserver(self, reactor, clock):
  26. return self.setup_test_homeserver(
  27. state_handler=Mock(spec=["get_current_hosts_in_room"]),
  28. federation_transport_client=Mock(spec=["send_transaction"]),
  29. )
  30. @override_config({"send_federation": True})
  31. def test_send_receipts(self):
  32. mock_state_handler = self.hs.get_state_handler()
  33. mock_state_handler.get_current_hosts_in_room.return_value = ["test", "host2"]
  34. mock_send_transaction = (
  35. self.hs.get_federation_transport_client().send_transaction
  36. )
  37. mock_send_transaction.return_value = defer.succeed({})
  38. sender = self.hs.get_federation_sender()
  39. receipt = ReadReceipt(
  40. "room_id", "m.read", "user_id", ["event_id"], {"ts": 1234}
  41. )
  42. self.successResultOf(sender.send_read_receipt(receipt))
  43. self.pump()
  44. # expect a call to send_transaction
  45. mock_send_transaction.assert_called_once()
  46. json_cb = mock_send_transaction.call_args[0][1]
  47. data = json_cb()
  48. self.assertEqual(
  49. data["edus"],
  50. [
  51. {
  52. "edu_type": "m.receipt",
  53. "content": {
  54. "room_id": {
  55. "m.read": {
  56. "user_id": {
  57. "event_ids": ["event_id"],
  58. "data": {"ts": 1234},
  59. }
  60. }
  61. }
  62. },
  63. }
  64. ],
  65. )
  66. @override_config({"send_federation": True})
  67. def test_send_receipts_with_backoff(self):
  68. """Send two receipts in quick succession; the second should be flushed, but
  69. only after 20ms"""
  70. mock_state_handler = self.hs.get_state_handler()
  71. mock_state_handler.get_current_hosts_in_room.return_value = ["test", "host2"]
  72. mock_send_transaction = (
  73. self.hs.get_federation_transport_client().send_transaction
  74. )
  75. mock_send_transaction.return_value = defer.succeed({})
  76. sender = self.hs.get_federation_sender()
  77. receipt = ReadReceipt(
  78. "room_id", "m.read", "user_id", ["event_id"], {"ts": 1234}
  79. )
  80. self.successResultOf(sender.send_read_receipt(receipt))
  81. self.pump()
  82. # expect a call to send_transaction
  83. mock_send_transaction.assert_called_once()
  84. json_cb = mock_send_transaction.call_args[0][1]
  85. data = json_cb()
  86. self.assertEqual(
  87. data["edus"],
  88. [
  89. {
  90. "edu_type": "m.receipt",
  91. "content": {
  92. "room_id": {
  93. "m.read": {
  94. "user_id": {
  95. "event_ids": ["event_id"],
  96. "data": {"ts": 1234},
  97. }
  98. }
  99. }
  100. },
  101. }
  102. ],
  103. )
  104. mock_send_transaction.reset_mock()
  105. # send the second RR
  106. receipt = ReadReceipt(
  107. "room_id", "m.read", "user_id", ["other_id"], {"ts": 1234}
  108. )
  109. self.successResultOf(sender.send_read_receipt(receipt))
  110. self.pump()
  111. mock_send_transaction.assert_not_called()
  112. self.reactor.advance(19)
  113. mock_send_transaction.assert_not_called()
  114. self.reactor.advance(10)
  115. mock_send_transaction.assert_called_once()
  116. json_cb = mock_send_transaction.call_args[0][1]
  117. data = json_cb()
  118. self.assertEqual(
  119. data["edus"],
  120. [
  121. {
  122. "edu_type": "m.receipt",
  123. "content": {
  124. "room_id": {
  125. "m.read": {
  126. "user_id": {
  127. "event_ids": ["other_id"],
  128. "data": {"ts": 1234},
  129. }
  130. }
  131. }
  132. },
  133. }
  134. ],
  135. )
  136. class FederationSenderDevicesTestCases(HomeserverTestCase):
  137. servlets = [
  138. admin.register_servlets,
  139. login.register_servlets,
  140. ]
  141. def make_homeserver(self, reactor, clock):
  142. return self.setup_test_homeserver(
  143. state_handler=Mock(spec=["get_current_hosts_in_room"]),
  144. federation_transport_client=Mock(spec=["send_transaction"]),
  145. )
  146. def default_config(self):
  147. c = super().default_config()
  148. c["send_federation"] = True
  149. return c
  150. def prepare(self, reactor, clock, hs):
  151. # stub out get_current_hosts_in_room
  152. mock_state_handler = hs.get_state_handler()
  153. mock_state_handler.get_current_hosts_in_room.return_value = ["test", "host2"]
  154. # stub out get_users_who_share_room_with_user so that it claims that
  155. # `@user2:host2` is in the room
  156. def get_users_who_share_room_with_user(user_id):
  157. return defer.succeed({"@user2:host2"})
  158. hs.get_datastore().get_users_who_share_room_with_user = (
  159. get_users_who_share_room_with_user
  160. )
  161. # whenever send_transaction is called, record the edu data
  162. self.edus = []
  163. self.hs.get_federation_transport_client().send_transaction.side_effect = (
  164. self.record_transaction
  165. )
  166. def record_transaction(self, txn, json_cb):
  167. data = json_cb()
  168. self.edus.extend(data["edus"])
  169. return defer.succeed({})
  170. def test_send_device_updates(self):
  171. """Basic case: each device update should result in an EDU"""
  172. # create a device
  173. u1 = self.register_user("user", "pass")
  174. self.login(u1, "pass", device_id="D1")
  175. # expect one edu
  176. self.assertEqual(len(self.edus), 1)
  177. stream_id = self.check_device_update_edu(self.edus.pop(0), u1, "D1", None)
  178. # a second call should produce no new device EDUs
  179. self.hs.get_federation_sender().send_device_messages("host2")
  180. self.pump()
  181. self.assertEqual(self.edus, [])
  182. # a second device
  183. self.login("user", "pass", device_id="D2")
  184. self.assertEqual(len(self.edus), 1)
  185. self.check_device_update_edu(self.edus.pop(0), u1, "D2", stream_id)
  186. def test_upload_signatures(self):
  187. """Uploading signatures on some devices should produce updates for that user"""
  188. e2e_handler = self.hs.get_e2e_keys_handler()
  189. # register two devices
  190. u1 = self.register_user("user", "pass")
  191. self.login(u1, "pass", device_id="D1")
  192. self.login(u1, "pass", device_id="D2")
  193. # expect two edus
  194. self.assertEqual(len(self.edus), 2)
  195. stream_id = None
  196. stream_id = self.check_device_update_edu(self.edus.pop(0), u1, "D1", stream_id)
  197. stream_id = self.check_device_update_edu(self.edus.pop(0), u1, "D2", stream_id)
  198. # upload signing keys for each device
  199. device1_signing_key = self.generate_and_upload_device_signing_key(u1, "D1")
  200. device2_signing_key = self.generate_and_upload_device_signing_key(u1, "D2")
  201. # expect two more edus
  202. self.assertEqual(len(self.edus), 2)
  203. stream_id = self.check_device_update_edu(self.edus.pop(0), u1, "D1", stream_id)
  204. stream_id = self.check_device_update_edu(self.edus.pop(0), u1, "D2", stream_id)
  205. # upload master key and self-signing key
  206. master_signing_key = generate_self_id_key()
  207. master_key = {
  208. "user_id": u1,
  209. "usage": ["master"],
  210. "keys": {key_id(master_signing_key): encode_pubkey(master_signing_key)},
  211. }
  212. # private key: HvQBbU+hc2Zr+JP1sE0XwBe1pfZZEYtJNPJLZJtS+F8
  213. selfsigning_signing_key = generate_self_id_key()
  214. selfsigning_key = {
  215. "user_id": u1,
  216. "usage": ["self_signing"],
  217. "keys": {
  218. key_id(selfsigning_signing_key): encode_pubkey(selfsigning_signing_key)
  219. },
  220. }
  221. sign.sign_json(selfsigning_key, u1, master_signing_key)
  222. cross_signing_keys = {
  223. "master_key": master_key,
  224. "self_signing_key": selfsigning_key,
  225. }
  226. self.get_success(
  227. e2e_handler.upload_signing_keys_for_user(u1, cross_signing_keys)
  228. )
  229. # expect signing key update edu
  230. self.assertEqual(len(self.edus), 1)
  231. self.assertEqual(self.edus.pop(0)["edu_type"], "org.matrix.signing_key_update")
  232. # sign the devices
  233. d1_json = build_device_dict(u1, "D1", device1_signing_key)
  234. sign.sign_json(d1_json, u1, selfsigning_signing_key)
  235. d2_json = build_device_dict(u1, "D2", device2_signing_key)
  236. sign.sign_json(d2_json, u1, selfsigning_signing_key)
  237. ret = self.get_success(
  238. e2e_handler.upload_signatures_for_device_keys(
  239. u1, {u1: {"D1": d1_json, "D2": d2_json}},
  240. )
  241. )
  242. self.assertEqual(ret["failures"], {})
  243. # expect two edus, in one or two transactions. We don't know what order the
  244. # devices will be updated.
  245. self.assertEqual(len(self.edus), 2)
  246. stream_id = None # FIXME: there is a discontinuity in the stream IDs: see #7142
  247. for edu in self.edus:
  248. self.assertEqual(edu["edu_type"], "m.device_list_update")
  249. c = edu["content"]
  250. if stream_id is not None:
  251. self.assertEqual(c["prev_id"], [stream_id])
  252. self.assertGreaterEqual(c["stream_id"], stream_id)
  253. stream_id = c["stream_id"]
  254. devices = {edu["content"]["device_id"] for edu in self.edus}
  255. self.assertEqual({"D1", "D2"}, devices)
  256. def test_delete_devices(self):
  257. """If devices are deleted, that should result in EDUs too"""
  258. # create devices
  259. u1 = self.register_user("user", "pass")
  260. self.login("user", "pass", device_id="D1")
  261. self.login("user", "pass", device_id="D2")
  262. self.login("user", "pass", device_id="D3")
  263. # expect three edus
  264. self.assertEqual(len(self.edus), 3)
  265. stream_id = self.check_device_update_edu(self.edus.pop(0), u1, "D1", None)
  266. stream_id = self.check_device_update_edu(self.edus.pop(0), u1, "D2", stream_id)
  267. stream_id = self.check_device_update_edu(self.edus.pop(0), u1, "D3", stream_id)
  268. # delete them again
  269. self.get_success(
  270. self.hs.get_device_handler().delete_devices(u1, ["D1", "D2", "D3"])
  271. )
  272. # expect three edus, in an unknown order
  273. self.assertEqual(len(self.edus), 3)
  274. for edu in self.edus:
  275. self.assertEqual(edu["edu_type"], "m.device_list_update")
  276. c = edu["content"]
  277. self.assertGreaterEqual(
  278. c.items(),
  279. {"user_id": u1, "prev_id": [stream_id], "deleted": True}.items(),
  280. )
  281. self.assertGreaterEqual(c["stream_id"], stream_id)
  282. stream_id = c["stream_id"]
  283. devices = {edu["content"]["device_id"] for edu in self.edus}
  284. self.assertEqual({"D1", "D2", "D3"}, devices)
  285. def test_unreachable_server(self):
  286. """If the destination server is unreachable, all the updates should get sent on
  287. recovery
  288. """
  289. mock_send_txn = self.hs.get_federation_transport_client().send_transaction
  290. mock_send_txn.side_effect = lambda t, cb: defer.fail("fail")
  291. # create devices
  292. u1 = self.register_user("user", "pass")
  293. self.login("user", "pass", device_id="D1")
  294. self.login("user", "pass", device_id="D2")
  295. self.login("user", "pass", device_id="D3")
  296. # delete them again
  297. self.get_success(
  298. self.hs.get_device_handler().delete_devices(u1, ["D1", "D2", "D3"])
  299. )
  300. self.assertGreaterEqual(mock_send_txn.call_count, 4)
  301. # recover the server
  302. mock_send_txn.side_effect = self.record_transaction
  303. self.hs.get_federation_sender().send_device_messages("host2")
  304. self.pump()
  305. # for each device, there should be a single update
  306. self.assertEqual(len(self.edus), 3)
  307. stream_id = None
  308. for edu in self.edus:
  309. self.assertEqual(edu["edu_type"], "m.device_list_update")
  310. c = edu["content"]
  311. self.assertEqual(c["prev_id"], [stream_id] if stream_id is not None else [])
  312. if stream_id is not None:
  313. self.assertGreaterEqual(c["stream_id"], stream_id)
  314. stream_id = c["stream_id"]
  315. devices = {edu["content"]["device_id"] for edu in self.edus}
  316. self.assertEqual({"D1", "D2", "D3"}, devices)
  317. def test_prune_outbound_device_pokes1(self):
  318. """If a destination is unreachable, and the updates are pruned, we should get
  319. a single update.
  320. This case tests the behaviour when the server has never been reachable.
  321. """
  322. mock_send_txn = self.hs.get_federation_transport_client().send_transaction
  323. mock_send_txn.side_effect = lambda t, cb: defer.fail("fail")
  324. # create devices
  325. u1 = self.register_user("user", "pass")
  326. self.login("user", "pass", device_id="D1")
  327. self.login("user", "pass", device_id="D2")
  328. self.login("user", "pass", device_id="D3")
  329. # delete them again
  330. self.get_success(
  331. self.hs.get_device_handler().delete_devices(u1, ["D1", "D2", "D3"])
  332. )
  333. self.assertGreaterEqual(mock_send_txn.call_count, 4)
  334. # run the prune job
  335. self.reactor.advance(10)
  336. self.get_success(
  337. self.hs.get_datastore()._prune_old_outbound_device_pokes(prune_age=1)
  338. )
  339. # recover the server
  340. mock_send_txn.side_effect = self.record_transaction
  341. self.hs.get_federation_sender().send_device_messages("host2")
  342. self.pump()
  343. # there should be a single update for this user.
  344. self.assertEqual(len(self.edus), 1)
  345. edu = self.edus.pop(0)
  346. self.assertEqual(edu["edu_type"], "m.device_list_update")
  347. c = edu["content"]
  348. # synapse uses an empty prev_id list to indicate "needs a full resync".
  349. self.assertEqual(c["prev_id"], [])
  350. def test_prune_outbound_device_pokes2(self):
  351. """If a destination is unreachable, and the updates are pruned, we should get
  352. a single update.
  353. This case tests the behaviour when the server was reachable, but then goes
  354. offline.
  355. """
  356. # create first device
  357. u1 = self.register_user("user", "pass")
  358. self.login("user", "pass", device_id="D1")
  359. # expect the update EDU
  360. self.assertEqual(len(self.edus), 1)
  361. self.check_device_update_edu(self.edus.pop(0), u1, "D1", None)
  362. # now the server goes offline
  363. mock_send_txn = self.hs.get_federation_transport_client().send_transaction
  364. mock_send_txn.side_effect = lambda t, cb: defer.fail("fail")
  365. self.login("user", "pass", device_id="D2")
  366. self.login("user", "pass", device_id="D3")
  367. # delete them again
  368. self.get_success(
  369. self.hs.get_device_handler().delete_devices(u1, ["D1", "D2", "D3"])
  370. )
  371. self.assertGreaterEqual(mock_send_txn.call_count, 3)
  372. # run the prune job
  373. self.reactor.advance(10)
  374. self.get_success(
  375. self.hs.get_datastore()._prune_old_outbound_device_pokes(prune_age=1)
  376. )
  377. # recover the server
  378. mock_send_txn.side_effect = self.record_transaction
  379. self.hs.get_federation_sender().send_device_messages("host2")
  380. self.pump()
  381. # ... and we should get a single update for this user.
  382. self.assertEqual(len(self.edus), 1)
  383. edu = self.edus.pop(0)
  384. self.assertEqual(edu["edu_type"], "m.device_list_update")
  385. c = edu["content"]
  386. # synapse uses an empty prev_id list to indicate "needs a full resync".
  387. self.assertEqual(c["prev_id"], [])
  388. def check_device_update_edu(
  389. self,
  390. edu: JsonDict,
  391. user_id: str,
  392. device_id: str,
  393. prev_stream_id: Optional[int],
  394. ) -> int:
  395. """Check that the given EDU is an update for the given device
  396. Returns the stream_id.
  397. """
  398. self.assertEqual(edu["edu_type"], "m.device_list_update")
  399. content = edu["content"]
  400. expected = {
  401. "user_id": user_id,
  402. "device_id": device_id,
  403. "prev_id": [prev_stream_id] if prev_stream_id is not None else [],
  404. }
  405. self.assertLessEqual(expected.items(), content.items())
  406. if prev_stream_id is not None:
  407. self.assertGreaterEqual(content["stream_id"], prev_stream_id)
  408. return content["stream_id"]
  409. def check_signing_key_update_txn(self, txn: JsonDict,) -> None:
  410. """Check that the txn has an EDU with a signing key update.
  411. """
  412. edus = txn["edus"]
  413. self.assertEqual(len(edus), 1)
  414. def generate_and_upload_device_signing_key(
  415. self, user_id: str, device_id: str
  416. ) -> SigningKey:
  417. """Generate a signing keypair for the given device, and upload it"""
  418. sk = key.generate_signing_key(device_id)
  419. device_dict = build_device_dict(user_id, device_id, sk)
  420. self.get_success(
  421. self.hs.get_e2e_keys_handler().upload_keys_for_user(
  422. user_id, device_id, {"device_keys": device_dict},
  423. )
  424. )
  425. return sk
  426. def generate_self_id_key() -> SigningKey:
  427. """generate a signing key whose version is its public key
  428. ... as used by the cross-signing-keys.
  429. """
  430. k = key.generate_signing_key("x")
  431. k.version = encode_pubkey(k)
  432. return k
  433. def key_id(k: BaseKey) -> str:
  434. return "%s:%s" % (k.alg, k.version)
  435. def encode_pubkey(sk: SigningKey) -> str:
  436. """Encode the public key corresponding to the given signing key as base64"""
  437. return key.encode_verify_key_base64(key.get_verify_key(sk))
  438. def build_device_dict(user_id: str, device_id: str, sk: SigningKey):
  439. """Build a dict representing the given device"""
  440. return {
  441. "user_id": user_id,
  442. "device_id": device_id,
  443. "algorithms": ["m.olm.curve25519-aes-sha256", "m.megolm.v1.aes-sha"],
  444. "keys": {
  445. "curve25519:" + device_id: "curve25519+key",
  446. key_id(sk): encode_pubkey(sk),
  447. },
  448. }