test_device.py 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538
  1. # Copyright 2020 Dirk Klimpel
  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. import urllib.parse
  15. from parameterized import parameterized
  16. from twisted.test.proto_helpers import MemoryReactor
  17. import synapse.rest.admin
  18. from synapse.api.errors import Codes
  19. from synapse.handlers.device import DeviceHandler
  20. from synapse.rest.client import login
  21. from synapse.server import HomeServer
  22. from synapse.util import Clock
  23. from tests import unittest
  24. class DeviceRestTestCase(unittest.HomeserverTestCase):
  25. servlets = [
  26. synapse.rest.admin.register_servlets,
  27. login.register_servlets,
  28. ]
  29. def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
  30. handler = hs.get_device_handler()
  31. assert isinstance(handler, DeviceHandler)
  32. self.handler = handler
  33. self.admin_user = self.register_user("admin", "pass", admin=True)
  34. self.admin_user_tok = self.login("admin", "pass")
  35. self.other_user = self.register_user("user", "pass")
  36. self.other_user_token = self.login("user", "pass")
  37. res = self.get_success(self.handler.get_devices_by_user(self.other_user))
  38. self.other_user_device_id = res[0]["device_id"]
  39. self.url = "/_synapse/admin/v2/users/%s/devices/%s" % (
  40. urllib.parse.quote(self.other_user),
  41. self.other_user_device_id,
  42. )
  43. @parameterized.expand(["GET", "PUT", "DELETE"])
  44. def test_no_auth(self, method: str) -> None:
  45. """
  46. Try to get a device of an user without authentication.
  47. """
  48. channel = self.make_request(method, self.url, b"{}")
  49. self.assertEqual(
  50. 401,
  51. channel.code,
  52. msg=channel.json_body,
  53. )
  54. self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
  55. @parameterized.expand(["GET", "PUT", "DELETE"])
  56. def test_requester_is_no_admin(self, method: str) -> None:
  57. """
  58. If the user is not a server admin, an error is returned.
  59. """
  60. channel = self.make_request(
  61. method,
  62. self.url,
  63. access_token=self.other_user_token,
  64. )
  65. self.assertEqual(
  66. 403,
  67. channel.code,
  68. msg=channel.json_body,
  69. )
  70. self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
  71. @parameterized.expand(["GET", "PUT", "DELETE"])
  72. def test_user_does_not_exist(self, method: str) -> None:
  73. """
  74. Tests that a lookup for a user that does not exist returns a 404
  75. """
  76. url = (
  77. "/_synapse/admin/v2/users/@unknown_person:test/devices/%s"
  78. % self.other_user_device_id
  79. )
  80. channel = self.make_request(
  81. method,
  82. url,
  83. access_token=self.admin_user_tok,
  84. )
  85. self.assertEqual(404, channel.code, msg=channel.json_body)
  86. self.assertEqual(Codes.NOT_FOUND, channel.json_body["errcode"])
  87. @parameterized.expand(["GET", "PUT", "DELETE"])
  88. def test_user_is_not_local(self, method: str) -> None:
  89. """
  90. Tests that a lookup for a user that is not a local returns a 400
  91. """
  92. url = (
  93. "/_synapse/admin/v2/users/@unknown_person:unknown_domain/devices/%s"
  94. % self.other_user_device_id
  95. )
  96. channel = self.make_request(
  97. method,
  98. url,
  99. access_token=self.admin_user_tok,
  100. )
  101. self.assertEqual(400, channel.code, msg=channel.json_body)
  102. self.assertEqual("Can only lookup local users", channel.json_body["error"])
  103. def test_unknown_device(self) -> None:
  104. """
  105. Tests that a lookup for a device that does not exist returns either 404 or 200.
  106. """
  107. url = "/_synapse/admin/v2/users/%s/devices/unknown_device" % urllib.parse.quote(
  108. self.other_user
  109. )
  110. channel = self.make_request(
  111. "GET",
  112. url,
  113. access_token=self.admin_user_tok,
  114. )
  115. self.assertEqual(404, channel.code, msg=channel.json_body)
  116. self.assertEqual(Codes.NOT_FOUND, channel.json_body["errcode"])
  117. channel = self.make_request(
  118. "PUT",
  119. url,
  120. access_token=self.admin_user_tok,
  121. )
  122. self.assertEqual(200, channel.code, msg=channel.json_body)
  123. channel = self.make_request(
  124. "DELETE",
  125. url,
  126. access_token=self.admin_user_tok,
  127. )
  128. # Delete unknown device returns status 200
  129. self.assertEqual(200, channel.code, msg=channel.json_body)
  130. def test_update_device_too_long_display_name(self) -> None:
  131. """
  132. Update a device with a display name that is invalid (too long).
  133. """
  134. # Set iniital display name.
  135. update = {"display_name": "new display"}
  136. self.get_success(
  137. self.handler.update_device(
  138. self.other_user, self.other_user_device_id, update
  139. )
  140. )
  141. # Request to update a device display name with a new value that is longer than allowed.
  142. update = {
  143. "display_name": "a"
  144. * (synapse.handlers.device.MAX_DEVICE_DISPLAY_NAME_LEN + 1)
  145. }
  146. channel = self.make_request(
  147. "PUT",
  148. self.url,
  149. access_token=self.admin_user_tok,
  150. content=update,
  151. )
  152. self.assertEqual(400, channel.code, msg=channel.json_body)
  153. self.assertEqual(Codes.TOO_LARGE, channel.json_body["errcode"])
  154. # Ensure the display name was not updated.
  155. channel = self.make_request(
  156. "GET",
  157. self.url,
  158. access_token=self.admin_user_tok,
  159. )
  160. self.assertEqual(200, channel.code, msg=channel.json_body)
  161. self.assertEqual("new display", channel.json_body["display_name"])
  162. def test_update_no_display_name(self) -> None:
  163. """
  164. Tests that a update for a device without JSON returns a 200
  165. """
  166. # Set iniital display name.
  167. update = {"display_name": "new display"}
  168. self.get_success(
  169. self.handler.update_device(
  170. self.other_user, self.other_user_device_id, update
  171. )
  172. )
  173. channel = self.make_request(
  174. "PUT",
  175. self.url,
  176. access_token=self.admin_user_tok,
  177. )
  178. self.assertEqual(200, channel.code, msg=channel.json_body)
  179. # Ensure the display name was not updated.
  180. channel = self.make_request(
  181. "GET",
  182. self.url,
  183. access_token=self.admin_user_tok,
  184. )
  185. self.assertEqual(200, channel.code, msg=channel.json_body)
  186. self.assertEqual("new display", channel.json_body["display_name"])
  187. def test_update_display_name(self) -> None:
  188. """
  189. Tests a normal successful update of display name
  190. """
  191. # Set new display_name
  192. channel = self.make_request(
  193. "PUT",
  194. self.url,
  195. access_token=self.admin_user_tok,
  196. content={"display_name": "new displayname"},
  197. )
  198. self.assertEqual(200, channel.code, msg=channel.json_body)
  199. # Check new display_name
  200. channel = self.make_request(
  201. "GET",
  202. self.url,
  203. access_token=self.admin_user_tok,
  204. )
  205. self.assertEqual(200, channel.code, msg=channel.json_body)
  206. self.assertEqual("new displayname", channel.json_body["display_name"])
  207. def test_get_device(self) -> None:
  208. """
  209. Tests that a normal lookup for a device is successfully
  210. """
  211. channel = self.make_request(
  212. "GET",
  213. self.url,
  214. access_token=self.admin_user_tok,
  215. )
  216. self.assertEqual(200, channel.code, msg=channel.json_body)
  217. self.assertEqual(self.other_user, channel.json_body["user_id"])
  218. # Check that all fields are available
  219. self.assertIn("user_id", channel.json_body)
  220. self.assertIn("device_id", channel.json_body)
  221. self.assertIn("display_name", channel.json_body)
  222. self.assertIn("last_seen_ip", channel.json_body)
  223. self.assertIn("last_seen_ts", channel.json_body)
  224. def test_delete_device(self) -> None:
  225. """
  226. Tests that a remove of a device is successfully
  227. """
  228. # Count number of devies of an user.
  229. res = self.get_success(self.handler.get_devices_by_user(self.other_user))
  230. number_devices = len(res)
  231. self.assertEqual(1, number_devices)
  232. # Delete device
  233. channel = self.make_request(
  234. "DELETE",
  235. self.url,
  236. access_token=self.admin_user_tok,
  237. )
  238. self.assertEqual(200, channel.code, msg=channel.json_body)
  239. # Ensure that the number of devices is decreased
  240. res = self.get_success(self.handler.get_devices_by_user(self.other_user))
  241. self.assertEqual(number_devices - 1, len(res))
  242. class DevicesRestTestCase(unittest.HomeserverTestCase):
  243. servlets = [
  244. synapse.rest.admin.register_servlets,
  245. login.register_servlets,
  246. ]
  247. def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
  248. self.admin_user = self.register_user("admin", "pass", admin=True)
  249. self.admin_user_tok = self.login("admin", "pass")
  250. self.other_user = self.register_user("user", "pass")
  251. self.url = "/_synapse/admin/v2/users/%s/devices" % urllib.parse.quote(
  252. self.other_user
  253. )
  254. def test_no_auth(self) -> None:
  255. """
  256. Try to list devices of an user without authentication.
  257. """
  258. channel = self.make_request("GET", self.url, b"{}")
  259. self.assertEqual(
  260. 401,
  261. channel.code,
  262. msg=channel.json_body,
  263. )
  264. self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
  265. def test_requester_is_no_admin(self) -> None:
  266. """
  267. If the user is not a server admin, an error is returned.
  268. """
  269. other_user_token = self.login("user", "pass")
  270. channel = self.make_request(
  271. "GET",
  272. self.url,
  273. access_token=other_user_token,
  274. )
  275. self.assertEqual(
  276. 403,
  277. channel.code,
  278. msg=channel.json_body,
  279. )
  280. self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
  281. def test_user_does_not_exist(self) -> None:
  282. """
  283. Tests that a lookup for a user that does not exist returns a 404
  284. """
  285. url = "/_synapse/admin/v2/users/@unknown_person:test/devices"
  286. channel = self.make_request(
  287. "GET",
  288. url,
  289. access_token=self.admin_user_tok,
  290. )
  291. self.assertEqual(404, channel.code, msg=channel.json_body)
  292. self.assertEqual(Codes.NOT_FOUND, channel.json_body["errcode"])
  293. def test_user_is_not_local(self) -> None:
  294. """
  295. Tests that a lookup for a user that is not a local returns a 400
  296. """
  297. url = "/_synapse/admin/v2/users/@unknown_person:unknown_domain/devices"
  298. channel = self.make_request(
  299. "GET",
  300. url,
  301. access_token=self.admin_user_tok,
  302. )
  303. self.assertEqual(400, channel.code, msg=channel.json_body)
  304. self.assertEqual("Can only lookup local users", channel.json_body["error"])
  305. def test_user_has_no_devices(self) -> None:
  306. """
  307. Tests that a normal lookup for devices is successfully
  308. if user has no devices
  309. """
  310. # Get devices
  311. channel = self.make_request(
  312. "GET",
  313. self.url,
  314. access_token=self.admin_user_tok,
  315. )
  316. self.assertEqual(200, channel.code, msg=channel.json_body)
  317. self.assertEqual(0, channel.json_body["total"])
  318. self.assertEqual(0, len(channel.json_body["devices"]))
  319. def test_get_devices(self) -> None:
  320. """
  321. Tests that a normal lookup for devices is successfully
  322. """
  323. # Create devices
  324. number_devices = 5
  325. for _ in range(number_devices):
  326. self.login("user", "pass")
  327. # Get devices
  328. channel = self.make_request(
  329. "GET",
  330. self.url,
  331. access_token=self.admin_user_tok,
  332. )
  333. self.assertEqual(200, channel.code, msg=channel.json_body)
  334. self.assertEqual(number_devices, channel.json_body["total"])
  335. self.assertEqual(number_devices, len(channel.json_body["devices"]))
  336. self.assertEqual(self.other_user, channel.json_body["devices"][0]["user_id"])
  337. # Check that all fields are available
  338. for d in channel.json_body["devices"]:
  339. self.assertIn("user_id", d)
  340. self.assertIn("device_id", d)
  341. self.assertIn("display_name", d)
  342. self.assertIn("last_seen_ip", d)
  343. self.assertIn("last_seen_ts", d)
  344. class DeleteDevicesRestTestCase(unittest.HomeserverTestCase):
  345. servlets = [
  346. synapse.rest.admin.register_servlets,
  347. login.register_servlets,
  348. ]
  349. def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
  350. self.handler = hs.get_device_handler()
  351. self.admin_user = self.register_user("admin", "pass", admin=True)
  352. self.admin_user_tok = self.login("admin", "pass")
  353. self.other_user = self.register_user("user", "pass")
  354. self.url = "/_synapse/admin/v2/users/%s/delete_devices" % urllib.parse.quote(
  355. self.other_user
  356. )
  357. def test_no_auth(self) -> None:
  358. """
  359. Try to delete devices of an user without authentication.
  360. """
  361. channel = self.make_request("POST", self.url, b"{}")
  362. self.assertEqual(
  363. 401,
  364. channel.code,
  365. msg=channel.json_body,
  366. )
  367. self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
  368. def test_requester_is_no_admin(self) -> None:
  369. """
  370. If the user is not a server admin, an error is returned.
  371. """
  372. other_user_token = self.login("user", "pass")
  373. channel = self.make_request(
  374. "POST",
  375. self.url,
  376. access_token=other_user_token,
  377. )
  378. self.assertEqual(
  379. 403,
  380. channel.code,
  381. msg=channel.json_body,
  382. )
  383. self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
  384. def test_user_does_not_exist(self) -> None:
  385. """
  386. Tests that a lookup for a user that does not exist returns a 404
  387. """
  388. url = "/_synapse/admin/v2/users/@unknown_person:test/delete_devices"
  389. channel = self.make_request(
  390. "POST",
  391. url,
  392. access_token=self.admin_user_tok,
  393. )
  394. self.assertEqual(404, channel.code, msg=channel.json_body)
  395. self.assertEqual(Codes.NOT_FOUND, channel.json_body["errcode"])
  396. def test_user_is_not_local(self) -> None:
  397. """
  398. Tests that a lookup for a user that is not a local returns a 400
  399. """
  400. url = "/_synapse/admin/v2/users/@unknown_person:unknown_domain/delete_devices"
  401. channel = self.make_request(
  402. "POST",
  403. url,
  404. access_token=self.admin_user_tok,
  405. )
  406. self.assertEqual(400, channel.code, msg=channel.json_body)
  407. self.assertEqual("Can only lookup local users", channel.json_body["error"])
  408. def test_unknown_devices(self) -> None:
  409. """
  410. Tests that a remove of a device that does not exist returns 200.
  411. """
  412. channel = self.make_request(
  413. "POST",
  414. self.url,
  415. access_token=self.admin_user_tok,
  416. content={"devices": ["unknown_device1", "unknown_device2"]},
  417. )
  418. # Delete unknown devices returns status 200
  419. self.assertEqual(200, channel.code, msg=channel.json_body)
  420. def test_delete_devices(self) -> None:
  421. """
  422. Tests that a remove of devices is successfully
  423. """
  424. # Create devices
  425. number_devices = 5
  426. for _ in range(number_devices):
  427. self.login("user", "pass")
  428. # Get devices
  429. res = self.get_success(self.handler.get_devices_by_user(self.other_user))
  430. self.assertEqual(number_devices, len(res))
  431. # Create list of device IDs
  432. device_ids = []
  433. for d in res:
  434. device_ids.append(str(d["device_id"]))
  435. # Delete devices
  436. channel = self.make_request(
  437. "POST",
  438. self.url,
  439. access_token=self.admin_user_tok,
  440. content={"devices": device_ids},
  441. )
  442. self.assertEqual(200, channel.code, msg=channel.json_body)
  443. res = self.get_success(self.handler.get_devices_by_user(self.other_user))
  444. self.assertEqual(0, len(res))