Merge branch 'develop' into e2e_backups

3 files changed, 254 insertions, 24 deletions

diff --git a/synapse/server_notices/resource_limits_server_notices.py b/synapse/server_notices/resource_limits_server_notices.py
new file mode 100644
index 0000000000..893b9001cd
--- /dev/null
+++ b/synapse/server_notices/resource_limits_server_notices.py
@@ -0,0 +1,204 @@
+# -*- coding: utf-8 -*-
+# 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.
+import logging
+
+from six import iteritems
+
+from twisted.internet import defer
+
+from synapse.api.constants import (
+    EventTypes,
+    ServerNoticeLimitReached,
+    ServerNoticeMsgType,
+)
+from synapse.api.errors import AuthError, ResourceLimitError, SynapseError
+from synapse.server_notices.server_notices_manager import SERVER_NOTICE_ROOM_TAG
+
+logger = logging.getLogger(__name__)
+
+
+class ResourceLimitsServerNotices(object):
+    """ Keeps track of whether the server has reached it's resource limit and
+    ensures that the client is kept up to date.
+    """
+    def __init__(self, hs):
+        """
+        Args:
+            hs (synapse.server.HomeServer):
+        """
+        self._server_notices_manager = hs.get_server_notices_manager()
+        self._store = hs.get_datastore()
+        self._auth = hs.get_auth()
+        self._config = hs.config
+        self._resouce_limited = False
+        self._message_handler = hs.get_message_handler()
+        self._state = hs.get_state_handler()
+
+        self._notifier = hs.get_notifier()
+
+    @defer.inlineCallbacks
+    def maybe_send_server_notice_to_user(self, user_id):
+        """Check if we need to send a notice to this user, this will be true in
+        two cases.
+        1. The server has reached its limit does not reflect this
+        2. The room state indicates that the server has reached its limit when
+        actually the server is fine
+
+        Args:
+            user_id (str): user to check
+
+        Returns:
+            Deferred
+        """
+        if self._config.hs_disabled is True:
+            return
+
+        if self._config.limit_usage_by_mau is False:
+            return
+
+        if not self._server_notices_manager.is_enabled():
+            # Don't try and send server notices unles they've been enabled
+            return
+
+        timestamp = yield self._store.user_last_seen_monthly_active(user_id)
+        if timestamp is None:
+            # This user will be blocked from receiving the notice anyway.
+            # In practice, not sure we can ever get here
+            return
+
+        # Determine current state of room
+
+        room_id = yield self._server_notices_manager.get_notice_room_for_user(user_id)
+
+        if not room_id:
+            logger.warn("Failed to get server notices room")
+            return
+
+        yield self._check_and_set_tags(user_id, room_id)
+        currently_blocked, ref_events = yield self._is_room_currently_blocked(room_id)
+
+        try:
+            # Normally should always pass in user_id if you have it, but in
+            # this case are checking what would happen to other users if they
+            # were to arrive.
+            try:
+                yield self._auth.check_auth_blocking()
+                is_auth_blocking = False
+            except ResourceLimitError as e:
+                is_auth_blocking = True
+                event_content = e.msg
+                event_limit_type = e.limit_type
+
+            if currently_blocked and not is_auth_blocking:
+                # Room is notifying of a block, when it ought not to be.
+                # Remove block notification
+                content = {
+                    "pinned": ref_events
+                }
+                yield self._server_notices_manager.send_notice(
+                    user_id, content, EventTypes.Pinned, '',
+                )
+
+            elif not currently_blocked and is_auth_blocking:
+                # Room is not notifying of a block, when it ought to be.
+                # Add block notification
+                content = {
+                    'body': event_content,
+                    'msgtype': ServerNoticeMsgType,
+                    'server_notice_type': ServerNoticeLimitReached,
+                    'admin_uri': self._config.admin_uri,
+                    'limit_type': event_limit_type
+                }
+                event = yield self._server_notices_manager.send_notice(
+                    user_id, content, EventTypes.Message,
+                )
+
+                content = {
+                    "pinned": [
+                        event.event_id,
+                    ]
+                }
+                yield self._server_notices_manager.send_notice(
+                    user_id, content, EventTypes.Pinned, '',
+                )
+
+        except SynapseError as e:
+            logger.error("Error sending resource limits server notice: %s", e)
+
+    @defer.inlineCallbacks
+    def _check_and_set_tags(self, user_id, room_id):
+        """
+        Since server notices rooms were originally not with tags,
+        important to check that tags have been set correctly
+        Args:
+            user_id(str): the user in question
+            room_id(str): the server notices room for that user
+        """
+        tags = yield self._store.get_tags_for_user(user_id)
+        server_notices_tags = tags.get(room_id)
+        need_to_set_tag = True
+        if server_notices_tags:
+            if server_notices_tags.get(SERVER_NOTICE_ROOM_TAG):
+                # tag already present, nothing to do here
+                need_to_set_tag = False
+        if need_to_set_tag:
+            max_id = yield self._store.add_tag_to_room(
+                user_id, room_id, SERVER_NOTICE_ROOM_TAG, {}
+            )
+            self._notifier.on_new_event(
+                "account_data_key", max_id, users=[user_id]
+            )
+
+    @defer.inlineCallbacks
+    def _is_room_currently_blocked(self, room_id):
+        """
+        Determines if the room is currently blocked
+
+        Args:
+            room_id(str): The room id of the server notices room
+
+        Returns:
+
+            bool: Is the room currently blocked
+            list: The list of pinned events that are unrelated to limit blocking
+            This list can be used as a convenience in the case where the block
+            is to be lifted and the remaining pinned event references need to be
+            preserved
+        """
+        currently_blocked = False
+        pinned_state_event = None
+        try:
+            pinned_state_event = yield self._state.get_current_state(
+                room_id, event_type=EventTypes.Pinned
+            )
+        except AuthError:
+            # The user has yet to join the server notices room
+            pass
+
+        referenced_events = []
+        if pinned_state_event is not None:
+            referenced_events = list(pinned_state_event.content.get('pinned', []))
+
+        events = yield self._store.get_events(referenced_events)
+        for event_id, event in iteritems(events):
+            if event.type != EventTypes.Message:
+                continue
+            if event.content.get("msgtype") == ServerNoticeMsgType:
+                currently_blocked = True
+                # remove event in case we need to disable blocking later on.
+                if event_id in referenced_events:
+                    referenced_events.remove(event.event_id)
+
+        defer.returnValue((currently_blocked, referenced_events))
diff --git a/synapse/server_notices/server_notices_manager.py b/synapse/server_notices/server_notices_manager.py
index a26deace53..c5cc6d728e 100644
--- a/synapse/server_notices/server_notices_manager.py
+++ b/synapse/server_notices/server_notices_manager.py
@@ -22,6 +22,8 @@ from synapse.util.caches.descriptors import cachedInlineCallbacks
 
 logger = logging.getLogger(__name__)
 
+SERVER_NOTICE_ROOM_TAG = "m.server_notice"
+
 
 class ServerNoticesManager(object):
     def __init__(self, hs):
@@ -37,6 +39,8 @@ class ServerNoticesManager(object):
         self._event_creation_handler = hs.get_event_creation_handler()
         self._is_mine_id = hs.is_mine_id
 
+        self._notifier = hs.get_notifier()
+
     def is_enabled(self):
         """Checks if server notices are enabled on this server.
 
@@ -46,7 +50,10 @@ class ServerNoticesManager(object):
         return self._config.server_notices_mxid is not None
 
     @defer.inlineCallbacks
-    def send_notice(self, user_id, event_content):
+    def send_notice(
+        self, user_id, event_content,
+        type=EventTypes.Message, state_key=None
+    ):
         """Send a notice to the given user
 
         Creates the server notices room, if none exists.
@@ -54,9 +61,11 @@ class ServerNoticesManager(object):
         Args:
             user_id (str): mxid of user to send event to.
             event_content (dict): content of event to send
+            type(EventTypes): type of event
+            is_state_event(bool): Is the event a state event
 
         Returns:
-            Deferred[None]
+            Deferred[FrozenEvent]
         """
         room_id = yield self.get_notice_room_for_user(user_id)
 
@@ -65,15 +74,20 @@ class ServerNoticesManager(object):
 
         logger.info("Sending server notice to %s", user_id)
 
-        yield self._event_creation_handler.create_and_send_nonmember_event(
-            requester, {
-                "type": EventTypes.Message,
-                "room_id": room_id,
-                "sender": system_mxid,
-                "content": event_content,
-            },
-            ratelimit=False,
+        event_dict = {
+            "type": type,
+            "room_id": room_id,
+            "sender": system_mxid,
+            "content": event_content,
+        }
+
+        if state_key is not None:
+            event_dict['state_key'] = state_key
+
+        res = yield self._event_creation_handler.create_and_send_nonmember_event(
+            requester, event_dict, ratelimit=False,
         )
+        defer.returnValue(res)
 
     @cachedInlineCallbacks()
     def get_notice_room_for_user(self, user_id):
@@ -142,5 +156,12 @@ class ServerNoticesManager(object):
         )
         room_id = info['room_id']
 
+        max_id = yield self._store.add_tag_to_room(
+            user_id, room_id, SERVER_NOTICE_ROOM_TAG, {},
+        )
+        self._notifier.on_new_event(
+            "account_data_key", max_id, users=[user_id]
+        )
+
         logger.info("Created server notices room %s for %s", room_id, user_id)
         defer.returnValue(room_id)
diff --git a/synapse/server_notices/server_notices_sender.py b/synapse/server_notices/server_notices_sender.py
index 5d23965f34..6121b2f267 100644
--- a/synapse/server_notices/server_notices_sender.py
+++ b/synapse/server_notices/server_notices_sender.py
@@ -12,7 +12,12 @@
 # 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.
+from twisted.internet import defer
+
 from synapse.server_notices.consent_server_notices import ConsentServerNotices
+from synapse.server_notices.resource_limits_server_notices import (
+    ResourceLimitsServerNotices,
+)
 
 
 class ServerNoticesSender(object):
@@ -25,34 +30,34 @@ class ServerNoticesSender(object):
         Args:
             hs (synapse.server.HomeServer):
         """
-        # todo: it would be nice to make this more dynamic
-        self._consent_server_notices = ConsentServerNotices(hs)
+        self._server_notices = (
+            ConsentServerNotices(hs),
+            ResourceLimitsServerNotices(hs)
+        )
 
+    @defer.inlineCallbacks
     def on_user_syncing(self, user_id):
         """Called when the user performs a sync operation.
 
         Args:
             user_id (str): mxid of user who synced
-
-        Returns:
-            Deferred
         """
-        return self._consent_server_notices.maybe_send_server_notice_to_user(
-            user_id,
-        )
+        for sn in self._server_notices:
+            yield sn.maybe_send_server_notice_to_user(
+                user_id,
+            )
 
+    @defer.inlineCallbacks
     def on_user_ip(self, user_id):
         """Called on the master when a worker process saw a client request.
 
         Args:
             user_id (str): mxid
-
-        Returns:
-            Deferred
         """
         # The synchrotrons use a stubbed version of ServerNoticesSender, so
         # we check for notices to send to the user in on_user_ip as well as
         # in on_user_syncing
-        return self._consent_server_notices.maybe_send_server_notice_to_user(
-            user_id,
-        )
+        for sn in self._server_notices:
+            yield sn.maybe_send_server_notice_to_user(
+                user_id,
+            )