Matrix-Mirrors/anonymousland-synapse
1
0
Fork 0
mirror of https://git.anonymousland.org/anonymousland/synapse.git synced 2025-05-02 22:14:55 -04:00
Code Issues Projects Releases Packages Wiki Activity

Add type hints to matrix federation client / agent. (#8806)

Browse source
This commit is contained in:
Patrick Cloke 2020-11-25 07:07:21 -05:00 • committed by GitHub
parent b08dc7effe
commit f38676d161
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 231 additions and 195 deletions
Download patch file Download diff file Expand all files Collapse all files

304
synapse/http/matrixfederationclient.py
View file

@ -17,8 +17,9 @@ import cgi
import logging
import random
import sys
import urllib
import urllib.parse
from io import BytesIO
from typing import BinaryIO, Callable, Dict, List, Optional, Tuple, Union
import attr
import treq
@ -31,9 +32,10 @@ from twisted.internet import defer, protocol
from twisted.internet.error import DNSLookupError
from twisted.internet.interfaces import IReactorPluggableNameResolver, IReactorTime
from twisted.internet.task import _EPSILON, Cooperator
from twisted.python.failure import Failure
from twisted.web._newclient import ResponseDone
from twisted.web.http_headers import Headers
from twisted.web.iweb import IResponse
from twisted.web.iweb import IBodyProducer, IResponse
import synapse.metrics
import synapse.util.retryutils
@ -54,6 +56,7 @@ from synapse.logging.opentracing import (
start_active_span,
tags,
)
from synapse.types import JsonDict
from synapse.util import json_decoder
from synapse.util.async_helpers import timeout_deferred
from synapse.util.metrics import Measure
@ -76,47 +79,44 @@ MAXINT = sys.maxsize
_next_id = 1
QueryArgs = Dict[str, Union[str, List[str]]]
@attr.s(slots=True, frozen=True)
class MatrixFederationRequest:
method = attr.ib()
method = attr.ib(type=str)
"""HTTP method
:type: str
"""
path = attr.ib()
path = attr.ib(type=str)
"""HTTP path
:type: str
"""
destination = attr.ib()
destination = attr.ib(type=str)
"""The remote server to send the HTTP request to.
:type: str"""
"""
json = attr.ib(default=None)
json = attr.ib(default=None, type=Optional[JsonDict])
"""JSON to send in the body.
:type: dict|None
"""
json_callback = attr.ib(default=None)
json_callback = attr.ib(default=None, type=Optional[Callable[[], JsonDict]])
"""A callback to generate the JSON.
:type: func|None
"""
query = attr.ib(default=None)
query = attr.ib(default=None, type=Optional[dict])
"""Query arguments.
:type: dict|None
"""
txn_id = attr.ib(default=None)
txn_id = attr.ib(default=None, type=Optional[str])
"""Unique ID for this request (for logging)
:type: str|None
"""
uri = attr.ib(init=False, type=bytes)
"""The URI of this request
"""
def __attrs_post_init__(self):
def __attrs_post_init__(self) -> None:
global _next_id
txn_id = "%s-O-%s" % (self.method, _next_id)
_next_id = (_next_id + 1) % (MAXINT - 1)
@ -136,7 +136,7 @@ class MatrixFederationRequest:
)
object.__setattr__(self, "uri", uri)
def get_json(self):
def get_json(self) -> Optional[JsonDict]:
if self.json_callback:
return self.json_callback()
return self.json
@ -148,7 +148,7 @@ async def _handle_json_response(
request: MatrixFederationRequest,
response: IResponse,
start_ms: int,
):
) -> JsonDict:
"""
Reads the JSON body of a response, with a timeout
@ -160,7 +160,7 @@ async def _handle_json_response(
start_ms: Timestamp when request was made
Returns:
dict: parsed JSON response
The parsed JSON response
"""
try:
check_content_type_is_json(response.headers)
@ -266,27 +266,29 @@ class MatrixFederationHttpClient:
self._cooperator = Cooperator(scheduler=schedule)
async def _send_request_with_optional_trailing_slash(
self, request, try_trailing_slash_on_400=False, **send_request_args
):
self,
request: MatrixFederationRequest,
try_trailing_slash_on_400: bool = False,
**send_request_args
) -> IResponse:
"""Wrapper for _send_request which can optionally retry the request
upon receiving a combination of a 400 HTTP response code and a
'M_UNRECOGNIZED' errcode. This is a workaround for Synapse <= v0.99.3
due to #3622.
Args:
request (MatrixFederationRequest): details of request to be sent
try_trailing_slash_on_400 (bool): Whether on receiving a 400
request: details of request to be sent
try_trailing_slash_on_400: Whether on receiving a 400
'M_UNRECOGNIZED' from the server to retry the request with a
trailing slash appended to the request path.
send_request_args (Dict): A dictionary of arguments to pass to
`_send_request()`.
send_request_args: A dictionary of arguments to pass to `_send_request()`.
Raises:
HttpResponseException: If we get an HTTP response code >= 300
(except 429).
Returns:
Dict: Parsed JSON response body.
Parsed JSON response body.
"""
try:
response = await self._send_request(request, **send_request_args)
@ -313,24 +315,26 @@ class MatrixFederationHttpClient:
async def _send_request(
self,
request,
retry_on_dns_fail=True,
timeout=None,
long_retries=False,
ignore_backoff=False,
backoff_on_404=False,
):
request: MatrixFederationRequest,
retry_on_dns_fail: bool = True,
timeout: Optional[int] = None,
long_retries: bool = False,
ignore_backoff: bool = False,
backoff_on_404: bool = False,
) -> IResponse:
"""
Sends a request to the given server.
Args:
request (MatrixFederationRequest): details of request to be sent
request: details of request to be sent
timeout (int|None): number of milliseconds to wait for the response headers
retry_on_dns_fail: true if the request should be retied on DNS failures
timeout: number of milliseconds to wait for the response headers
(including connecting to the server), *for each attempt*.
60s by default.
long_retries (bool): whether to use the long retry algorithm.
long_retries: whether to use the long retry algorithm.
The regular retry algorithm makes 4 attempts, with intervals
[0.5s, 1s, 2s].
@ -346,14 +350,13 @@ class MatrixFederationHttpClient:
NB: the long retry algorithm takes over 20 minutes to complete, with
a default timeout of 60s!
ignore_backoff (bool): true to ignore the historical backoff data
ignore_backoff: true to ignore the historical backoff data
and try the request anyway.
backoff_on_404 (bool): Back off if we get a 404
backoff_on_404: Back off if we get a 404
Returns:
twisted.web.client.Response: resolves with the HTTP
response object on success.
Resolves with the HTTP response object on success.
Raises:
HttpResponseException: If we get an HTTP response code >= 300
@ -404,7 +407,7 @@ class MatrixFederationHttpClient:
)
# Inject the span into the headers
headers_dict = {}
headers_dict = {} # type: Dict[bytes, List[bytes]]
inject_active_span_byte_dict(headers_dict, request.destination)
headers_dict[b"User-Agent"] = [self.version_string_bytes]
@ -435,7 +438,7 @@ class MatrixFederationHttpClient:
data = encode_canonical_json(json)
producer = QuieterFileBodyProducer(
BytesIO(data), cooperator=self._cooperator
)
) # type: Optional[IBodyProducer]
else:
producer = None
auth_headers = self.build_auth_headers(
@ -524,14 +527,16 @@ class MatrixFederationHttpClient:
)
body = None
e = HttpResponseException(response.code, response_phrase, body)
exc = HttpResponseException(
response.code, response_phrase, body
)
# Retry if the error is a 429 (Too Many Requests),
# otherwise just raise a standard HttpResponseException
if response.code == 429:
raise RequestSendFailed(e, can_retry=True) from e
raise RequestSendFailed(exc, can_retry=True) from exc
else:
raise e
raise exc
break
except RequestSendFailed as e:
@ -582,22 +587,27 @@ class MatrixFederationHttpClient:
return response
def build_auth_headers(
self, destination, method, url_bytes, content=None, destination_is=None
):
self,
destination: Optional[bytes],
method: bytes,
url_bytes: bytes,
content: Optional[JsonDict] = None,
destination_is: Optional[bytes] = None,
) -> List[bytes]:
"""
Builds the Authorization headers for a federation request
Args:
destination (bytes|None): The destination homeserver of the request.
destination: The destination homeserver of the request.
May be None if the destination is an identity server, in which case
destination_is must be non-None.
method (bytes): The HTTP method of the request
url_bytes (bytes): The URI path of the request
content (object): The body of the request
destination_is (bytes): As 'destination', but if the destination is an
method: The HTTP method of the request
url_bytes: The URI path of the request
content: The body of the request
destination_is: As 'destination', but if the destination is an
identity server
Returns:
list[bytes]: a list of headers to be added as "Authorization:" headers
A list of headers to be added as "Authorization:" headers
"""
request = {
"method": method.decode("ascii"),
@ -629,33 +639,32 @@ class MatrixFederationHttpClient:
async def put_json(
self,
destination,
path,
args={},
data={},
json_data_callback=None,
long_retries=False,
timeout=None,
ignore_backoff=False,
backoff_on_404=False,
try_trailing_slash_on_400=False,
):
destination: str,
path: str,
args: Optional[QueryArgs] = None,
data: Optional[JsonDict] = None,
json_data_callback: Optional[Callable[[], JsonDict]] = None,
long_retries: bool = False,
timeout: Optional[int] = None,
ignore_backoff: bool = False,
backoff_on_404: bool = False,
try_trailing_slash_on_400: bool = False,
) -> Union[JsonDict, list]:
""" Sends the specified json data using PUT
Args:
destination (str): The remote server to send the HTTP request
to.
path (str): The HTTP path.
args (dict): query params
data (dict): A dict containing the data that will be used as
destination: The remote server to send the HTTP request to.
path: The HTTP path.
args: query params
data: A dict containing the data that will be used as
the request body. This will be encoded as JSON.
json_data_callback (callable): A callable returning the dict to
json_data_callback: A callable returning the dict to
use as the request body.
long_retries (bool): whether to use the long retry algorithm. See
long_retries: whether to use the long retry algorithm. See
docs on _send_request for details.
timeout (int|None): number of milliseconds to wait for the response.
timeout: number of milliseconds to wait for the response.
self._default_timeout (60s) by default.
Note that we may make several attempts to send the request; this
@ -663,19 +672,19 @@ class MatrixFederationHttpClient:
*each* attempt (including connection time) as well as the time spent
reading the response body after a 200 response.
ignore_backoff (bool): true to ignore the historical backoff data
ignore_backoff: true to ignore the historical backoff data
and try the request anyway.
backoff_on_404 (bool): True if we should count a 404 response as
backoff_on_404: True if we should count a 404 response as
a failure of the server (and should therefore back off future
requests).
try_trailing_slash_on_400 (bool): True if on a 400 M_UNRECOGNIZED
try_trailing_slash_on_400: True if on a 400 M_UNRECOGNIZED
response we should try appending a trailing slash to the end
of the request. Workaround for #3622 in Synapse <= v0.99.3. This
will be attempted before backing off if backing off has been
enabled.
Returns:
dict|list: Succeeds when we get a 2xx HTTP response. The
Succeeds when we get a 2xx HTTP response. The
result will be the decoded JSON body.
Raises:
@ -721,29 +730,28 @@ class MatrixFederationHttpClient:
async def post_json(
self,
destination,
path,
data={},
long_retries=False,
timeout=None,
ignore_backoff=False,
args={},
):
destination: str,
path: str,
data: Optional[JsonDict] = None,
long_retries: bool = False,
timeout: Optional[int] = None,
ignore_backoff: bool = False,
args: Optional[QueryArgs] = None,
) -> Union[JsonDict, list]:
""" Sends the specified json data using POST
Args:
destination (str): The remote server to send the HTTP request
to.
destination: The remote server to send the HTTP request to.
path (str): The HTTP path.
path: The HTTP path.
data (dict): A dict containing the data that will be used as
data: A dict containing the data that will be used as
the request body. This will be encoded as JSON.
long_retries (bool): whether to use the long retry algorithm. See
long_retries: whether to use the long retry algorithm. See
docs on _send_request for details.
timeout (int|None): number of milliseconds to wait for the response.
timeout: number of milliseconds to wait for the response.
self._default_timeout (60s) by default.
Note that we may make several attempts to send the request; this
@ -751,10 +759,10 @@ class MatrixFederationHttpClient:
*each* attempt (including connection time) as well as the time spent
reading the response body after a 200 response.
ignore_backoff (bool): true to ignore the historical backoff data and
ignore_backoff: true to ignore the historical backoff data and
try the request anyway.
args (dict): query params
args: query params
Returns:
dict|list: Succeeds when we get a 2xx HTTP response. The
result will be the decoded JSON body.
@ -795,26 +803,25 @@ class MatrixFederationHttpClient:
async def get_json(
self,
destination,
path,
args=None,
retry_on_dns_fail=True,
timeout=None,
ignore_backoff=False,
try_trailing_slash_on_400=False,
):
destination: str,
path: str,
args: Optional[QueryArgs] = None,
retry_on_dns_fail: bool = True,
timeout: Optional[int] = None,
ignore_backoff: bool = False,
try_trailing_slash_on_400: bool = False,
) -> Union[JsonDict, list]:
""" GETs some json from the given host homeserver and path
Args:
destination (str): The remote server to send the HTTP request
to.
destination: The remote server to send the HTTP request to.
path (str): The HTTP path.
path: The HTTP path.
args (dict|None): A dictionary used to create query strings, defaults to
args: A dictionary used to create query strings, defaults to
None.
timeout (int|None): number of milliseconds to wait for the response.
timeout: number of milliseconds to wait for the response.
self._default_timeout (60s) by default.
Note that we may make several attempts to send the request; this
@ -822,14 +829,14 @@ class MatrixFederationHttpClient:
*each* attempt (including connection time) as well as the time spent
reading the response body after a 200 response.
ignore_backoff (bool): true to ignore the historical backoff data
ignore_backoff: true to ignore the historical backoff data
and try the request anyway.
try_trailing_slash_on_400 (bool): True if on a 400 M_UNRECOGNIZED
try_trailing_slash_on_400: True if on a 400 M_UNRECOGNIZED
response we should try appending a trailing slash to the end of
the request. Workaround for #3622 in Synapse <= v0.99.3.
Returns:
dict|list: Succeeds when we get a 2xx HTTP response. The
Succeeds when we get a 2xx HTTP response. The
result will be the decoded JSON body.
Raises:
@ -870,24 +877,23 @@ class MatrixFederationHttpClient:
async def delete_json(
self,
destination,
path,
long_retries=False,
timeout=None,
ignore_backoff=False,
args={},
):
destination: str,
path: str,
long_retries: bool = False,
timeout: Optional[int] = None,
ignore_backoff: bool = False,
args: Optional[QueryArgs] = None,
) -> Union[JsonDict, list]:
"""Send a DELETE request to the remote expecting some json response
Args:
destination (str): The remote server to send the HTTP request
to.
path (str): The HTTP path.
destination: The remote server to send the HTTP request to.
path: The HTTP path.
long_retries (bool): whether to use the long retry algorithm. See
long_retries: whether to use the long retry algorithm. See
docs on _send_request for details.
timeout (int|None): number of milliseconds to wait for the response.
timeout: number of milliseconds to wait for the response.
self._default_timeout (60s) by default.
Note that we may make several attempts to send the request; this
@ -895,12 +901,12 @@ class MatrixFederationHttpClient:
*each* attempt (including connection time) as well as the time spent
reading the response body after a 200 response.
ignore_backoff (bool): true to ignore the historical backoff data and
ignore_backoff: true to ignore the historical backoff data and
try the request anyway.
args (dict): query params
args: query params
Returns:
dict|list: Succeeds when we get a 2xx HTTP response. The
Succeeds when we get a 2xx HTTP response. The
result will be the decoded JSON body.
Raises:
@ -938,25 +944,25 @@ class MatrixFederationHttpClient:
async def get_file(
self,
destination,
path,
destination: str,
path: str,
output_stream,
args={},
retry_on_dns_fail=True,
max_size=None,
ignore_backoff=False,
):
args: Optional[QueryArgs] = None,
retry_on_dns_fail: bool = True,
max_size: Optional[int] = None,
ignore_backoff: bool = False,
) -> Tuple[int, Dict[bytes, List[bytes]]]:
"""GETs a file from a given homeserver
Args:
destination (str): The remote server to send the HTTP request to.
path (str): The HTTP path to GET.
output_stream (file): File to write the response body to.
args (dict): Optional dictionary used to create the query string.
ignore_backoff (bool): true to ignore the historical backoff data
destination: The remote server to send the HTTP request to.
path: The HTTP path to GET.
output_stream: File to write the response body to.
args: Optional dictionary used to create the query string.
ignore_backoff: true to ignore the historical backoff data
and try the request anyway.
Returns:
tuple[int, dict]: Resolves with an (int,dict) tuple of
Resolves with an (int,dict) tuple of
the file length and a dict of the response headers.
Raises:
@ -1005,13 +1011,15 @@ class MatrixFederationHttpClient:
class _ReadBodyToFileProtocol(protocol.Protocol):
def __init__(self, stream, deferred, max_size):
def __init__(
self, stream: BinaryIO, deferred: defer.Deferred, max_size: Optional[int]
):
self.stream = stream
self.deferred = deferred
self.length = 0
self.max_size = max_size
def dataReceived(self, data):
def dataReceived(self, data: bytes) -> None:
self.stream.write(data)
self.length += len(data)
if self.max_size is not None and self.length >= self.max_size:
@ -1025,14 +1033,16 @@ class _ReadBodyToFileProtocol(protocol.Protocol):
self.deferred = defer.Deferred()
self.transport.loseConnection()
def connectionLost(self, reason):
def connectionLost(self, reason: Failure) -> None:
if reason.check(ResponseDone):
self.deferred.callback(self.length)
else:
self.deferred.errback(reason)
def _readBodyToFile(response, stream, max_size):
def _readBodyToFile(
response: IResponse, stream: BinaryIO, max_size: Optional[int]
) -> defer.Deferred:
d = defer.Deferred()
response.deliverBody(_ReadBodyToFileProtocol(stream, d, max_size))
return d
@ -1049,13 +1059,13 @@ def _flatten_response_never_received(e):
return repr(e)
def check_content_type_is_json(headers):
def check_content_type_is_json(headers: Headers) -> None:
"""
Check that a set of HTTP headers have a Content-Type header, and that it
is application/json.
Args:
headers (twisted.web.http_headers.Headers): headers to check
headers: headers to check
Raises:
RequestSendFailed: if the Content-Type header is missing or isn't JSON
@ -1080,7 +1090,7 @@ def check_content_type_is_json(headers):
)
def encode_query_args(args):
def encode_query_args(args: Optional[QueryArgs]) -> bytes:
if args is None:
return b""
@ -1088,8 +1098,8 @@ def encode_query_args(args):
for k, vs in args.items():
if isinstance(vs, str):
vs = [vs]
encoded_args[k] = [v.encode("UTF-8") for v in vs]
encoded_args[k] = [v.encode("utf8") for v in vs]
query_bytes = urllib.parse.urlencode(encoded_args, True)
query_str = urllib.parse.urlencode(encoded_args, True)
return query_bytes.encode("utf8")
return query_str.encode("utf8")