diff options
Diffstat (limited to 'synapse/handlers')
| -rw-r--r-- | synapse/handlers/_base.py | 64 | ||||
| -rw-r--r-- | synapse/handlers/directory.py | 10 | ||||
| -rw-r--r-- | synapse/handlers/federation.py | 547 | ||||
| -rw-r--r-- | synapse/handlers/message.py | 89 | ||||
| -rw-r--r-- | synapse/handlers/profile.py | 22 | ||||
| -rw-r--r-- | synapse/handlers/room.py | 144 |
6 files changed, 579 insertions, 297 deletions
diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index de4d23bbb3..30c6733063 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -14,7 +14,18 @@ # limitations under the License. from twisted.internet import defer + from synapse.api.errors import LimitExceededError +from synapse.util.async import run_on_reactor +from synapse.crypto.event_signing import add_hashes_and_signatures +from synapse.api.events.room import RoomMemberEvent +from synapse.api.constants import Membership + +import logging + + +logger = logging.getLogger(__name__) + class BaseHandler(object): @@ -30,6 +41,9 @@ class BaseHandler(object): self.clock = hs.get_clock() self.hs = hs + self.signing_key = hs.config.signing_key[0] + self.server_name = hs.hostname + def ratelimit(self, user_id): time_now = self.clock.time() allowed, time_allowed = self.ratelimiter.send_message( @@ -44,16 +58,58 @@ class BaseHandler(object): @defer.inlineCallbacks def _on_new_room_event(self, event, snapshot, extra_destinations=[], - extra_users=[]): + extra_users=[], suppress_auth=False, + do_invite_host=None): + yield run_on_reactor() + snapshot.fill_out_prev_events(event) + yield self.state_handler.annotate_event_with_state(event) + + yield self.auth.add_auth_events(event) + + logger.debug("Signing event...") + + add_hashes_and_signatures( + event, self.server_name, self.signing_key + ) + + logger.debug("Signed event.") + + if not suppress_auth: + logger.debug("Authing...") + self.auth.check(event, raises=True) + logger.debug("Authed") + else: + logger.debug("Suppressed auth.") + + if do_invite_host: + federation_handler = self.hs.get_handlers().federation_handler + invite_event = yield federation_handler.send_invite( + do_invite_host, + event + ) + + # FIXME: We need to check if the remote changed anything else + event.signatures = invite_event.signatures + yield self.store.persist_event(event) destinations = set(extra_destinations) # Send a PDU to all hosts who have joined the room. - destinations.update((yield self.store.get_joined_hosts_for_room( - event.room_id - ))) + + for k, s in event.state_events.items(): + try: + if k[0] == RoomMemberEvent.TYPE: + if s.content["membership"] == Membership.JOIN: + destinations.add( + self.hs.parse_userid(s.state_key).domain + ) + except: + logger.warn( + "Failed to get destination from event %s", s.event_id + ) + event.destinations = list(destinations) self.notifier.on_new_room_event(event, extra_users=extra_users) diff --git a/synapse/handlers/directory.py b/synapse/handlers/directory.py index a56830d520..164363cdc5 100644 --- a/synapse/handlers/directory.py +++ b/synapse/handlers/directory.py @@ -147,10 +147,8 @@ class DirectoryHandler(BaseHandler): content={"aliases": aliases}, ) - snapshot = yield self.store.snapshot_room( - room_id=room_id, - user_id=user_id, - ) + snapshot = yield self.store.snapshot_room(event) - yield self.state_handler.handle_new_event(event, snapshot) - yield self._on_new_room_event(event, snapshot, extra_users=[user_id]) + yield self._on_new_room_event( + event, snapshot, extra_users=[user_id], suppress_auth=True + ) diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index f52591d2a3..492005a170 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -17,13 +17,18 @@ from ._base import BaseHandler -from synapse.api.events.room import InviteJoinEvent, RoomMemberEvent +from synapse.api.events.utils import prune_event +from synapse.api.errors import AuthError, FederationError, SynapseError +from synapse.api.events.room import RoomMemberEvent from synapse.api.constants import Membership from synapse.util.logutils import log_function -from synapse.federation.pdu_codec import PduCodec -from synapse.api.errors import SynapseError +from synapse.util.async import run_on_reactor +from synapse.crypto.event_signing import ( + compute_event_signature, check_event_content_hash +) +from syutil.jsonutil import encode_canonical_json -from twisted.internet import defer, reactor +from twisted.internet import defer import logging @@ -38,6 +43,8 @@ class FederationHandler(BaseHandler): 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): @@ -55,12 +62,14 @@ class FederationHandler(BaseHandler): self.state_handler = hs.get_state_handler() # self.auth_handler = gs.get_auth_handler() self.server_name = hs.hostname + self.keyring = hs.get_keyring() self.lock_manager = hs.get_room_lock_manager() self.replication_layer.set_handler(self) - self.pdu_codec = PduCodec(hs) + # When joining a room we need to queue any events for that room up + self.room_queues = {} @log_function @defer.inlineCallbacks @@ -78,7 +87,9 @@ class FederationHandler(BaseHandler): processing. """ - pdu = self.pdu_codec.pdu_from_event(event) + yield run_on_reactor() + + pdu = event if not hasattr(pdu, "destinations") or not pdu.destinations: pdu.destinations = [] @@ -87,98 +98,114 @@ class FederationHandler(BaseHandler): @log_function @defer.inlineCallbacks - def on_receive_pdu(self, pdu, backfilled): + def on_receive_pdu(self, pdu, backfilled, state=None): """ Called by the ReplicationLayer when we have a new pdu. We need to - do auth checks and put it throught the StateHandler. + do auth checks and put it through the StateHandler. """ - event = self.pdu_codec.event_from_pdu(pdu) + event = pdu logger.debug("Got event: %s", event.event_id) - with (yield self.lock_manager.lock(pdu.context)): - if event.is_state and not backfilled: - is_new_state = yield self.state_handler.handle_new_state( - pdu - ) - else: - is_new_state = False - # TODO: Implement something in federation that allows us to - # respond to PDU. + # If we are currently in the process of joining this room, then we + # queue up events for later processing. + if event.room_id in self.room_queues: + self.room_queues[event.room_id].append(pdu) + return + + logger.debug("Processing event: %s", event.event_id) - target_is_mine = False - if hasattr(event, "target_host"): - target_is_mine = event.target_host == self.hs.hostname - - if event.type == InviteJoinEvent.TYPE: - if not target_is_mine: - logger.debug("Ignoring invite/join event %s", event) - return - - # If we receive an invite/join event then we need to join the - # sender to the given room. - # TODO: We should probably auth this or some such - content = event.content - content.update({"membership": Membership.JOIN}) - new_event = self.event_factory.create_event( - etype=RoomMemberEvent.TYPE, - state_key=event.user_id, - room_id=event.room_id, - user_id=event.user_id, - membership=Membership.JOIN, - content=content + redacted_event = prune_event(event) + + redacted_pdu_json = redacted_event.get_pdu_json() + try: + yield self.keyring.verify_json_for_server( + event.origin, redacted_pdu_json + ) + except SynapseError as e: + logger.warn("Signature check failed for %s redacted to %s", + encode_canonical_json(pdu.get_pdu_json()), + encode_canonical_json(redacted_pdu_json), + ) + raise FederationError( + "ERROR", + e.code, + e.msg, + affected=event.event_id, ) - yield self.hs.get_handlers().room_member_handler.change_membership( - new_event, - do_auth=False, + if not check_event_content_hash(event): + logger.warn( + "Event content has been tampered, redacting %s, %s", + event.event_id, encode_canonical_json(event.get_full_dict()) ) + event = redacted_event - else: - with (yield self.room_lock.lock(event.room_id)): - yield self.store.persist_event( - event, - backfilled, - is_new_state=is_new_state - ) + is_new_state = yield self.state_handler.annotate_event_with_state( + event, + old_state=state + ) - room = yield self.store.get_room(event.room_id) + logger.debug("Event: %s", event) - if not room: - # Huh, let's try and get the current state - try: - yield self.replication_layer.get_state_for_context( - event.origin, event.room_id - ) + try: + self.auth.check(event, raises=True) + except AuthError as e: + raise FederationError( + "ERROR", + e.code, + e.msg, + affected=event.event_id, + ) - hosts = yield self.store.get_joined_hosts_for_room( - event.room_id - ) - if self.hs.hostname in hosts: - try: - yield self.store.store_room( - room_id=event.room_id, - room_creator_user_id="", - is_public=False, - ) - except: - pass - except: - logger.exception( - "Failed to get current state for room %s", - event.room_id - ) + is_new_state = is_new_state and not backfilled - if not backfilled: - extra_users = [] - if event.type == RoomMemberEvent.TYPE: - target_user_id = event.state_key - target_user = self.hs.parse_userid(target_user_id) - extra_users.append(target_user) + # TODO: Implement something in federation that allows us to + # respond to PDU. + + yield self.store.persist_event( + event, + backfilled, + is_new_state=is_new_state + ) - yield self.notifier.on_new_room_event( - event, extra_users=extra_users + room = yield self.store.get_room(event.room_id) + + if not room: + # Huh, let's try and get the current state + try: + yield self.replication_layer.get_state_for_context( + event.origin, event.room_id, event.event_id, + ) + + hosts = yield self.store.get_joined_hosts_for_room( + event.room_id + ) + if self.hs.hostname in hosts: + try: + yield self.store.store_room( + room_id=event.room_id, + room_creator_user_id="", + is_public=False, + ) + except: + pass + except: + logger.exception( + "Failed to get current state for room %s", + event.room_id ) + if not backfilled: + extra_users = [] + if event.type == RoomMemberEvent.TYPE: + target_user_id = event.state_key + target_user = self.hs.parse_userid(target_user_id) + extra_users.append(target_user) + + yield self.notifier.on_new_room_event( + event, extra_users=extra_users + ) + if event.type == RoomMemberEvent.TYPE: if event.membership == Membership.JOIN: user = self.hs.parse_userid(event.state_key) @@ -189,79 +216,349 @@ class FederationHandler(BaseHandler): @log_function @defer.inlineCallbacks def backfill(self, dest, room_id, limit): - pdus = yield self.replication_layer.backfill(dest, room_id, limit) + """ Trigger a backfill request to `dest` for the given `room_id` + """ + extremities = yield self.store.get_oldest_events_in_room(room_id) + + pdus = yield self.replication_layer.backfill( + dest, + room_id, + limit, + extremities=extremities, + ) events = [] for pdu in pdus: - event = self.pdu_codec.event_from_pdu(pdu) + event = pdu + + # FIXME (erikj): Not sure this actually works :/ + yield self.state_handler.annotate_event_with_state(event) + events.append(event) + yield self.store.persist_event(event, backfilled=True) defer.returnValue(events) + @defer.inlineCallbacks + def send_invite(self, target_host, event): + """ Sends the invite to the remote server for signing. + + Invites must be signed by the invitee's server before distribution. + """ + pdu = yield self.replication_layer.send_invite( + destination=target_host, + context=event.room_id, + event_id=event.event_id, + pdu=event + ) + + defer.returnValue(pdu) + + @defer.inlineCallbacks + def on_event_auth(self, event_id): + auth = yield self.store.get_auth_chain(event_id) + defer.returnValue([e for e in auth]) + @log_function @defer.inlineCallbacks def do_invite_join(self, target_host, room_id, joinee, content, snapshot): + """ Attempts to join the `joinee` to the room `room_id` via the + server `target_host`. + + This first triggers a /make_join/ request that returns a partial + event that we can fill out and sign. This is then sent to the + remote server via /send_join/ which responds with the state at that + event and the auth_chains. - hosts = yield self.store.get_joined_hosts_for_room(room_id) - if self.hs.hostname in hosts: - # We are already in the room. - logger.debug("We're already in the room apparently") - defer.returnValue(False) + We suspend processing of any received events from this room until we + have finished processing the join. + """ + pdu = yield self.replication_layer.make_join( + target_host, + room_id, + joinee + ) + + logger.debug("Got response to make_join: %s", pdu) + + event = pdu + + # We should assert some things. + assert(event.type == RoomMemberEvent.TYPE) + assert(event.user_id == joinee) + assert(event.state_key == joinee) + assert(event.room_id == room_id) + + event.outlier = False + + self.room_queues[room_id] = [] - # First get current state to see if we are already joined. try: - yield self.replication_layer.get_state_for_context( - target_host, room_id + event.event_id = self.event_factory.create_event_id() + event.content = content + + state = yield self.replication_layer.send_join( + target_host, + event ) - hosts = yield self.store.get_joined_hosts_for_room(room_id) - if self.hs.hostname in hosts: - # Oh, we were actually in the room already. - logger.debug("We're already in the room apparently") - defer.returnValue(False) - except Exception: - logger.exception("Failed to get current state") - - new_event = self.event_factory.create_event( - etype=InviteJoinEvent.TYPE, - target_host=target_host, - room_id=room_id, - user_id=joinee, - content=content - ) + logger.debug("do_invite_join state: %s", state) - new_event.destinations = [target_host] + yield self.state_handler.annotate_event_with_state( + event, + old_state=state + ) - snapshot.fill_out_prev_events(new_event) - yield self.handle_new_event(new_event, snapshot) + logger.debug("do_invite_join event: %s", event) - # TODO (erikj): Time out here. - d = defer.Deferred() - self.waiting_for_join_list.setdefault((joinee, room_id), []).append(d) - reactor.callLater(10, d.cancel) + try: + yield self.store.store_room( + room_id=room_id, + room_creator_user_id="", + is_public=False + ) + except: + # FIXME + pass - try: - yield d - except defer.CancelledError: - raise SynapseError(500, "Unable to join remote room") + for e in state: + # FIXME: Auth these. + e.outlier = True - try: - yield self.store.store_room( - room_id=room_id, - room_creator_user_id="", - is_public=False + yield self.state_handler.annotate_event_with_state( + e, + ) + + yield self.store.persist_event( + e, + backfilled=False, + is_new_state=True + ) + + yield self.store.persist_event( + event, + backfilled=False, + is_new_state=True ) - except: - pass + finally: + room_queue = self.room_queues[room_id] + del self.room_queues[room_id] + for p in room_queue: + try: + yield self.on_receive_pdu(p, backfilled=False) + except: + pass defer.returnValue(True) + @defer.inlineCallbacks + @log_function + def on_make_join_request(self, context, user_id): + """ We've received a /make_join/ request, so we create a partial + join event for the room and return that. We don *not* persist or + process it until the other server has signed it and sent it back. + """ + event = self.event_factory.create_event( + etype=RoomMemberEvent.TYPE, + content={"membership": Membership.JOIN}, + room_id=context, + user_id=user_id, + state_key=user_id, + ) + + snapshot = yield self.store.snapshot_room(event) + snapshot.fill_out_prev_events(event) + + yield self.state_handler.annotate_event_with_state(event) + yield self.auth.add_auth_events(event) + self.auth.check(event, raises=True) + + pdu = event + + defer.returnValue(pdu) + + @defer.inlineCallbacks + @log_function + def on_send_join_request(self, origin, pdu): + """ We have received a join event for a room. Fully process it and + respond with the current state and auth chains. + """ + event = pdu + + event.outlier = False + + is_new_state = yield self.state_handler.annotate_event_with_state(event) + self.auth.check(event, raises=True) + + # FIXME (erikj): All this is duplicated above :( + + yield self.store.persist_event( + event, + backfilled=False, + is_new_state=is_new_state + ) + + extra_users = [] + if event.type == RoomMemberEvent.TYPE: + target_user_id = event.state_key + target_user = self.hs.parse_userid(target_user_id) + extra_users.append(target_user) + + yield self.notifier.on_new_room_event( + event, extra_users=extra_users + ) + + if event.type == RoomMemberEvent.TYPE: + if event.membership == Membership.JOIN: + user = self.hs.parse_userid(event.state_key) + self.distributor.fire( + "user_joined_room", user=user, room_id=event.room_id + ) + + new_pdu = event + + destinations = set() + + for k, s in event.state_events.items(): + try: + if k[0] == RoomMemberEvent.TYPE: + if s.content["membership"] == Membership.JOIN: + destinations.add( + self.hs.parse_userid(s.state_key).domain + ) + except: + logger.warn( + "Failed to get destination from event %s", s.event_id + ) + + new_pdu.destinations = list(destinations) + + yield self.replication_layer.send_pdu(new_pdu) + + auth_chain = yield self.store.get_auth_chain(event.event_id) + + defer.returnValue({ + "state": event.state_events.values(), + "auth_chain": auth_chain, + }) + + @defer.inlineCallbacks + def on_invite_request(self, origin, pdu): + """ We've got an invite event. Process and persist it. Sign it. + + Respond with the now signed event. + """ + event = pdu + + event.outlier = True + + event.signatures.update( + compute_event_signature( + event, + self.hs.hostname, + self.hs.config.signing_key[0] + ) + ) + + yield self.state_handler.annotate_event_with_state(event) + + yield self.store.persist_event( + event, + backfilled=False, + ) + + target_user = self.hs.parse_userid(event.state_key) + yield self.notifier.on_new_room_event( + event, extra_users=[target_user], + ) + + defer.returnValue(event) + + @defer.inlineCallbacks + def get_state_for_pdu(self, origin, room_id, event_id): + yield run_on_reactor() + + in_room = yield self.auth.check_host_in_room(room_id, origin) + if not in_room: + raise AuthError(403, "Host not in room.") + + state_groups = yield self.store.get_state_groups( + [event_id] + ) + + if state_groups: + _, state = state_groups.items().pop() + results = { + (e.type, e.state_key): e for e in state + } + + event = yield self.store.get_event(event_id) + if hasattr(event, "state_key"): + # Get previous state + if hasattr(event, "replaces_state") and event.replaces_state: + prev_event = yield self.store.get_event( + event.replaces_state + ) + results[(event.type, event.state_key)] = prev_event + else: + del results[(event.type, event.state_key)] + + defer.returnValue(results.values()) + else: + defer.returnValue([]) + + @defer.inlineCallbacks + @log_function + def on_backfill_request(self, origin, context, pdu_list, limit): + in_room = yield self.auth.check_host_in_room(context, origin) + if not in_room: + raise AuthError(403, "Host not in room.") + + events = yield self.store.get_backfill_events( + context, + pdu_list, + limit + ) + + defer.returnValue(events) + + @defer.inlineCallbacks + @log_function + def get_persisted_pdu(self, origin, event_id): + """ Get a PDU from the database with given origin and id. + + Returns: + Deferred: Results in a `Pdu`. + """ + event = yield self.store.get_event( + event_id, + allow_none=True, + ) + + if event: + in_room = yield self.auth.check_host_in_room( + event.room_id, + origin + ) + if not in_room: + raise AuthError(403, "Host not in room.") + + defer.returnValue(event) + else: + defer.returnValue(None) + + @log_function + def get_min_depth_for_context(self, context): + return self.store.get_min_depth(context) @log_function def _on_user_joined(self, user, room_id): - waiters = self.waiting_for_join_list.get((user.to_string(), room_id), []) + waiters = self.waiting_for_join_list.get( + (user.to_string(), room_id), + [] + ) while waiters: waiters.pop().callback(None) diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py index 811e280c2d..b77d9d1644 100644 --- a/synapse/handlers/message.py +++ b/synapse/handlers/message.py @@ -16,7 +16,6 @@ from twisted.internet import defer from synapse.api.constants import Membership -from synapse.api.events.room import RoomTopicEvent from synapse.api.errors import RoomError from synapse.streams.config import PaginationConfig from ._base import BaseHandler @@ -26,7 +25,6 @@ import logging logger = logging.getLogger(__name__) - class MessageHandler(BaseHandler): def __init__(self, hs): @@ -59,7 +57,8 @@ class MessageHandler(BaseHandler): # user_id=sender_id # ) - # TODO (erikj): Once we work out the correct c-s api we need to think on how to do this. + # TODO (erikj): Once we work out the correct c-s api we need to think + # on how to do this. defer.returnValue(None) @@ -81,12 +80,11 @@ class MessageHandler(BaseHandler): user = self.hs.parse_userid(event.user_id) assert user.is_mine, "User must be our own: %s" % (user,) - snapshot = yield self.store.snapshot_room(event.room_id, event.user_id) - - if not suppress_auth: - yield self.auth.check(event, snapshot, raises=True) + snapshot = yield self.store.snapshot_room(event) - yield self._on_new_room_event(event, snapshot) + yield self._on_new_room_event( + event, snapshot, suppress_auth=suppress_auth + ) self.hs.get_handlers().presence_handler.bump_presence_active_time( user @@ -111,7 +109,9 @@ class MessageHandler(BaseHandler): data_source = self.hs.get_event_sources().sources["room"] if not pagin_config.from_token: - pagin_config.from_token = yield self.hs.get_event_sources().get_current_token() + pagin_config.from_token = ( + yield self.hs.get_event_sources().get_current_token() + ) user = self.hs.parse_userid(user_id) @@ -142,66 +142,27 @@ class MessageHandler(BaseHandler): SynapseError if something went wrong. """ - snapshot = yield self.store.snapshot_room( - event.room_id, - event.user_id, - state_type=event.type, - state_key=event.state_key, - ) - - yield self.auth.check(event, snapshot, raises=True) - - yield self.state_handler.handle_new_event(event, snapshot) + snapshot = yield self.store.snapshot_room(event) yield self._on_new_room_event(event, snapshot) @defer.inlineCallbacks def get_room_data(self, user_id=None, room_id=None, - event_type=None, state_key="", - public_room_rules=[], - private_room_rules=["join"]): + event_type=None, state_key=""): """ Get data from a room. Args: event : The room path event - public_room_rules : A list of membership states the user can be in, - in order to read this data IN A PUBLIC ROOM. An empty list means - 'any state'. - private_room_rules : A list of membership states the user can be - in, in order to read this data IN A PRIVATE ROOM. An empty list - means 'any state'. Returns: The path data content. Raises: SynapseError if something went wrong. """ - if event_type == RoomTopicEvent.TYPE: - # anyone invited/joined can read the topic - private_room_rules = ["invite", "join"] - - # does this room exist - room = yield self.store.get_room(room_id) - if not room: - raise RoomError(403, "Room does not exist.") - - # does this user exist in this room - member = yield self.store.get_room_member( - room_id=room_id, - user_id="" if not user_id else user_id) - - member_state = member.membership if member else None - - if room.is_public and public_room_rules: - # make sure the user meets public room rules - if member_state not in public_room_rules: - raise RoomError(403, "Member does not meet public room rules.") - elif not room.is_public and private_room_rules: - # make sure the user meets private room rules - if member_state not in private_room_rules: - raise RoomError( - 403, "Member does not meet private room rules.") - - data = yield self.store.get_current_state( + have_joined = yield self.auth.check_joined_room(room_id, user_id) + if not have_joined: + raise RoomError(403, "User not in room.") + + data = yield self.state_handler.get_current_state( room_id, event_type, state_key ) defer.returnValue(data) @@ -219,9 +180,7 @@ class MessageHandler(BaseHandler): @defer.inlineCallbacks def send_feedback(self, event): - snapshot = yield self.store.snapshot_room(event.room_id, event.user_id) - - yield self.auth.check(event, snapshot, raises=True) + snapshot = yield self.store.snapshot_room(event) # store message in db yield self._on_new_room_event(event, snapshot) @@ -239,7 +198,7 @@ class MessageHandler(BaseHandler): yield self.auth.check_joined_room(room_id, user_id) # TODO: This is duplicating logic from snapshot_all_rooms - current_state = yield self.store.get_current_state(room_id) + current_state = yield self.state_handler.get_current_state(room_id) defer.returnValue([self.hs.serialize_event(c) for c in current_state]) @defer.inlineCallbacks @@ -289,8 +248,10 @@ class MessageHandler(BaseHandler): d = { "room_id": event.room_id, "membership": event.membership, - "visibility": ("public" if event.room_id in - public_room_ids else "private"), + "visibility": ( + "public" if event.room_id in public_room_ids + else "private" + ), } if event.membership == Membership.INVITE: @@ -316,10 +277,12 @@ class MessageHandler(BaseHandler): "end": end_token.to_string(), } - current_state = yield self.store.get_current_state( + current_state = yield self.state_handler.get_current_state( event.room_id ) - d["state"] = [self.hs.serialize_event(c) for c in current_state] + d["state"] = [ + self.hs.serialize_event(c) for c in current_state + ] except: logger.exception("Failed to get snapshot") diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py index dab9b03f04..834b37f5f3 100644 --- a/synapse/handlers/profile.py +++ b/synapse/handlers/profile.py @@ -17,7 +17,6 @@ from twisted.internet import defer from synapse.api.errors import SynapseError, AuthError, CodeMessageException from synapse.api.constants import Membership -from synapse.api.events.room import RoomMemberEvent from ._base import BaseHandler @@ -153,10 +152,13 @@ class ProfileHandler(BaseHandler): if not user.is_mine: defer.returnValue(None) - (displayname, avatar_url) = yield defer.gatherResults([ - self.store.get_profile_displayname(user.localpart), - self.store.get_profile_avatar_url(user.localpart), - ]) + (displayname, avatar_url) = yield defer.gatherResults( + [ + self.store.get_profile_displayname(user.localpart), + self.store.get_profile_avatar_url(user.localpart), + ], + consumeErrors=True + ) state["displayname"] = displayname state["avatar_url"] = avatar_url @@ -196,10 +198,7 @@ class ProfileHandler(BaseHandler): ) for j in joins: - snapshot = yield self.store.snapshot_room( - j.room_id, j.state_key, RoomMemberEvent.TYPE, - j.state_key - ) + snapshot = yield self.store.snapshot_room(j) content = { "membership": j.content["membership"], @@ -218,5 +217,6 @@ class ProfileHandler(BaseHandler): user_id=j.state_key, ) - yield self.state_handler.handle_new_event(new_event, snapshot) - yield self._on_new_room_event(new_event, snapshot) + yield self._on_new_room_event( + new_event, snapshot, suppress_auth=True + ) diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index 81ce1a5907..cfe1061ed3 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -21,10 +21,10 @@ from synapse.api.constants import Membership, JoinRules from synapse.api.errors import StoreError, SynapseError from synapse.api.events.room import ( RoomMemberEvent, RoomCreateEvent, RoomPowerLevelsEvent, - RoomJoinRulesEvent, RoomAddStateLevelEvent, RoomTopicEvent, - RoomSendEventLevelEvent, RoomOpsPowerLevelsEvent, RoomNameEvent, + RoomTopicEvent, RoomNameEvent, RoomJoinRulesEvent, ) from synapse.util import stringutils +from synapse.util.async import run_on_reactor from ._base import BaseHandler import logging @@ -111,26 +111,15 @@ class RoomCreationHandler(BaseHandler): user, room_id, is_public=is_public ) - if room_alias: - directory_handler = self.hs.get_handlers().directory_handler - yield directory_handler.create_association( - user_id=user_id, - room_id=room_id, - room_alias=room_alias, - servers=[self.hs.hostname], - ) - @defer.inlineCallbacks def handle_event(event): - snapshot = yield self.store.snapshot_room( - room_id=room_id, - user_id=user_id, - ) + snapshot = yield self.store.snapshot_room(event) logger.debug("Event: %s", event) - yield self.state_handler.handle_new_event(event, snapshot) - yield self._on_new_room_event(event, snapshot, extra_users=[user]) + yield self._on_new_room_event( + event, snapshot, extra_users=[user], suppress_auth=True + ) for event in creation_events: yield handle_event(event) @@ -141,7 +130,6 @@ class RoomCreationHandler(BaseHandler): etype=RoomNameEvent.TYPE, room_id=room_id, user_id=user_id, - required_power_level=50, content={"name": name}, ) @@ -153,7 +141,6 @@ class RoomCreationHandler(BaseHandler): etype=RoomTopicEvent.TYPE, room_id=room_id, user_id=user_id, - required_power_level=50, content={"topic": topic}, ) @@ -188,9 +175,18 @@ class RoomCreationHandler(BaseHandler): join_event, do_auth=False ) + result = {"room_id": room_id} + if room_alias: result["room_alias"] = room_alias.to_string() + directory_handler = self.hs.get_handlers().directory_handler + yield directory_handler.create_association( + user_id=user_id, + room_id=room_id, + room_alias=room_alias, + servers=[self.hs.hostname], + ) defer.returnValue(result) @@ -198,7 +194,6 @@ class RoomCreationHandler(BaseHandler): event_keys = { "room_id": room_id, "user_id": creator.to_string(), - "required_power_level": 100, } def create(etype, **content): @@ -215,7 +210,21 @@ class RoomCreationHandler(BaseHandler): power_levels_event = self.event_factory.create_event( etype=RoomPowerLevelsEvent.TYPE, - content={creator.to_string(): 100, "default": 0}, + content={ + "users": { + creator.to_string(): 100, + }, + "users_default": 0, + "events": { + RoomNameEvent.TYPE: 100, + RoomPowerLevelsEvent.TYPE: 100, + }, + "events_default": 0, + "state_default": 50, + "ban": 50, + "kick": 50, + "redact": 50 + }, **event_keys ) @@ -225,30 +234,10 @@ class RoomCreationHandler(BaseHandler): join_rule=join_rule, ) - add_state_event = create( - etype=RoomAddStateLevelEvent.TYPE, - level=100, - ) - - send_event = create( - etype=RoomSendEventLevelEvent.TYPE, - level=0, - ) - - ops = create( - etype=RoomOpsPowerLevelsEvent.TYPE, - ban_level=50, - kick_level=50, - redact_level=50, - ) - return [ creation_event, power_levels_event, join_rules_event, - add_state_event, - send_event, - ops, ] @@ -363,10 +352,8 @@ class RoomMemberHandler(BaseHandler): """ target_user_id = event.state_key - snapshot = yield self.store.snapshot_room( - event.room_id, event.user_id, - RoomMemberEvent.TYPE, target_user_id - ) + snapshot = yield self.store.snapshot_room(event) + ## TODO(markjh): get prev state from snapshot. prev_state = yield self.store.get_room_member( target_user_id, event.room_id @@ -375,13 +362,6 @@ class RoomMemberHandler(BaseHandler): if prev_state: event.content["prev"] = prev_state.membership -# if prev_state and prev_state.membership == event.membership: -# # treat this event as a NOOP. -# if do_auth: # This is mainly to fix a unit test. -# yield self.auth.check(event, raises=True) -# defer.returnValue({}) -# return - room_id = event.room_id # If we're trying to join a room then we have to do this differently @@ -391,29 +371,17 @@ class RoomMemberHandler(BaseHandler): yield self._do_join(event, snapshot, do_auth=do_auth) else: # This is not a JOIN, so we can handle it normally. - if do_auth: - yield self.auth.check(event, snapshot, raises=True) - - # If we're banning someone, set a req power level - if event.membership == Membership.BAN: - if not hasattr(event, "required_power_level") or event.required_power_level is None: - # Add some default required_power_level - user_level = yield self.store.get_power_level( - event.room_id, - event.user_id, - ) - event.required_power_level = user_level if prev_state and prev_state.membership == event.membership: # double same action, treat this event as a NOOP. defer.returnValue({}) return - yield self.state_handler.handle_new_event(event, snapshot) yield self._do_local_membership_update( event, membership=event.content["membership"], snapshot=snapshot, + do_auth=do_auth, ) defer.returnValue({"room_id": room_id}) @@ -443,10 +411,7 @@ class RoomMemberHandler(BaseHandler): content=content, ) - snapshot = yield self.store.snapshot_room( - room_id, joinee.to_string(), RoomMemberEvent.TYPE, - joinee.to_string() - ) + snapshot = yield self.store.snapshot_room(new_event) yield self._do_join(new_event, snapshot, room_host=host, do_auth=True) @@ -468,9 +433,12 @@ class RoomMemberHandler(BaseHandler): # that we are allowed to join when we decide whether or not we # need to do the invite/join dance. - hosts = yield self.store.get_joined_hosts_for_room(room_id) + is_host_in_room = yield self.auth.check_host_in_room( + event.room_id, + self.hs.hostname + ) - if self.hs.hostname in hosts: + if is_host_in_room: should_do_dance = False elif room_host: should_do_dance = True @@ -502,14 +470,11 @@ class RoomMemberHandler(BaseHandler): if not have_joined: logger.debug("Doing normal join") - if do_auth: - yield self.auth.check(event, snapshot, raises=True) - - yield self.state_handler.handle_new_event(event, snapshot) yield self._do_local_membership_update( event, membership=event.content["membership"], snapshot=snapshot, + do_auth=do_auth, ) user = self.hs.parse_userid(event.user_id) @@ -553,26 +518,29 @@ class RoomMemberHandler(BaseHandler): defer.returnValue([r.room_id for r in rooms]) - def _do_local_membership_update(self, event, membership, snapshot): - destinations = [] + @defer.inlineCallbacks + def _do_local_membership_update(self, event, membership, snapshot, + do_auth): + yield run_on_reactor() # If we're inviting someone, then we should also send it to that # HS. target_user_id = event.state_key target_user = self.hs.parse_userid(target_user_id) - if membership == Membership.INVITE: - host = target_user.domain - destinations.append(host) - - # Always include target domain - host = target_user.domain - destinations.append(host) - - return self._on_new_room_event( - event, snapshot, extra_destinations=destinations, - extra_users=[target_user] + if membership == Membership.INVITE and not target_user.is_mine: + do_invite_host = target_user.domain + else: + do_invite_host = None + + yield self._on_new_room_event( + event, + snapshot, + extra_users=[target_user], + suppress_auth=(not do_auth), + do_invite_host=do_invite_host, ) + class RoomListHandler(BaseHandler): @defer.inlineCallbacks |