Merge branch 'develop' of github.com:matrix-org/synapse into matrix-org-hotfixes

AUTHORS.rst

@@ -69,3 +69,6 @@ Serban Constantin <serban.constantin at gmail dot com>
 
 Jason Robinson <jasonr at matrix.org>
  * Minor fixes
+
+Joseph Weston <joseph at weston.cloud>
+ + Add admin API for querying HS version

changelog.d/4772.feature

@@ -0,0 +1 @@
+Add an endpoint to the admin API for querying the server version. Contributed by Joseph Weston.

changelog.d/4792.bugfix

@@ -0,0 +1 @@
+Handle batch updates in worker replication protocol.

changelog.d/4804.feature

@@ -0,0 +1 @@
+Add configurable rate limiting to the /register endpoint.

changelog.d/4815.misc

@@ -0,0 +1 @@
+Add some docstrings.

changelog.d/4816.misc

@@ -0,0 +1 @@
+Add debug logger to try and track down #4422.

changelog.d/4817.misc

@@ -0,0 +1 @@
+Make shutdown API send explanation message to room after users have been forced joined.

changelog.d/4818.bugfix

@@ -0,0 +1 @@
+Fix bug where we didn't correctly throttle sending of USER_IP commands over replication.

docs/admin_api/version_api.rst

@@ -0,0 +1,22 @@
+Version API
+===========
+
+This API returns the running Synapse version and the Python version
+on which Synapse is being run. This is useful when a Synapse instance
+is behind a proxy that does not forward the 'Server' header (which also
+contains Synapse version information).
+
+The api is::
+
+    GET /_matrix/client/r0/admin/server_version
+
+including an ``access_token`` of a server admin.
+
+It returns a JSON body like the following:
+
+.. code:: json
+
+    {
+        "server_version": "0.99.2rc1 (b=develop, abcdef123)",
+        "python_version": "3.6.8"
+    }

docs/sample_config.yaml

@@ -393,6 +393,17 @@ federation_rc_reject_limit: 50
 #
 federation_rc_concurrent: 3
 
+# Number of registration requests a client can send per second.
+# Defaults to 1/minute (0.17).
+#
+#rc_registration_requests_per_second: 0.17
+
+# Number of registration requests a client can send before being
+# throttled.
+# Defaults to 3.
+#
+#rc_registration_request_burst_count: 3.0
+
 
 
 # Directory where uploaded images and attachments are stored.
@@ -580,6 +591,8 @@ turn_allow_guests: True
 
 
 ## Registration ##
+# Registration can be rate-limited using the parameters in the "Ratelimiting"
+# section of this file.
 
 # Enable registration for new users.
 enable_registration: False
@@ -657,17 +670,6 @@ trusted_third_party_id_servers:
 #
 autocreate_auto_join_rooms: true
 
-# Number of registration requests a client can send per second.
-# Defaults to 1/minute (0.17).
-#
-#rc_registration_requests_per_second: 0.17
-
-# Number of registration requests a client can send before being
-# throttled.
-# Defaults to 3.
-#
-#rc_registration_request_burst_count: 3.0
-
 
 ## Metrics ###
 

synapse/config/ratelimiting.py

@@ -27,6 +27,13 @@ class RatelimitConfig(Config):
         self.federation_rc_reject_limit = config["federation_rc_reject_limit"]
         self.federation_rc_concurrent = config["federation_rc_concurrent"]
 
+        self.rc_registration_requests_per_second = config.get(
+            "rc_registration_requests_per_second", 0.17,
+        )
+        self.rc_registration_request_burst_count = config.get(
+            "rc_registration_request_burst_count", 3,
+        )
+
     def default_config(self, **kwargs):
         return """\
         ## Ratelimiting ##
@@ -62,4 +69,15 @@ class RatelimitConfig(Config):
         # single server
         #
         federation_rc_concurrent: 3
+
+        # Number of registration requests a client can send per second.
+        # Defaults to 1/minute (0.17).
+        #
+        #rc_registration_requests_per_second: 0.17
+
+        # Number of registration requests a client can send before being
+        # throttled.
+        # Defaults to 3.
+        #
+        #rc_registration_request_burst_count: 3.0
         """

synapse/config/registration.py

@@ -54,13 +54,6 @@ class RegistrationConfig(Config):
             config.get("disable_msisdn_registration", False)
         )
 
-        self.rc_registration_requests_per_second = config.get(
-            "rc_registration_requests_per_second", 0.17,
-        )
-        self.rc_registration_request_burst_count = config.get(
-            "rc_registration_request_burst_count", 3,
-        )
-
     def default_config(self, generate_secrets=False, **kwargs):
         if generate_secrets:
             registration_shared_secret = 'registration_shared_secret: "%s"' % (
@@ -71,6 +64,8 @@ class RegistrationConfig(Config):
 
         return """\
         ## Registration ##
+        # Registration can be rate-limited using the parameters in the "Ratelimiting"
+        # section of this file.
 
         # Enable registration for new users.
         enable_registration: False
@@ -147,17 +142,6 @@ class RegistrationConfig(Config):
         # users cannot be auto-joined since they do not exist.
         #
         autocreate_auto_join_rooms: true
-
-        # Number of registration requests a client can send per second.
-        # Defaults to 1/minute (0.17).
-        #
-        #rc_registration_requests_per_second: 0.17
-
-        # Number of registration requests a client can send before being
-        # throttled.
-        # Defaults to 3.
-        #
-        #rc_registration_request_burst_count: 3.0
         """ % locals()
 
     def add_arguments(self, parser):

synapse/handlers/register.py

@@ -61,7 +61,7 @@ class RegistrationHandler(BaseHandler):
         self.user_directory_handler = hs.get_user_directory_handler()
         self.captcha_client = CaptchaServerHttpClient(hs)
         self.identity_handler = self.hs.get_handlers().identity_handler
-        self.ratelimiter = hs.get_ratelimiter()
+        self.ratelimiter = hs.get_registration_ratelimiter()
 
         self._next_generated_user_id = None
 

synapse/handlers/sync.py

@@ -39,6 +39,9 @@ from synapse.visibility import filter_events_for_client
 
 logger = logging.getLogger(__name__)
 
+# Debug logger for https://github.com/matrix-org/synapse/issues/4422
+issue4422_logger = logging.getLogger("synapse.handler.sync.4422_debug")
+
 SYNC_RESPONSE_CACHE_MS = 2 * 60 * 1000
 
 # Counts the number of times we returned a non-empty sync. `type` is one of
@@ -969,7 +972,7 @@ class SyncHandler(object):
         for joined_room in sync_result_builder.joined:
             room_id = joined_room.room_id
             if room_id in newly_joined_rooms:
-                logger.info(
+                issue4422_logger.debug(
                     "Sync result for newly joined room %s: %r",
                     room_id, joined_room,
                 )
@@ -1443,7 +1446,7 @@ class SyncHandler(object):
                     prev_membership = None
                     if old_mem_ev:
                         prev_membership = old_mem_ev.membership
-                    logger.info(
+                    issue4422_logger.debug(
                         "Previous membership for room %s with join: %s (event %s)",
                         room_id, prev_membership, old_mem_ev_id,
                     )
@@ -1570,7 +1573,7 @@ class SyncHandler(object):
 
             if newly_joined:
                 # debugging for https://github.com/matrix-org/synapse/issues/4422
-                logger.info(
+                issue4422_logger.debug(
                     "RoomSyncResultBuilder events for newly joined room %s: %r",
                     room_id, entry.events,
                 )
@@ -1697,7 +1700,7 @@ class SyncHandler(object):
 
         if newly_joined:
             # debug for https://github.com/matrix-org/synapse/issues/4422
-            logger.info(
+            issue4422_logger.debug(
                 "Timeline events after filtering in newly-joined room %s: %r",
                 room_id, batch,
             )
@@ -1936,18 +1939,31 @@ class SyncResultBuilder(object):
     """Used to help build up a new SyncResult for a user
 
     Attributes:
-        joined (list[JoinedSyncResult]):
+        sync_config (SyncConfig)
-        archived (list[ArchivedSyncResult]):
+        full_state (bool)
+        since_token (StreamToken)
+        now_token (StreamToken)
+        joined_room_ids (list[str])
+
+        # The following mirror the fields in a sync response
+        presence (list)
+        account_data (list)
+        joined (list[JoinedSyncResult])
+        invited (list[InvitedSyncResult])
+        archived (list[ArchivedSyncResult])
+        device (list)
+        groups (GroupsSyncResult|None)
+        to_device (list)
     """
     def __init__(self, sync_config, full_state, since_token, now_token,
                  joined_room_ids):
         """
         Args:
-            sync_config(SyncConfig)
+            sync_config (SyncConfig)
-            full_state(bool): The full_state flag as specified by user
+            full_state (bool): The full_state flag as specified by user
-            since_token(StreamToken): The token supplied by user, or None.
+            since_token (StreamToken): The token supplied by user, or None.
-            now_token(StreamToken): The token to sync up to.
+            now_token (StreamToken): The token to sync up to.
-
+            joined_room_ids (list[str]): List of rooms the user is joined to
         """
         self.sync_config = sync_config
         self.full_state = full_state
@@ -1975,8 +1991,8 @@ class RoomSyncResultBuilder(object):
         Args:
             room_id(str)
             rtype(str): One of `"joined"` or `"archived"`
-            events(list[FrozenEvent]): List of events to include in the room (more events
+            events(list[FrozenEvent]): List of events to include in the room
-                may be added when generating result).
+                (more events may be added when generating result).
             newly_joined(bool): If the user has newly joined the room
             full_state(bool): Whether the full state should be sent in result
             since_token(StreamToken): Earliest point to return events from, or None

synapse/replication/tcp/protocol.py

@@ -451,7 +451,7 @@ class ServerReplicationStreamProtocol(BaseReplicationStreamProtocol):
 
     @defer.inlineCallbacks
     def subscribe_to_stream(self, stream_name, token):
-        """Subscribe the remote to a streams.
+        """Subscribe the remote to a stream.
 
         This invloves checking if they've missed anything and sending those
         updates down if they have. During that time new updates for the stream
@@ -478,11 +478,36 @@ class ServerReplicationStreamProtocol(BaseReplicationStreamProtocol):
 
             # Now we can send any updates that came in while we were subscribing
             pending_rdata = self.pending_rdata.pop(stream_name, [])
+            updates = []
             for token, update in pending_rdata:
-                # Only send updates newer than the current token
+                # If the token is null, it is part of a batch update. Batches
-                if token > current_token:
+                # are multiple updates that share a single token. To denote
+                # this, the token is set to None for all tokens in the batch
+                # except for the last. If we find a None token, we keep looking
+                # through tokens until we find one that is not None and then
+                # process all previous updates in the batch as if they had the
+                # final token.
+                if token is None:
+                    # Store this update as part of a batch
+                    updates.append(update)
+                    continue
+
+                if token <= current_token:
+                    # This update or batch of updates is older than
+                    # current_token, dismiss it
+                    updates = []
+                    continue
+
+                updates.append(update)
+
+                # Send all updates that are part of this batch with the
+                # found token
+                for update in updates:
                     self.send_command(RdataCommand(stream_name, token, update))
 
+                # Clear stored updates
+                updates = []
+
             # They're now fully subscribed
             self.replication_streams.add(stream_name)
         except Exception as e:

synapse/rest/client/v1/admin.py

@@ -17,12 +17,14 @@
 import hashlib
 import hmac
 import logging
+import platform
 
 from six import text_type
 from six.moves import http_client
 
 from twisted.internet import defer
 
+import synapse
 from synapse.api.constants import Membership, UserTypes
 from synapse.api.errors import AuthError, Codes, NotFoundError, SynapseError
 from synapse.http.servlet import (
@@ -32,6 +34,7 @@ from synapse.http.servlet import (
     parse_string,
 )
 from synapse.types import UserID, create_requester
+from synapse.util.versionstring import get_version_string
 
 from .base import ClientV1RestServlet, client_path_patterns
 
@@ -66,6 +69,25 @@ class UsersRestServlet(ClientV1RestServlet):
         defer.returnValue((200, ret))
 
 
+class VersionServlet(ClientV1RestServlet):
+    PATTERNS = client_path_patterns("/admin/server_version")
+
+    @defer.inlineCallbacks
+    def on_GET(self, request):
+        requester = yield self.auth.get_user_by_req(request)
+        is_admin = yield self.auth.is_server_admin(requester.user)
+
+        if not is_admin:
+            raise AuthError(403, "You are not a server admin")
+
+        ret = {
+            'server_version': get_version_string(synapse),
+            'python_version': platform.python_version(),
+        }
+
+        defer.returnValue((200, ret))
+
+
 class UserRegisterServlet(ClientV1RestServlet):
     """
     Attributes:
@@ -763,3 +785,4 @@ def register_servlets(hs, http_server):
     QuarantineMediaInRoom(hs).register(http_server)
     ListMediaInRoom(hs).register(http_server)
     UserRegisterServlet(hs).register(http_server)
+    VersionServlet(hs).register(http_server)

synapse/rest/client/v2_alpha/register.py

@@ -196,7 +196,7 @@ class RegisterRestServlet(RestServlet):
         self.identity_handler = hs.get_handlers().identity_handler
         self.room_member_handler = hs.get_room_member_handler()
         self.macaroon_gen = hs.get_macaroon_generator()
-        self.ratelimiter = hs.get_ratelimiter()
+        self.ratelimiter = hs.get_registration_ratelimiter()
         self.clock = hs.get_clock()
 
     @interactive_auth_handler

synapse/server.py

@@ -206,6 +206,7 @@ class HomeServer(object):
         self.clock = Clock(reactor)
         self.distributor = Distributor()
         self.ratelimiter = Ratelimiter()
+        self.registration_ratelimiter = Ratelimiter()
 
         self.datastore = None
 
@@ -251,6 +252,9 @@ class HomeServer(object):
     def get_ratelimiter(self):
         return self.ratelimiter
 
+    def get_registration_ratelimiter(self):
+        return self.registration_ratelimiter
+
     def build_federation_client(self):
         return FederationClient(self)
 

synapse/storage/stream.py

@@ -191,14 +191,18 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
     @defer.inlineCallbacks
     def get_room_events_stream_for_rooms(self, room_ids, from_key, to_key, limit=0,
                                          order='DESC'):
-        """
+        """Get new room events in stream ordering since `from_key`.
 
         Args:
-            room_ids:
+            room_id (str)
-            from_key:
+            from_key (str): Token from which no events are returned before
-            to_key:
+            to_key (str): Token from which no events are returned after. (This
-            limit:
+                is typically the current stream token)
-            order:
+            limit (int): Maximum number of events to return
+            order (str): Either "DESC" or "ASC". Determines which events are
+                returned when the result is limited. If "DESC" then the most
+                recent `limit` events are returned, otherwise returns the
+                oldest `limit` events.
 
         Returns:
             Deferred[dict[str,tuple[list[FrozenEvent], str]]]

tests/rest/client/v1/test_admin.py

@@ -20,14 +20,48 @@ import json
 from mock import Mock
 
 from synapse.api.constants import UserTypes
-from synapse.rest.client.v1.admin import register_servlets
+from synapse.rest.client.v1 import admin, login
 
 from tests import unittest
 
 
+class VersionTestCase(unittest.HomeserverTestCase):
+
+    servlets = [
+        admin.register_servlets,
+        login.register_servlets,
+    ]
+
+    url = '/_matrix/client/r0/admin/server_version'
+
+    def test_version_string(self):
+        self.register_user("admin", "pass", admin=True)
+        self.admin_token = self.login("admin", "pass")
+
+        request, channel = self.make_request("GET", self.url,
+                                             access_token=self.admin_token)
+        self.render(request)
+
+        self.assertEqual(200, int(channel.result["code"]),
+                         msg=channel.result["body"])
+        self.assertEqual({'server_version', 'python_version'},
+                         set(channel.json_body.keys()))
+
+    def test_inaccessible_to_non_admins(self):
+        self.register_user("unprivileged-user", "pass", admin=False)
+        user_token = self.login("unprivileged-user", "pass")
+
+        request, channel = self.make_request("GET", self.url,
+                                             access_token=user_token)
+        self.render(request)
+
+        self.assertEqual(403, int(channel.result['code']),
+                         msg=channel.result['body'])
+
+
 class UserRegisterTestCase(unittest.HomeserverTestCase):
 
-    servlets = [register_servlets]
+    servlets = [admin.register_servlets]
 
     def make_homeserver(self, reactor, clock):