Send the opentracing span information also to appservices

changelog.d/16227.feature

@@ -0,0 +1 @@
+Add span information to requests sent to appservices. Contributed by MTRNord.

synapse/appservice/api.py

@@ -40,6 +40,7 @@
 from synapse.events import EventBase
 from synapse.events.utils import SerializeEventConfig, serialize_event
 from synapse.http.client import SimpleHttpClient, is_unknown_endpoint
+from synapse.logging import opentracing
 from synapse.types import DeviceListUpdates, JsonDict, ThirdPartyInstanceID
 from synapse.util.caches.response_cache import ResponseCache
 
@@ -136,10 +137,15 @@ async def query_user(self, service: "ApplicationService", user_id: str) -> bool:
             args = None
             if self.config.use_appservice_legacy_authorization:
                 args = {"access_token": service.hs_token}
+
+            headers: Dict[bytes, List[bytes]] = {
+                b"Authorization": [b"Bearer " + service.hs_token.encode("ascii")]
+            }
+            opentracing.inject_header_dict(headers, check_destination=False)
             response = await self.get_json(
                 f"{service.url}{APP_SERVICE_PREFIX}/users/{urllib.parse.quote(user_id)}",
                 args,
-                headers={"Authorization": [f"Bearer {service.hs_token}"]},
+                headers=headers,
             )
             if response is not None:  # just an empty json object
                 return True
@@ -162,10 +168,15 @@ async def query_alias(self, service: "ApplicationService", alias: str) -> bool:
             args = None
             if self.config.use_appservice_legacy_authorization:
                 args = {"access_token": service.hs_token}
+
+            headers: Dict[bytes, List[bytes]] = {
+                b"Authorization": [b"Bearer " + service.hs_token.encode("ascii")]
+            }
+            opentracing.inject_header_dict(headers, check_destination=False)
             response = await self.get_json(
                 f"{service.url}{APP_SERVICE_PREFIX}/rooms/{urllib.parse.quote(alias)}",
                 args,
-                headers={"Authorization": [f"Bearer {service.hs_token}"]},
+                headers=headers,
             )
             if response is not None:  # just an empty json object
                 return True
@@ -203,10 +214,15 @@ async def query_3pe(
                     **fields,
                     b"access_token": service.hs_token,
                 }
+
+            headers: Dict[bytes, List[bytes]] = {
+                b"Authorization": [b"Bearer " + service.hs_token.encode("ascii")]
+            }
+            opentracing.inject_header_dict(headers, check_destination=False)
             response = await self.get_json(
                 f"{service.url}{APP_SERVICE_PREFIX}/thirdparty/{kind}/{urllib.parse.quote(protocol)}",
                 args=args,
-                headers={"Authorization": [f"Bearer {service.hs_token}"]},
+                headers=headers,
             )
             if not isinstance(response, list):
                 logger.warning(
@@ -243,10 +259,15 @@ async def _get() -> Optional[JsonDict]:
                 args = None
                 if self.config.use_appservice_legacy_authorization:
                     args = {"access_token": service.hs_token}
+
+                headers: Dict[bytes, List[bytes]] = {
+                    b"Authorization": [b"Bearer " + service.hs_token.encode("ascii")]
+                }
+                opentracing.inject_header_dict(headers, check_destination=False)
                 info = await self.get_json(
                     f"{service.url}{APP_SERVICE_PREFIX}/thirdparty/protocol/{urllib.parse.quote(protocol)}",
                     args,
-                    headers={"Authorization": [f"Bearer {service.hs_token}"]},
+                    headers=headers,
                 )
 
                 if not _is_valid_3pe_metadata(info):
@@ -280,10 +301,14 @@ async def ping(self, service: "ApplicationService", txn_id: Optional[str]) -> No
         # This is required by the configuration.
         assert service.hs_token is not None
 
+        headers: Dict[bytes, List[bytes]] = {
+            b"Authorization": [b"Bearer " + service.hs_token.encode("ascii")]
+        }
+        opentracing.inject_header_dict(headers, check_destination=False)
         await self.post_json_get_json(
             uri=f"{service.url}{APP_SERVICE_PREFIX}/ping",
             post_json={"transaction_id": txn_id},
-            headers={"Authorization": [f"Bearer {service.hs_token}"]},
+            headers=headers,
         )
 
     async def push_bulk(
@@ -360,11 +385,15 @@ async def push_bulk(
             if self.config.use_appservice_legacy_authorization:
                 args = {"access_token": service.hs_token}
 
+            headers: Dict[bytes, List[bytes]] = {
+                b"Authorization": [b"Bearer " + service.hs_token.encode("ascii")]
+            }
+            opentracing.inject_header_dict(headers, check_destination=False)
             await self.put_json(
                 f"{service.url}{APP_SERVICE_PREFIX}/transactions/{urllib.parse.quote(str(txn_id))}",
                 json_body=body,
                 args=args,
-                headers={"Authorization": [f"Bearer {service.hs_token}"]},
+                headers=headers,
             )
             if logger.isEnabledFor(logging.DEBUG):
                 logger.debug(
@@ -432,12 +461,16 @@ async def claim_client_keys(
                 [algorithm] * count
             )
 
+        headers: Dict[bytes, List[bytes]] = {
+            b"Authorization": [b"Bearer " + service.hs_token.encode("ascii")]
+        }
+        opentracing.inject_header_dict(headers, check_destination=False)
         uri = f"{service.url}/_matrix/app/unstable/org.matrix.msc3983/keys/claim"
         try:
             response = await self.post_json_get_json(
                 uri,
                 body,
-                headers={"Authorization": [f"Bearer {service.hs_token}"]},
+                headers=headers,
             )
         except HttpResponseException as e:
             # The appservice doesn't support this endpoint.
@@ -492,13 +525,16 @@ async def query_keys(
 
         # This is required by the configuration.
         assert service.hs_token is not None
-
+        headers: Dict[bytes, List[bytes]] = {
+            b"Authorization": [b"Bearer " + service.hs_token.encode("ascii")]
+        }
+        opentracing.inject_header_dict(headers, check_destination=False)
         uri = f"{service.url}/_matrix/app/unstable/org.matrix.msc3984/keys/query"
         try:
             response = await self.post_json_get_json(
                 uri,
                 query,
-                headers={"Authorization": [f"Bearer {service.hs_token}"]},
+                headers=headers,
             )
         except HttpResponseException as e:
             # The appservice doesn't support this endpoint.

tests/appservice/test_api.py

@@ -76,15 +76,17 @@ async def get_json(
             headers: Mapping[Union[str, bytes], Sequence[Union[str, bytes]]],
         ) -> List[JsonDict]:
             # Ensure the access token is passed as a header.
-            if not headers or not headers.get("Authorization"):
+            if not headers or not headers.get(b"Authorization"):
                 raise RuntimeError("Access token not provided")
             # ... and not as a query param
             if b"access_token" in args:
                 raise RuntimeError(
                     "Access token should not be passed as a query param."
                 )
 
-            self.assertEqual(headers.get("Authorization"), [f"Bearer {TOKEN}"])
+            self.assertEqual(
+                headers.get(b"Authorization"), [f"Bearer {TOKEN}".encode()]
+            )
             self.request_url = url
             if url == URL_USER:
                 return SUCCESS_RESULT_USER
@@ -152,11 +154,13 @@ async def get_json(
             # Ensure the access token is passed as a both a query param and in the headers.
             if not args.get(b"access_token"):
                 raise RuntimeError("Access token should be provided in query params.")
-            if not headers or not headers.get("Authorization"):
+            if not headers or not headers.get(b"Authorization"):
                 raise RuntimeError("Access token should be provided in auth headers.")
 
             self.assertEqual(args.get(b"access_token"), TOKEN)
-            self.assertEqual(headers.get("Authorization"), [f"Bearer {TOKEN}"])
+            self.assertEqual(
+                headers.get(b"Authorization"), [f"Bearer {TOKEN}".encode()]
+            )
             self.request_url = url
             if url == URL_USER:
                 return SUCCESS_RESULT_USER
@@ -208,10 +212,12 @@ async def post_json_get_json(
             headers: Mapping[Union[str, bytes], Sequence[Union[str, bytes]]],
         ) -> JsonDict:
             # Ensure the access token is passed as both a header and query arg.
-            if not headers.get("Authorization"):
+            if not headers.get(b"Authorization"):
                 raise RuntimeError("Access token not provided")
 
-            self.assertEqual(headers.get("Authorization"), [f"Bearer {TOKEN}"])
+            self.assertEqual(
+                headers.get(b"Authorization"), [f"Bearer {TOKEN}".encode()]
+            )
             return RESPONSE
 
         # We assign to a method, which mypy doesn't like.