def verify_attestation(self, attestation, group_id, user_id, server_name=None):
        """Verifies that the given attestation matches the given parameters.

        An optional server_name can be supplied to explicitly set which server's
        signature is expected. Otherwise assumes that either the group_id or user_id
        is local and uses the other's server as the one to check.
        """

        if not server_name:
            if get_domain_from_id(group_id) == self.server_name:
                server_name = get_domain_from_id(user_id)
            elif get_domain_from_id(user_id) == self.server_name:
                server_name = get_domain_from_id(group_id)
            else:
                raise Exception("Expected either group_id or user_id to be local")

        if user_id != attestation["user_id"]:
            raise SynapseError(400, "Attestation has incorrect user_id")

        if group_id != attestation["group_id"]:
            raise SynapseError(400, "Attestation has incorrect group_id")
        valid_until_ms = attestation["valid_until_ms"]

        # TODO: We also want to check that *new* attestations that people give
        # us to store are valid for at least a little while.
        if valid_until_ms < self.clock.time_msec():
            raise SynapseError(400, "Attestation expired")

        yield self.keyring.verify_json_for_server(server_name, attestation)

    def get_users_in_group(self, group_id, requester_user_id):
        """Get users in a group
        """
        if self.is_mine_id(group_id):
            res = yield self.groups_server_handler.get_users_in_group(
                group_id, requester_user_id
            )
            defer.returnValue(res)

        group_server_name = get_domain_from_id(group_id)

        res = yield self.transport_client.get_users_in_group(
            get_domain_from_id(group_id), group_id, requester_user_id,
        )

        chunk = res["chunk"]
        valid_entries = []
        for entry in chunk:
            g_user_id = entry["user_id"]
            attestation = entry.pop("attestation", {})
            try:
                if get_domain_from_id(g_user_id) != group_server_name:
                    yield self.attestations.verify_attestation(
                        attestation,
                        group_id=group_id,
                        user_id=g_user_id,
                        server_name=get_domain_from_id(g_user_id),
                    )
                valid_entries.append(entry)
            except Exception as e:
                logger.info("Failed to verify user is in group: %s", e)

        res["chunk"] = valid_entries

        defer.returnValue(res)

        def _renew_attestation(group_id, user_id):
            try:
                if not self.is_mine_id(group_id):
                    destination = get_domain_from_id(group_id)
                elif not self.is_mine_id(user_id):
                    destination = get_domain_from_id(user_id)
                else:
                    logger.warn(
                        "Incorrectly trying to do attestations for user: %r in %r",
                        user_id, group_id,
                    )
                    yield self.store.remove_attestation_renewal(group_id, user_id)
                    return

                attestation = self.attestations.create_attestation(group_id, user_id)

                yield self.transport_client.renew_group_attestation(
                    destination, group_id, user_id,
                    content={"attestation": attestation},
                )

                yield self.store.update_attestation_renewal(
                    group_id, user_id, attestation
                )
            except Exception:
                logger.exception("Error renewing attestation of %r in %r",
                                 user_id, group_id)

def check_redaction(event, auth_events):
    """Check whether the event sender is allowed to redact the target event.

    Returns:
        True if the the sender is allowed to redact the target event if the
        target event was created by them.
        False if the sender is allowed to redact the target event with no
        further checks.

    Raises:
        AuthError if the event sender is definitely not allowed to redact
        the target event.
    """
    user_level = get_user_power_level(event.user_id, auth_events)

    redact_level = _get_named_level(auth_events, "redact", 50)

    if user_level >= redact_level:
        return False

    redacter_domain = get_domain_from_id(event.event_id)
    redactee_domain = get_domain_from_id(event.redacts)
    if redacter_domain == redactee_domain:
        return True

    raise AuthError(
        403,
        "You don't have permission to redact events"
    )

        def _update_profile_in_user_dir_txn(txn):
            new_entry = self._simple_upsert_txn(
                txn,
                table="user_directory",
                keyvalues={"user_id": user_id},
                insertion_values={"room_id": room_id},
                values={"display_name": display_name, "avatar_url": avatar_url},
                lock=False,  # We're only inserter
            )

            if isinstance(self.database_engine, PostgresEngine):
                # We weight the localpart most highly, then display name and finally
                # server name
                if new_entry:
                    sql = """
                        INSERT INTO user_directory_search(user_id, vector)
                        VALUES (?,
                            setweight(to_tsvector('english', ?), 'A')
                            || setweight(to_tsvector('english', ?), 'D')
                            || setweight(to_tsvector('english', COALESCE(?, '')), 'B')
                        )
                    """
                    txn.execute(
                        sql,
                        (
                            user_id, get_localpart_from_id(user_id),
                            get_domain_from_id(user_id), display_name,
                        )
                    )
                else:
                    sql = """
                        UPDATE user_directory_search
                        SET vector = setweight(to_tsvector('english', ?), 'A')
                            || setweight(to_tsvector('english', ?), 'D')
                            || setweight(to_tsvector('english', COALESCE(?, '')), 'B')
                        WHERE user_id = ?
                    """
                    txn.execute(
                        sql,
                        (
                            get_localpart_from_id(user_id), get_domain_from_id(user_id),
                            display_name, user_id,
                        )
                    )
            elif isinstance(self.database_engine, Sqlite3Engine):
                value = "%s %s" % (user_id, display_name,) if display_name else user_id
                self._simple_upsert_txn(
                    txn,
                    table="user_directory_search",
                    keyvalues={"user_id": user_id},
                    values={"value": value},
                    lock=False,  # We're only inserter
                )
            else:
                # This should be unreachable.
                raise Exception("Unrecognized database engine")

            txn.call_after(self.get_user_in_directory.invalidate, (user_id,))

    def is_host_joined(self, room_id, host):
        if '%' in host or '_' in host:
            raise Exception("Invalid host name")

        sql = """
            SELECT state_key FROM current_state_events AS c
            INNER JOIN room_memberships USING (event_id)
            WHERE membership = 'join'
                AND type = 'm.room.member'
                AND c.room_id = ?
                AND state_key LIKE ?
            LIMIT 1
        """

        # We do need to be careful to ensure that host doesn't have any wild cards
        # in it, but we checked above for known ones and we'll check below that
        # the returned user actually has the correct domain.
        like_clause = "%:" + host

        rows = yield self._execute("is_host_joined", None, sql, room_id, like_clause)

        if not rows:
            defer.returnValue(False)

        user_id = rows[0][0]
        if get_domain_from_id(user_id) != host:
            # This can only happen if the host name has something funky in it
            raise Exception("Invalid host name")

        defer.returnValue(True)

    def _create_association(self, room_alias, room_id, servers=None, creator=None):
        # general association creation for both human users and app services

        for wchar in string.whitespace:
                if wchar in room_alias.localpart:
                    raise SynapseError(400, "Invalid characters in room alias")

        if not self.hs.is_mine(room_alias):
            raise SynapseError(400, "Room alias must be local")
            # TODO(erikj): Change this.

        # TODO(erikj): Add transactions.
        # TODO(erikj): Check if there is a current association.
        if not servers:
            users = yield self.state.get_current_user_in_room(room_id)
            servers = set(get_domain_from_id(u) for u in users)

        if not servers:
            raise SynapseError(400, "Failed to get server list")

        yield self.store.create_room_alias_association(
            room_alias,
            room_id,
            servers,
            creator=creator,
        )

    def _push_remote(self, member, typing):
        try:
            users = yield self.state.get_current_user_in_room(member.room_id)
            self._member_last_federation_poke[member] = self.clock.time_msec()

            now = self.clock.time_msec()
            self.wheel_timer.insert(
                now=now,
                obj=member,
                then=now + FEDERATION_PING_INTERVAL,
            )

            for domain in set(get_domain_from_id(u) for u in users):
                if domain != self.server_name:
                    self.federation.send_edu(
                        destination=domain,
                        edu_type="m.typing",
                        content={
                            "room_id": member.room_id,
                            "user_id": member.user_id,
                            "typing": typing,
                        },
                        key=member,
                    )
        except Exception:
            logger.exception("Error pushing typing notif to remotes")

    def remove_user_from_group(self, group_id, user_id, requester_user_id, content):
        """Remove a user from the group; either a user is leaving or an admin
        kicked them.
        """

        yield self.check_group_is_ours(group_id, requester_user_id, and_exists=True)

        is_kick = False
        if requester_user_id != user_id:
            is_admin = yield self.store.is_user_admin_in_group(
                group_id, requester_user_id
            )
            if not is_admin:
                raise SynapseError(403, "User is not admin in group")

            is_kick = True

        yield self.store.remove_user_from_group(
            group_id, user_id,
        )

        if is_kick:
            if self.hs.is_mine_id(user_id):
                groups_local = self.hs.get_groups_local_handler()
                yield groups_local.user_removed_from_group(group_id, user_id, {})
            else:
                yield self.transport_client.remove_user_from_group_notification(
                    get_domain_from_id(user_id), group_id, user_id, {}
                )

        if not self.hs.is_mine_id(user_id):
            yield self.store.maybe_delete_remote_profile_cache(user_id)

        defer.returnValue({})

    def on_direct_to_device_edu(self, origin, content):
        local_messages = {}
        sender_user_id = content["sender"]
        if origin != get_domain_from_id(sender_user_id):
            logger.warn(
                "Dropping device message from %r with spoofed sender %r",
                origin, sender_user_id
            )
        message_type = content["type"]
        message_id = content["message_id"]
        for user_id, by_device in content["messages"].items():
            messages_by_device = {
                device_id: {
                    "content": message_content,
                    "type": message_type,
                    "sender": sender_user_id,
                }
                for device_id, message_content in by_device.items()
            }
            if messages_by_device:
                local_messages[user_id] = messages_by_device

        stream_id = yield self.store.add_messages_from_remote_to_device_inbox(
            origin, message_id, local_messages
        )

        self.notifier.on_new_event(
            "to_device_key", stream_id, users=local_messages.keys()
        )

    def _add_new_user(self, room_id, user_id):
        """Add new user to the room by creating an event and poking the federation API.
        """

        hostname = get_domain_from_id(user_id)

        room_version = self.get_success(self.store.get_room_version(room_id))

        builder = EventBuilder(
            state=self.state,
            auth=self.auth,
            store=self.store,
            clock=self.clock,
            hostname=hostname,
            signing_key=self.random_signing_key,
            format_version=room_version_to_event_format(room_version),
            room_id=room_id,
            type=EventTypes.Member,
            sender=user_id,
            state_key=user_id,
            content={"membership": Membership.JOIN},
        )

        prev_event_ids = self.get_success(
            self.store.get_latest_event_ids_in_room(room_id)
        )

        event = self.get_success(builder.build(prev_event_ids))

        self.get_success(self.federation_handler.on_receive_pdu(hostname, event))

        # Check that it was successfully persisted.
        self.get_success(self.store.get_event(event.event_id))
        self.get_success(self.store.get_event(event.event_id))

    def notify_device_update(self, user_id, device_ids):
        """Notify that a user's device(s) has changed. Pokes the notifier, and
        remote servers if the user is local.
        """
        users_who_share_room = yield self.store.get_users_who_share_room_with_user(
            user_id
        )

        hosts = set()
        if self.hs.is_mine_id(user_id):
            hosts.update(get_domain_from_id(u) for u in users_who_share_room)
            hosts.discard(self.server_name)

        position = yield self.store.add_device_change_to_streams(
            user_id, device_ids, list(hosts)
        )

        room_ids = yield self.store.get_rooms_for_user(user_id)

        yield self.notifier.on_new_event(
            "device_list_key", position, rooms=room_ids,
        )

        if hosts:
            logger.info("Sending device list update notif to: %r", hosts)
            for host in hosts:
                self.federation_sender.send_device_messages(host)

def get_interested_remotes(store, states, state_handler):
    """Given a list of presence states figure out which remote servers
    should be sent which.

    All the presence states should be for local users only.

    Args:
        store (DataStore)
        states (list(UserPresenceState))

    Returns:
        Deferred list of ([destinations], [UserPresenceState]), where for
        each row the list of UserPresenceState should be sent to each
        destination
    """
    hosts_and_states = []

    # First we look up the rooms each user is in (as well as any explicit
    # subscriptions), then for each distinct room we look up the remote
    # hosts in those rooms.
    room_ids_to_states, users_to_states = yield get_interested_parties(store, states)

    for room_id, states in iteritems(room_ids_to_states):
        hosts = yield state_handler.get_current_hosts_in_room(room_id)
        hosts_and_states.append((hosts, states))

    for user_id, states in iteritems(users_to_states):
        host = get_domain_from_id(user_id)
        hosts_and_states.append(([host], states))

    defer.returnValue(hosts_and_states)

    def _push_update(self, room_id, user_id, typing):
        users = yield self.state.get_current_user_in_room(room_id)
        domains = set(get_domain_from_id(u) for u in users)

        deferreds = []
        for domain in domains:
            if domain == self.server_name:
                preserve_fn(self._push_update_local)(
                    room_id=room_id,
                    user_id=user_id,
                    typing=typing
                )
            else:
                deferreds.append(preserve_fn(self.federation.send_edu)(
                    destination=domain,
                    edu_type="m.typing",
                    content={
                        "room_id": room_id,
                        "user_id": user_id,
                        "typing": typing,
                    },
                    key=(room_id, user_id),
                ))

        yield preserve_context_over_deferred(
            defer.DeferredList(deferreds, consumeErrors=True)
        )

    def f(self, group_id, *args, **kwargs):
        if self.is_mine_id(group_id):
            return getattr(self.groups_server_handler, func_name)(
                group_id, *args, **kwargs
            )
        else:
            destination = get_domain_from_id(group_id)
            d = getattr(self.transport_client, func_name)(
                destination, group_id, *args, **kwargs
            )

            # Capture errors returned by the remote homeserver and
            # re-throw specific errors as SynapseErrors. This is so
            # when the remote end responds with things like 403 Not
            # In Group, we can communicate that to the client instead
            # of a 500.
            def http_response_errback(failure):
                failure.trap(HttpResponseException)
                e = failure.value
                if e.code == 403:
                    raise e.to_synapse_error()
                return failure

            def request_failed_errback(failure):
                failure.trap(RequestSendFailed)
                raise SynapseError(502, "Failed to contact group server")

            d.addErrback(http_response_errback)
            d.addErrback(request_failed_errback)
            return d

    def incoming_device_list_update(self, origin, edu_content):
        """Called on incoming device list update from federation. Responsible
        for parsing the EDU and adding to pending updates list.
        """

        user_id = edu_content.pop("user_id")
        device_id = edu_content.pop("device_id")
        stream_id = str(edu_content.pop("stream_id"))  # They may come as ints
        prev_ids = edu_content.pop("prev_id", [])
        prev_ids = [str(p) for p in prev_ids]   # They may come as ints

        if get_domain_from_id(user_id) != origin:
            # TODO: Raise?
            logger.warning("Got device list update edu for %r from %r", user_id, origin)
            return

        room_ids = yield self.store.get_rooms_for_user(user_id)
        if not room_ids:
            # We don't share any rooms with this user. Ignore update, as we
            # probably won't get any further updates.
            return

        self._pending_updates.setdefault(user_id, []).append(
            (device_id, stream_id, prev_ids, edu_content)
        )

        yield self._handle_device_updates(user_id)

    def _recv_edu(self, origin, content):
        room_id = content["room_id"]
        user_id = content["user_id"]

        member = RoomMember(user_id=user_id, room_id=room_id)

        # Check that the string is a valid user id
        user = UserID.from_string(user_id)

        if user.domain != origin:
            logger.info(
                "Got typing update from %r with bad 'user_id': %r",
                origin, user_id,
            )
            return

        users = yield self.state.get_current_user_in_room(room_id)
        domains = set(get_domain_from_id(u) for u in users)

        if self.server_name in domains:
            logger.info("Got typing update from %s: %r", user_id, content)
            now = self.clock.time_msec()
            self._member_typing_until[member] = now + FEDERATION_TIMEOUT
            self.wheel_timer.insert(
                now=now,
                obj=member,
                then=now + FEDERATION_TIMEOUT,
            )
            self._push_update_local(
                member=member,
                typing=content["typing"]
            )

    def check_event_is_visible(event, state):
        history = state.get((EventTypes.RoomHistoryVisibility, ''), None)
        if history:
            visibility = history.content.get("history_visibility", "shared")
            if visibility in ["invited", "joined"]:
                # We now loop through all state events looking for
                # membership states for the requesting server to determine
                # if the server is either in the room or has been invited
                # into the room.
                for ev in itervalues(state):
                    if ev.type != EventTypes.Member:
                        continue
                    try:
                        domain = get_domain_from_id(ev.state_key)
                    except Exception:
                        continue

                    if domain != server_name:
                        continue

                    memtype = ev.membership
                    if memtype == Membership.JOIN:
                        return True
                    elif memtype == Membership.INVITE:
                        if visibility == "invited":
                            return True
                else:
                    # server has no users in the room: redact
                    return False

        return True

    def _push_remotes(self, receipts):
        """Given a list of receipts, works out which remote servers should be
        poked and pokes them.
        """
        # TODO: Some of this stuff should be coallesced.
        for receipt in receipts:
            room_id = receipt["room_id"]
            receipt_type = receipt["receipt_type"]
            user_id = receipt["user_id"]
            event_ids = receipt["event_ids"]
            data = receipt["data"]

            users = yield self.state.get_current_user_in_room(room_id)
            remotedomains = set(get_domain_from_id(u) for u in users)
            remotedomains = remotedomains.copy()
            remotedomains.discard(self.server_name)

            logger.debug("Sending receipt to: %r", remotedomains)

            for domain in remotedomains:
                self.federation.send_edu(
                    destination=domain,
                    edu_type="m.receipt",
                    content={
                        room_id: {
                            receipt_type: {
                                user_id: {
                                    "event_ids": event_ids,
                                    "data": data,
                                }
                            }
                        },
                    },
                    key=(room_id, receipt_type, user_id),
                )

    def remove_user_from_group(self, group_id, user_id, requester_user_id, content):
        """Remove a user from a group
        """
        if user_id == requester_user_id:
            token = yield self.store.register_user_group_membership(
                group_id, user_id,
                membership="leave",
            )
            self.notifier.on_new_event(
                "groups_key", token, users=[user_id],
            )

            # TODO: Should probably remember that we tried to leave so that we can
            # retry if the group server is currently down.

        if self.is_mine_id(group_id):
            res = yield self.groups_server_handler.remove_user_from_group(
                group_id, user_id, requester_user_id, content,
            )
        else:
            content["requester_user_id"] = requester_user_id
            res = yield self.transport_client.remove_user_from_group(
                get_domain_from_id(group_id), group_id, requester_user_id,
                user_id, content,
            )

        defer.returnValue(res)