synapse/synapse/federation/transport/client.py

# -*- coding: utf-8 -*-
# Copyright 2014-2016 OpenMarket Ltd
# Copyright 2018 New Vector Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# 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 logging

from six.moves import urllib

from twisted.internet import defer

from synapse.api.constants import Membership
from synapse.api.urls import FEDERATION_V1_PREFIX, FEDERATION_V2_PREFIX
from synapse.logging.utils import log_function

logger = logging.getLogger(__name__)


class TransportLayerClient(object):
    """Sends federation HTTP requests to other servers"""

    def __init__(self, hs):
        self.server_name = hs.hostname
        self.client = hs.get_http_client()

    @log_function
    def get_room_state(self, destination, room_id, event_id):
        """ Requests all state for a given room from the given server at the
        given event.

        Args:
            destination (str): The host name of the remote home server we want
                to get the state from.
            context (str): The name of the context we want the state of
            event_id (str): The event we want the context at.

        Returns:
            Deferred: Results in a dict received from the remote homeserver.
        """
        logger.debug("get_room_state dest=%s, room=%s", destination, room_id)

        path = _create_v1_path("/state/%s", room_id)
        return self.client.get_json(
            destination,
            path=path,
            args={"event_id": event_id},
            try_trailing_slash_on_400=True,
        )

    @log_function
    def get_room_state_ids(self, destination, room_id, event_id):
        """ Requests all state for a given room from the given server at the
        given event. Returns the state's event_id's

        Args:
            destination (str): The host name of the remote home server we want
                to get the state from.
            context (str): The name of the context we want the state of
            event_id (str): The event we want the context at.

        Returns:
            Deferred: Results in a dict received from the remote homeserver.
        """
        logger.debug("get_room_state_ids dest=%s, room=%s", destination, room_id)

        path = _create_v1_path("/state_ids/%s", room_id)
        return self.client.get_json(
            destination,
            path=path,
            args={"event_id": event_id},
            try_trailing_slash_on_400=True,
        )

    @log_function
    def get_event(self, destination, event_id, timeout=None):
        """ Requests the pdu with give id and origin from the given server.

        Args:
            destination (str): The host name of the remote home server we want
                to get the state from.
            event_id (str): The id of the event being requested.
            timeout (int): How long to try (in ms) the destination for before
                giving up. None indicates no timeout.

        Returns:
            Deferred: Results in a dict received from the remote homeserver.
        """
        logger.debug("get_pdu dest=%s, event_id=%s", destination, event_id)

        path = _create_v1_path("/event/%s", event_id)
        return self.client.get_json(
            destination, path=path, timeout=timeout, try_trailing_slash_on_400=True
        )

    @log_function
    def backfill(self, destination, room_id, event_tuples, limit):
        """ Requests `limit` previous PDUs in a given context before list of
        PDUs.

        Args:
            dest (str)
            room_id (str)
            event_tuples (list)
            limit (int)

        Returns:
            Deferred: Results in a dict received from the remote homeserver.
        """
        logger.debug(
            "backfill dest=%s, room_id=%s, event_tuples=%s, limit=%s",
            destination,
            room_id,
            repr(event_tuples),
            str(limit),
        )

        if not event_tuples:
            # TODO: raise?
            return

        path = _create_v1_path("/backfill/%s", room_id)

        args = {"v": event_tuples, "limit": [str(limit)]}

        return self.client.get_json(
            destination, path=path, args=args, try_trailing_slash_on_400=True
        )

    @defer.inlineCallbacks
    @log_function
    def send_transaction(self, transaction, json_data_callback=None):
        """ Sends the given Transaction to its destination

        Args:
            transaction (Transaction)

        Returns:
            Deferred: Succeeds when we get a 2xx HTTP response. The result
            will be the decoded JSON body.

            Fails with ``HTTPRequestException`` if we get an HTTP response
            code >= 300.

            Fails with ``NotRetryingDestination`` if we are not yet ready
            to retry this server.

            Fails with ``FederationDeniedError`` if this destination
            is not on our federation whitelist
        """
        logger.debug(
            "send_data dest=%s, txid=%s",
            transaction.destination,
            transaction.transaction_id,
        )

        if transaction.destination == self.server_name:
            raise RuntimeError("Transport layer cannot send to itself!")

        # FIXME: This is only used by the tests. The actual json sent is
        # generated by the json_data_callback.
        json_data = transaction.get_dict()

        path = _create_v1_path("/send/%s", transaction.transaction_id)

        response = yield self.client.put_json(
            transaction.destination,
            path=path,
            data=json_data,
            json_data_callback=json_data_callback,
            long_retries=True,
            backoff_on_404=True,  # If we get a 404 the other side has gone
            try_trailing_slash_on_400=True,
        )

        return response

    @defer.inlineCallbacks
    @log_function
    def make_query(
        self, destination, query_type, args, retry_on_dns_fail, ignore_backoff=False
    ):
        path = _create_v1_path("/query/%s", query_type)

        content = yield self.client.get_json(
            destination=destination,
            path=path,
            args=args,
            retry_on_dns_fail=retry_on_dns_fail,
            timeout=10000,
            ignore_backoff=ignore_backoff,
        )

        return content

    @defer.inlineCallbacks
    @log_function
    def make_membership_event(self, destination, room_id, user_id, membership, params):
        """Asks a remote server to build and sign us a membership event

        Note that this does not append any events to any graphs.

        Args:
            destination (str): address of remote homeserver
            room_id (str): room to join/leave
            user_id (str): user to be joined/left
            membership (str): one of join/leave
            params (dict[str, str|Iterable[str]]): Query parameters to include in the
                request.

        Returns:
            Deferred: Succeeds when we get a 2xx HTTP response. The result
            will be the decoded JSON body (ie, the new event).

            Fails with ``HTTPRequestException`` if we get an HTTP response
            code >= 300.

            Fails with ``NotRetryingDestination`` if we are not yet ready
            to retry this server.

            Fails with ``FederationDeniedError`` if the remote destination
            is not in our federation whitelist
        """
        valid_memberships = {Membership.JOIN, Membership.LEAVE}
        if membership not in valid_memberships:
            raise RuntimeError(
                "make_membership_event called with membership='%s', must be one of %s"
                % (membership, ",".join(valid_memberships))
            )
        path = _create_v1_path("/make_%s/%s/%s", membership, room_id, user_id)

        ignore_backoff = False
        retry_on_dns_fail = False

        if membership == Membership.LEAVE:
            # we particularly want to do our best to send leave events. The
            # problem is that if it fails, we won't retry it later, so if the
            # remote server was just having a momentary blip, the room will be
            # out of sync.
            ignore_backoff = True
            retry_on_dns_fail = True

        content = yield self.client.get_json(
            destination=destination,
            path=path,
            args=params,
            retry_on_dns_fail=retry_on_dns_fail,
            timeout=20000,
            ignore_backoff=ignore_backoff,
        )

        return content

    @defer.inlineCallbacks
    @log_function
    def send_join(self, destination, room_id, event_id, content):
        path = _create_v1_path("/send_join/%s/%s", room_id, event_id)

        response = yield self.client.put_json(
            destination=destination, path=path, data=content
        )

        return response

    @defer.inlineCallbacks
    @log_function
    def send_leave(self, destination, room_id, event_id, content):
        path = _create_v1_path("/send_leave/%s/%s", room_id, event_id)

        response = yield self.client.put_json(
            destination=destination,
            path=path,
            data=content,
            # we want to do our best to send this through. The problem is
            # that if it fails, we won't retry it later, so if the remote
            # server was just having a momentary blip, the room will be out of
            # sync.
            ignore_backoff=True,
        )

        return response

    @defer.inlineCallbacks
    @log_function
    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(
            destination=destination, path=path, data=content, ignore_backoff=True
        )

        return response

    @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
        )

        return 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,
    ):
        path = _create_v1_path("/publicRooms")

        args = {"include_all_networks": "true" if include_all_networks else "false"}
        if third_party_instance_id:
            args["third_party_instance_id"] = (third_party_instance_id,)
        if limit:
            args["limit"] = [str(limit)]
        if since_token:
            args["since"] = [since_token]

        # TODO(erikj): Actually send the search_filter across federation.

        response = yield self.client.get_json(
            destination=remote_server, path=path, args=args, ignore_backoff=True
        )

        return response

    @defer.inlineCallbacks
    @log_function
    def exchange_third_party_invite(self, destination, room_id, event_dict):
        path = _create_v1_path("/exchange_third_party_invite/%s", room_id)

        response = yield self.client.put_json(
            destination=destination, path=path, data=event_dict
        )

        return response

    @defer.inlineCallbacks
    @log_function
    def get_event_auth(self, destination, room_id, event_id):
        path = _create_v1_path("/event_auth/%s/%s", room_id, event_id)

        content = yield self.client.get_json(destination=destination, path=path)

        return content

    @defer.inlineCallbacks
    @log_function
    def send_query_auth(self, destination, room_id, event_id, content):
        path = _create_v1_path("/query_auth/%s/%s", room_id, event_id)

        content = yield self.client.post_json(
            destination=destination, path=path, data=content
        )

        return content

    @defer.inlineCallbacks
    @log_function
    def query_client_keys(self, destination, query_content, timeout):
        """Query the device keys for a list of user ids hosted on a remote
        server.

        Request:
            {
              "device_keys": {
                "<user_id>": ["<device_id>"]
            } }

        Response:
            {
              "device_keys": {
                "<user_id>": {
                  "<device_id>": {...}
            } } }

        Args:
            destination(str): The server to query.
            query_content(dict): The user ids to query.
        Returns:
            A dict containg the device keys.
        """
        path = _create_v1_path("/user/keys/query")

        content = yield self.client.post_json(
            destination=destination, path=path, data=query_content, timeout=timeout
        )
        return content

    @defer.inlineCallbacks
    @log_function
    def query_user_devices(self, destination, user_id, timeout):
        """Query the devices for a user id hosted on a remote server.

        Response:
            {
              "stream_id": "...",
              "devices": [ { ... } ]
            }

        Args:
            destination(str): The server to query.
            query_content(dict): The user ids to query.
        Returns:
            A dict containg the device keys.
        """
        path = _create_v1_path("/user/devices/%s", user_id)

        content = yield self.client.get_json(
            destination=destination, path=path, timeout=timeout
        )
        return content

    @defer.inlineCallbacks
    @log_function
    def claim_client_keys(self, destination, query_content, timeout):
        """Claim one-time keys for a list of devices hosted on a remote server.

        Request:
            {
              "one_time_keys": {
                "<user_id>": {
                    "<device_id>": "<algorithm>"
            } } }

        Response:
            {
              "device_keys": {
                "<user_id>": {
                  "<device_id>": {
                    "<algorithm>:<key_id>": "<key_base64>"
            } } } }

        Args:
            destination(str): The server to query.
            query_content(dict): The user ids to query.
        Returns:
            A dict containg the one-time keys.
        """

        path = _create_v1_path("/user/keys/claim")

        content = yield self.client.post_json(
            destination=destination, path=path, data=query_content, timeout=timeout
        )
        return content

    @defer.inlineCallbacks
    @log_function
    def get_missing_events(
        self,
        destination,
        room_id,
        earliest_events,
        latest_events,
        limit,
        min_depth,
        timeout,
    ):
        path = _create_v1_path("/get_missing_events/%s", room_id)

        content = yield self.client.post_json(
            destination=destination,
            path=path,
            data={
                "limit": int(limit),
                "min_depth": int(min_depth),
                "earliest_events": earliest_events,
                "latest_events": latest_events,
            },
            timeout=timeout,
        )

        return content

    @log_function
    def get_group_profile(self, destination, group_id, requester_user_id):
        """Get a group profile
        """
        path = _create_v1_path("/groups/%s/profile", group_id)

        return self.client.get_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            ignore_backoff=True,
        )

    @log_function
    def update_group_profile(self, destination, group_id, requester_user_id, content):
        """Update a remote group profile

        Args:
            destination (str)
            group_id (str)
            requester_user_id (str)
            content (dict): The new profile of the group
        """
        path = _create_v1_path("/groups/%s/profile", group_id)

        return self.client.post_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            data=content,
            ignore_backoff=True,
        )

    @log_function
    def get_group_summary(self, destination, group_id, requester_user_id):
        """Get a group summary
        """
        path = _create_v1_path("/groups/%s/summary", group_id)

        return self.client.get_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            ignore_backoff=True,
        )

    @log_function
    def get_rooms_in_group(self, destination, group_id, requester_user_id):
        """Get all rooms in a group
        """
        path = _create_v1_path("/groups/%s/rooms", group_id)

        return self.client.get_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            ignore_backoff=True,
        )

    def add_room_to_group(
        self, destination, group_id, requester_user_id, room_id, content
    ):
        """Add a room to a group
        """
        path = _create_v1_path("/groups/%s/room/%s", group_id, room_id)

        return self.client.post_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            data=content,
            ignore_backoff=True,
        )

    def update_room_in_group(
        self, destination, group_id, requester_user_id, room_id, config_key, content
    ):
        """Update room in group
        """
        path = _create_v1_path(
            "/groups/%s/room/%s/config/%s", group_id, room_id, config_key
        )

        return self.client.post_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            data=content,
            ignore_backoff=True,
        )

    def remove_room_from_group(self, destination, group_id, requester_user_id, room_id):
        """Remove a room from a group
        """
        path = _create_v1_path("/groups/%s/room/%s", group_id, room_id)

        return self.client.delete_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            ignore_backoff=True,
        )

    @log_function
    def get_users_in_group(self, destination, group_id, requester_user_id):
        """Get users in a group
        """
        path = _create_v1_path("/groups/%s/users", group_id)

        return self.client.get_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            ignore_backoff=True,
        )

    @log_function
    def get_invited_users_in_group(self, destination, group_id, requester_user_id):
        """Get users that have been invited to a group
        """
        path = _create_v1_path("/groups/%s/invited_users", group_id)

        return self.client.get_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            ignore_backoff=True,
        )

    @log_function
    def accept_group_invite(self, destination, group_id, user_id, content):
        """Accept a group invite
        """
        path = _create_v1_path("/groups/%s/users/%s/accept_invite", group_id, user_id)

        return self.client.post_json(
            destination=destination, path=path, data=content, ignore_backoff=True
        )

    @log_function
    def join_group(self, destination, group_id, user_id, content):
        """Attempts to join a group
        """
        path = _create_v1_path("/groups/%s/users/%s/join", group_id, user_id)

        return self.client.post_json(
            destination=destination, path=path, data=content, ignore_backoff=True
        )

    @log_function
    def invite_to_group(
        self, destination, group_id, user_id, requester_user_id, content
    ):
        """Invite a user to a group
        """
        path = _create_v1_path("/groups/%s/users/%s/invite", group_id, user_id)

        return self.client.post_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            data=content,
            ignore_backoff=True,
        )

    @log_function
    def invite_to_group_notification(self, destination, group_id, user_id, content):
        """Sent by group server to inform a user's server that they have been
        invited.
        """

        path = _create_v1_path("/groups/local/%s/users/%s/invite", group_id, user_id)

        return self.client.post_json(
            destination=destination, path=path, data=content, ignore_backoff=True
        )

    @log_function
    def remove_user_from_group(
        self, destination, group_id, requester_user_id, user_id, content
    ):
        """Remove a user fron a group
        """
        path = _create_v1_path("/groups/%s/users/%s/remove", group_id, user_id)

        return self.client.post_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            data=content,
            ignore_backoff=True,
        )

    @log_function
    def remove_user_from_group_notification(
        self, destination, group_id, user_id, content
    ):
        """Sent by group server to inform a user's server that they have been
        kicked from the group.
        """

        path = _create_v1_path("/groups/local/%s/users/%s/remove", group_id, user_id)

        return self.client.post_json(
            destination=destination, path=path, data=content, ignore_backoff=True
        )

    @log_function
    def renew_group_attestation(self, destination, group_id, user_id, content):
        """Sent by either a group server or a user's server to periodically update
        the attestations
        """

        path = _create_v1_path("/groups/%s/renew_attestation/%s", group_id, user_id)

        return self.client.post_json(
            destination=destination, path=path, data=content, ignore_backoff=True
        )

    @log_function
    def update_group_summary_room(
        self, destination, group_id, user_id, room_id, category_id, content
    ):
        """Update a room entry in a group summary
        """
        if category_id:
            path = _create_v1_path(
                "/groups/%s/summary/categories/%s/rooms/%s",
                group_id,
                category_id,
                room_id,
            )
        else:
            path = _create_v1_path("/groups/%s/summary/rooms/%s", group_id, room_id)

        return self.client.post_json(
            destination=destination,
            path=path,
            args={"requester_user_id": user_id},
            data=content,
            ignore_backoff=True,
        )

    @log_function
    def delete_group_summary_room(
        self, destination, group_id, user_id, room_id, category_id
    ):
        """Delete a room entry in a group summary
        """
        if category_id:
            path = _create_v1_path(
                "/groups/%s/summary/categories/%s/rooms/%s",
                group_id,
                category_id,
                room_id,
            )
        else:
            path = _create_v1_path("/groups/%s/summary/rooms/%s", group_id, room_id)

        return self.client.delete_json(
            destination=destination,
            path=path,
            args={"requester_user_id": user_id},
            ignore_backoff=True,
        )

    @log_function
    def get_group_categories(self, destination, group_id, requester_user_id):
        """Get all categories in a group
        """
        path = _create_v1_path("/groups/%s/categories", group_id)

        return self.client.get_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            ignore_backoff=True,
        )

    @log_function
    def get_group_category(self, destination, group_id, requester_user_id, category_id):
        """Get category info in a group
        """
        path = _create_v1_path("/groups/%s/categories/%s", group_id, category_id)

        return self.client.get_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            ignore_backoff=True,
        )

    @log_function
    def update_group_category(
        self, destination, group_id, requester_user_id, category_id, content
    ):
        """Update a category in a group
        """
        path = _create_v1_path("/groups/%s/categories/%s", group_id, category_id)

        return self.client.post_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            data=content,
            ignore_backoff=True,
        )

    @log_function
    def delete_group_category(
        self, destination, group_id, requester_user_id, category_id
    ):
        """Delete a category in a group
        """
        path = _create_v1_path("/groups/%s/categories/%s", group_id, category_id)

        return self.client.delete_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            ignore_backoff=True,
        )

    @log_function
    def get_group_roles(self, destination, group_id, requester_user_id):
        """Get all roles in a group
        """
        path = _create_v1_path("/groups/%s/roles", group_id)

        return self.client.get_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            ignore_backoff=True,
        )

    @log_function
    def get_group_role(self, destination, group_id, requester_user_id, role_id):
        """Get a roles info
        """
        path = _create_v1_path("/groups/%s/roles/%s", group_id, role_id)

        return self.client.get_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            ignore_backoff=True,
        )

    @log_function
    def update_group_role(
        self, destination, group_id, requester_user_id, role_id, content
    ):
        """Update a role in a group
        """
        path = _create_v1_path("/groups/%s/roles/%s", group_id, role_id)

        return self.client.post_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            data=content,
            ignore_backoff=True,
        )

    @log_function
    def delete_group_role(self, destination, group_id, requester_user_id, role_id):
        """Delete a role in a group
        """
        path = _create_v1_path("/groups/%s/roles/%s", group_id, role_id)

        return self.client.delete_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            ignore_backoff=True,
        )

    @log_function
    def update_group_summary_user(
        self, destination, group_id, requester_user_id, user_id, role_id, content
    ):
        """Update a users entry in a group
        """
        if role_id:
            path = _create_v1_path(
                "/groups/%s/summary/roles/%s/users/%s", group_id, role_id, user_id
            )
        else:
            path = _create_v1_path("/groups/%s/summary/users/%s", group_id, user_id)

        return self.client.post_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            data=content,
            ignore_backoff=True,
        )

    @log_function
    def set_group_join_policy(self, destination, group_id, requester_user_id, content):
        """Sets the join policy for a group
        """
        path = _create_v1_path("/groups/%s/settings/m.join_policy", group_id)

        return self.client.put_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            data=content,
            ignore_backoff=True,
        )

    @log_function
    def delete_group_summary_user(
        self, destination, group_id, requester_user_id, user_id, role_id
    ):
        """Delete a users entry in a group
        """
        if role_id:
            path = _create_v1_path(
                "/groups/%s/summary/roles/%s/users/%s", group_id, role_id, user_id
            )
        else:
            path = _create_v1_path("/groups/%s/summary/users/%s", group_id, user_id)

        return self.client.delete_json(
            destination=destination,
            path=path,
            args={"requester_user_id": requester_user_id},
            ignore_backoff=True,
        )

    def bulk_get_publicised_groups(self, destination, user_ids):
        """Get the groups a list of users are publicising
        """

        path = _create_v1_path("/get_groups_publicised")

        content = {"user_ids": user_ids}

        return self.client.post_json(
            destination=destination, path=path, data=content, ignore_backoff=True
        )


def _create_v1_path(path, *args):
    """Creates a path against V1 federation API from the path template and
    args. Ensures that all args are url encoded.

    Example:

        _create_v1_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_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
    )