Fix registration on workers (#4682)

* Move RegistrationHandler init to HomeServer

* Move post registration actions to RegistrationHandler

* Add post regisration replication endpoint

* Newsfile

changelog.d/4682.feature

@@ -0,0 +1 @@
+Allow registration and login to be handled by a worker instance.

synapse/handlers/__init__.py

@@ -17,7 +17,6 @@ from .admin import AdminHandler
 from .directory import DirectoryHandler
 from .federation import FederationHandler
 from .identity import IdentityHandler
-from .register import RegistrationHandler
 from .search import SearchHandler
 
 
@@ -41,7 +40,6 @@ class Handlers(object):
     """
 
     def __init__(self, hs):
-        self.registration_handler = RegistrationHandler(hs)
         self.federation_handler = FederationHandler(hs)
         self.directory_handler = DirectoryHandler(hs)
         self.admin_handler = AdminHandler(hs)

synapse/handlers/register.py

@@ -19,6 +19,7 @@ import logging
 from twisted.internet import defer
 
 from synapse import types
+from synapse.api.constants import LoginType
 from synapse.api.errors import (
     AuthError,
     Codes,
@@ -26,9 +27,14 @@ from synapse.api.errors import (
     RegistrationError,
     SynapseError,
 )
+from synapse.config.server import is_threepid_reserved
 from synapse.http.client import CaptchaServerHttpClient
+from synapse.http.servlet import assert_params_in_dict
 from synapse.replication.http.login import RegisterDeviceReplicationServlet
-from synapse.replication.http.register import ReplicationRegisterServlet
+from synapse.replication.http.register import (
+    ReplicationPostRegisterActionsServlet,
+    ReplicationRegisterServlet,
+)
 from synapse.types import RoomAlias, RoomID, UserID, create_requester
 from synapse.util.async_helpers import Linearizer
 from synapse.util.threepids import check_3pid_allowed
@@ -53,6 +59,7 @@ class RegistrationHandler(BaseHandler):
         self.profile_handler = hs.get_profile_handler()
         self.user_directory_handler = hs.get_user_directory_handler()
         self.captcha_client = CaptchaServerHttpClient(hs)
+        self.identity_handler = self.hs.get_handlers().identity_handler
 
         self._next_generated_user_id = None
 
@@ -68,8 +75,12 @@ class RegistrationHandler(BaseHandler):
             self._register_device_client = (
                 RegisterDeviceReplicationServlet.make_client(hs)
             )
+            self._post_registration_client = (
+                ReplicationPostRegisterActionsServlet.make_client(hs)
+            )
         else:
             self.device_handler = hs.get_device_handler()
+            self.pusher_pool = hs.get_pusherpool()
 
     @defer.inlineCallbacks
     def check_username(self, localpart, guest_access_token=None,
@@ -369,8 +380,7 @@ class RegistrationHandler(BaseHandler):
             logger.info("validating threepidcred sid %s on id server %s",
                         c['sid'], c['idServer'])
             try:
-                identity_handler = self.hs.get_handlers().identity_handler
-                threepid = yield identity_handler.threepid_from_creds(c)
+                threepid = yield self.identity_handler.threepid_from_creds(c)
             except Exception:
                 logger.exception("Couldn't validate 3pid")
                 raise RegistrationError(400, "Couldn't validate 3pid")
@@ -394,9 +404,8 @@ class RegistrationHandler(BaseHandler):
 
         # Now we have a matrix ID, bind it to the threepids we were given
         for c in threepidCreds:
-            identity_handler = self.hs.get_handlers().identity_handler
             # XXX: This should be a deferred list, shouldn't it?
-            yield identity_handler.bind_threepid(c, user_id)
+            yield self.identity_handler.bind_threepid(c, user_id)
 
     def check_user_id_not_appservice_exclusive(self, user_id, allowed_appservice=None):
         # don't allow people to register the server notices mxid
@@ -671,3 +680,184 @@ class RegistrationHandler(BaseHandler):
                 )
 
             defer.returnValue((device_id, access_token))
+
+    @defer.inlineCallbacks
+    def post_registration_actions(self, user_id, auth_result, access_token,
+                                  bind_email, bind_msisdn):
+        """A user has completed registration
+
+        Args:
+            user_id (str): The user ID that consented
+            auth_result (dict): The authenticated credentials of the newly
+                registered user.
+            access_token (str|None): The access token of the newly logged in
+                device, or None if `inhibit_login` enabled.
+            bind_email (bool): Whether to bind the email with the identity
+                server
+            bind_msisdn (bool): Whether to bind the msisdn with the identity
+                server
+        """
+        if self.hs.config.worker_app:
+            yield self._post_registration_client(
+                user_id=user_id,
+                auth_result=auth_result,
+                access_token=access_token,
+                bind_email=bind_email,
+                bind_msisdn=bind_msisdn,
+            )
+            return
+
+        if auth_result and LoginType.EMAIL_IDENTITY in auth_result:
+            threepid = auth_result[LoginType.EMAIL_IDENTITY]
+            # Necessary due to auth checks prior to the threepid being
+            # written to the db
+            if is_threepid_reserved(
+                self.hs.config.mau_limits_reserved_threepids, threepid
+            ):
+                yield self.store.upsert_monthly_active_user(user_id)
+
+            yield self._register_email_threepid(
+                user_id, threepid, access_token,
+                bind_email,
+            )
+
+        if auth_result and LoginType.MSISDN in auth_result:
+            threepid = auth_result[LoginType.MSISDN]
+            yield self._register_msisdn_threepid(
+                user_id, threepid, bind_msisdn,
+            )
+
+        if auth_result and LoginType.TERMS in auth_result:
+            yield self._on_user_consented(
+                user_id, self.hs.config.user_consent_version,
+            )
+
+    @defer.inlineCallbacks
+    def _on_user_consented(self, user_id, consent_version):
+        """A user consented to the terms on registration
+
+        Args:
+            user_id (str): The user ID that consented
+            consent_version (str): version of the policy the user has
+                consented to.
+        """
+        logger.info("%s has consented to the privacy policy", user_id)
+        yield self.store.user_set_consent_version(
+            user_id, consent_version,
+        )
+        yield self.post_consent_actions(user_id)
+
+    @defer.inlineCallbacks
+    def _register_email_threepid(self, user_id, threepid, token, bind_email):
+        """Add an email address as a 3pid identifier
+
+        Also adds an email pusher for the email address, if configured in the
+        HS config
+
+        Also optionally binds emails to the given user_id on the identity server
+
+        Must be called on master.
+
+        Args:
+            user_id (str): id of user
+            threepid (object): m.login.email.identity auth response
+            token (str|None): access_token for the user, or None if not logged
+                in.
+            bind_email (bool): true if the client requested the email to be
+                bound at the identity server
+        Returns:
+            defer.Deferred:
+        """
+        reqd = ('medium', 'address', 'validated_at')
+        if any(x not in threepid for x in reqd):
+            # This will only happen if the ID server returns a malformed response
+            logger.info("Can't add incomplete 3pid")
+            return
+
+        yield self._auth_handler.add_threepid(
+            user_id,
+            threepid['medium'],
+            threepid['address'],
+            threepid['validated_at'],
+        )
+
+        # And we add an email pusher for them by default, but only
+        # if email notifications are enabled (so people don't start
+        # getting mail spam where they weren't before if email
+        # notifs are set up on a home server)
+        if (self.hs.config.email_enable_notifs and
+                self.hs.config.email_notif_for_new_users
+                and token):
+            # Pull the ID of the access token back out of the db
+            # It would really make more sense for this to be passed
+            # up when the access token is saved, but that's quite an
+            # invasive change I'd rather do separately.
+            user_tuple = yield self.store.get_user_by_access_token(
+                token
+            )
+            token_id = user_tuple["token_id"]
+
+            yield self.pusher_pool.add_pusher(
+                user_id=user_id,
+                access_token=token_id,
+                kind="email",
+                app_id="m.email",
+                app_display_name="Email Notifications",
+                device_display_name=threepid["address"],
+                pushkey=threepid["address"],
+                lang=None,  # We don't know a user's language here
+                data={},
+            )
+
+        if bind_email:
+            logger.info("bind_email specified: binding")
+            logger.debug("Binding emails %s to %s" % (
+                threepid, user_id
+            ))
+            yield self.identity_handler.bind_threepid(
+                threepid['threepid_creds'], user_id
+            )
+        else:
+            logger.info("bind_email not specified: not binding email")
+
+    @defer.inlineCallbacks
+    def _register_msisdn_threepid(self, user_id, threepid, bind_msisdn):
+        """Add a phone number as a 3pid identifier
+
+        Also optionally binds msisdn to the given user_id on the identity server
+
+        Must be called on master.
+
+        Args:
+            user_id (str): id of user
+            threepid (object): m.login.msisdn auth response
+            token (str): access_token for the user
+            bind_email (bool): true if the client requested the email to be
+                bound at the identity server
+        Returns:
+            defer.Deferred:
+        """
+        try:
+            assert_params_in_dict(threepid, ['medium', 'address', 'validated_at'])
+        except SynapseError as ex:
+            if ex.errcode == Codes.MISSING_PARAM:
+                # This will only happen if the ID server returns a malformed response
+                logger.info("Can't add incomplete 3pid")
+                defer.returnValue(None)
+            raise
+
+        yield self._auth_handler.add_threepid(
+            user_id,
+            threepid['medium'],
+            threepid['address'],
+            threepid['validated_at'],
+        )
+
+        if bind_msisdn:
+            logger.info("bind_msisdn specified: binding")
+            logger.debug("Binding msisdn %s to %s", threepid, user_id)
+            yield self.identity_handler.bind_threepid(
+                threepid['threepid_creds'], user_id
+            )
+        else:
+            logger.info("bind_msisdn not specified: not binding msisdn")

synapse/handlers/room_member.py

@@ -61,7 +61,7 @@ class RoomMemberHandler(object):
 
         self.federation_handler = hs.get_handlers().federation_handler
         self.directory_handler = hs.get_handlers().directory_handler
-        self.registration_handler = hs.get_handlers().registration_handler
+        self.registration_handler = hs.get_registration_handler()
         self.profile_handler = hs.get_profile_handler()
         self.event_creation_handler = hs.get_event_creation_handler()
 

synapse/module_api/__init__.py

@@ -79,7 +79,7 @@ class ModuleApi(object):
         Returns:
             Deferred: a 2-tuple of (user_id, access_token)
         """
-        reg = self.hs.get_handlers().registration_handler
+        reg = self.hs.get_registration_handler()
         return reg.register(localpart=localpart)
 
     @defer.inlineCallbacks

synapse/replication/http/login.py

@@ -35,7 +35,7 @@ class RegisterDeviceReplicationServlet(ReplicationEndpoint):
 
     def __init__(self, hs):
         super(RegisterDeviceReplicationServlet, self).__init__(hs)
-        self.registration_handler = hs.get_handlers().registration_handler
+        self.registration_handler = hs.get_registration_handler()
 
     @staticmethod
     def _serialize_payload(user_id, device_id, initial_display_name, is_guest):

synapse/replication/http/membership.py

@@ -191,7 +191,7 @@ class ReplicationRegister3PIDGuestRestServlet(ReplicationEndpoint):
     def __init__(self, hs):
         super(ReplicationRegister3PIDGuestRestServlet, self).__init__(hs)
 
-        self.registeration_handler = hs.get_handlers().registration_handler
+        self.registeration_handler = hs.get_registration_handler()
         self.store = hs.get_datastore()
         self.clock = hs.get_clock()
 
@@ -251,7 +251,7 @@ class ReplicationUserJoinedLeftRoomRestServlet(ReplicationEndpoint):
     def __init__(self, hs):
         super(ReplicationUserJoinedLeftRoomRestServlet, self).__init__(hs)
 
-        self.registeration_handler = hs.get_handlers().registration_handler
+        self.registeration_handler = hs.get_registration_handler()
         self.store = hs.get_datastore()
         self.clock = hs.get_clock()
         self.distributor = hs.get_distributor()

synapse/replication/http/register.py

@@ -87,5 +87,60 @@ class ReplicationRegisterServlet(ReplicationEndpoint):
         defer.returnValue((200, {}))
 
 
+class ReplicationPostRegisterActionsServlet(ReplicationEndpoint):
+    """Run any post registration actions
+    """
+
+    NAME = "post_register"
+    PATH_ARGS = ("user_id",)
+
+    def __init__(self, hs):
+        super(ReplicationPostRegisterActionsServlet, self).__init__(hs)
+        self.store = hs.get_datastore()
+        self.registration_handler = hs.get_registration_handler()
+
+    @staticmethod
+    def _serialize_payload(user_id, auth_result, access_token, bind_email,
+                           bind_msisdn):
+        """
+        Args:
+            user_id (str): The user ID that consented
+            auth_result (dict): The authenticated credentials of the newly
+                registered user.
+            access_token (str|None): The access token of the newly logged in
+                device, or None if `inhibit_login` enabled.
+            bind_email (bool): Whether to bind the email with the identity
+                server
+            bind_msisdn (bool): Whether to bind the msisdn with the identity
+                server
+        """
+        return {
+            "auth_result": auth_result,
+            "access_token": access_token,
+            "bind_email": bind_email,
+            "bind_msisdn": bind_msisdn,
+        }
+
+    @defer.inlineCallbacks
+    def _handle_request(self, request, user_id):
+        content = parse_json_object_from_request(request)
+
+        auth_result = content["auth_result"]
+        access_token = content["access_token"]
+        bind_email = content["bind_email"]
+        bind_msisdn = content["bind_msisdn"]
+
+        yield self.registration_handler.post_registration_actions(
+            user_id=user_id,
+            auth_result=auth_result,
+            access_token=access_token,
+            bind_email=bind_email,
+            bind_msisdn=bind_msisdn,
+        )
+
+        defer.returnValue((200, {}))
+
+
 def register_servlets(hs, http_server):
     ReplicationRegisterServlet(hs).register(http_server)
+    ReplicationPostRegisterActionsServlet(hs).register(http_server)

synapse/rest/client/v1/login.py

@@ -94,7 +94,7 @@ class LoginRestServlet(ClientV1RestServlet):
         self.jwt_algorithm = hs.config.jwt_algorithm
         self.cas_enabled = hs.config.cas_enabled
         self.auth_handler = self.hs.get_auth_handler()
-        self.registration_handler = hs.get_handlers().registration_handler
+        self.registration_handler = hs.get_registration_handler()
         self.handlers = hs.get_handlers()
         self._well_known_builder = WellKnownBuilder(hs)
 
@@ -434,7 +434,7 @@ class SSOAuthHandler(object):
     def __init__(self, hs):
         self._hostname = hs.hostname
         self._auth_handler = hs.get_auth_handler()
-        self._registration_handler = hs.get_handlers().registration_handler
+        self._registration_handler = hs.get_registration_handler()
         self._macaroon_gen = hs.get_macaroon_generator()
 
     @defer.inlineCallbacks

synapse/rest/client/v2_alpha/auth.py

@@ -129,7 +129,7 @@ class AuthRestServlet(RestServlet):
         self.hs = hs
         self.auth = hs.get_auth()
         self.auth_handler = hs.get_auth_handler()
-        self.registration_handler = hs.get_handlers().registration_handler
+        self.registration_handler = hs.get_registration_handler()
 
     def on_GET(self, request, stagetype):
         session = parse_string(request, "session")

synapse/rest/client/v2_alpha/register.py

@@ -145,7 +145,7 @@ class UsernameAvailabilityRestServlet(RestServlet):
         """
         super(UsernameAvailabilityRestServlet, self).__init__()
         self.hs = hs
-        self.registration_handler = hs.get_handlers().registration_handler
+        self.registration_handler = hs.get_registration_handler()
         self.ratelimiter = FederationRateLimiter(
             hs.get_clock(),
             # Time window of 2s
@@ -187,7 +187,7 @@ class RegisterRestServlet(RestServlet):
         self.auth = hs.get_auth()
         self.store = hs.get_datastore()
         self.auth_handler = hs.get_auth_handler()
-        self.registration_handler = hs.get_handlers().registration_handler
+        self.registration_handler = hs.get_registration_handler()
         self.identity_handler = hs.get_handlers().identity_handler
         self.room_member_handler = hs.get_room_member_handler()
         self.macaroon_gen = hs.get_macaroon_generator()
@@ -389,8 +389,7 @@ class RegisterRestServlet(RestServlet):
                 registered_user_id
             )
             # don't re-register the threepids
-            add_email = False
-            add_msisdn = False
+            registered = False
         else:
             # NB: This may be from the auth handler and NOT from the POST
             assert_params_in_dict(params, ["password"])
@@ -427,34 +426,21 @@ class RegisterRestServlet(RestServlet):
                 session_id, "registered_user_id", registered_user_id
             )
 
-            add_email = True
-            add_msisdn = True
+            registered = True
 
         return_dict = yield self._create_registration_details(
             registered_user_id, params
         )
 
-        if add_email and auth_result and LoginType.EMAIL_IDENTITY in auth_result:
-            threepid = auth_result[LoginType.EMAIL_IDENTITY]
-            yield self._register_email_threepid(
-                registered_user_id, threepid, return_dict["access_token"],
-                params.get("bind_email")
+        if registered:
+            yield self.registration_handler.post_registration_actions(
+                user_id=registered_user_id,
+                auth_result=auth_result,
+                access_token=return_dict.get("access_token"),
+                bind_email=params.get("bind_email"),
+                bind_msisdn=params.get("bind_msisdn"),
             )
 
-        if add_msisdn and auth_result and LoginType.MSISDN in auth_result:
-            threepid = auth_result[LoginType.MSISDN]
-            yield self._register_msisdn_threepid(
-                registered_user_id, threepid, return_dict["access_token"],
-                params.get("bind_msisdn")
-            )
-
-        if auth_result and LoginType.TERMS in auth_result:
-            logger.info("%s has consented to the privacy policy" % registered_user_id)
-            yield self.store.user_set_consent_version(
-                registered_user_id, self.hs.config.user_consent_version,
-            )
-            yield self.registration_handler.post_consent_actions(registered_user_id)
-
         defer.returnValue((200, return_dict))
 
     def on_OPTIONS(self, _):
@@ -505,115 +491,6 @@ class RegisterRestServlet(RestServlet):
         result = yield self._create_registration_details(user_id, body)
         defer.returnValue(result)
 
-    @defer.inlineCallbacks
-    def _register_email_threepid(self, user_id, threepid, token, bind_email):
-        """Add an email address as a 3pid identifier
-
-        Also adds an email pusher for the email address, if configured in the
-        HS config
-
-        Also optionally binds emails to the given user_id on the identity server
-
-        Args:
-            user_id (str): id of user
-            threepid (object): m.login.email.identity auth response
-            token (str): access_token for the user
-            bind_email (bool): true if the client requested the email to be
-                bound at the identity server
-        Returns:
-            defer.Deferred:
-        """
-        reqd = ('medium', 'address', 'validated_at')
-        if any(x not in threepid for x in reqd):
-            # This will only happen if the ID server returns a malformed response
-            logger.info("Can't add incomplete 3pid")
-            return
-
-        yield self.auth_handler.add_threepid(
-            user_id,
-            threepid['medium'],
-            threepid['address'],
-            threepid['validated_at'],
-        )
-
-        # And we add an email pusher for them by default, but only
-        # if email notifications are enabled (so people don't start
-        # getting mail spam where they weren't before if email
-        # notifs are set up on a home server)
-        if (self.hs.config.email_enable_notifs and
-                self.hs.config.email_notif_for_new_users):
-            # Pull the ID of the access token back out of the db
-            # It would really make more sense for this to be passed
-            # up when the access token is saved, but that's quite an
-            # invasive change I'd rather do separately.
-            user_tuple = yield self.store.get_user_by_access_token(
-                token
-            )
-            token_id = user_tuple["token_id"]
-
-            yield self.hs.get_pusherpool().add_pusher(
-                user_id=user_id,
-                access_token=token_id,
-                kind="email",
-                app_id="m.email",
-                app_display_name="Email Notifications",
-                device_display_name=threepid["address"],
-                pushkey=threepid["address"],
-                lang=None,  # We don't know a user's language here
-                data={},
-            )
-
-        if bind_email:
-            logger.info("bind_email specified: binding")
-            logger.debug("Binding emails %s to %s" % (
-                threepid, user_id
-            ))
-            yield self.identity_handler.bind_threepid(
-                threepid['threepid_creds'], user_id
-            )
-        else:
-            logger.info("bind_email not specified: not binding email")
-
-    @defer.inlineCallbacks
-    def _register_msisdn_threepid(self, user_id, threepid, token, bind_msisdn):
-        """Add a phone number as a 3pid identifier
-
-        Also optionally binds msisdn to the given user_id on the identity server
-
-        Args:
-            user_id (str): id of user
-            threepid (object): m.login.msisdn auth response
-            token (str): access_token for the user
-            bind_email (bool): true if the client requested the email to be
-                bound at the identity server
-        Returns:
-            defer.Deferred:
-        """
-        try:
-            assert_params_in_dict(threepid, ['medium', 'address', 'validated_at'])
-        except SynapseError as ex:
-            if ex.errcode == Codes.MISSING_PARAM:
-                # This will only happen if the ID server returns a malformed response
-                logger.info("Can't add incomplete 3pid")
-                defer.returnValue(None)
-            raise
-
-        yield self.auth_handler.add_threepid(
-            user_id,
-            threepid['medium'],
-            threepid['address'],
-            threepid['validated_at'],
-        )
-
-        if bind_msisdn:
-            logger.info("bind_msisdn specified: binding")
-            logger.debug("Binding msisdn %s to %s", threepid, user_id)
-            yield self.identity_handler.bind_threepid(
-                threepid['threepid_creds'], user_id
-            )
-        else:
-            logger.info("bind_msisdn not specified: not binding msisdn")
-
     @defer.inlineCallbacks
     def _create_registration_details(self, user_id, params):
         """Complete registration of newly-registered user

synapse/rest/consent/consent_resource.py

@@ -89,7 +89,7 @@ class ConsentResource(Resource):
 
         self.hs = hs
         self.store = hs.get_datastore()
-        self.registration_handler = hs.get_handlers().registration_handler
+        self.registration_handler = hs.get_registration_handler()
 
         # this is required by the request_handler wrapper
         self.clock = hs.get_clock()

synapse/server.py

@@ -64,6 +64,7 @@ from synapse.handlers.presence import PresenceHandler
 from synapse.handlers.profile import BaseProfileHandler, MasterProfileHandler
 from synapse.handlers.read_marker import ReadMarkerHandler
 from synapse.handlers.receipts import ReceiptsHandler
+from synapse.handlers.register import RegistrationHandler
 from synapse.handlers.room import RoomContextHandler, RoomCreationHandler
 from synapse.handlers.room_list import RoomListHandler
 from synapse.handlers.room_member import RoomMemberMasterHandler
@@ -181,6 +182,7 @@ class HomeServer(object):
         'pagination_handler',
         'room_context_handler',
         'sendmail',
+        'registration_handler',
     ]
 
     # This is overridden in derived application classes
@@ -481,6 +483,9 @@ class HomeServer(object):
     def build_room_context_handler(self):
         return RoomContextHandler(self)
 
+    def build_registration_handler(self):
+        return RegistrationHandler(self)
+
     def remove_pusher(self, app_id, push_key, user_id):
         return self.get_pusherpool().remove_pusher(app_id, push_key, user_id)
 

tests/handlers/test_register.py

@@ -48,7 +48,7 @@ class RegistrationTestCase(unittest.TestCase):
             generate_access_token=Mock(return_value='secret')
         )
         self.hs.get_macaroon_generator = Mock(return_value=self.macaroon_generator)
-        self.handler = self.hs.get_handlers().registration_handler
+        self.handler = self.hs.get_registration_handler()
         self.store = self.hs.get_datastore()
         self.hs.config.max_mau_value = 50
         self.lots_of_users = 100