<commit_before># -*- 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.

"""Contains handlers for federation events."""

import itertools
import logging

import six
from six import iteritems, itervalues
from six.moves import http_client, zip

from signedjson.key import decode_verify_key_bytes
from signedjson.sign import verify_signed_json
from unpaddedbase64 import decode_base64

from twisted.internet import defer

from synapse.api.constants import (
    KNOWN_ROOM_VERSIONS,
    EventTypes,
    Membership,
    RejectedReason,
)
from synapse.api.errors import (
    AuthError,
    CodeMessageException,
    FederationDeniedError,
    FederationError,
    StoreError,
    SynapseError,
)
from synapse.crypto.event_signing import (
    add_hashes_and_signatures,
    compute_event_signature,
)
from synapse.events.validator import EventValidator
from synapse.replication.http.federation import (
    ReplicationCleanRoomRestServlet,
    ReplicationFederationSendEventsRestServlet,
)
from synapse.replication.http.membership import ReplicationUserJoinedLeftRoomRestServlet
from synapse.state import resolve_events_with_factory
from synapse.types import UserID, get_domain_from_id
from synapse.util import logcontext, unwrapFirstError
from synapse.util.async_helpers import Linearizer
from synapse.util.distributor import user_joined_room
from synapse.util.frozenutils import unfreeze
from synapse.util.logutils import log_function
from synapse.util.retryutils import NotRetryingDestination
from synapse.visibility import filter_events_for_server

from ._base import BaseHandler

logger = logging.getLogger(__name__)


def shortstr(iterable, maxitems=5):
    """If iterable has maxitems or fewer, return the stringification of a list
    containing those items.

    Otherwise, return the stringification of a a list with the first maxitems items,
    followed by "...".

    Args:
        iterable (Iterable): iterable to truncate
        maxitems (int): number of items to return before truncating

    Returns:
        unicode
    """

    items = list(itertools.islice(iterable, maxitems + 1))
    if len(items) <= maxitems:
        return str(items)
    return u"[" + u", ".join(repr(r) for r in items[:maxitems]) + u", ...]"


class FederationHandler(BaseHandler):
    """Handles events that originated from federation.
        Responsible for:
        a) handling received Pdus before handing them on as Events to the rest
        of the home server (including auth and state conflict resoultion)
        b) converting events that were produced by local clients that may need
        to be sent to remote home servers.
        c) doing the necessary dances to invite remote users and join remote
        rooms.
    """

    def __init__(self, hs):
        super(FederationHandler, self).__init__(hs)

        self.hs = hs

        self.store = hs.get_datastore()  # type: synapse.storage.DataStore
        self.federation_client = hs.get_federation_client()
        self.state_handler = hs.get_state_handler()
        self.server_name = hs.hostname
        self.keyring = hs.get_keyring()
        self.action_generator = hs.get_action_generator()
        self.is_mine_id = hs.is_mine_id
        self.pusher_pool = hs.get_pusherpool()
        self.spam_checker = hs.get_spam_checker()
        self.event_creation_handler = hs.get_event_creation_handler()
        self._server_notices_mxid = hs.config.server_notices_mxid
        self.config = hs.config
        self.http_client = hs.get_simple_http_client()

        self._send_events_to_master = (
            ReplicationFederationSendEventsRestServlet.make_client(hs)
        )
        self._notify_user_membership_change = (
            ReplicationUserJoinedLeftRoomRestServlet.make_client(hs)
        )
        self._clean_room_for_join_client = (
            ReplicationCleanRoomRestServlet.make_client(hs)
        )

        # When joining a room we need to queue any events for that room up
        self.room_queues = {}
        self._room_pdu_linearizer = Linearizer("fed_room_pdu")

    @defer.inlineCallbacks
    def on_receive_pdu(
            self, origin, pdu, sent_to_us_directly=False,
    ):
        """ Process a PDU received via a federation /send/ transaction, or
        via backfill of missing prev_events

        Args:
            origin (str): server which initiated the /send/ transaction. Will
                be used to fetch missing events or state.
            pdu (FrozenEvent): received PDU
            sent_to_us_directly (bool): True if this event was pushed to us; False if
                we pulled it as the result of a missing prev_event.

        Returns (Deferred): completes with None
        """

        room_id = pdu.room_id
        event_id = pdu.event_id

        logger.info(
            "[%s %s] handling received PDU: %s",
            room_id, event_id, pdu,
        )

        # We reprocess pdus when we have seen them only as outliers
        existing = yield self.store.get_event(
            event_id,
            allow_none=True,
            allow_rejected=True,
        )

        # FIXME: Currently we fetch an event again when we already have it
        # if it has been marked as an outlier.

        already_seen = (
            existing and (
                not existing.internal_metadata.is_outlier()
                or pdu.internal_metadata.is_outlier()
            )
        )
        if already_seen:
            logger.debug("[%s %s]: Already seen pdu", room_id, event_id)
            return

        # do some initial sanity-checking of the event. In particular, make
        # sure it doesn't have hundreds of prev_events or auth_events, which
        # could cause a huge state resolution or cascade of event fetches.
        try:
            self._sanity_check_event(pdu)
        except SynapseError as err:
            logger.warn("[%s %s] Received event failed sanity checks", room_id, event_id)
            raise FederationError(
                "ERROR",
                err.code,
                err.msg,
                affected=pdu.event_id,
            )

        # If we are currently in the process of joining this room, then we
        # queue up events for later processing.
        if room_id in self.room_queues:
            logger.info(
                "[%s %s] Queuing PDU from %s for now: join in progress",
                room_id, event_id, origin,
            )
            self.room_queues[room_id].append((pdu, origin))
            return

        # If we're no longer in the room just ditch the event entirely. This
        # is probably an old server that has come back and thinks we're still
        # in the room (or we've been rejoined to the room by a state reset).
        #
        # If we were never in the room then maybe our database got vaped and
        # we should check if we *are* in fact in the room. If we are then we
        # can magically rejoin the room.
        is_in_room = yield self.auth.check_host_in_room(
            room_id,
            self.server_name
        )
        if not is_in_room:
            was_in_room = yield self.store.was_host_joined(
                pdu.room_id, self.server_name,
            )
            if was_in_room:
                logger.info(
                    "[%s %s] Ignoring PDU from %s as we've left the room",
                    room_id, event_id, origin,
                )
                defer.returnValue(None)

        state = None
        auth_chain = []

        # Get missing pdus if necessary.
        if not pdu.internal_metadata.is_outlier():
            # We only backfill backwards to the min depth.
            min_depth = yield self.get_min_depth_for_context(
                pdu.room_id
            )

            logger.debug(
                "[%s %s] min_depth: %d",
                room_id, event_id, min_depth,
            )

            prevs = {e_id for e_id, _ in pdu.prev_events}
            seen = yield self.store.have_seen_events(prevs)

            if min_depth and pdu.depth < min_depth:
                # This is so that we don't notify the user about this
                # message, to work around the fact that some events will
                # reference really really old events we really don't want to
                # send to the clients.
                pdu.internal_metadata.outlier = True
            elif min_depth and pdu.depth > min_depth:
                missing_prevs = prevs - seen
                if sent_to_us_directly and missing_prevs:
                    # If we're missing stuff, ensure we only fetch stuff one
                    # at a time.
                    logger.info(
                        "[%s %s] Acquiring room lock to fetch %d missing prev_events: %s",
                        room_id, event_id, len(missing_prevs), shortstr(missing_prevs),
                    )
                    with (yield self._room_pdu_linearizer.queue(pdu.room_id)):
                        logger.info(
                            "[%s %s] Acquired room lock to fetch %d missing prev_events",
                            room_id, event_id, len(missing_prevs),
                        )

                        yield self._get_missing_events_for_pdu(
                            origin, pdu, prevs, min_depth
                        )

                        # Update the set of things we've seen after trying to
                        # fetch the missing stuff
                        seen = yield self.store.have_seen_events(prevs)

                        if not prevs - seen:
                            logger.info(
                                "[%s %s] Found all missing prev_events",
                                room_id, event_id,
                            )
                elif missing_prevs:
                    logger.info(
                        "[%s %s] Not recursively fetching %d missing prev_events: %s",
                        room_id, event_id, len(missing_prevs), shortstr(missing_prevs),
                    )

            if prevs - seen:
                # We've still not been able to get all of the prev_events for this event.
                #
                # In this case, we need to fall back to asking another server in the
                # federation for the state at this event. That's ok provided we then
                # resolve the state against other bits of the DAG before using it (which
                # will ensure that you can't just take over a room by sending an event,
                # withholding its prev_events, and declaring yourself to be an admin in
                # the subsequent state request).
                #
                # Now, if we're pulling this event as a missing prev_event, then clearly
                # this event is not going to become the only forward-extremity and we are
                # guaranteed to resolve its state against our existing forward
                # extremities, so that should be fine.
                #
                # On the other hand, if this event was pushed to us, it is possible for
                # it to become the only forward-extremity in the room, and we would then
                # trust its state to be the state for the whole room. This is very bad.
                # Further, if the event was pushed to us, there is no excuse for us not to
                # have all the prev_events. We therefore reject any such events.
                #
                # XXX this really feels like it could/should be merged with the above,
                # but there is an interaction with min_depth that I'm not really
                # following.

                if sent_to_us_directly:
                    logger.warn(
                        "[%s %s] Failed to fetch %d prev events: rejecting",
                        room_id, event_id, len(prevs - seen),
                    )
                    raise FederationError(
                        "ERROR",
                        403,
                        (
                            "Your server isn't divulging details about prev_events "
                            "referenced in this event."
                        ),
                        affected=pdu.event_id,
                    )

                # Calculate the state after each of the previous events, and
                # resolve them to find the correct state at the current event.
                auth_chains = set()
                event_map = {
                    event_id: pdu,
                }
                try:
                    # Get the state of the events we know about
                    ours = yield self.store.get_state_groups_ids(room_id, seen)

                    # state_maps is a list of mappings from (type, state_key) to event_id
                    # type: list[dict[tuple[str, str], str]]
                    state_maps = list(ours.values())

                    # we don't need this any more, let's delete it.
                    del ours

                    # Ask the remote server for the states we don't
                    # know about
                    for p in prevs - seen:
                        logger.info(
                            "[%s %s] Requesting state at missing prev_event %s",
                            room_id, event_id, p,
                        )

                        with logcontext.nested_logging_context(p):
                            # note that if any of the missing prevs share missing state or
                            # auth events, the requests to fetch those events are deduped
                            # by the get_pdu_cache in federation_client.
                            remote_state, got_auth_chain = (
                                yield self.federation_client.get_state_for_room(
                                    origin, room_id, p,
                                )
                            )

                            # we want the state *after* p; get_state_for_room returns the
                            # state *before* p.
                            remote_event = yield self.federation_client.get_pdu(
                                [origin], p, outlier=True,
                            )

                            if remote_event is None:
                                raise Exception(
                                    "Unable to get missing prev_event %s" % (p, )
                                )

                            if remote_event.is_state():
                                remote_state.append(remote_event)

                            # XXX hrm I'm not convinced that duplicate events will compare
                            # for equality, so I'm not sure this does what the author
                            # hoped.
                            auth_chains.update(got_auth_chain)

                            remote_state_map = {
                                (x.type, x.state_key): x.event_id for x in remote_state
                            }
                            state_maps.append(remote_state_map)

                            for x in remote_state:
                                event_map[x.event_id] = x

                    # Resolve any conflicting state
                    @defer.inlineCallbacks
                    def fetch(ev_ids):
                        fetched = yield self.store.get_events(
                            ev_ids, get_prev_content=False, check_redacted=False,
                        )
                        # add any events we fetch here to the `event_map` so that we
                        # can use them to build the state event list below.
                        event_map.update(fetched)
                        defer.returnValue(fetched)

                    room_version = yield self.store.get_room_version(room_id)
                    state_map = yield resolve_events_with_factory(
                        room_version, state_maps, event_map, fetch,
                    )

                    # we need to give _process_received_pdu the actual state events
                    # rather than event ids, so generate that now.
                    state = [
                        event_map[e] for e in six.itervalues(state_map)
                    ]
                    auth_chain = list(auth_chains)
                except Exception:
                    logger.warn(
                        "[%s %s] Error attempting to resolve state at missing "
                        "prev_events",
                        room_id, event_id, exc_info=True,
                    )
                    raise FederationError(
                        "ERROR",
                        403,
                        "We can't get valid state history.",
                        affected=event_id,
                    )

        yield self._process_received_pdu(
            origin,
            pdu,
            state=state,
            auth_chain=auth_chain,<commit_msg>Fix up use of resolve_events_with_factory<commit_after># -*- 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.

"""Contains handlers for federation events."""

import itertools
import logging

import six
from six import iteritems, itervalues
from six.moves import http_client, zip

from signedjson.key import decode_verify_key_bytes
from signedjson.sign import verify_signed_json
from unpaddedbase64 import decode_base64

from twisted.internet import defer

from synapse.api.constants import (
    KNOWN_ROOM_VERSIONS,
    EventTypes,
    Membership,
    RejectedReason,
)
from synapse.api.errors import (
    AuthError,
    CodeMessageException,
    FederationDeniedError,
    FederationError,
    StoreError,
    SynapseError,
)
from synapse.crypto.event_signing import (
    add_hashes_and_signatures,
    compute_event_signature,
)
from synapse.events.validator import EventValidator
from synapse.replication.http.federation import (
    ReplicationCleanRoomRestServlet,
    ReplicationFederationSendEventsRestServlet,
)
from synapse.replication.http.membership import ReplicationUserJoinedLeftRoomRestServlet
from synapse.state import StateResolutionStore, resolve_events_with_factory
from synapse.types import UserID, get_domain_from_id
from synapse.util import logcontext, unwrapFirstError
from synapse.util.async_helpers import Linearizer
from synapse.util.distributor import user_joined_room
from synapse.util.frozenutils import unfreeze
from synapse.util.logutils import log_function
from synapse.util.retryutils import NotRetryingDestination
from synapse.visibility import filter_events_for_server

from ._base import BaseHandler

logger = logging.getLogger(__name__)


def shortstr(iterable, maxitems=5):
    """If iterable has maxitems or fewer, return the stringification of a list
    containing those items.

    Otherwise, return the stringification of a a list with the first maxitems items,
    followed by "...".

    Args:
        iterable (Iterable): iterable to truncate
        maxitems (int): number of items to return before truncating

    Returns:
        unicode
    """

    items = list(itertools.islice(iterable, maxitems + 1))
    if len(items) <= maxitems:
        return str(items)
    return u"[" + u", ".join(repr(r) for r in items[:maxitems]) + u", ...]"


class FederationHandler(BaseHandler):
    """Handles events that originated from federation.
        Responsible for:
        a) handling received Pdus before handing them on as Events to the rest
        of the home server (including auth and state conflict resoultion)
        b) converting events that were produced by local clients that may need
        to be sent to remote home servers.
        c) doing the necessary dances to invite remote users and join remote
        rooms.
    """

    def __init__(self, hs):
        super(FederationHandler, self).__init__(hs)

        self.hs = hs

        self.store = hs.get_datastore()  # type: synapse.storage.DataStore
        self.federation_client = hs.get_federation_client()
        self.state_handler = hs.get_state_handler()
        self.server_name = hs.hostname
        self.keyring = hs.get_keyring()
        self.action_generator = hs.get_action_generator()
        self.is_mine_id = hs.is_mine_id
        self.pusher_pool = hs.get_pusherpool()
        self.spam_checker = hs.get_spam_checker()
        self.event_creation_handler = hs.get_event_creation_handler()
        self._server_notices_mxid = hs.config.server_notices_mxid
        self.config = hs.config
        self.http_client = hs.get_simple_http_client()

        self._send_events_to_master = (
            ReplicationFederationSendEventsRestServlet.make_client(hs)
        )
        self._notify_user_membership_change = (
            ReplicationUserJoinedLeftRoomRestServlet.make_client(hs)
        )
        self._clean_room_for_join_client = (
            ReplicationCleanRoomRestServlet.make_client(hs)
        )

        # When joining a room we need to queue any events for that room up
        self.room_queues = {}
        self._room_pdu_linearizer = Linearizer("fed_room_pdu")

    @defer.inlineCallbacks
    def on_receive_pdu(
            self, origin, pdu, sent_to_us_directly=False,
    ):
        """ Process a PDU received via a federation /send/ transaction, or
        via backfill of missing prev_events

        Args:
            origin (str): server which initiated the /send/ transaction. Will
                be used to fetch missing events or state.
            pdu (FrozenEvent): received PDU
            sent_to_us_directly (bool): True if this event was pushed to us; False if
                we pulled it as the result of a missing prev_event.

        Returns (Deferred): completes with None
        """

        room_id = pdu.room_id
        event_id = pdu.event_id

        logger.info(
            "[%s %s] handling received PDU: %s",
            room_id, event_id, pdu,
        )

        # We reprocess pdus when we have seen them only as outliers
        existing = yield self.store.get_event(
            event_id,
            allow_none=True,
            allow_rejected=True,
        )

        # FIXME: Currently we fetch an event again when we already have it
        # if it has been marked as an outlier.

        already_seen = (
            existing and (
                not existing.internal_metadata.is_outlier()
                or pdu.internal_metadata.is_outlier()
            )
        )
        if already_seen:
            logger.debug("[%s %s]: Already seen pdu", room_id, event_id)
            return

        # do some initial sanity-checking of the event. In particular, make
        # sure it doesn't have hundreds of prev_events or auth_events, which
        # could cause a huge state resolution or cascade of event fetches.
        try:
            self._sanity_check_event(pdu)
        except SynapseError as err:
            logger.warn("[%s %s] Received event failed sanity checks", room_id, event_id)
            raise FederationError(
                "ERROR",
                err.code,
                err.msg,
                affected=pdu.event_id,
            )

        # If we are currently in the process of joining this room, then we
        # queue up events for later processing.
        if room_id in self.room_queues:
            logger.info(
                "[%s %s] Queuing PDU from %s for now: join in progress",
                room_id, event_id, origin,
            )
            self.room_queues[room_id].append((pdu, origin))
            return

        # If we're no longer in the room just ditch the event entirely. This
        # is probably an old server that has come back and thinks we're still
        # in the room (or we've been rejoined to the room by a state reset).
        #
        # If we were never in the room then maybe our database got vaped and
        # we should check if we *are* in fact in the room. If we are then we
        # can magically rejoin the room.
        is_in_room = yield self.auth.check_host_in_room(
            room_id,
            self.server_name
        )
        if not is_in_room:
            was_in_room = yield self.store.was_host_joined(
                pdu.room_id, self.server_name,
            )
            if was_in_room:
                logger.info(
                    "[%s %s] Ignoring PDU from %s as we've left the room",
                    room_id, event_id, origin,
                )
                defer.returnValue(None)

        state = None
        auth_chain = []

        # Get missing pdus if necessary.
        if not pdu.internal_metadata.is_outlier():
            # We only backfill backwards to the min depth.
            min_depth = yield self.get_min_depth_for_context(
                pdu.room_id
            )

            logger.debug(
                "[%s %s] min_depth: %d",
                room_id, event_id, min_depth,
            )

            prevs = {e_id for e_id, _ in pdu.prev_events}
            seen = yield self.store.have_seen_events(prevs)

            if min_depth and pdu.depth < min_depth:
                # This is so that we don't notify the user about this
                # message, to work around the fact that some events will
                # reference really really old events we really don't want to
                # send to the clients.
                pdu.internal_metadata.outlier = True
            elif min_depth and pdu.depth > min_depth:
                missing_prevs = prevs - seen
                if sent_to_us_directly and missing_prevs:
                    # If we're missing stuff, ensure we only fetch stuff one
                    # at a time.
                    logger.info(
                        "[%s %s] Acquiring room lock to fetch %d missing prev_events: %s",
                        room_id, event_id, len(missing_prevs), shortstr(missing_prevs),
                    )
                    with (yield self._room_pdu_linearizer.queue(pdu.room_id)):
                        logger.info(
                            "[%s %s] Acquired room lock to fetch %d missing prev_events",
                            room_id, event_id, len(missing_prevs),
                        )

                        yield self._get_missing_events_for_pdu(
                            origin, pdu, prevs, min_depth
                        )

                        # Update the set of things we've seen after trying to
                        # fetch the missing stuff
                        seen = yield self.store.have_seen_events(prevs)

                        if not prevs - seen:
                            logger.info(
                                "[%s %s] Found all missing prev_events",
                                room_id, event_id,
                            )
                elif missing_prevs:
                    logger.info(
                        "[%s %s] Not recursively fetching %d missing prev_events: %s",
                        room_id, event_id, len(missing_prevs), shortstr(missing_prevs),
                    )

            if prevs - seen:
                # We've still not been able to get all of the prev_events for this event.
                #
                # In this case, we need to fall back to asking another server in the
                # federation for the state at this event. That's ok provided we then
                # resolve the state against other bits of the DAG before using it (which
                # will ensure that you can't just take over a room by sending an event,
                # withholding its prev_events, and declaring yourself to be an admin in
                # the subsequent state request).
                #
                # Now, if we're pulling this event as a missing prev_event, then clearly
                # this event is not going to become the only forward-extremity and we are
                # guaranteed to resolve its state against our existing forward
                # extremities, so that should be fine.
                #
                # On the other hand, if this event was pushed to us, it is possible for
                # it to become the only forward-extremity in the room, and we would then
                # trust its state to be the state for the whole room. This is very bad.
                # Further, if the event was pushed to us, there is no excuse for us not to
                # have all the prev_events. We therefore reject any such events.
                #
                # XXX this really feels like it could/should be merged with the above,
                # but there is an interaction with min_depth that I'm not really
                # following.

                if sent_to_us_directly:
                    logger.warn(
                        "[%s %s] Failed to fetch %d prev events: rejecting",
                        room_id, event_id, len(prevs - seen),
                    )
                    raise FederationError(
                        "ERROR",
                        403,
                        (
                            "Your server isn't divulging details about prev_events "
                            "referenced in this event."
                        ),
                        affected=pdu.event_id,
                    )

                # Calculate the state after each of the previous events, and
                # resolve them to find the correct state at the current event.
                auth_chains = set()
                event_map = {
                    event_id: pdu,
                }
                try:
                    # Get the state of the events we know about
                    ours = yield self.store.get_state_groups_ids(room_id, seen)

                    # state_maps is a list of mappings from (type, state_key) to event_id
                    # type: list[dict[tuple[str, str], str]]
                    state_maps = list(ours.values())

                    # we don't need this any more, let's delete it.
                    del ours

                    # Ask the remote server for the states we don't
                    # know about
                    for p in prevs - seen:
                        logger.info(
                            "[%s %s] Requesting state at missing prev_event %s",
                            room_id, event_id, p,
                        )

                        with logcontext.nested_logging_context(p):
                            # note that if any of the missing prevs share missing state or
                            # auth events, the requests to fetch those events are deduped
                            # by the get_pdu_cache in federation_client.
                            remote_state, got_auth_chain = (
                                yield self.federation_client.get_state_for_room(
                                    origin, room_id, p,
                                )
                            )

                            # we want the state *after* p; get_state_for_room returns the
                            # state *before* p.
                            remote_event = yield self.federation_client.get_pdu(
                                [origin], p, outlier=True,
                            )

                            if remote_event is None:
                                raise Exception(
                                    "Unable to get missing prev_event %s" % (p, )
                                )

                            if remote_event.is_state():
                                remote_state.append(remote_event)

                            # XXX hrm I'm not convinced that duplicate events will compare
                            # for equality, so I'm not sure this does what the author
                            # hoped.
                            auth_chains.update(got_auth_chain)

                            remote_state_map = {
                                (x.type, x.state_key): x.event_id for x in remote_state
                            }
                            state_maps.append(remote_state_map)

                            for x in remote_state:
                                event_map[x.event_id] = x

                    room_version = yield self.store.get_room_version(room_id)
                    state_map = yield resolve_events_with_factory(
                        room_version, state_maps, event_map,
                        state_res_store=StateResolutionStore(self.store),
                    )

                    evs = yield self.store.get_events(
                        list(state_map.values()),
                        get_prev_content=False,
                        check_redacted=False,
                    )
                    event_map.update(evs)

                    # we need to give _process_received_pdu the actual state events
                    # rather than event ids, so generate that now.
                    state = [
                        event_map[e] for e in six.itervalues(state_map)
                    ]
                    auth_chain = list(auth_chains)
                except Exception:
                    logger.warn(
                        "[%s %s] Error attempting to resolve state at missing "
                        "prev_events",
                        room_id, event_id, exc_info=True,
                    )
                    raise FederationError(
                        "ERROR",
                        403,
                        "We can't get valid state history.",
                        affected=event_id,
                    )

        yield self._process_received_pdu(
            origin,
            pdu,
            state=state,
            auth_chain=auth_chain,
{'fraction_non_alphanumeric': 0.05880834217849049, 'fraction_numerical': 0.0016638457787085115, 'mean_word_length': 1.5111655381069011, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3078951', 'n_tokens_mistral': 9096, 'n_tokens_neox': 8513, 'n_words': 2944}