1
0

test_device.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344
  1. # Copyright 2016 OpenMarket Ltd
  2. # Copyright 2018 New Vector Ltd
  3. # Copyright 2020 The Matrix.org Foundation C.I.C.
  4. #
  5. # Licensed under the Apache License, Version 2.0 (the "License");
  6. # you may not use this file except in compliance with the License.
  7. # You may obtain a copy of the License at
  8. #
  9. # http://www.apache.org/licenses/LICENSE-2.0
  10. #
  11. # Unless required by applicable law or agreed to in writing, software
  12. # distributed under the License is distributed on an "AS IS" BASIS,
  13. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14. # See the License for the specific language governing permissions and
  15. # limitations under the License.
  16. from typing import Optional
  17. from twisted.test.proto_helpers import MemoryReactor
  18. from synapse.api.errors import NotFoundError, SynapseError
  19. from synapse.handlers.device import MAX_DEVICE_DISPLAY_NAME_LEN
  20. from synapse.server import HomeServer
  21. from synapse.util import Clock
  22. from tests import unittest
  23. user1 = "@boris:aaa"
  24. user2 = "@theresa:bbb"
  25. class DeviceTestCase(unittest.HomeserverTestCase):
  26. def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
  27. hs = self.setup_test_homeserver("server", federation_http_client=None)
  28. self.handler = hs.get_device_handler()
  29. self.store = hs.get_datastores().main
  30. return hs
  31. def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
  32. # These tests assume that it starts 1000 seconds in.
  33. self.reactor.advance(1000)
  34. def test_device_is_created_with_invalid_name(self) -> None:
  35. self.get_failure(
  36. self.handler.check_device_registered(
  37. user_id="@boris:foo",
  38. device_id="foo",
  39. initial_device_display_name="a" * (MAX_DEVICE_DISPLAY_NAME_LEN + 1),
  40. ),
  41. SynapseError,
  42. )
  43. def test_device_is_created_if_doesnt_exist(self) -> None:
  44. res = self.get_success(
  45. self.handler.check_device_registered(
  46. user_id="@boris:foo",
  47. device_id="fco",
  48. initial_device_display_name="display name",
  49. )
  50. )
  51. self.assertEqual(res, "fco")
  52. dev = self.get_success(self.handler.store.get_device("@boris:foo", "fco"))
  53. self.assertEqual(dev["display_name"], "display name")
  54. def test_device_is_preserved_if_exists(self) -> None:
  55. res1 = self.get_success(
  56. self.handler.check_device_registered(
  57. user_id="@boris:foo",
  58. device_id="fco",
  59. initial_device_display_name="display name",
  60. )
  61. )
  62. self.assertEqual(res1, "fco")
  63. res2 = self.get_success(
  64. self.handler.check_device_registered(
  65. user_id="@boris:foo",
  66. device_id="fco",
  67. initial_device_display_name="new display name",
  68. )
  69. )
  70. self.assertEqual(res2, "fco")
  71. dev = self.get_success(self.handler.store.get_device("@boris:foo", "fco"))
  72. self.assertEqual(dev["display_name"], "display name")
  73. def test_device_id_is_made_up_if_unspecified(self) -> None:
  74. device_id = self.get_success(
  75. self.handler.check_device_registered(
  76. user_id="@theresa:foo",
  77. device_id=None,
  78. initial_device_display_name="display",
  79. )
  80. )
  81. dev = self.get_success(self.handler.store.get_device("@theresa:foo", device_id))
  82. self.assertEqual(dev["display_name"], "display")
  83. def test_get_devices_by_user(self) -> None:
  84. self._record_users()
  85. res = self.get_success(self.handler.get_devices_by_user(user1))
  86. self.assertEqual(3, len(res))
  87. device_map = {d["device_id"]: d for d in res}
  88. self.assertDictContainsSubset(
  89. {
  90. "user_id": user1,
  91. "device_id": "xyz",
  92. "display_name": "display 0",
  93. "last_seen_ip": None,
  94. "last_seen_ts": None,
  95. },
  96. device_map["xyz"],
  97. )
  98. self.assertDictContainsSubset(
  99. {
  100. "user_id": user1,
  101. "device_id": "fco",
  102. "display_name": "display 1",
  103. "last_seen_ip": "ip1",
  104. "last_seen_ts": 1000000,
  105. },
  106. device_map["fco"],
  107. )
  108. self.assertDictContainsSubset(
  109. {
  110. "user_id": user1,
  111. "device_id": "abc",
  112. "display_name": "display 2",
  113. "last_seen_ip": "ip3",
  114. "last_seen_ts": 3000000,
  115. },
  116. device_map["abc"],
  117. )
  118. def test_get_device(self) -> None:
  119. self._record_users()
  120. res = self.get_success(self.handler.get_device(user1, "abc"))
  121. self.assertDictContainsSubset(
  122. {
  123. "user_id": user1,
  124. "device_id": "abc",
  125. "display_name": "display 2",
  126. "last_seen_ip": "ip3",
  127. "last_seen_ts": 3000000,
  128. },
  129. res,
  130. )
  131. def test_delete_device(self) -> None:
  132. self._record_users()
  133. # delete the device
  134. self.get_success(self.handler.delete_device(user1, "abc"))
  135. # check the device was deleted
  136. self.get_failure(self.handler.get_device(user1, "abc"), NotFoundError)
  137. # we'd like to check the access token was invalidated, but that's a
  138. # bit of a PITA.
  139. def test_delete_device_and_device_inbox(self) -> None:
  140. self._record_users()
  141. # add an device_inbox
  142. self.get_success(
  143. self.store.db_pool.simple_insert(
  144. "device_inbox",
  145. {
  146. "user_id": user1,
  147. "device_id": "abc",
  148. "stream_id": 1,
  149. "message_json": "{}",
  150. },
  151. )
  152. )
  153. # delete the device
  154. self.get_success(self.handler.delete_device(user1, "abc"))
  155. # check that the device_inbox was deleted
  156. res = self.get_success(
  157. self.store.db_pool.simple_select_one(
  158. table="device_inbox",
  159. keyvalues={"user_id": user1, "device_id": "abc"},
  160. retcols=("user_id", "device_id"),
  161. allow_none=True,
  162. desc="get_device_id_from_device_inbox",
  163. )
  164. )
  165. self.assertIsNone(res)
  166. def test_update_device(self) -> None:
  167. self._record_users()
  168. update = {"display_name": "new display"}
  169. self.get_success(self.handler.update_device(user1, "abc", update))
  170. res = self.get_success(self.handler.get_device(user1, "abc"))
  171. self.assertEqual(res["display_name"], "new display")
  172. def test_update_device_too_long_display_name(self) -> None:
  173. """Update a device with a display name that is invalid (too long)."""
  174. self._record_users()
  175. # Request to update a device display name with a new value that is longer than allowed.
  176. update = {"display_name": "a" * (MAX_DEVICE_DISPLAY_NAME_LEN + 1)}
  177. self.get_failure(
  178. self.handler.update_device(user1, "abc", update),
  179. SynapseError,
  180. )
  181. # Ensure the display name was not updated.
  182. res = self.get_success(self.handler.get_device(user1, "abc"))
  183. self.assertEqual(res["display_name"], "display 2")
  184. def test_update_unknown_device(self) -> None:
  185. update = {"display_name": "new_display"}
  186. self.get_failure(
  187. self.handler.update_device("user_id", "unknown_device_id", update),
  188. NotFoundError,
  189. )
  190. def _record_users(self) -> None:
  191. # check this works for both devices which have a recorded client_ip,
  192. # and those which don't.
  193. self._record_user(user1, "xyz", "display 0")
  194. self._record_user(user1, "fco", "display 1", "token1", "ip1")
  195. self._record_user(user1, "abc", "display 2", "token2", "ip2")
  196. self._record_user(user1, "abc", "display 2", "token3", "ip3")
  197. self._record_user(user2, "def", "dispkay", "token4", "ip4")
  198. self.reactor.advance(10000)
  199. def _record_user(
  200. self,
  201. user_id: str,
  202. device_id: str,
  203. display_name: str,
  204. access_token: Optional[str] = None,
  205. ip: Optional[str] = None,
  206. ) -> None:
  207. device_id = self.get_success(
  208. self.handler.check_device_registered(
  209. user_id=user_id,
  210. device_id=device_id,
  211. initial_device_display_name=display_name,
  212. )
  213. )
  214. if access_token is not None and ip is not None:
  215. self.get_success(
  216. self.store.insert_client_ip(
  217. user_id, access_token, ip, "user_agent", device_id
  218. )
  219. )
  220. self.reactor.advance(1000)
  221. class DehydrationTestCase(unittest.HomeserverTestCase):
  222. def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
  223. hs = self.setup_test_homeserver("server", federation_http_client=None)
  224. self.handler = hs.get_device_handler()
  225. self.registration = hs.get_registration_handler()
  226. self.auth = hs.get_auth()
  227. self.store = hs.get_datastores().main
  228. return hs
  229. def test_dehydrate_and_rehydrate_device(self) -> None:
  230. user_id = "@boris:dehydration"
  231. self.get_success(self.store.register_user(user_id, "foobar"))
  232. # First check if we can store and fetch a dehydrated device
  233. stored_dehydrated_device_id = self.get_success(
  234. self.handler.store_dehydrated_device(
  235. user_id=user_id,
  236. device_data={"device_data": {"foo": "bar"}},
  237. initial_device_display_name="dehydrated device",
  238. )
  239. )
  240. retrieved_device_id, device_data = self.get_success(
  241. self.handler.get_dehydrated_device(user_id=user_id)
  242. )
  243. self.assertEqual(retrieved_device_id, stored_dehydrated_device_id)
  244. self.assertEqual(device_data, {"device_data": {"foo": "bar"}})
  245. # Create a new login for the user and dehydrated the device
  246. device_id, access_token, _expiration_time, _refresh_token = self.get_success(
  247. self.registration.register_device(
  248. user_id=user_id,
  249. device_id=None,
  250. initial_display_name="new device",
  251. )
  252. )
  253. # Trying to claim a nonexistent device should throw an error
  254. self.get_failure(
  255. self.handler.rehydrate_device(
  256. user_id=user_id,
  257. access_token=access_token,
  258. device_id="not the right device ID",
  259. ),
  260. NotFoundError,
  261. )
  262. # dehydrating the right devices should succeed and change our device ID
  263. # to the dehydrated device's ID
  264. res = self.get_success(
  265. self.handler.rehydrate_device(
  266. user_id=user_id,
  267. access_token=access_token,
  268. device_id=retrieved_device_id,
  269. )
  270. )
  271. self.assertEqual(res, {"success": True})
  272. # make sure that our device ID has changed
  273. user_info = self.get_success(self.auth.get_user_by_access_token(access_token))
  274. self.assertEqual(user_info.device_id, retrieved_device_id)
  275. # make sure the device has the display name that was set from the login
  276. res = self.get_success(self.handler.get_device(user_id, retrieved_device_id))
  277. self.assertEqual(res["display_name"], "new device")
  278. # make sure that the device ID that we were initially assigned no longer exists
  279. self.get_failure(
  280. self.handler.get_device(user_id, device_id),
  281. NotFoundError,
  282. )
  283. # make sure that there's no device available for dehydrating now
  284. ret = self.get_success(self.handler.get_dehydrated_device(user_id=user_id))
  285. self.assertIsNone(ret)