1
0

test_api.py 8.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248
  1. # Copyright 2022 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 Any, List, Mapping, Optional, Sequence, Union
  15. from unittest.mock import Mock
  16. from twisted.test.proto_helpers import MemoryReactor
  17. from synapse.appservice import ApplicationService
  18. from synapse.server import HomeServer
  19. from synapse.types import JsonDict
  20. from synapse.util import Clock
  21. from tests import unittest
  22. from tests.unittest import override_config
  23. PROTOCOL = "myproto"
  24. TOKEN = "myastoken"
  25. URL = "http://mytestservice"
  26. class ApplicationServiceApiTestCase(unittest.HomeserverTestCase):
  27. def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
  28. self.api = hs.get_application_service_api()
  29. self.service = ApplicationService(
  30. id="unique_identifier",
  31. sender="@as:test",
  32. url=URL,
  33. token="unused",
  34. hs_token=TOKEN,
  35. )
  36. def test_query_3pe_authenticates_token_via_header(self) -> None:
  37. """
  38. Tests that 3pe queries to the appservice are authenticated
  39. with the appservice's token.
  40. """
  41. SUCCESS_RESULT_USER = [
  42. {
  43. "protocol": PROTOCOL,
  44. "userid": "@a:user",
  45. "fields": {
  46. "more": "fields",
  47. },
  48. }
  49. ]
  50. SUCCESS_RESULT_LOCATION = [
  51. {
  52. "protocol": PROTOCOL,
  53. "alias": "#a:room",
  54. "fields": {
  55. "more": "fields",
  56. },
  57. }
  58. ]
  59. URL_USER = f"{URL}/_matrix/app/v1/thirdparty/user/{PROTOCOL}"
  60. URL_LOCATION = f"{URL}/_matrix/app/v1/thirdparty/location/{PROTOCOL}"
  61. self.request_url = None
  62. async def get_json(
  63. url: str,
  64. args: Mapping[Any, Any],
  65. headers: Mapping[Union[str, bytes], Sequence[Union[str, bytes]]],
  66. ) -> List[JsonDict]:
  67. # Ensure the access token is passed as a header.
  68. if not headers or not headers.get(b"Authorization"):
  69. raise RuntimeError("Access token not provided")
  70. # ... and not as a query param
  71. if b"access_token" in args:
  72. raise RuntimeError(
  73. "Access token should not be passed as a query param."
  74. )
  75. self.assertEqual(
  76. headers.get(b"Authorization"), [f"Bearer {TOKEN}".encode()]
  77. )
  78. self.request_url = url
  79. if url == URL_USER:
  80. return SUCCESS_RESULT_USER
  81. elif url == URL_LOCATION:
  82. return SUCCESS_RESULT_LOCATION
  83. else:
  84. raise RuntimeError(
  85. "URL provided was invalid. This should never be seen."
  86. )
  87. # We assign to a method, which mypy doesn't like.
  88. self.api.get_json = Mock(side_effect=get_json) # type: ignore[method-assign]
  89. result = self.get_success(
  90. self.api.query_3pe(self.service, "user", PROTOCOL, {b"some": [b"field"]})
  91. )
  92. self.assertEqual(self.request_url, URL_USER)
  93. self.assertEqual(result, SUCCESS_RESULT_USER)
  94. result = self.get_success(
  95. self.api.query_3pe(
  96. self.service, "location", PROTOCOL, {b"some": [b"field"]}
  97. )
  98. )
  99. self.assertEqual(self.request_url, URL_LOCATION)
  100. self.assertEqual(result, SUCCESS_RESULT_LOCATION)
  101. @override_config({"use_appservice_legacy_authorization": True})
  102. def test_query_3pe_authenticates_token_via_param(self) -> None:
  103. """
  104. Tests that 3pe queries to the appservice are authenticated
  105. with the appservice's token.
  106. """
  107. SUCCESS_RESULT_USER = [
  108. {
  109. "protocol": PROTOCOL,
  110. "userid": "@a:user",
  111. "fields": {
  112. "more": "fields",
  113. },
  114. }
  115. ]
  116. SUCCESS_RESULT_LOCATION = [
  117. {
  118. "protocol": PROTOCOL,
  119. "alias": "#a:room",
  120. "fields": {
  121. "more": "fields",
  122. },
  123. }
  124. ]
  125. URL_USER = f"{URL}/_matrix/app/v1/thirdparty/user/{PROTOCOL}"
  126. URL_LOCATION = f"{URL}/_matrix/app/v1/thirdparty/location/{PROTOCOL}"
  127. self.request_url = None
  128. async def get_json(
  129. url: str,
  130. args: Mapping[Any, Any],
  131. headers: Optional[
  132. Mapping[Union[str, bytes], Sequence[Union[str, bytes]]]
  133. ] = None,
  134. ) -> List[JsonDict]:
  135. # Ensure the access token is passed as a both a query param and in the headers.
  136. if not args.get(b"access_token"):
  137. raise RuntimeError("Access token should be provided in query params.")
  138. if not headers or not headers.get(b"Authorization"):
  139. raise RuntimeError("Access token should be provided in auth headers.")
  140. self.assertEqual(args.get(b"access_token"), TOKEN)
  141. self.assertEqual(
  142. headers.get(b"Authorization"), [f"Bearer {TOKEN}".encode()]
  143. )
  144. self.request_url = url
  145. if url == URL_USER:
  146. return SUCCESS_RESULT_USER
  147. elif url == URL_LOCATION:
  148. return SUCCESS_RESULT_LOCATION
  149. else:
  150. raise RuntimeError(
  151. "URL provided was invalid. This should never be seen."
  152. )
  153. # We assign to a method, which mypy doesn't like.
  154. self.api.get_json = Mock(side_effect=get_json) # type: ignore[method-assign]
  155. result = self.get_success(
  156. self.api.query_3pe(self.service, "user", PROTOCOL, {b"some": [b"field"]})
  157. )
  158. self.assertEqual(self.request_url, URL_USER)
  159. self.assertEqual(result, SUCCESS_RESULT_USER)
  160. result = self.get_success(
  161. self.api.query_3pe(
  162. self.service, "location", PROTOCOL, {b"some": [b"field"]}
  163. )
  164. )
  165. self.assertEqual(self.request_url, URL_LOCATION)
  166. self.assertEqual(result, SUCCESS_RESULT_LOCATION)
  167. def test_claim_keys(self) -> None:
  168. """
  169. Tests that the /keys/claim response is properly parsed for missing
  170. keys.
  171. """
  172. RESPONSE: JsonDict = {
  173. "@alice:example.org": {
  174. "DEVICE_1": {
  175. "signed_curve25519:AAAAHg": {
  176. # We don't really care about the content of the keys,
  177. # they get passed back transparently.
  178. },
  179. "signed_curve25519:BBBBHg": {},
  180. },
  181. "DEVICE_2": {"signed_curve25519:CCCCHg": {}},
  182. },
  183. }
  184. async def post_json_get_json(
  185. uri: str,
  186. post_json: Any,
  187. headers: Mapping[Union[str, bytes], Sequence[Union[str, bytes]]],
  188. ) -> JsonDict:
  189. # Ensure the access token is passed as both a header and query arg.
  190. if not headers.get(b"Authorization"):
  191. raise RuntimeError("Access token not provided")
  192. self.assertEqual(
  193. headers.get(b"Authorization"), [f"Bearer {TOKEN}".encode()]
  194. )
  195. return RESPONSE
  196. # We assign to a method, which mypy doesn't like.
  197. self.api.post_json_get_json = Mock(side_effect=post_json_get_json) # type: ignore[method-assign]
  198. MISSING_KEYS = [
  199. # Known user, known device, missing algorithm.
  200. ("@alice:example.org", "DEVICE_2", "xyz", 1),
  201. # Known user, missing device.
  202. ("@alice:example.org", "DEVICE_3", "signed_curve25519", 1),
  203. # Unknown user.
  204. ("@bob:example.org", "DEVICE_4", "signed_curve25519", 1),
  205. ]
  206. claimed_keys, missing = self.get_success(
  207. self.api.claim_client_keys(
  208. self.service,
  209. [
  210. # Found devices
  211. ("@alice:example.org", "DEVICE_1", "signed_curve25519", 1),
  212. ("@alice:example.org", "DEVICE_2", "signed_curve25519", 1),
  213. ]
  214. + MISSING_KEYS,
  215. )
  216. )
  217. self.assertEqual(claimed_keys, RESPONSE)
  218. self.assertEqual(missing, MISSING_KEYS)