test_registration_tokens.py 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832
  1. # Copyright 2021 Callum Brown
  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 random
  15. import string
  16. from http import HTTPStatus
  17. from typing import Optional
  18. from twisted.test.proto_helpers import MemoryReactor
  19. import synapse.rest.admin
  20. from synapse.api.errors import Codes
  21. from synapse.rest.client import login
  22. from synapse.server import HomeServer
  23. from synapse.util import Clock
  24. from tests import unittest
  25. class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase):
  26. servlets = [
  27. synapse.rest.admin.register_servlets,
  28. login.register_servlets,
  29. ]
  30. def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
  31. self.store = hs.get_datastores().main
  32. self.admin_user = self.register_user("admin", "pass", admin=True)
  33. self.admin_user_tok = self.login("admin", "pass")
  34. self.other_user = self.register_user("user", "pass")
  35. self.other_user_tok = self.login("user", "pass")
  36. self.url = "/_synapse/admin/v1/registration_tokens"
  37. def _new_token(
  38. self,
  39. token: Optional[str] = None,
  40. uses_allowed: Optional[int] = None,
  41. pending: int = 0,
  42. completed: int = 0,
  43. expiry_time: Optional[int] = None,
  44. ) -> str:
  45. """Helper function to create a token."""
  46. if token is None:
  47. token = "".join(random.choices(string.ascii_letters, k=8))
  48. self.get_success(
  49. self.store.db_pool.simple_insert(
  50. "registration_tokens",
  51. {
  52. "token": token,
  53. "uses_allowed": uses_allowed,
  54. "pending": pending,
  55. "completed": completed,
  56. "expiry_time": expiry_time,
  57. },
  58. )
  59. )
  60. return token
  61. # CREATION
  62. def test_create_no_auth(self) -> None:
  63. """Try to create a token without authentication."""
  64. channel = self.make_request("POST", self.url + "/new", {})
  65. self.assertEqual(
  66. HTTPStatus.UNAUTHORIZED,
  67. channel.code,
  68. msg=channel.json_body,
  69. )
  70. self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
  71. def test_create_requester_not_admin(self) -> None:
  72. """Try to create a token while not an admin."""
  73. channel = self.make_request(
  74. "POST",
  75. self.url + "/new",
  76. {},
  77. access_token=self.other_user_tok,
  78. )
  79. self.assertEqual(
  80. HTTPStatus.FORBIDDEN,
  81. channel.code,
  82. msg=channel.json_body,
  83. )
  84. self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
  85. def test_create_using_defaults(self) -> None:
  86. """Create a token using all the defaults."""
  87. channel = self.make_request(
  88. "POST",
  89. self.url + "/new",
  90. {},
  91. access_token=self.admin_user_tok,
  92. )
  93. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
  94. self.assertEqual(len(channel.json_body["token"]), 16)
  95. self.assertIsNone(channel.json_body["uses_allowed"])
  96. self.assertIsNone(channel.json_body["expiry_time"])
  97. self.assertEqual(channel.json_body["pending"], 0)
  98. self.assertEqual(channel.json_body["completed"], 0)
  99. def test_create_specifying_fields(self) -> None:
  100. """Create a token specifying the value of all fields."""
  101. # As many of the allowed characters as possible with length <= 64
  102. token = "adefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._~-"
  103. data = {
  104. "token": token,
  105. "uses_allowed": 1,
  106. "expiry_time": self.clock.time_msec() + 1000000,
  107. }
  108. channel = self.make_request(
  109. "POST",
  110. self.url + "/new",
  111. data,
  112. access_token=self.admin_user_tok,
  113. )
  114. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
  115. self.assertEqual(channel.json_body["token"], token)
  116. self.assertEqual(channel.json_body["uses_allowed"], 1)
  117. self.assertEqual(channel.json_body["expiry_time"], data["expiry_time"])
  118. self.assertEqual(channel.json_body["pending"], 0)
  119. self.assertEqual(channel.json_body["completed"], 0)
  120. def test_create_with_null_value(self) -> None:
  121. """Create a token specifying unlimited uses and no expiry."""
  122. data = {
  123. "uses_allowed": None,
  124. "expiry_time": None,
  125. }
  126. channel = self.make_request(
  127. "POST",
  128. self.url + "/new",
  129. data,
  130. access_token=self.admin_user_tok,
  131. )
  132. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
  133. self.assertEqual(len(channel.json_body["token"]), 16)
  134. self.assertIsNone(channel.json_body["uses_allowed"])
  135. self.assertIsNone(channel.json_body["expiry_time"])
  136. self.assertEqual(channel.json_body["pending"], 0)
  137. self.assertEqual(channel.json_body["completed"], 0)
  138. def test_create_token_too_long(self) -> None:
  139. """Check token longer than 64 chars is invalid."""
  140. data = {"token": "a" * 65}
  141. channel = self.make_request(
  142. "POST",
  143. self.url + "/new",
  144. data,
  145. access_token=self.admin_user_tok,
  146. )
  147. self.assertEqual(
  148. HTTPStatus.BAD_REQUEST,
  149. channel.code,
  150. msg=channel.json_body,
  151. )
  152. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  153. def test_create_token_invalid_chars(self) -> None:
  154. """Check you can't create token with invalid characters."""
  155. data = {
  156. "token": "abc/def",
  157. }
  158. channel = self.make_request(
  159. "POST",
  160. self.url + "/new",
  161. data,
  162. access_token=self.admin_user_tok,
  163. )
  164. self.assertEqual(
  165. HTTPStatus.BAD_REQUEST,
  166. channel.code,
  167. msg=channel.json_body,
  168. )
  169. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  170. def test_create_token_already_exists(self) -> None:
  171. """Check you can't create token that already exists."""
  172. data = {
  173. "token": "abcd",
  174. }
  175. channel1 = self.make_request(
  176. "POST",
  177. self.url + "/new",
  178. data,
  179. access_token=self.admin_user_tok,
  180. )
  181. self.assertEqual(HTTPStatus.OK, channel1.code, msg=channel1.json_body)
  182. channel2 = self.make_request(
  183. "POST",
  184. self.url + "/new",
  185. data,
  186. access_token=self.admin_user_tok,
  187. )
  188. self.assertEqual(HTTPStatus.BAD_REQUEST, channel2.code, msg=channel2.json_body)
  189. self.assertEqual(channel2.json_body["errcode"], Codes.INVALID_PARAM)
  190. def test_create_unable_to_generate_token(self) -> None:
  191. """Check right error is raised when server can't generate unique token."""
  192. # Create all possible single character tokens
  193. tokens = []
  194. for c in string.ascii_letters + string.digits + "._~-":
  195. tokens.append((c, None, 0, 0, None))
  196. self.get_success(
  197. self.store.db_pool.simple_insert_many(
  198. "registration_tokens",
  199. keys=("token", "uses_allowed", "pending", "completed", "expiry_time"),
  200. values=tokens,
  201. desc="create_all_registration_tokens",
  202. )
  203. )
  204. # Check creating a single character token fails with a 500 status code
  205. channel = self.make_request(
  206. "POST",
  207. self.url + "/new",
  208. {"length": 1},
  209. access_token=self.admin_user_tok,
  210. )
  211. self.assertEqual(500, channel.code, msg=channel.json_body)
  212. def test_create_uses_allowed(self) -> None:
  213. """Check you can only create a token with good values for uses_allowed."""
  214. # Should work with 0 (token is invalid from the start)
  215. channel = self.make_request(
  216. "POST",
  217. self.url + "/new",
  218. {"uses_allowed": 0},
  219. access_token=self.admin_user_tok,
  220. )
  221. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
  222. self.assertEqual(channel.json_body["uses_allowed"], 0)
  223. # Should fail with negative integer
  224. channel = self.make_request(
  225. "POST",
  226. self.url + "/new",
  227. {"uses_allowed": -5},
  228. access_token=self.admin_user_tok,
  229. )
  230. self.assertEqual(
  231. HTTPStatus.BAD_REQUEST,
  232. channel.code,
  233. msg=channel.json_body,
  234. )
  235. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  236. # Should fail with float
  237. channel = self.make_request(
  238. "POST",
  239. self.url + "/new",
  240. {"uses_allowed": 1.5},
  241. access_token=self.admin_user_tok,
  242. )
  243. self.assertEqual(
  244. HTTPStatus.BAD_REQUEST,
  245. channel.code,
  246. msg=channel.json_body,
  247. )
  248. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  249. def test_create_expiry_time(self) -> None:
  250. """Check you can't create a token with an invalid expiry_time."""
  251. # Should fail with a time in the past
  252. channel = self.make_request(
  253. "POST",
  254. self.url + "/new",
  255. {"expiry_time": self.clock.time_msec() - 10000},
  256. access_token=self.admin_user_tok,
  257. )
  258. self.assertEqual(
  259. HTTPStatus.BAD_REQUEST,
  260. channel.code,
  261. msg=channel.json_body,
  262. )
  263. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  264. # Should fail with float
  265. channel = self.make_request(
  266. "POST",
  267. self.url + "/new",
  268. {"expiry_time": self.clock.time_msec() + 1000000.5},
  269. access_token=self.admin_user_tok,
  270. )
  271. self.assertEqual(
  272. HTTPStatus.BAD_REQUEST,
  273. channel.code,
  274. msg=channel.json_body,
  275. )
  276. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  277. def test_create_length(self) -> None:
  278. """Check you can only generate a token with a valid length."""
  279. # Should work with 64
  280. channel = self.make_request(
  281. "POST",
  282. self.url + "/new",
  283. {"length": 64},
  284. access_token=self.admin_user_tok,
  285. )
  286. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
  287. self.assertEqual(len(channel.json_body["token"]), 64)
  288. # Should fail with 0
  289. channel = self.make_request(
  290. "POST",
  291. self.url + "/new",
  292. {"length": 0},
  293. access_token=self.admin_user_tok,
  294. )
  295. self.assertEqual(
  296. HTTPStatus.BAD_REQUEST,
  297. channel.code,
  298. msg=channel.json_body,
  299. )
  300. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  301. # Should fail with a negative integer
  302. channel = self.make_request(
  303. "POST",
  304. self.url + "/new",
  305. {"length": -5},
  306. access_token=self.admin_user_tok,
  307. )
  308. self.assertEqual(
  309. HTTPStatus.BAD_REQUEST,
  310. channel.code,
  311. msg=channel.json_body,
  312. )
  313. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  314. # Should fail with a float
  315. channel = self.make_request(
  316. "POST",
  317. self.url + "/new",
  318. {"length": 8.5},
  319. access_token=self.admin_user_tok,
  320. )
  321. self.assertEqual(
  322. HTTPStatus.BAD_REQUEST,
  323. channel.code,
  324. msg=channel.json_body,
  325. )
  326. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  327. # Should fail with 65
  328. channel = self.make_request(
  329. "POST",
  330. self.url + "/new",
  331. {"length": 65},
  332. access_token=self.admin_user_tok,
  333. )
  334. self.assertEqual(
  335. HTTPStatus.BAD_REQUEST,
  336. channel.code,
  337. msg=channel.json_body,
  338. )
  339. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  340. # UPDATING
  341. def test_update_no_auth(self) -> None:
  342. """Try to update a token without authentication."""
  343. channel = self.make_request(
  344. "PUT",
  345. self.url + "/1234", # Token doesn't exist but that doesn't matter
  346. {},
  347. )
  348. self.assertEqual(
  349. HTTPStatus.UNAUTHORIZED,
  350. channel.code,
  351. msg=channel.json_body,
  352. )
  353. self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
  354. def test_update_requester_not_admin(self) -> None:
  355. """Try to update a token while not an admin."""
  356. channel = self.make_request(
  357. "PUT",
  358. self.url + "/1234", # Token doesn't exist but that doesn't matter
  359. {},
  360. access_token=self.other_user_tok,
  361. )
  362. self.assertEqual(
  363. HTTPStatus.FORBIDDEN,
  364. channel.code,
  365. msg=channel.json_body,
  366. )
  367. self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
  368. def test_update_non_existent(self) -> None:
  369. """Try to update a token that doesn't exist."""
  370. channel = self.make_request(
  371. "PUT",
  372. self.url + "/1234",
  373. {"uses_allowed": 1},
  374. access_token=self.admin_user_tok,
  375. )
  376. self.assertEqual(
  377. HTTPStatus.NOT_FOUND,
  378. channel.code,
  379. msg=channel.json_body,
  380. )
  381. self.assertEqual(channel.json_body["errcode"], Codes.NOT_FOUND)
  382. def test_update_uses_allowed(self) -> None:
  383. """Test updating just uses_allowed."""
  384. # Create new token using default values
  385. token = self._new_token()
  386. # Should succeed with 1
  387. channel = self.make_request(
  388. "PUT",
  389. self.url + "/" + token,
  390. {"uses_allowed": 1},
  391. access_token=self.admin_user_tok,
  392. )
  393. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
  394. self.assertEqual(channel.json_body["uses_allowed"], 1)
  395. self.assertIsNone(channel.json_body["expiry_time"])
  396. # Should succeed with 0 (makes token invalid)
  397. channel = self.make_request(
  398. "PUT",
  399. self.url + "/" + token,
  400. {"uses_allowed": 0},
  401. access_token=self.admin_user_tok,
  402. )
  403. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
  404. self.assertEqual(channel.json_body["uses_allowed"], 0)
  405. self.assertIsNone(channel.json_body["expiry_time"])
  406. # Should succeed with null
  407. channel = self.make_request(
  408. "PUT",
  409. self.url + "/" + token,
  410. {"uses_allowed": None},
  411. access_token=self.admin_user_tok,
  412. )
  413. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
  414. self.assertIsNone(channel.json_body["uses_allowed"])
  415. self.assertIsNone(channel.json_body["expiry_time"])
  416. # Should fail with a float
  417. channel = self.make_request(
  418. "PUT",
  419. self.url + "/" + token,
  420. {"uses_allowed": 1.5},
  421. access_token=self.admin_user_tok,
  422. )
  423. self.assertEqual(
  424. HTTPStatus.BAD_REQUEST,
  425. channel.code,
  426. msg=channel.json_body,
  427. )
  428. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  429. # Should fail with a negative integer
  430. channel = self.make_request(
  431. "PUT",
  432. self.url + "/" + token,
  433. {"uses_allowed": -5},
  434. access_token=self.admin_user_tok,
  435. )
  436. self.assertEqual(
  437. HTTPStatus.BAD_REQUEST,
  438. channel.code,
  439. msg=channel.json_body,
  440. )
  441. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  442. def test_update_expiry_time(self) -> None:
  443. """Test updating just expiry_time."""
  444. # Create new token using default values
  445. token = self._new_token()
  446. new_expiry_time = self.clock.time_msec() + 1000000
  447. # Should succeed with a time in the future
  448. channel = self.make_request(
  449. "PUT",
  450. self.url + "/" + token,
  451. {"expiry_time": new_expiry_time},
  452. access_token=self.admin_user_tok,
  453. )
  454. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
  455. self.assertEqual(channel.json_body["expiry_time"], new_expiry_time)
  456. self.assertIsNone(channel.json_body["uses_allowed"])
  457. # Should succeed with null
  458. channel = self.make_request(
  459. "PUT",
  460. self.url + "/" + token,
  461. {"expiry_time": None},
  462. access_token=self.admin_user_tok,
  463. )
  464. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
  465. self.assertIsNone(channel.json_body["expiry_time"])
  466. self.assertIsNone(channel.json_body["uses_allowed"])
  467. # Should fail with a time in the past
  468. past_time = self.clock.time_msec() - 10000
  469. channel = self.make_request(
  470. "PUT",
  471. self.url + "/" + token,
  472. {"expiry_time": past_time},
  473. access_token=self.admin_user_tok,
  474. )
  475. self.assertEqual(
  476. HTTPStatus.BAD_REQUEST,
  477. channel.code,
  478. msg=channel.json_body,
  479. )
  480. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  481. # Should fail a float
  482. channel = self.make_request(
  483. "PUT",
  484. self.url + "/" + token,
  485. {"expiry_time": new_expiry_time + 0.5},
  486. access_token=self.admin_user_tok,
  487. )
  488. self.assertEqual(
  489. HTTPStatus.BAD_REQUEST,
  490. channel.code,
  491. msg=channel.json_body,
  492. )
  493. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  494. def test_update_both(self) -> None:
  495. """Test updating both uses_allowed and expiry_time."""
  496. # Create new token using default values
  497. token = self._new_token()
  498. new_expiry_time = self.clock.time_msec() + 1000000
  499. data = {
  500. "uses_allowed": 1,
  501. "expiry_time": new_expiry_time,
  502. }
  503. channel = self.make_request(
  504. "PUT",
  505. self.url + "/" + token,
  506. data,
  507. access_token=self.admin_user_tok,
  508. )
  509. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
  510. self.assertEqual(channel.json_body["uses_allowed"], 1)
  511. self.assertEqual(channel.json_body["expiry_time"], new_expiry_time)
  512. def test_update_invalid_type(self) -> None:
  513. """Test using invalid types doesn't work."""
  514. # Create new token using default values
  515. token = self._new_token()
  516. data = {
  517. "uses_allowed": False,
  518. "expiry_time": "1626430124000",
  519. }
  520. channel = self.make_request(
  521. "PUT",
  522. self.url + "/" + token,
  523. data,
  524. access_token=self.admin_user_tok,
  525. )
  526. self.assertEqual(
  527. HTTPStatus.BAD_REQUEST,
  528. channel.code,
  529. msg=channel.json_body,
  530. )
  531. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  532. # DELETING
  533. def test_delete_no_auth(self) -> None:
  534. """Try to delete a token without authentication."""
  535. channel = self.make_request(
  536. "DELETE",
  537. self.url + "/1234", # Token doesn't exist but that doesn't matter
  538. {},
  539. )
  540. self.assertEqual(
  541. HTTPStatus.UNAUTHORIZED,
  542. channel.code,
  543. msg=channel.json_body,
  544. )
  545. self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
  546. def test_delete_requester_not_admin(self) -> None:
  547. """Try to delete a token while not an admin."""
  548. channel = self.make_request(
  549. "DELETE",
  550. self.url + "/1234", # Token doesn't exist but that doesn't matter
  551. {},
  552. access_token=self.other_user_tok,
  553. )
  554. self.assertEqual(
  555. HTTPStatus.FORBIDDEN,
  556. channel.code,
  557. msg=channel.json_body,
  558. )
  559. self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
  560. def test_delete_non_existent(self) -> None:
  561. """Try to delete a token that doesn't exist."""
  562. channel = self.make_request(
  563. "DELETE",
  564. self.url + "/1234",
  565. {},
  566. access_token=self.admin_user_tok,
  567. )
  568. self.assertEqual(
  569. HTTPStatus.NOT_FOUND,
  570. channel.code,
  571. msg=channel.json_body,
  572. )
  573. self.assertEqual(channel.json_body["errcode"], Codes.NOT_FOUND)
  574. def test_delete(self) -> None:
  575. """Test deleting a token."""
  576. # Create new token using default values
  577. token = self._new_token()
  578. channel = self.make_request(
  579. "DELETE",
  580. self.url + "/" + token,
  581. {},
  582. access_token=self.admin_user_tok,
  583. )
  584. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
  585. # GETTING ONE
  586. def test_get_no_auth(self) -> None:
  587. """Try to get a token without authentication."""
  588. channel = self.make_request(
  589. "GET",
  590. self.url + "/1234", # Token doesn't exist but that doesn't matter
  591. {},
  592. )
  593. self.assertEqual(
  594. HTTPStatus.UNAUTHORIZED,
  595. channel.code,
  596. msg=channel.json_body,
  597. )
  598. self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
  599. def test_get_requester_not_admin(self) -> None:
  600. """Try to get a token while not an admin."""
  601. channel = self.make_request(
  602. "GET",
  603. self.url + "/1234", # Token doesn't exist but that doesn't matter
  604. {},
  605. access_token=self.other_user_tok,
  606. )
  607. self.assertEqual(
  608. HTTPStatus.FORBIDDEN,
  609. channel.code,
  610. msg=channel.json_body,
  611. )
  612. self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
  613. def test_get_non_existent(self) -> None:
  614. """Try to get a token that doesn't exist."""
  615. channel = self.make_request(
  616. "GET",
  617. self.url + "/1234",
  618. {},
  619. access_token=self.admin_user_tok,
  620. )
  621. self.assertEqual(
  622. HTTPStatus.NOT_FOUND,
  623. channel.code,
  624. msg=channel.json_body,
  625. )
  626. self.assertEqual(channel.json_body["errcode"], Codes.NOT_FOUND)
  627. def test_get(self) -> None:
  628. """Test getting a token."""
  629. # Create new token using default values
  630. token = self._new_token()
  631. channel = self.make_request(
  632. "GET",
  633. self.url + "/" + token,
  634. {},
  635. access_token=self.admin_user_tok,
  636. )
  637. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
  638. self.assertEqual(channel.json_body["token"], token)
  639. self.assertIsNone(channel.json_body["uses_allowed"])
  640. self.assertIsNone(channel.json_body["expiry_time"])
  641. self.assertEqual(channel.json_body["pending"], 0)
  642. self.assertEqual(channel.json_body["completed"], 0)
  643. # LISTING
  644. def test_list_no_auth(self) -> None:
  645. """Try to list tokens without authentication."""
  646. channel = self.make_request("GET", self.url, {})
  647. self.assertEqual(
  648. HTTPStatus.UNAUTHORIZED,
  649. channel.code,
  650. msg=channel.json_body,
  651. )
  652. self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
  653. def test_list_requester_not_admin(self) -> None:
  654. """Try to list tokens while not an admin."""
  655. channel = self.make_request(
  656. "GET",
  657. self.url,
  658. {},
  659. access_token=self.other_user_tok,
  660. )
  661. self.assertEqual(
  662. HTTPStatus.FORBIDDEN,
  663. channel.code,
  664. msg=channel.json_body,
  665. )
  666. self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
  667. def test_list_all(self) -> None:
  668. """Test listing all tokens."""
  669. # Create new token using default values
  670. token = self._new_token()
  671. channel = self.make_request(
  672. "GET",
  673. self.url,
  674. {},
  675. access_token=self.admin_user_tok,
  676. )
  677. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
  678. self.assertEqual(len(channel.json_body["registration_tokens"]), 1)
  679. token_info = channel.json_body["registration_tokens"][0]
  680. self.assertEqual(token_info["token"], token)
  681. self.assertIsNone(token_info["uses_allowed"])
  682. self.assertIsNone(token_info["expiry_time"])
  683. self.assertEqual(token_info["pending"], 0)
  684. self.assertEqual(token_info["completed"], 0)
  685. def test_list_invalid_query_parameter(self) -> None:
  686. """Test with `valid` query parameter not `true` or `false`."""
  687. channel = self.make_request(
  688. "GET",
  689. self.url + "?valid=x",
  690. {},
  691. access_token=self.admin_user_tok,
  692. )
  693. self.assertEqual(
  694. HTTPStatus.BAD_REQUEST,
  695. channel.code,
  696. msg=channel.json_body,
  697. )
  698. def _test_list_query_parameter(self, valid: str) -> None:
  699. """Helper used to test both valid=true and valid=false."""
  700. # Create 2 valid and 2 invalid tokens.
  701. now = self.hs.get_clock().time_msec()
  702. # Create always valid token
  703. valid1 = self._new_token()
  704. # Create token that hasn't been used up
  705. valid2 = self._new_token(uses_allowed=1)
  706. # Create token that has expired
  707. invalid1 = self._new_token(expiry_time=now - 10000)
  708. # Create token that has been used up but hasn't expired
  709. invalid2 = self._new_token(
  710. uses_allowed=2,
  711. pending=1,
  712. completed=1,
  713. expiry_time=now + 1000000,
  714. )
  715. if valid == "true":
  716. tokens = [valid1, valid2]
  717. else:
  718. tokens = [invalid1, invalid2]
  719. channel = self.make_request(
  720. "GET",
  721. self.url + "?valid=" + valid,
  722. {},
  723. access_token=self.admin_user_tok,
  724. )
  725. self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
  726. self.assertEqual(len(channel.json_body["registration_tokens"]), 2)
  727. token_info_1 = channel.json_body["registration_tokens"][0]
  728. token_info_2 = channel.json_body["registration_tokens"][1]
  729. self.assertIn(token_info_1["token"], tokens)
  730. self.assertIn(token_info_2["token"], tokens)
  731. def test_list_valid(self) -> None:
  732. """Test listing just valid tokens."""
  733. self._test_list_query_parameter(valid="true")
  734. def test_list_invalid(self) -> None:
  735. """Test listing just invalid tokens."""
  736. self._test_list_query_parameter(valid="false")