diff options
Diffstat (limited to 'synapse')
| -rw-r--r-- | synapse/federation/federation_client.py | 64 | ||||
| -rw-r--r-- | synapse/federation/federation_server.py | 2 | ||||
| -rw-r--r-- | synapse/federation/transport/client.py | 39 | ||||
| -rw-r--r-- | synapse/handlers/federation.py | 2 | ||||
| -rw-r--r-- | synapse/http/federation/matrix_federation_agent.py | 114 | ||||
| -rw-r--r-- | synapse/util/async_helpers.py | 10 |
6 files changed, 206 insertions, 25 deletions
diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py index be3bb59431..9b4acd2ed7 100644 --- a/synapse/federation/federation_client.py +++ b/synapse/federation/federation_client.py @@ -743,18 +743,9 @@ class FederationClient(FederationBase): @defer.inlineCallbacks def send_invite(self, destination, room_id, event_id, pdu): - time_now = self._clock.time_msec() - try: - code, content = yield self.transport_layer.send_invite( - destination=destination, - room_id=room_id, - event_id=event_id, - content=pdu.get_pdu_json(time_now), - ) - except HttpResponseException as e: - if e.code == 403: - raise e.to_synapse_error() - raise + room_version = yield self.store.get_room_version(room_id) + + content = yield self._do_send_invite(destination, pdu, room_version) pdu_dict = content["event"] @@ -772,6 +763,55 @@ class FederationClient(FederationBase): defer.returnValue(pdu) + @defer.inlineCallbacks + def _do_send_invite(self, destination, pdu, room_version): + """Actually sends the invite, first trying v2 API and falling back to + v1 API if necessary. + + Args: + destination (str): Target server + pdu (FrozenEvent) + room_version (str) + + Returns: + dict: The event as a dict as returned by the remote server + """ + time_now = self._clock.time_msec() + + try: + content = yield self.transport_layer.send_invite_v2( + destination=destination, + room_id=pdu.room_id, + event_id=pdu.event_id, + content={ + "event": pdu.get_pdu_json(time_now), + "room_version": room_version, + "invite_room_state": pdu.unsigned.get("invite_room_state", []), + }, + ) + defer.returnValue(content) + except HttpResponseException as e: + if e.code in [400, 404]: + if room_version in (RoomVersions.V1, RoomVersions.V2): + pass # We'll fall through + else: + raise Exception("Remote server is too old") + elif e.code == 403: + raise e.to_synapse_error() + else: + raise + + # Didn't work, try v1 API. + # Note the v1 API returns a tuple of `(200, content)` + + _, content = yield self.transport_layer.send_invite_v1( + destination=destination, + room_id=pdu.room_id, + event_id=pdu.event_id, + content=pdu.get_pdu_json(time_now), + ) + defer.returnValue(content) + def send_leave(self, destinations, pdu): """Sends a leave event to one of a list of homeservers. diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index 4aa04b9588..6681614232 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -322,7 +322,7 @@ class FederationServer(FederationBase): if self.hs.is_mine_id(event.event_id): event.signatures.update( compute_event_signature( - event, + event.get_pdu_json(), self.hs.hostname, self.hs.config.signing_key[0] ) diff --git a/synapse/federation/transport/client.py b/synapse/federation/transport/client.py index 260178c47b..8e2be218e2 100644 --- a/synapse/federation/transport/client.py +++ b/synapse/federation/transport/client.py @@ -21,7 +21,7 @@ from six.moves import urllib from twisted.internet import defer from synapse.api.constants import Membership -from synapse.api.urls import FEDERATION_V1_PREFIX +from synapse.api.urls import FEDERATION_V1_PREFIX, FEDERATION_V2_PREFIX from synapse.util.logutils import log_function logger = logging.getLogger(__name__) @@ -289,7 +289,7 @@ class TransportLayerClient(object): @defer.inlineCallbacks @log_function - def send_invite(self, destination, room_id, event_id, content): + def send_invite_v1(self, destination, room_id, event_id, content): path = _create_v1_path("/invite/%s/%s", room_id, event_id) response = yield self.client.put_json( @@ -303,6 +303,20 @@ class TransportLayerClient(object): @defer.inlineCallbacks @log_function + def send_invite_v2(self, destination, room_id, event_id, content): + path = _create_v2_path("/invite/%s/%s", room_id, event_id) + + response = yield self.client.put_json( + destination=destination, + path=path, + data=content, + ignore_backoff=True, + ) + + defer.returnValue(response) + + @defer.inlineCallbacks + @log_function def get_public_rooms(self, remote_server, limit, since_token, search_filter=None, include_all_networks=False, third_party_instance_id=None): @@ -958,3 +972,24 @@ def _create_v1_path(path, *args): FEDERATION_V1_PREFIX + path % tuple(urllib.parse.quote(arg, "") for arg in args) ) + + +def _create_v2_path(path, *args): + """Creates a path against V2 federation API from the path template and + args. Ensures that all args are url encoded. + + Example: + + _create_v2_path("/event/%s/", event_id) + + Args: + path (str): String template for the path + args: ([str]): Args to insert into path. Each arg will be url encoded + + Returns: + str + """ + return ( + FEDERATION_V2_PREFIX + + path % tuple(urllib.parse.quote(arg, "") for arg in args) + ) diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index adf59db7a8..fcaf7530b0 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -1300,7 +1300,7 @@ class FederationHandler(BaseHandler): event.signatures.update( compute_event_signature( - event, + event.get_pdu_json(), self.hs.hostname, self.hs.config.signing_key[0] ) diff --git a/synapse/http/federation/matrix_federation_agent.py b/synapse/http/federation/matrix_federation_agent.py index 4a6f634c8b..07c72c9351 100644 --- a/synapse/http/federation/matrix_federation_agent.py +++ b/synapse/http/federation/matrix_federation_agent.py @@ -12,6 +12,8 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import cgi +import json import logging import attr @@ -20,7 +22,7 @@ from zope.interface import implementer from twisted.internet import defer from twisted.internet.endpoints import HostnameEndpoint, wrapClientTLS -from twisted.web.client import URI, Agent, HTTPConnectionPool +from twisted.web.client import URI, Agent, HTTPConnectionPool, readBody from twisted.web.http_headers import Headers from twisted.web.iweb import IAgent @@ -43,13 +45,19 @@ class MatrixFederationAgent(object): tls_client_options_factory (ClientTLSOptionsFactory|None): factory to use for fetching client tls options, or none to disable TLS. + _well_known_tls_policy (IPolicyForHTTPS|None): + TLS policy to use for fetching .well-known files. None to use a default + (browser-like) implementation. + srv_resolver (SrvResolver|None): SRVResolver impl to use for looking up SRV records. None to use a default implementation. """ def __init__( - self, reactor, tls_client_options_factory, _srv_resolver=None, + self, reactor, tls_client_options_factory, + _well_known_tls_policy=None, + _srv_resolver=None, ): self._reactor = reactor self._tls_client_options_factory = tls_client_options_factory @@ -62,6 +70,14 @@ class MatrixFederationAgent(object): self._pool.maxPersistentPerHost = 5 self._pool.cachedConnectionTimeout = 2 * 60 + agent_args = {} + if _well_known_tls_policy is not None: + # the param is called 'contextFactory', but actually passing a + # contextfactory is deprecated, and it expects an IPolicyForHTTPS. + agent_args['contextFactory'] = _well_known_tls_policy + _well_known_agent = Agent(self._reactor, pool=self._pool, **agent_args) + self._well_known_agent = _well_known_agent + @defer.inlineCallbacks def request(self, method, uri, headers=None, bodyProducer=None): """ @@ -114,7 +130,11 @@ class MatrixFederationAgent(object): class EndpointFactory(object): @staticmethod def endpointForURI(_uri): - logger.info("Connecting to %s:%s", res.target_host, res.target_port) + logger.info( + "Connecting to %s:%i", + res.target_host.decode("ascii"), + res.target_port, + ) ep = HostnameEndpoint(self._reactor, res.target_host, res.target_port) if tls_options is not None: ep = wrapClientTLS(tls_options, ep) @@ -127,7 +147,7 @@ class MatrixFederationAgent(object): defer.returnValue(res) @defer.inlineCallbacks - def _route_matrix_uri(self, parsed_uri): + def _route_matrix_uri(self, parsed_uri, lookup_well_known=True): """Helper for `request`: determine the routing for a Matrix URI Args: @@ -135,6 +155,9 @@ class MatrixFederationAgent(object): parsed with URI.fromBytes(uri, defaultPort=-1) to set the `port` to -1 if there is no explicit port given. + lookup_well_known (bool): True if we should look up the .well-known file if + there is no SRV record. + Returns: Deferred[_RoutingResult] """ @@ -169,6 +192,42 @@ class MatrixFederationAgent(object): service_name = b"_matrix._tcp.%s" % (parsed_uri.host,) server_list = yield self._srv_resolver.resolve_service(service_name) + if not server_list and lookup_well_known: + # try a .well-known lookup + well_known_server = yield self._get_well_known(parsed_uri.host) + + if well_known_server: + # if we found a .well-known, start again, but don't do another + # .well-known lookup. + + # parse the server name in the .well-known response into host/port. + # (This code is lifted from twisted.web.client.URI.fromBytes). + if b':' in well_known_server: + well_known_host, well_known_port = well_known_server.rsplit(b':', 1) + try: + well_known_port = int(well_known_port) + except ValueError: + # the part after the colon could not be parsed as an int + # - we assume it is an IPv6 literal with no port (the closing + # ']' stops it being parsed as an int) + well_known_host, well_known_port = well_known_server, -1 + else: + well_known_host, well_known_port = well_known_server, -1 + + new_uri = URI( + scheme=parsed_uri.scheme, + netloc=well_known_server, + host=well_known_host, + port=well_known_port, + path=parsed_uri.path, + params=parsed_uri.params, + query=parsed_uri.query, + fragment=parsed_uri.fragment, + ) + + res = yield self._route_matrix_uri(new_uri, lookup_well_known=False) + defer.returnValue(res) + if not server_list: target_host = parsed_uri.host port = 8448 @@ -190,6 +249,53 @@ class MatrixFederationAgent(object): target_port=port, )) + @defer.inlineCallbacks + def _get_well_known(self, server_name): + """Attempt to fetch and parse a .well-known file for the given server + + Args: + server_name (bytes): name of the server, from the requested url + + Returns: + Deferred[bytes|None]: either the new server name, from the .well-known, or + None if there was no .well-known file. + """ + # FIXME: add a cache + + uri = b"https://%s/.well-known/matrix/server" % (server_name, ) + logger.info("Fetching %s", uri.decode("ascii")) + try: + response = yield make_deferred_yieldable( + self._well_known_agent.request(b"GET", uri), + ) + except Exception as e: + logger.info( + "Connection error fetching %s: %s", + uri.decode("ascii"), e, + ) + defer.returnValue(None) + + body = yield make_deferred_yieldable(readBody(response)) + + if response.code != 200: + logger.info( + "Error response %i from %s: %s", + response.code, uri.decode("ascii"), body, + ) + defer.returnValue(None) + + content_types = response.headers.getRawHeaders(u'content-type') + if content_types is None: + raise Exception("no content-type header on .well-known response") + content_type, _opts = cgi.parse_header(content_types[-1]) + if content_type != 'application/json': + raise Exception("content-type not application/json on .well-known response") + parsed_body = json.loads(body.decode('utf-8')) + logger.info("Response from .well-known: %s", parsed_body) + if not isinstance(parsed_body, dict) or "m.server" not in parsed_body: + raise Exception("invalid .well-known response") + defer.returnValue(parsed_body["m.server"].encode("ascii")) + @attr.s class _RoutingResult(object): diff --git a/synapse/util/async_helpers.py b/synapse/util/async_helpers.py index 430bb15f51..f0e4a0e10c 100644 --- a/synapse/util/async_helpers.py +++ b/synapse/util/async_helpers.py @@ -201,7 +201,7 @@ class Linearizer(object): if entry[0] >= self.max_count: res = self._await_lock(key) else: - logger.info( + logger.debug( "Acquired uncontended linearizer lock %r for key %r", self.name, key, ) entry[0] += 1 @@ -215,7 +215,7 @@ class Linearizer(object): try: yield finally: - logger.info("Releasing linearizer lock %r for key %r", self.name, key) + logger.debug("Releasing linearizer lock %r for key %r", self.name, key) # We've finished executing so check if there are any things # blocked waiting to execute and start one of them @@ -247,7 +247,7 @@ class Linearizer(object): """ entry = self.key_to_defer[key] - logger.info( + logger.debug( "Waiting to acquire linearizer lock %r for key %r", self.name, key, ) @@ -255,7 +255,7 @@ class Linearizer(object): entry[1][new_defer] = 1 def cb(_r): - logger.info("Acquired linearizer lock %r for key %r", self.name, key) + logger.debug("Acquired linearizer lock %r for key %r", self.name, key) entry[0] += 1 # if the code holding the lock completes synchronously, then it @@ -273,7 +273,7 @@ class Linearizer(object): def eb(e): logger.info("defer %r got err %r", new_defer, e) if isinstance(e, CancelledError): - logger.info( + logger.debug( "Cancelling wait for linearizer lock %r for key %r", self.name, key, ) |