diff options
Diffstat (limited to 'synapse/handlers')
| -rw-r--r-- | synapse/handlers/_base.py | 113 | ||||
| -rw-r--r-- | synapse/handlers/directory.py | 30 | ||||
| -rw-r--r-- | synapse/handlers/federation.py | 265 | ||||
| -rw-r--r-- | synapse/handlers/message.py | 54 | ||||
| -rw-r--r-- | synapse/handlers/presence.py | 38 | ||||
| -rw-r--r-- | synapse/handlers/profile.py | 35 | ||||
| -rw-r--r-- | synapse/handlers/register.py | 6 | ||||
| -rw-r--r-- | synapse/handlers/room.py | 211 | ||||
| -rw-r--r-- | synapse/handlers/typing.py | 6 |
9 files changed, 413 insertions, 345 deletions
diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index 15adc9dc2c..af8eb5f0f5 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -15,11 +15,12 @@ from twisted.internet import defer -from synapse.api.errors import LimitExceededError +from synapse.api.errors import LimitExceededError, SynapseError 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 +from synapse.api.constants import Membership, EventTypes + +from synapse.events.snapshot import EventContext import logging @@ -31,10 +32,8 @@ class BaseHandler(object): def __init__(self, hs): self.store = hs.get_datastore() - self.event_factory = hs.get_event_factory() self.auth = hs.get_auth() self.notifier = hs.get_notifier() - self.room_lock = hs.get_room_lock_manager() self.state_handler = hs.get_state_handler() self.distributor = hs.get_distributor() self.ratelimiter = hs.get_ratelimiter() @@ -44,6 +43,8 @@ class BaseHandler(object): self.signing_key = hs.config.signing_key[0] self.server_name = hs.hostname + self.event_builder_factory = hs.get_event_builder_factory() + def ratelimit(self, user_id): time_now = self.clock.time() allowed, time_allowed = self.ratelimiter.send_message( @@ -57,62 +58,100 @@ class BaseHandler(object): ) @defer.inlineCallbacks - def _on_new_room_event(self, event, snapshot, extra_destinations=[], - extra_users=[], suppress_auth=False, - do_invite_host=None): + def _create_new_client_event(self, builder): yield run_on_reactor() - snapshot.fill_out_prev_events(event) + context = EventContext() + + latest_ret = yield self.store.get_latest_events_in_room( + builder.room_id, + ) + + if latest_ret: + depth = max([d for _, _, d in latest_ret]) + 1 + else: + depth = 1 + + prev_events = [(e, h) for e, h, _ in latest_ret] + + builder.prev_events = prev_events + builder.depth = depth - yield self.state_handler.annotate_event_with_state(event) + state_handler = self.state_handler + ret = yield state_handler.annotate_context_with_state( + builder, + context, + ) + prev_state = ret - yield self.auth.add_auth_events(event) + if builder.is_state(): + builder.prev_state = prev_state - logger.debug("Signing event...") + yield self.auth.add_auth_events(builder, context) add_hashes_and_signatures( - event, self.server_name, self.signing_key + builder, self.server_name, self.signing_key + ) + + event = builder.build() + + logger.debug( + "Created event %s with auth_events: %s, current state: %s", + event.event_id, context.auth_events, context.current_state, ) - logger.debug("Signed event.") + defer.returnValue( + (event, context,) + ) + + @defer.inlineCallbacks + def handle_new_client_event(self, event, context, extra_destinations=[], + extra_users=[], suppress_auth=False): + yield run_on_reactor() + + # We now need to go and hit out to wherever we need to hit out to. if not suppress_auth: - logger.debug("Authing...") - self.auth.check(event, auth_events=event.old_state_events) - logger.debug("Authed") - else: - logger.debug("Suppressed auth.") + self.auth.check(event, auth_events=context.auth_events) - if do_invite_host: - federation_handler = self.hs.get_handlers().federation_handler - invite_event = yield federation_handler.send_invite( - do_invite_host, - event - ) + yield self.store.persist_event(event, context=context) - # FIXME: We need to check if the remote changed anything else - event.signatures = invite_event.signatures + federation_handler = self.hs.get_handlers().federation_handler - yield self.store.persist_event(event) + if event.type == EventTypes.Member: + if event.content["membership"] == Membership.INVITE: + invitee = self.hs.parse_userid(event.state_key) + if not self.hs.is_mine(invitee): + # TODO: Can we add signature from remote server in a nicer + # way? If we have been invited by a remote server, we need + # to get them to sign the event. + returned_invite = yield federation_handler.send_invite( + invitee.domain, + event, + ) + + # TODO: Make sure the signatures actually are correct. + event.signatures.update( + returned_invite.signatures + ) destinations = set(extra_destinations) - # Send a PDU to all hosts who have joined the room. - - for k, s in event.state_events.items(): + for k, s in context.current_state.items(): try: - if k[0] == RoomMemberEvent.TYPE: + if k[0] == EventTypes.Member: if s.content["membership"] == Membership.JOIN: destinations.add( self.hs.parse_userid(s.state_key).domain ) - except: + except SynapseError: logger.warn( "Failed to get destination from event %s", s.event_id ) - event.destinations = list(destinations) - yield self.notifier.on_new_room_event(event, extra_users=extra_users) - federation_handler = self.hs.get_handlers().federation_handler - yield federation_handler.handle_new_event(event, snapshot) + yield federation_handler.handle_new_event( + event, + None, + destinations=destinations, + ) diff --git a/synapse/handlers/directory.py b/synapse/handlers/directory.py index 3b37e49e6f..404baea796 100644 --- a/synapse/handlers/directory.py +++ b/synapse/handlers/directory.py @@ -18,7 +18,7 @@ from twisted.internet import defer from ._base import BaseHandler from synapse.api.errors import SynapseError, Codes, CodeMessageException -from synapse.api.events.room import RoomAliasesEvent +from synapse.api.constants import EventTypes import logging @@ -40,7 +40,7 @@ class DirectoryHandler(BaseHandler): # TODO(erikj): Do auth. - if not room_alias.is_mine: + if not self.hs.is_mine(room_alias): raise SynapseError(400, "Room alias must be local") # TODO(erikj): Change this. @@ -64,7 +64,7 @@ class DirectoryHandler(BaseHandler): def delete_association(self, user_id, room_alias): # TODO Check if server admin - if not room_alias.is_mine: + if not self.hs.is_mine(room_alias): raise SynapseError(400, "Room alias must be local") room_id = yield self.store.delete_room_alias(room_alias) @@ -75,7 +75,7 @@ class DirectoryHandler(BaseHandler): @defer.inlineCallbacks def get_association(self, room_alias): room_id = None - if room_alias.is_mine: + if self.hs.is_mine(room_alias): result = yield self.store.get_association_from_room_alias( room_alias ) @@ -123,7 +123,7 @@ class DirectoryHandler(BaseHandler): @defer.inlineCallbacks def on_directory_query(self, args): room_alias = self.hs.parse_roomalias(args["room_alias"]) - if not room_alias.is_mine: + if not self.hs.is_mine(room_alias): raise SynapseError( 400, "Room Alias is not hosted on this Home Server" ) @@ -148,16 +148,12 @@ class DirectoryHandler(BaseHandler): def send_room_alias_update_event(self, user_id, room_id): aliases = yield self.store.get_aliases_for_room(room_id) - event = self.event_factory.create_event( - etype=RoomAliasesEvent.TYPE, - state_key=self.hs.hostname, - room_id=room_id, - user_id=user_id, - content={"aliases": aliases}, - ) - - snapshot = yield self.store.snapshot_room(event) + msg_handler = self.hs.get_handlers().message_handler + yield msg_handler.create_and_send_event({ + "type": EventTypes.Aliases, + "state_key": self.hs.hostname, + "room_id": room_id, + "sender": user_id, + "content": {"aliases": aliases}, + }) - 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 cfb5029774..16a104c0e2 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -17,12 +17,12 @@ from ._base import BaseHandler -from synapse.api.events.utils import prune_event +from synapse.events.snapshot import EventContext +from synapse.events.utils import prune_event from synapse.api.errors import ( AuthError, FederationError, SynapseError, StoreError, ) -from synapse.api.events.room import RoomMemberEvent, RoomCreateEvent -from synapse.api.constants import Membership +from synapse.api.constants import EventTypes, Membership from synapse.util.logutils import log_function from synapse.util.async import run_on_reactor from synapse.crypto.event_signing import ( @@ -76,7 +76,7 @@ class FederationHandler(BaseHandler): @log_function @defer.inlineCallbacks - def handle_new_event(self, event, snapshot): + def handle_new_event(self, event, snapshot, destinations): """ Takes in an event from the client to server side, that has already been authed and handled by the state module, and sends it to any remote home servers that may be interested. @@ -92,12 +92,7 @@ class FederationHandler(BaseHandler): yield run_on_reactor() - pdu = event - - if not hasattr(pdu, "destinations") or not pdu.destinations: - pdu.destinations = [] - - yield self.replication_layer.send_pdu(pdu) + yield self.replication_layer.send_pdu(event, destinations) @log_function @defer.inlineCallbacks @@ -140,7 +135,7 @@ class FederationHandler(BaseHandler): 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.event_id, encode_canonical_json(event.get_dict()) ) event = redacted_event @@ -153,7 +148,7 @@ class FederationHandler(BaseHandler): event.room_id, self.server_name ) - if not is_in_room and not event.outlier: + if not is_in_room and not event.internal_metadata.outlier: logger.debug("Got event for room we're not in.") replication_layer = self.replication_layer @@ -164,7 +159,7 @@ class FederationHandler(BaseHandler): ) for e in auth_chain: - e.outlier = True + e.internal_metadata.outlier = True try: yield self._handle_new_event(e, fetch_missing=False) except: @@ -184,7 +179,7 @@ class FederationHandler(BaseHandler): if state: for e in state: - e.outlier = True + e.internal_metadata.outlier = True try: yield self._handle_new_event(e) except: @@ -229,7 +224,7 @@ class FederationHandler(BaseHandler): if not backfilled: extra_users = [] - if event.type == RoomMemberEvent.TYPE: + if event.type == EventTypes.Member: target_user_id = event.state_key target_user = self.hs.parse_userid(target_user_id) extra_users.append(target_user) @@ -238,7 +233,7 @@ class FederationHandler(BaseHandler): event, extra_users=extra_users ) - if event.type == RoomMemberEvent.TYPE: + if event.type == EventTypes.Member: if event.membership == Membership.JOIN: user = self.hs.parse_userid(event.state_key) yield self.distributor.fire( @@ -265,11 +260,18 @@ class FederationHandler(BaseHandler): event = pdu # FIXME (erikj): Not sure this actually works :/ - yield self.state_handler.annotate_event_with_state(event) + context = EventContext() + yield self.state_handler.annotate_context_with_state(event, context) - events.append(event) + events.append( + (event, context) + ) - yield self.store.persist_event(event, backfilled=True) + yield self.store.persist_event( + event, + context=context, + backfilled=True + ) defer.returnValue(events) @@ -286,8 +288,6 @@ class FederationHandler(BaseHandler): pdu=event ) - - defer.returnValue(pdu) @defer.inlineCallbacks @@ -332,42 +332,55 @@ class FederationHandler(BaseHandler): event = pdu # We should assert some things. - assert(event.type == RoomMemberEvent.TYPE) + # FIXME: Do this in a nicer way + assert(event.type == EventTypes.Member) assert(event.user_id == joinee) assert(event.state_key == joinee) assert(event.room_id == room_id) - event.outlier = False + event.internal_metadata.outlier = False self.room_queues[room_id] = [] + builder = self.event_builder_factory.new( + event.get_pdu_json() + ) + + handled_events = set() + try: - event.event_id = self.event_factory.create_event_id() - event.origin = self.hs.hostname - event.content = content + builder.event_id = self.event_builder_factory.create_event_id() + builder.origin = self.hs.hostname + builder.content = content if not hasattr(event, "signatures"): - event.signatures = {} + builder.signatures = {} add_hashes_and_signatures( - event, + builder, self.hs.hostname, self.hs.config.signing_key[0], ) + new_event = builder.build() + ret = yield self.replication_layer.send_join( target_host, - event + new_event ) state = ret["state"] auth_chain = ret["auth_chain"] auth_chain.sort(key=lambda e: e.depth) + handled_events.update([s.event_id for s in state]) + handled_events.update([a.event_id for a in auth_chain]) + handled_events.add(new_event.event_id) + logger.debug("do_invite_join auth_chain: %s", auth_chain) logger.debug("do_invite_join state: %s", state) - logger.debug("do_invite_join event: %s", event) + logger.debug("do_invite_join event: %s", new_event) try: yield self.store.store_room( @@ -380,7 +393,7 @@ class FederationHandler(BaseHandler): pass for e in auth_chain: - e.outlier = True + e.internal_metadata.outlier = True try: yield self._handle_new_event(e, fetch_missing=False) except: @@ -391,7 +404,7 @@ class FederationHandler(BaseHandler): for e in state: # FIXME: Auth these. - e.outlier = True + e.internal_metadata.outlier = True try: yield self._handle_new_event( e, @@ -404,13 +417,13 @@ class FederationHandler(BaseHandler): ) yield self._handle_new_event( - event, + new_event, state=state, current_state=state, ) yield self.notifier.on_new_room_event( - event, extra_users=[joinee] + new_event, extra_users=[joinee] ) logger.debug("Finished joining %s to %s", joinee, room_id) @@ -419,6 +432,9 @@ class FederationHandler(BaseHandler): del self.room_queues[room_id] for p, origin in room_queue: + if p.event_id in handled_events: + continue + try: self.on_receive_pdu(origin, p, backfilled=False) except: @@ -428,25 +444,24 @@ class FederationHandler(BaseHandler): @defer.inlineCallbacks @log_function - def on_make_join_request(self, context, user_id): + def on_make_join_request(self, room_id, 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, - ) + builder = self.event_builder_factory.new({ + "type": EventTypes.Member, + "content": {"membership": Membership.JOIN}, + "room_id": room_id, + "sender": user_id, + "state_key": user_id, + }) - snapshot = yield self.store.snapshot_room(event) - snapshot.fill_out_prev_events(event) + event, context = yield self._create_new_client_event( + builder=builder, + ) - yield self.state_handler.annotate_event_with_state(event) - yield self.auth.add_auth_events(event) - self.auth.check(event, auth_events=event.old_state_events) + self.auth.check(event, auth_events=context.auth_events) pdu = event @@ -460,12 +475,24 @@ class FederationHandler(BaseHandler): """ event = pdu - event.outlier = False + logger.debug( + "on_send_join_request: Got event: %s, signatures: %s", + event.event_id, + event.signatures, + ) + + event.internal_metadata.outlier = False - yield self._handle_new_event(event) + context = yield self._handle_new_event(event) + + logger.debug( + "on_send_join_request: After _handle_new_event: %s, sigs: %s", + event.event_id, + event.signatures, + ) extra_users = [] - if event.type == RoomMemberEvent.TYPE: + if event.type == EventTypes.Member: target_user_id = event.state_key target_user = self.hs.parse_userid(target_user_id) extra_users.append(target_user) @@ -474,7 +501,7 @@ class FederationHandler(BaseHandler): event, extra_users=extra_users ) - if event.type == RoomMemberEvent.TYPE: + if event.type == EventTypes.Member: if event.content["membership"] == Membership.JOIN: user = self.hs.parse_userid(event.state_key) yield self.distributor.fire( @@ -485,9 +512,9 @@ class FederationHandler(BaseHandler): destinations = set() - for k, s in event.state_events.items(): + for k, s in context.current_state.items(): try: - if k[0] == RoomMemberEvent.TYPE: + if k[0] == EventTypes.Member: if s.content["membership"] == Membership.JOIN: destinations.add( self.hs.parse_userid(s.state_key).domain @@ -497,14 +524,18 @@ class FederationHandler(BaseHandler): "Failed to get destination from event %s", s.event_id ) - new_pdu.destinations = list(destinations) + logger.debug( + "on_send_join_request: Sending event: %s, signatures: %s", + event.event_id, + event.signatures, + ) - yield self.replication_layer.send_pdu(new_pdu) + yield self.replication_layer.send_pdu(new_pdu, destinations) auth_chain = yield self.store.get_auth_chain(event.event_id) defer.returnValue({ - "state": event.state_events.values(), + "state": context.current_state.values(), "auth_chain": auth_chain, }) @@ -516,7 +547,9 @@ class FederationHandler(BaseHandler): """ event = pdu - event.outlier = True + context = EventContext() + + event.internal_metadata.outlier = True event.signatures.update( compute_event_signature( @@ -526,10 +559,11 @@ class FederationHandler(BaseHandler): ) ) - yield self.state_handler.annotate_event_with_state(event) + yield self.state_handler.annotate_context_with_state(event, context) yield self.store.persist_event( event, + context=context, backfilled=False, ) @@ -559,13 +593,13 @@ class FederationHandler(BaseHandler): } event = yield self.store.get_event(event_id) - if hasattr(event, "state_key"): + if event and event.is_state(): # 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 + if "replaces_state" in event.unsigned: + prev_id = event.unsigned["replaces_state"] + if prev_id != event.event_id: + prev_event = yield self.store.get_event(prev_id) + results[(event.type, event.state_key)] = prev_event else: del results[(event.type, event.state_key)] @@ -651,74 +685,81 @@ class FederationHandler(BaseHandler): @defer.inlineCallbacks def _handle_new_event(self, event, state=None, backfilled=False, current_state=None, fetch_missing=True): - is_new_state = yield self.state_handler.annotate_event_with_state( + context = EventContext() + + logger.debug( + "_handle_new_event: Before annotate: %s, sigs: %s", + event.event_id, event.signatures, + ) + + yield self.state_handler.annotate_context_with_state( event, + context, old_state=state ) - if event.old_state_events: - known_ids = set( - [s.event_id for s in event.old_state_events.values()] - ) - for e_id, _ in event.auth_events: - if e_id not in known_ids: - e = yield self.store.get_event( - e_id, - allow_none=True, - ) - - if not e: - # TODO: Do some conflict res to make sure that we're - # not the ones who are wrong. - logger.info( - "Rejecting %s as %s not in %s", - event.event_id, e_id, known_ids, - ) - raise AuthError(403, "Auth events are stale") - - auth_events = event.old_state_events - else: - # We need to get the auth events from somewhere. + logger.debug( + "_handle_new_event: Before auth fetch: %s, sigs: %s", + event.event_id, event.signatures, + ) - # TODO: Don't just hit the DBs? + is_new_state = not event.internal_metadata.is_outlier() - auth_events = {} - for e_id, _ in event.auth_events: + known_ids = set( + [s.event_id for s in context.auth_events.values()] + ) + for e_id, _ in event.auth_events: + if e_id not in known_ids: e = yield self.store.get_event( - e_id, - allow_none=True, + e_id, allow_none=True, ) if not e: - e = yield self.replication_layer.get_pdu( - event.origin, e_id, outlier=True + # TODO: Do some conflict res to make sure that we're + # not the ones who are wrong. + logger.info( + "Rejecting %s as %s not in db or %s", + event.event_id, e_id, known_ids, ) + # FIXME: How does raising AuthError work with federation? + raise AuthError(403, "Auth events are stale") - if e and fetch_missing: - try: - yield self.on_receive_pdu(event.origin, e, False) - except: - logger.exception( - "Failed to parse auth event %s", - e_id, - ) + context.auth_events[(e.type, e.state_key)] = e - if not e: - logger.warn("Can't find auth event %s.", e_id) + logger.debug( + "_handle_new_event: Before hack: %s, sigs: %s", + event.event_id, event.signatures, + ) - auth_events[(e.type, e.state_key)] = e + if event.type == EventTypes.Member and not event.auth_events: + if len(event.prev_events) == 1: + c = yield self.store.get_event(event.prev_events[0][0]) + if c.type == EventTypes.Create: + context.auth_events[(c.type, c.state_key)] = c - if event.type == RoomMemberEvent.TYPE and not event.auth_events: - if len(event.prev_events) == 1: - c = yield self.store.get_event(event.prev_events[0][0]) - if c.type == RoomCreateEvent.TYPE: - auth_events[(c.type, c.state_key)] = c + logger.debug( + "_handle_new_event: Before auth check: %s, sigs: %s", + event.event_id, event.signatures, + ) - self.auth.check(event, auth_events=auth_events) + self.auth.check(event, auth_events=context.auth_events) + + logger.debug( + "_handle_new_event: Before persist_event: %s, sigs: %s", + event.event_id, event.signatures, + ) yield self.store.persist_event( event, + context=context, backfilled=backfilled, is_new_state=(is_new_state and not backfilled), current_state=current_state, ) + + logger.debug( + "_handle_new_event: After persist_event: %s, sigs: %s", + event.event_id, event.signatures, + ) + + defer.returnValue(context) diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py index 42dc4d46f3..baf372fdad 100644 --- a/synapse/handlers/message.py +++ b/synapse/handlers/message.py @@ -15,10 +15,13 @@ from twisted.internet import defer -from synapse.api.constants import Membership +from synapse.api.constants import EventTypes, Membership from synapse.api.errors import RoomError from synapse.streams.config import PaginationConfig from synapse.util.logcontext import PreserveLoggingContext + +from synapse.events.validator import EventValidator + from ._base import BaseHandler import logging @@ -32,7 +35,7 @@ class MessageHandler(BaseHandler): super(MessageHandler, self).__init__(hs) self.hs = hs self.clock = hs.get_clock() - self.event_factory = hs.get_event_factory() + self.validator = EventValidator() @defer.inlineCallbacks def get_message(self, msg_id=None, room_id=None, sender_id=None, @@ -79,7 +82,7 @@ class MessageHandler(BaseHandler): self.ratelimit(event.user_id) # TODO(paul): Why does 'event' not have a 'user' object? user = self.hs.parse_userid(event.user_id) - assert user.is_mine, "User must be our own: %s" % (user,) + assert self.hs.is_mine(user), "User must be our own: %s" % (user,) snapshot = yield self.store.snapshot_room(event) @@ -134,19 +137,48 @@ class MessageHandler(BaseHandler): defer.returnValue(chunk) @defer.inlineCallbacks - def store_room_data(self, event=None): - """ Stores data for a room. + def create_and_send_event(self, event_dict): + """ Given a dict from a client, create and handle a new event. + + Creates an FrozenEvent object, filling out auth_events, prev_events, + etc. + + Adds display names to Join membership events. + + Persists and notifies local clients and federation. Args: - event : The room path event - stamp_event (bool) : True to stamp event content with server keys. - Raises: - SynapseError if something went wrong. + event_dict (dict): An entire event """ + builder = self.event_builder_factory.new(event_dict) + + self.validator.validate_new(builder) + + if builder.type == EventTypes.Member: + membership = builder.content.get("membership", None) + if membership == Membership.JOIN: + joinee = self.hs.parse_userid(builder.state_key) + # If event doesn't include a display name, add one. + yield self.distributor.fire( + "collect_presencelike_data", + joinee, + builder.content + ) - snapshot = yield self.store.snapshot_room(event) + event, context = yield self._create_new_client_event( + builder=builder, + ) - yield self._on_new_room_event(event, snapshot) + if event.type == EventTypes.Member: + member_handler = self.hs.get_handlers().room_member_handler + yield member_handler.change_membership(event, context) + else: + yield self.handle_new_client_event( + event=event, + context=context, + ) + + defer.returnValue(event) @defer.inlineCallbacks def get_room_data(self, user_id=None, room_id=None, diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py index 84a039489f..f3abfecdee 100644 --- a/synapse/handlers/presence.py +++ b/synapse/handlers/presence.py @@ -147,7 +147,7 @@ class PresenceHandler(BaseHandler): @defer.inlineCallbacks def is_presence_visible(self, observer_user, observed_user): - assert(observed_user.is_mine) + assert(self.hs.is_mine(observed_user)) if observer_user == observed_user: defer.returnValue(True) @@ -165,7 +165,7 @@ class PresenceHandler(BaseHandler): @defer.inlineCallbacks def get_state(self, target_user, auth_user, as_event=False): - if target_user.is_mine: + if self.hs.is_mine(target_user): visible = yield self.is_presence_visible( observer_user=auth_user, observed_user=target_user @@ -212,7 +212,7 @@ class PresenceHandler(BaseHandler): # TODO (erikj): Turn this back on. Why did we end up sending EDUs # everywhere? - if not target_user.is_mine: + if not self.hs.is_mine(target_user): raise SynapseError(400, "User is not hosted on this Home Server") if target_user != auth_user: @@ -291,7 +291,7 @@ class PresenceHandler(BaseHandler): @defer.inlineCallbacks def user_joined_room(self, user, room_id): - if user.is_mine: + if self.hs.is_mine(user): statuscache = self._get_or_make_usercache(user) # No actual update but we need to bump the serial anyway for the @@ -309,7 +309,7 @@ class PresenceHandler(BaseHandler): rm_handler = self.homeserver.get_handlers().room_member_handler curr_users = yield rm_handler.get_room_members(room_id) - for local_user in [c for c in curr_users if c.is_mine]: + for local_user in [c for c in curr_users if self.hs.is_mine(c)]: self.push_update_to_local_and_remote( observed_user=local_user, users_to_push=[user], @@ -318,14 +318,14 @@ class PresenceHandler(BaseHandler): @defer.inlineCallbacks def send_invite(self, observer_user, observed_user): - if not observer_user.is_mine: + if not self.hs.is_mine(observer_user): raise SynapseError(400, "User is not hosted on this Home Server") yield self.store.add_presence_list_pending( observer_user.localpart, observed_user.to_string() ) - if observed_user.is_mine: + if self.hs.is_mine(observed_user): yield self.invite_presence(observed_user, observer_user) else: yield self.federation.send_edu( @@ -339,7 +339,7 @@ class PresenceHandler(BaseHandler): @defer.inlineCallbacks def _should_accept_invite(self, observed_user, observer_user): - if not observed_user.is_mine: + if not self.hs.is_mine(observed_user): defer.returnValue(False) row = yield self.store.has_presence_state(observed_user.localpart) @@ -359,7 +359,7 @@ class PresenceHandler(BaseHandler): observed_user.localpart, observer_user.to_string() ) - if observer_user.is_mine: + if self.hs.is_mine(observer_user): if accept: yield self.accept_presence(observed_user, observer_user) else: @@ -396,7 +396,7 @@ class PresenceHandler(BaseHandler): @defer.inlineCallbacks def drop(self, observed_user, observer_user): - if not observer_user.is_mine: + if not self.hs.is_mine(observer_user): raise SynapseError(400, "User is not hosted on this Home Server") yield self.store.del_presence_list( @@ -410,7 +410,7 @@ class PresenceHandler(BaseHandler): @defer.inlineCallbacks def get_presence_list(self, observer_user, accepted=None): - if not observer_user.is_mine: + if not self.hs.is_mine(observer_user): raise SynapseError(400, "User is not hosted on this Home Server") presence = yield self.store.get_presence_list( @@ -465,7 +465,7 @@ class PresenceHandler(BaseHandler): ) for target_user in target_users: - if target_user.is_mine: + if self.hs.is_mine(target_user): self._start_polling_local(user, target_user) # We want to tell the person that just came online @@ -477,7 +477,7 @@ class PresenceHandler(BaseHandler): ) deferreds = [] - remote_users = [u for u in target_users if not u.is_mine] + remote_users = [u for u in target_users if not self.hs.is_mine(u)] remoteusers_by_domain = partition(remote_users, lambda u: u.domain) # Only poll for people in our get_presence_list for domain in remoteusers_by_domain: @@ -520,7 +520,7 @@ class PresenceHandler(BaseHandler): def stop_polling_presence(self, user, target_user=None): logger.debug("Stop polling for presence from %s", user) - if not target_user or target_user.is_mine: + if not target_user or self.hs.is_mine(target_user): self._stop_polling_local(user, target_user=target_user) deferreds = [] @@ -579,7 +579,7 @@ class PresenceHandler(BaseHandler): @defer.inlineCallbacks @log_function def push_presence(self, user, statuscache): - assert(user.is_mine) + assert(self.hs.is_mine(user)) logger.debug("Pushing presence update from %s", user) @@ -696,7 +696,7 @@ class PresenceHandler(BaseHandler): for poll in content.get("poll", []): user = self.hs.parse_userid(poll) - if not user.is_mine: + if not self.hs.is_mine(user): continue # TODO(paul) permissions checks @@ -711,7 +711,7 @@ class PresenceHandler(BaseHandler): for unpoll in content.get("unpoll", []): user = self.hs.parse_userid(unpoll) - if not user.is_mine: + if not self.hs.is_mine(user): continue if user in self._remote_sendmap: @@ -730,7 +730,7 @@ class PresenceHandler(BaseHandler): localusers, remoteusers = partitionbool( users_to_push, - lambda u: u.is_mine + lambda u: self.hs.is_mine(u) ) localusers = set(localusers) @@ -788,7 +788,7 @@ class PresenceEventSource(object): [u.to_string() for u in observer_user, observed_user])): defer.returnValue(True) - if observed_user.is_mine: + if self.hs.is_mine(observed_user): pushmap = presence._local_pushmap defer.returnValue( diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py index 814b3b68fe..33a2c167ec 100644 --- a/synapse/handlers/profile.py +++ b/synapse/handlers/profile.py @@ -51,7 +51,7 @@ class ProfileHandler(BaseHandler): @defer.inlineCallbacks def get_displayname(self, target_user): - if target_user.is_mine: + if self.hs.is_mine(target_user): displayname = yield self.store.get_profile_displayname( target_user.localpart ) @@ -81,7 +81,7 @@ class ProfileHandler(BaseHandler): def set_displayname(self, target_user, auth_user, new_displayname): """target_user is the user whose displayname is to be changed; auth_user is the user attempting to make this change.""" - if not target_user.is_mine: + if not self.hs.is_mine(target_user): raise SynapseError(400, "User is not hosted on this Home Server") if target_user != auth_user: @@ -101,7 +101,7 @@ class ProfileHandler(BaseHandler): @defer.inlineCallbacks def get_avatar_url(self, target_user): - if target_user.is_mine: + if self.hs.is_mine(target_user): avatar_url = yield self.store.get_profile_avatar_url( target_user.localpart ) @@ -130,7 +130,7 @@ class ProfileHandler(BaseHandler): def set_avatar_url(self, target_user, auth_user, new_avatar_url): """target_user is the user whose avatar_url is to be changed; auth_user is the user attempting to make this change.""" - if not target_user.is_mine: + if not self.hs.is_mine(target_user): raise SynapseError(400, "User is not hosted on this Home Server") if target_user != auth_user: @@ -150,7 +150,7 @@ class ProfileHandler(BaseHandler): @defer.inlineCallbacks def collect_presencelike_data(self, user, state): - if not user.is_mine: + if not self.hs.is_mine(user): defer.returnValue(None) with PreserveLoggingContext(): @@ -170,7 +170,7 @@ class ProfileHandler(BaseHandler): @defer.inlineCallbacks def on_profile_query(self, args): user = self.hs.parse_userid(args["user_id"]) - if not user.is_mine: + if not self.hs.is_mine(user): raise SynapseError(400, "User is not hosted on this Home Server") just_field = args.get("field", None) @@ -191,7 +191,7 @@ class ProfileHandler(BaseHandler): @defer.inlineCallbacks def _update_join_states(self, user): - if not user.is_mine: + if not self.hs.is_mine(user): return joins = yield self.store.get_rooms_for_user_where_membership_is( @@ -200,8 +200,6 @@ class ProfileHandler(BaseHandler): ) for j in joins: - snapshot = yield self.store.snapshot_room(j) - content = { "membership": j.content["membership"], } @@ -210,14 +208,11 @@ class ProfileHandler(BaseHandler): "collect_presencelike_data", user, content ) - new_event = self.event_factory.create_event( - etype=j.type, - room_id=j.room_id, - state_key=j.state_key, - content=content, - user_id=j.state_key, - ) - - yield self._on_new_room_event( - new_event, snapshot, suppress_auth=True - ) + msg_handler = self.hs.get_handlers().message_handler + yield msg_handler.create_and_send_event({ + "type": j.type, + "room_id": j.room_id, + "state_key": j.state_key, + "content": content, + "sender": j.state_key, + }) diff --git a/synapse/handlers/register.py b/synapse/handlers/register.py index 48c326ebf0..15d8716455 100644 --- a/synapse/handlers/register.py +++ b/synapse/handlers/register.py @@ -22,6 +22,7 @@ from synapse.api.errors import ( ) from ._base import BaseHandler import synapse.util.stringutils as stringutils +from synapse.util.async import run_on_reactor from synapse.http.client import SimpleHttpClient from synapse.http.client import CaptchaServerHttpClient @@ -54,12 +55,13 @@ class RegistrationHandler(BaseHandler): Raises: RegistrationError if there was a problem registering. """ + yield run_on_reactor() password_hash = None if password: password_hash = bcrypt.hashpw(password, bcrypt.gensalt()) if localpart: - user = UserID(localpart, self.hs.hostname, True) + user = UserID(localpart, self.hs.hostname) user_id = user.to_string() token = self._generate_token(user_id) @@ -78,7 +80,7 @@ class RegistrationHandler(BaseHandler): while not user_id and not token: try: localpart = self._generate_user_id() - user = UserID(localpart, self.hs.hostname, True) + user = UserID(localpart, self.hs.hostname) user_id = user.to_string() token = self._generate_token(user_id) diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index c802e8f69d..8567d7409d 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -17,12 +17,8 @@ from twisted.internet import defer from synapse.types import UserID, RoomAlias, RoomID -from synapse.api.constants import Membership, JoinRules +from synapse.api.constants import EventTypes, Membership, JoinRules from synapse.api.errors import StoreError, SynapseError -from synapse.api.events.room import ( - RoomMemberEvent, RoomCreateEvent, RoomPowerLevelsEvent, - RoomTopicEvent, RoomNameEvent, RoomJoinRulesEvent, -) from synapse.util import stringutils from synapse.util.async import run_on_reactor from ._base import BaseHandler @@ -52,9 +48,9 @@ class RoomCreationHandler(BaseHandler): self.ratelimit(user_id) if "room_alias_name" in config: - room_alias = RoomAlias.create_local( + room_alias = RoomAlias.create( config["room_alias_name"], - self.hs + self.hs.hostname, ) mapping = yield self.store.get_association_from_room_alias( room_alias @@ -76,8 +72,8 @@ class RoomCreationHandler(BaseHandler): if room_id: # Ensure room_id is the correct type - room_id_obj = RoomID.from_string(room_id, self.hs) - if not room_id_obj.is_mine: + room_id_obj = RoomID.from_string(room_id) + if not self.hs.is_mine(room_id_obj): raise SynapseError(400, "Room id must be local") yield self.store.store_room( @@ -93,7 +89,10 @@ class RoomCreationHandler(BaseHandler): while attempts < 5: try: random_string = stringutils.random_string(18) - gen_room_id = RoomID.create_local(random_string, self.hs) + gen_room_id = RoomID.create( + random_string, + self.hs.hostname, + ) yield self.store.store_room( room_id=gen_room_id.to_string(), room_creator_user_id=user_id, @@ -120,59 +119,37 @@ class RoomCreationHandler(BaseHandler): user, room_id, is_public=is_public ) - room_member_handler = self.hs.get_handlers().room_member_handler - - @defer.inlineCallbacks - def handle_event(event): - snapshot = yield self.store.snapshot_room(event) - - logger.debug("Event: %s", event) - - if event.type == RoomMemberEvent.TYPE: - yield room_member_handler.change_membership( - event, - do_auth=True - ) - else: - yield self._on_new_room_event( - event, snapshot, extra_users=[user], suppress_auth=True - ) + msg_handler = self.hs.get_handlers().message_handler for event in creation_events: - yield handle_event(event) + yield msg_handler.create_and_send_event(event) if "name" in config: name = config["name"] - name_event = self.event_factory.create_event( - etype=RoomNameEvent.TYPE, - room_id=room_id, - user_id=user_id, - content={"name": name}, - ) - - yield handle_event(name_event) + yield msg_handler.create_and_send_event({ + "type": EventTypes.Name, + "room_id": room_id, + "sender": user_id, + "content": {"name": name}, + }) if "topic" in config: topic = config["topic"] - topic_event = self.event_factory.create_event( - etype=RoomTopicEvent.TYPE, - room_id=room_id, - user_id=user_id, - content={"topic": topic}, - ) - - yield handle_event(topic_event) + yield msg_handler.create_and_send_event({ + "type": EventTypes.Topic, + "room_id": room_id, + "sender": user_id, + "content": {"topic": topic}, + }) - content = {"membership": Membership.INVITE} for invitee in invite_list: - invite_event = self.event_factory.create_event( - etype=RoomMemberEvent.TYPE, - state_key=invitee, - room_id=room_id, - user_id=user_id, - content=content - ) - yield handle_event(invite_event) + yield msg_handler.create_and_send_event({ + "type": EventTypes.Member, + "state_key": invitee, + "room_id": room_id, + "user_id": user_id, + "content": {"membership": Membership.INVITE}, + }) result = {"room_id": room_id} @@ -189,40 +166,44 @@ class RoomCreationHandler(BaseHandler): event_keys = { "room_id": room_id, - "user_id": creator_id, + "sender": creator_id, + "state_key": "", } - def create(etype, **content): - return self.event_factory.create_event( - etype=etype, - content=content, - **event_keys - ) + def create(etype, content, **kwargs): + e = { + "type": etype, + "content": content, + } + + e.update(event_keys) + e.update(kwargs) + + return e creation_event = create( - etype=RoomCreateEvent.TYPE, - creator=creator.to_string(), + etype=EventTypes.Create, + content={"creator": creator.to_string()}, ) - join_event = self.event_factory.create_event( - etype=RoomMemberEvent.TYPE, + join_event = create( + etype=EventTypes.Member, state_key=creator_id, content={ "membership": Membership.JOIN, }, - **event_keys ) - power_levels_event = self.event_factory.create_event( - etype=RoomPowerLevelsEvent.TYPE, + power_levels_event = create( + etype=EventTypes.PowerLevels, content={ "users": { creator.to_string(): 100, }, "users_default": 0, "events": { - RoomNameEvent.TYPE: 100, - RoomPowerLevelsEvent.TYPE: 100, + EventTypes.Name: 100, + EventTypes.PowerLevels: 100, }, "events_default": 0, "state_default": 50, @@ -230,13 +211,12 @@ class RoomCreationHandler(BaseHandler): "kick": 50, "redact": 50 }, - **event_keys ) join_rule = JoinRules.PUBLIC if is_public else JoinRules.INVITE join_rules_event = create( - etype=RoomJoinRulesEvent.TYPE, - join_rule=join_rule, + etype=EventTypes.JoinRules, + content={"join_rule": join_rule}, ) return [ @@ -288,7 +268,7 @@ class RoomMemberHandler(BaseHandler): if ignore_user is not None and member == ignore_user: continue - if member.is_mine: + if self.hs.is_mine(member): if localusers is not None: localusers.add(member) else: @@ -349,7 +329,7 @@ class RoomMemberHandler(BaseHandler): defer.returnValue(member) @defer.inlineCallbacks - def change_membership(self, event=None, do_auth=True): + def change_membership(self, event, context, do_auth=True): """ Change the membership status of a user in a room. Args: @@ -359,11 +339,9 @@ class RoomMemberHandler(BaseHandler): """ target_user_id = event.state_key - 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 + prev_state = context.current_state.get( + (EventTypes.Member, target_user_id), + None ) room_id = event.room_id @@ -372,10 +350,11 @@ class RoomMemberHandler(BaseHandler): # if this HS is not currently in the room, i.e. we have to do the # invite/join dance. if event.membership == Membership.JOIN: - yield self._do_join(event, snapshot, do_auth=do_auth) + yield self._do_join(event, context, do_auth=do_auth) else: # This is not a JOIN, so we can handle it normally. + # FIXME: This isn't idempotency. if prev_state and prev_state.membership == event.membership: # double same action, treat this event as a NOOP. defer.returnValue({}) @@ -384,7 +363,7 @@ class RoomMemberHandler(BaseHandler): yield self._do_local_membership_update( event, membership=event.content["membership"], - snapshot=snapshot, + context=context, do_auth=do_auth, ) @@ -412,32 +391,26 @@ class RoomMemberHandler(BaseHandler): host = hosts[0] content.update({"membership": Membership.JOIN}) - new_event = self.event_factory.create_event( - etype=RoomMemberEvent.TYPE, - state_key=joinee.to_string(), - room_id=room_id, - user_id=joinee.to_string(), - membership=Membership.JOIN, - content=content, - ) - - snapshot = yield self.store.snapshot_room(new_event) + builder = self.event_builder_factory.new({ + "type": EventTypes.Member, + "state_key": joinee.to_string(), + "room_id": room_id, + "sender": joinee.to_string(), + "membership": Membership.JOIN, + "content": content, + }) + event, context = yield self._create_new_client_event(builder) - yield self._do_join(new_event, snapshot, room_host=host, do_auth=True) + yield self._do_join(event, context, room_host=host, do_auth=True) defer.returnValue({"room_id": room_id}) @defer.inlineCallbacks - def _do_join(self, event, snapshot, room_host=None, do_auth=True): + def _do_join(self, event, context, room_host=None, do_auth=True): joinee = self.hs.parse_userid(event.state_key) # room_id = RoomID.from_string(event.room_id, self.hs) room_id = event.room_id - # If event doesn't include a display name, add one. - yield self.distributor.fire( - "collect_presencelike_data", joinee, event.content - ) - # XXX: We don't do an auth check if we are doing an invite # join dance for now, since we're kinda implicitly checking # that we are allowed to join when we decide whether or not we @@ -459,31 +432,29 @@ class RoomMemberHandler(BaseHandler): ) if prev_state and prev_state.membership == Membership.INVITE: - room = yield self.store.get_room(room_id) - inviter = UserID.from_string( - prev_state.user_id, self.hs - ) + inviter = UserID.from_string(prev_state.user_id) - should_do_dance = not inviter.is_mine and not room + should_do_dance = not self.hs.is_mine(inviter) room_host = inviter.domain else: should_do_dance = False - have_joined = False if should_do_dance: handler = self.hs.get_handlers().federation_handler - have_joined = yield handler.do_invite_join( - room_host, room_id, event.user_id, event.content, snapshot + yield handler.do_invite_join( + room_host, + room_id, + event.user_id, + event.get_dict()["content"], # FIXME To get a non-frozen dict + context ) - - # We want to do the _do_update inside the room lock. - if not have_joined: + else: logger.debug("Doing normal join") yield self._do_local_membership_update( event, membership=event.content["membership"], - snapshot=snapshot, + context=context, do_auth=do_auth, ) @@ -508,10 +479,10 @@ class RoomMemberHandler(BaseHandler): if prev_state and prev_state.membership == Membership.INVITE: room = yield self.store.get_room(room_id) inviter = UserID.from_string( - prev_state.sender, self.hs + prev_state.sender ) - is_remote_invite_join = not inviter.is_mine and not room + is_remote_invite_join = not self.hs.is_mine(inviter) and not room room_host = inviter.domain else: is_remote_invite_join = False @@ -533,25 +504,17 @@ class RoomMemberHandler(BaseHandler): defer.returnValue(room_ids) @defer.inlineCallbacks - def _do_local_membership_update(self, event, membership, snapshot, + def _do_local_membership_update(self, event, membership, context, 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 and not target_user.is_mine: - do_invite_host = target_user.domain - else: - do_invite_host = None + target_user = self.hs.parse_userid(event.state_key) - yield self._on_new_room_event( + yield self.handle_new_client_event( event, - snapshot, + context, extra_users=[target_user], suppress_auth=(not do_auth), - do_invite_host=do_invite_host, ) diff --git a/synapse/handlers/typing.py b/synapse/handlers/typing.py index 3f2f250dd2..77d66f66ff 100644 --- a/synapse/handlers/typing.py +++ b/synapse/handlers/typing.py @@ -63,7 +63,7 @@ class TypingNotificationHandler(BaseHandler): @defer.inlineCallbacks def started_typing(self, target_user, auth_user, room_id, timeout): - if not target_user.is_mine: + if not self.hs.is_mine(target_user): raise SynapseError(400, "User is not hosted on this Home Server") if target_user != auth_user: @@ -100,7 +100,7 @@ class TypingNotificationHandler(BaseHandler): @defer.inlineCallbacks def stopped_typing(self, target_user, auth_user, room_id): - if not target_user.is_mine: + if not self.hs.is_mine(target_user): raise SynapseError(400, "User is not hosted on this Home Server") if target_user != auth_user: @@ -118,7 +118,7 @@ class TypingNotificationHandler(BaseHandler): @defer.inlineCallbacks def user_left_room(self, user, room_id): - if user.is_mine: + if self.hs.is_mine(user): member = RoomMember(room_id=room_id, user=user) yield self._stopped_typing(member) |