diff options
| -rw-r--r-- | synapse/api/auth.py | 13 | ||||
| -rw-r--r-- | synapse/handlers/_base.py | 6 | ||||
| -rw-r--r-- | synapse/handlers/federation.py | 34 | ||||
| -rw-r--r-- | synapse/handlers/room.py | 13 | ||||
| -rw-r--r-- | synapse/push/pusherpool.py | 1 | ||||
| -rw-r--r-- | synapse/storage/event_push_actions.py | 59 |
6 files changed, 93 insertions, 33 deletions
diff --git a/synapse/api/auth.py b/synapse/api/auth.py index 4f9c3c9db8..9e912fdfbe 100644 --- a/synapse/api/auth.py +++ b/synapse/api/auth.py @@ -25,6 +25,7 @@ from synapse.api.errors import AuthError, Codes, SynapseError, EventSizeError from synapse.types import Requester, RoomID, UserID, EventID from synapse.util.logutils import log_function from synapse.util.logcontext import preserve_context_over_fn +from synapse.util.metrics import Measure from unpaddedbase64 import decode_base64 import logging @@ -44,6 +45,7 @@ class Auth(object): def __init__(self, hs): self.hs = hs + self.clock = hs.get_clock() self.store = hs.get_datastore() self.state = hs.get_state_handler() self.TOKEN_NOT_FOUND_HTTP_STATUS = 401 @@ -66,9 +68,9 @@ class Auth(object): Returns: True if the auth checks pass. """ - self.check_size_limits(event) + with Measure(self.clock, "auth.check"): + self.check_size_limits(event) - try: if not hasattr(event, "room_id"): raise AuthError(500, "Event has no room_id: %s" % event) if auth_events is None: @@ -127,13 +129,6 @@ class Auth(object): self.check_redaction(event, auth_events) logger.debug("Allowing! %s", event) - except AuthError as e: - logger.info( - "Event auth check failed on event %s with msg: %s", - event, e.msg - ) - logger.info("Denying! %s", event) - raise def check_size_limits(self, event): def too_big(field): diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index aaf6b1b837..13a675b208 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -316,7 +316,11 @@ class BaseHandler(object): if ratelimit: self.ratelimit(requester) - self.auth.check(event, auth_events=context.current_state) + try: + self.auth.check(event, auth_events=context.current_state) + except AuthError as err: + logger.warn("Denying new event %r because %s", event, err) + raise err yield self.maybe_kick_guest_users(event, context.current_state.values()) diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 5ac55e10f3..d95e0b23b1 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -681,9 +681,13 @@ class FederationHandler(BaseHandler): "state_key": user_id, }) - event, context = yield self._create_new_client_event( - builder=builder, - ) + try: + event, context = yield self._create_new_client_event( + builder=builder, + ) + except AuthError as e: + logger.warn("Failed to create join %r because %s", event, e) + raise e self.auth.check(event, auth_events=context.current_state) @@ -915,7 +919,11 @@ class FederationHandler(BaseHandler): builder=builder, ) - self.auth.check(event, auth_events=context.current_state) + try: + self.auth.check(event, auth_events=context.current_state) + except AuthError as e: + logger.warn("Failed to create new leave %r because %s", event, e) + raise e defer.returnValue(event) @@ -1512,8 +1520,9 @@ class FederationHandler(BaseHandler): try: self.auth.check(event, auth_events=auth_events) - except AuthError: - raise + except AuthError as e: + logger.warn("Failed auth resolution for %r because %s", event, e) + raise e @defer.inlineCallbacks def construct_auth_difference(self, local_auth, remote_auth): @@ -1689,7 +1698,12 @@ class FederationHandler(BaseHandler): event_dict, event, context ) - self.auth.check(event, context.current_state) + try: + self.auth.check(event, context.current_state) + except AuthError as e: + logger.warn("Denying new third party invite %r because %s", event, e) + raise e + yield self._check_signature(event, auth_events=context.current_state) member_handler = self.hs.get_handlers().room_member_handler yield member_handler.send_membership_event(None, event, context) @@ -1714,7 +1728,11 @@ class FederationHandler(BaseHandler): event_dict, event, context ) - self.auth.check(event, auth_events=context.current_state) + try: + self.auth.check(event, auth_events=context.current_state) + except AuthError as e: + logger.warn("Denying third party invite %r because %s", event, e) + raise e yield self._check_signature(event, auth_events=context.current_state) returned_invite = yield self.send_invite(origin, event) diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index ea306cd42a..dd9c18df84 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -358,8 +358,6 @@ class RoomListHandler(BaseHandler): @defer.inlineCallbacks def handle_room(room_id): - aliases = yield self.store.get_aliases_for_room(room_id) - # We pull each bit of state out indvidually to avoid pulling the # full state into memory. Due to how the caching works this should # be fairly quick, even if not originally in the cache. @@ -374,6 +372,14 @@ class RoomListHandler(BaseHandler): defer.returnValue(None) result = {"room_id": room_id} + + joined_users = yield self.store.get_users_in_room(room_id) + if len(joined_users) == 0: + return + + result["num_joined_members"] = len(joined_users) + + aliases = yield self.store.get_aliases_for_room(room_id) if aliases: result["aliases"] = aliases @@ -413,9 +419,6 @@ class RoomListHandler(BaseHandler): if avatar_url: result["avatar_url"] = avatar_url - joined_users = yield self.store.get_users_in_room(room_id) - result["num_joined_members"] = len(joined_users) - results.append(result) yield concurrently_execute(handle_room, room_ids, 10) diff --git a/synapse/push/pusherpool.py b/synapse/push/pusherpool.py index ba513601e7..aa095f9d9b 100644 --- a/synapse/push/pusherpool.py +++ b/synapse/push/pusherpool.py @@ -32,7 +32,6 @@ class PusherPool: self.store = self.hs.get_datastore() self.clock = self.hs.get_clock() self.pushers = {} - self.last_pusher_started = -1 @defer.inlineCallbacks def start(self): diff --git a/synapse/storage/event_push_actions.py b/synapse/storage/event_push_actions.py index 355478957d..88737b7a6c 100644 --- a/synapse/storage/event_push_actions.py +++ b/synapse/storage/event_push_actions.py @@ -116,26 +116,67 @@ class EventPushActionsStore(SQLBaseStore): def get_unread_push_actions_for_user_in_range(self, user_id, min_stream_ordering, max_stream_ordering=None): - def f(txn): + def get_after_receipt(txn): sql = ( - "SELECT event_id, stream_ordering, actions" - " FROM event_push_actions" - " WHERE user_id = ? AND stream_ordering > ?" + "SELECT ep.event_id, ep.stream_ordering, ep.actions " + "FROM event_push_actions AS ep, (" + " SELECT room_id, user_id," + " max(topological_ordering) as topological_ordering," + " max(stream_ordering) as stream_ordering" + " FROM events" + " NATURAL JOIN receipts_linearized WHERE receipt_type = 'm.read'" + " GROUP BY room_id, user_id" + ") AS rl " + "WHERE" + " ep.room_id = rl.room_id" + " AND (" + " ep.topological_ordering > rl.topological_ordering" + " OR (" + " ep.topological_ordering = rl.topological_ordering" + " AND ep.stream_ordering > ?" + " )" + " )" + " AND ep.user_id = ?" + " AND ep.user_id = rl.user_id" ) - args = [user_id, min_stream_ordering] + args = [min_stream_ordering, user_id] if max_stream_ordering is not None: - sql += " AND stream_ordering <= ?" + sql += " AND ep.stream_ordering <= ?" args.append(max_stream_ordering) - sql += " ORDER BY stream_ordering ASC" + sql += " ORDER BY ep.stream_ordering ASC" txn.execute(sql, args) return txn.fetchall() - ret = yield self.runInteraction("get_unread_push_actions_for_user_in_range", f) + after_read_receipt = yield self.runInteraction( + "get_unread_push_actions_for_user_in_range", get_after_receipt + ) + + def get_no_receipt(txn): + sql = ( + "SELECT ep.event_id, ep.stream_ordering, ep.actions " + "FROM event_push_actions AS ep " + "WHERE ep.room_id not in (" + " SELECT room_id FROM events NATURAL JOIN receipts_linearized" + " WHERE receipt_type = 'm.read' AND user_id = ? " + " GROUP BY room_id" + ") AND ep.user_id = ? AND ep.stream_ordering > ?" + ) + args = [user_id, user_id, min_stream_ordering] + if max_stream_ordering is not None: + sql += " AND ep.stream_ordering <= ?" + args.append(max_stream_ordering) + sql += " ORDER BY ep.stream_ordering ASC" + txn.execute(sql, args) + return txn.fetchall() + no_read_receipt = yield self.runInteraction( + "get_unread_push_actions_for_user_in_range", get_no_receipt + ) + defer.returnValue([ { "event_id": row[0], "stream_ordering": row[1], "actions": json.loads(row[2]), - } for row in ret + } for row in after_read_receipt + no_read_receipt ]) @defer.inlineCallbacks |