The SubscriptionWorkflow and Registar classes now have both a token and a

"token owner".  The latter describes who owns the token --i.e. which phase of
the workflow is being waited on.  It can either be no one, the subscriber, or
the moderator.

Tokens and token owners are properly initialized and reset when the workflow
is completed, so we always know which step of the process is being waited on.

Also, remove ISubscriptionService.join() since this will now be handled by the
IRegistrar adapter.

4 files changed, 120 insertions, 80 deletions

diff --git a/src/mailman/app/registrar.py b/src/mailman/app/registrar.py
index ae4322d22..a3699de91 100644
--- a/src/mailman/app/registrar.py
+++ b/src/mailman/app/registrar.py
@@ -63,7 +63,7 @@ class Registrar:
             pre_confirmed=pre_confirmed,
             pre_approved=pre_approved)
         list(workflow)
-        return workflow.token
+        return workflow.token, workflow.token_owner
 
     def confirm(self, token):
         """See `IRegistrar`."""
@@ -71,7 +71,7 @@ class Registrar:
         workflow.token = token
         workflow.restore()
         list(workflow)
-        return workflow.token
+        return workflow.token, workflow.token_owner
 
     def discard(self, token):
         """See `IRegistrar`."""
diff --git a/src/mailman/app/subscriptions.py b/src/mailman/app/subscriptions.py
index 3138c513b..de68a2a46 100644
--- a/src/mailman/app/subscriptions.py
+++ b/src/mailman/app/subscriptions.py
@@ -31,9 +31,8 @@ import logging
 from email.utils import formataddr
 from enum import Enum
 from datetime import timedelta
-from mailman.app.membership import add_member, delete_member
+from mailman.app.membership import delete_member
 from mailman.app.workflow import Workflow
-from mailman.core.constants import system_preferences
 from mailman.core.i18n import _
 from mailman.database.transaction import dbconnection
 from mailman.email.message import UserNotification
@@ -42,12 +41,10 @@ from mailman.interfaces.bans import IBanManager
 from mailman.interfaces.listmanager import (
     IListManager, ListDeletingEvent, NoSuchListError)
 from mailman.interfaces.mailinglist import SubscriptionPolicy
-from mailman.interfaces.member import (
-    DeliveryMode, MemberRole, MembershipIsBannedError)
+from mailman.interfaces.member import MembershipIsBannedError
 from mailman.interfaces.pending import IPendable, IPendings
 from mailman.interfaces.registrar import ConfirmationNeededEvent
-from mailman.interfaces.subscriptions import (
-    ISubscriptionService, MissingUserError, RequestRecord)
+from mailman.interfaces.subscriptions import ISubscriptionService, TokenOwner
 from mailman.interfaces.user import IUser
 from mailman.interfaces.usermanager import IUserManager
 from mailman.interfaces.workflow import IWorkflowStateManager
@@ -56,7 +53,6 @@ from mailman.utilities.datetime import now
 from mailman.utilities.i18n import make
 from operator import attrgetter
 from sqlalchemy import and_, or_
-from uuid import UUID
 from zope.component import getUtility
 from zope.event import notify
 from zope.interface import implementer
@@ -97,6 +93,7 @@ class SubscriptionWorkflow(Workflow):
         'address_key',
         'subscriber_key',
         'user_key',
+        'token_owner_key',
         )
 
     def __init__(self, mlist, subscriber=None, *,
@@ -106,6 +103,7 @@ class SubscriptionWorkflow(Workflow):
         self.address = None
         self.user = None
         self.which = None
+        self._set_token(TokenOwner.no_one)
         # The subscriber must be either an IUser or IAddress.
         if IAddress.providedBy(subscriber):
             self.address = subscriber
@@ -151,6 +149,29 @@ class SubscriptionWorkflow(Workflow):
     def subscriber_key(self, key):
         self.which = WhichSubscriber(key)
 
+    @property
+    def token_owner_key(self):
+        return self.token_owner.value
+
+    @token_owner_key.setter
+    def token_owner_key(self, value):
+        self.token_owner = TokenOwner(value)
+
+    def _set_token(self, token_owner):
+        assert isinstance(token_owner, TokenOwner)
+        # Create a new token to prevent replay attacks.  It seems like this
+        # should produce the same token, but it won't because the pending adds
+        # a bit of randomization.
+        self.token_owner = token_owner
+        if token_owner is TokenOwner.no_one:
+            self.token = None
+            return
+        pendable = Pendable(
+            list_id=self.mlist.list_id,
+            address=self.address.email,
+            )
+        self.token = getUtility(IPendings).add(pendable, timedelta(days=3650))
+
     def _step_sanity_checks(self):
         # Ensure that we have both an address and a user, even if the address
         # is not verified.  We can't set the preferred address until it is
@@ -174,13 +195,7 @@ class SubscriptionWorkflow(Workflow):
         # Is this email address banned?
         if IBanManager(self.mlist).is_banned(self.address.email):
             raise MembershipIsBannedError(self.mlist, self.address.email)
-        # Create a pending record.  This will give us the hash token we can use
-        # to uniquely name this workflow.
-        pendable = Pendable(
-            list_id=self.mlist.list_id,
-            address=self.address.email,
-            )
-        self.token = getUtility(IPendings).add(pendable, timedelta(days=3650))
+        # Start out with the subscriber being the token owner.
         self.push('verification_checks')
 
     def _step_verification_checks(self):
@@ -229,6 +244,7 @@ class SubscriptionWorkflow(Workflow):
         # Here's the next step in the workflow, assuming the moderator
         # approves of the subscription.  If they don't, the workflow and
         # subscription request will just be thrown away.
+        self._set_token(TokenOwner.moderator)
         self.push('subscribe_from_restored')
         self.save()
         log.info('{}: held subscription request from {}'.format(
@@ -255,6 +271,8 @@ class SubscriptionWorkflow(Workflow):
         raise StopIteration
 
     def _step_subscribe_from_restored(self):
+        # Prevent replay attacks.
+        self._set_token(TokenOwner.no_one)
         # Restore a little extra state that can't be stored in the database
         # (because the order of setattr() on restore is indeterminate), then
         # subscribe the user.
@@ -270,9 +288,9 @@ class SubscriptionWorkflow(Workflow):
         self.mlist.subscribe(self.subscriber)
         # This workflow is done so throw away any associated state.
         getUtility(IWorkflowStateManager).restore(self.name, self.token)
-        self.token = None
 
     def _step_send_confirmation(self):
+        self._set_token(TokenOwner.subscriber)
         self.push('do_confirm_verify')
         self.save()
         # Triggering this event causes the confirmation message to be sent.
@@ -290,14 +308,8 @@ class SubscriptionWorkflow(Workflow):
         else:
             assert self.which is WhichSubscriber.user
             self.subscriber = self.user
-        # Create a new token to prevent replay attacks.  It seems like this
-        # should produce the same token, but it won't because the pending adds
-        # a bit of randomization.
-        pendable = Pendable(
-            list_id=self.mlist.list_id,
-            address=self.address.email,
-            )
-        self.token = getUtility(IPendings).add(pendable, timedelta(days=3650))
+        # Reset the token so it can't be used in a replay attack.
+        self._set_token(TokenOwner.no_one)
         # The user has confirmed their subscription request, and also verified
         # their email address if necessary.  This latter needs to be set on the
         # IAddress, but there's nothing more to do about the confirmation step.
@@ -396,31 +408,6 @@ class SubscriptionService:
         for member in self.get_members():
             yield member
 
-    def join(self, list_id, subscriber,
-             display_name=None,
-             delivery_mode=DeliveryMode.regular,
-             role=MemberRole.member):
-        """See `ISubscriptionService`."""
-        mlist = getUtility(IListManager).get_by_list_id(list_id)
-        if mlist is None:
-            raise NoSuchListError(list_id)
-        # Is the subscriber an email address or user id?
-        if isinstance(subscriber, str):
-            if display_name is None:
-                display_name, at, domain = subscriber.partition('@')
-            return add_member(
-                mlist,
-                RequestRecord(subscriber, display_name, delivery_mode,
-                              system_preferences.preferred_language),
-                role)
-        else:
-            # We have to assume it's a UUID.
-            assert isinstance(subscriber, UUID), 'Not a UUID'
-            user = getUtility(IUserManager).get_user_by_id(subscriber)
-            if user is None:
-                raise MissingUserError(subscriber)
-            return mlist.subscribe(user, role)
-
     def leave(self, list_id, email):
         """See `ISubscriptionService`."""
         mlist = getUtility(IListManager).get_by_list_id(list_id)
diff --git a/src/mailman/app/tests/test_registrar.py b/src/mailman/app/tests/test_registrar.py
index 4f5e1e3f9..1431b5e6f 100644
--- a/src/mailman/app/tests/test_registrar.py
+++ b/src/mailman/app/tests/test_registrar.py
@@ -28,6 +28,7 @@ from mailman.app.lifecycle import create_list
 from mailman.interfaces.mailinglist import SubscriptionPolicy
 from mailman.interfaces.pending import IPendings
 from mailman.interfaces.registrar import IRegistrar
+from mailman.interfaces.subscriptions import TokenOwner
 from mailman.interfaces.usermanager import IUserManager
 from mailman.testing.layers import ConfigLayer
 from mailman.utilities.datetime import now
@@ -49,35 +50,30 @@ class TestRegistrar(unittest.TestCase):
 
     def test_unique_token(self):
         # Registering a subscription request provides a unique token associated
-        # with a pendable.
+        # with a pendable, and the owner of the token.
         self.assertEqual(self._pendings.count, 0)
-        token = self._registrar.register(self._anne)
+        token, token_owner = self._registrar.register(self._anne)
         self.assertIsNotNone(token)
+        self.assertEqual(token_owner, TokenOwner.subscriber)
         self.assertEqual(self._pendings.count, 1)
         record = self._pendings.confirm(token, expunge=False)
         self.assertEqual(record['list_id'], self._mlist.list_id)
         self.assertEqual(record['address'], 'anne@example.com')
 
-    def test_no_token(self):
+    def test_subscribe(self):
         # Registering a subscription request where no confirmation or
-        # moderation steps are needed, leaves us with no token, since there's
-        # nothing more to do.
+        # moderation steps are needed, leaves us with no token or owner, since
+        # there's nothing more to do.
         self._mlist.subscription_policy = SubscriptionPolicy.open
         self._anne.verified_on = now()
-        token = self._registrar.register(self._anne)
+        token, token_owner = self._registrar.register(self._anne)
         self.assertIsNone(token)
-        record = self._pendings.confirm(token, expunge=False)
-        self.assertIsNone(record)
-
-    def test_is_subscribed(self):
-        # Where no confirmation or moderation steps are needed, registration
-        # happens immediately.
-        self._mlist.subscription_policy = SubscriptionPolicy.open
-        self._anne.verified_on = now()
-        status = self._registrar.register(self._anne)
-        self.assertIsNone(status)
+        self.assertEqual(token_owner, TokenOwner.no_one)
         member = self._mlist.regular_members.get_member('anne@example.com')
         self.assertEqual(member.address, self._anne)
+        # There's nothing to confirm.
+        record = self._pendings.confirm(token, expunge=False)
+        self.assertIsNone(record)
 
     def test_no_such_token(self):
         # Given a token which is not in the database, a LookupError is raised.
@@ -90,12 +86,15 @@ class TestRegistrar(unittest.TestCase):
         # to approve.  Running the workflow gives us a token.  Confirming the
         # token subscribes the user.
         self._mlist.subscription_policy = SubscriptionPolicy.open
-        token = self._registrar.register(self._anne)
+        token, token_owner = self._registrar.register(self._anne)
         self.assertIsNotNone(token)
+        self.assertEqual(token_owner, TokenOwner.subscriber)
         member = self._mlist.regular_members.get_member('anne@example.com')
         self.assertIsNone(member)
         # Now confirm the subscription.
-        self._registrar.confirm(token)
+        token, token_owner = self._registrar.confirm(token)
+        self.assertIsNone(token)
+        self.assertEqual(token_owner, TokenOwner.no_one)
         member = self._mlist.regular_members.get_member('anne@example.com')
         self.assertEqual(member.address, self._anne)
 
@@ -106,12 +105,15 @@ class TestRegistrar(unittest.TestCase):
         # user.
         self._mlist.subscription_policy = SubscriptionPolicy.confirm
         self._anne.verified_on = now()
-        token = self._registrar.register(self._anne)
+        token, token_owner = self._registrar.register(self._anne)
         self.assertIsNotNone(token)
+        self.assertEqual(token_owner, TokenOwner.subscriber)
         member = self._mlist.regular_members.get_member('anne@example.com')
         self.assertIsNone(member)
         # Now confirm the subscription.
-        self._registrar.confirm(token)
+        token, token_owner = self._registrar.confirm(token)
+        self.assertIsNone(token)
+        self.assertEqual(token_owner, TokenOwner.no_one)
         member = self._mlist.regular_members.get_member('anne@example.com')
         self.assertEqual(member.address, self._anne)
 
@@ -121,12 +123,15 @@ class TestRegistrar(unittest.TestCase):
         # token subscribes the user.
         self._mlist.subscription_policy = SubscriptionPolicy.moderate
         self._anne.verified_on = now()
-        token = self._registrar.register(self._anne)
+        token, token_owner = self._registrar.register(self._anne)
         self.assertIsNotNone(token)
+        self.assertEqual(token_owner, TokenOwner.moderator)
         member = self._mlist.regular_members.get_member('anne@example.com')
         self.assertIsNone(member)
         # Now confirm the subscription.
-        self._registrar.confirm(token)
+        token, token_owner = self._registrar.confirm(token)
+        self.assertIsNone(token)
+        self.assertEqual(token_owner, TokenOwner.no_one)
         member = self._mlist.regular_members.get_member('anne@example.com')
         self.assertEqual(member.address, self._anne)
 
@@ -140,21 +145,26 @@ class TestRegistrar(unittest.TestCase):
           SubscriptionPolicy.confirm_then_moderate
         self._anne.verified_on = now()
         # Runs until subscription confirmation.
-        token = self._registrar.register(self._anne)
+        token, token_owner = self._registrar.register(self._anne)
         self.assertIsNotNone(token)
+        self.assertEqual(token_owner, TokenOwner.subscriber)
         member = self._mlist.regular_members.get_member('anne@example.com')
         self.assertIsNone(member)
         # Now confirm the subscription, and wait for the moderator to approve
         # the subscription.  She is still not subscribed.
-        new_token = self._registrar.confirm(token)
+        new_token, token_owner = self._registrar.confirm(token)
         # The new token, used for the moderator to approve the message, is not
         # the same as the old token.
         self.assertNotEqual(new_token, token)
+        self.assertIsNotNone(new_token)
+        self.assertEqual(token_owner, TokenOwner.moderator)
         member = self._mlist.regular_members.get_member('anne@example.com')
         self.assertIsNone(member)
         # Confirm once more, this time as the moderator approving the
         # subscription.  Now she's a member.
-        self._registrar.confirm(new_token)
+        token, token_owner = self._registrar.confirm(new_token)
+        self.assertIsNone(token)
+        self.assertEqual(token_owner, TokenOwner.no_one)
         member = self._mlist.regular_members.get_member('anne@example.com')
         self.assertEqual(member.address, self._anne)
 
@@ -167,16 +177,18 @@ class TestRegistrar(unittest.TestCase):
           SubscriptionPolicy.confirm_then_moderate
         self._anne.verified_on = now()
         # Runs until subscription confirmation.
-        token = self._registrar.register(self._anne)
+        token, token_owner = self._registrar.register(self._anne)
         self.assertIsNotNone(token)
+        self.assertEqual(token_owner, TokenOwner.subscriber)
         member = self._mlist.regular_members.get_member('anne@example.com')
         self.assertIsNone(member)
         # Now confirm the subscription, and wait for the moderator to approve
         # the subscription.  She is still not subscribed.
-        new_token = self._registrar.confirm(token)
+        new_token, token_owner = self._registrar.confirm(token)
         # The status is not true because the user has not yet been subscribed
         # to the mailing list.
         self.assertIsNotNone(new_token)
+        self.assertEqual(token_owner, TokenOwner.moderator)
         member = self._mlist.regular_members.get_member('anne@example.com')
         self.assertIsNone(member)
         # The new token is different than the old token.
@@ -185,9 +197,10 @@ class TestRegistrar(unittest.TestCase):
         self.assertRaises(LookupError, self._registrar.confirm, token)
         # Confirm once more, this time with the new token, as the moderator
         # approving the subscription.  Now she's a member.
-        done_token = self._registrar.confirm(new_token)
+        done_token, token_owner = self._registrar.confirm(new_token)
         # The token is None, signifying that the member has been subscribed.
         self.assertIsNone(done_token)
+        self.assertEqual(token_owner, TokenOwner.no_one)
         member = self._mlist.regular_members.get_member('anne@example.com')
         self.assertEqual(member.address, self._anne)
 
@@ -197,8 +210,9 @@ class TestRegistrar(unittest.TestCase):
         self._mlist.subscription_policy = SubscriptionPolicy.confirm
         self._anne.verified_on = now()
         # Runs until subscription confirmation.
-        token = self._registrar.register(self._anne)
+        token, token_owner = self._registrar.register(self._anne)
         self.assertIsNotNone(token)
+        self.assertEqual(token_owner, TokenOwner.subscriber)
         member = self._mlist.regular_members.get_member('anne@example.com')
         self.assertIsNone(member)
         # Now discard the subscription request.
diff --git a/src/mailman/app/tests/test_subscriptions.py b/src/mailman/app/tests/test_subscriptions.py
index a4971d793..77f308b9c 100644
--- a/src/mailman/app/tests/test_subscriptions.py
+++ b/src/mailman/app/tests/test_subscriptions.py
@@ -34,7 +34,7 @@ from mailman.interfaces.member import (
     MemberRole, MembershipIsBannedError, MissingPreferredAddressError)
 from mailman.interfaces.pending import IPendings
 from mailman.interfaces.subscriptions import (
-    MissingUserError, ISubscriptionService)
+    ISubscriptionService, MissingUserError, TokenOwner)
 from mailman.testing.helpers import LogFileMark, get_queue_messages
 from mailman.testing.layers import ConfigLayer
 from mailman.interfaces.mailinglist import SubscriptionPolicy
@@ -45,6 +45,7 @@ from zope.component import getUtility
 
 
 
+@unittest.skip('XXX -- no more .join()')
 class TestJoin(unittest.TestCase):
     layer = ConfigLayer
 
@@ -88,6 +89,12 @@ class TestSubscriptionWorkflow(unittest.TestCase):
         self._anne = 'anne@example.com'
         self._user_manager = getUtility(IUserManager)
 
+    def test_no_token_to_start_with(self):
+        # The workflow starts with no tokens.
+        workflow = SubscriptionWorkflow(self._mlist)
+        self.assertIsNone(workflow.token)
+        self.assertEqual(workflow.token_owner, TokenOwner.no_one)
+
     def test_user_or_address_required(self):
         # The `subscriber` attribute must be a user or address.
         workflow = SubscriptionWorkflow(self._mlist)
@@ -311,6 +318,9 @@ class TestSubscriptionWorkflow(unittest.TestCase):
         # Anne is now a member of the mailing list.
         member = self._mlist.regular_members.get_member(self._anne)
         self.assertEqual(member.address, anne)
+        # No further token is needed.
+        self.assertIsNone(workflow.token)
+        self.assertEqual(workflow.token_owner, TokenOwner.no_one)
 
     def test_do_subscription_pre_approved(self):
         # An moderation-requiring subscription policy plus a pre-verified and
@@ -326,6 +336,9 @@ class TestSubscriptionWorkflow(unittest.TestCase):
         # Anne is now a member of the mailing list.
         member = self._mlist.regular_members.get_member(self._anne)
         self.assertEqual(member.address, anne)
+        # No further token is needed.
+        self.assertIsNone(workflow.token)
+        self.assertEqual(workflow.token_owner, TokenOwner.no_one)
 
     def test_do_subscription_pre_approved_pre_confirmed(self):
         # An moderation-requiring subscription policy plus a pre-verified and
@@ -343,6 +356,9 @@ class TestSubscriptionWorkflow(unittest.TestCase):
         # Anne is now a member of the mailing list.
         member = self._mlist.regular_members.get_member(self._anne)
         self.assertEqual(member.address, anne)
+        # No further token is needed.
+        self.assertIsNone(workflow.token)
+        self.assertEqual(workflow.token_owner, TokenOwner.no_one)
 
     def test_do_subscription_cleanups(self):
         # Once the user is subscribed, the token, and its associated pending
@@ -362,6 +378,7 @@ class TestSubscriptionWorkflow(unittest.TestCase):
         self.assertEqual(member.address, anne)
         # The workflow is done, so it has no token.
         self.assertIsNone(workflow.token)
+        self.assertEqual(workflow.token_owner, TokenOwner.no_one)
         # The pendable associated with the token has been evicted.
         self.assertIsNone(getUtility(IPendings).confirm(token, expunge=False))
         # There is no saved workflow associated with the token.  This shows up
@@ -384,6 +401,9 @@ class TestSubscriptionWorkflow(unittest.TestCase):
         # The user is not currently subscribed to the mailing list.
         member = self._mlist.regular_members.get_member(self._anne)
         self.assertIsNone(member)
+        # The token is owned by the moderator.
+        self.assertIsNotNone(workflow.token)
+        self.assertEqual(workflow.token_owner, TokenOwner.moderator)
         # Create a new workflow with the previous workflow's save token, and
         # restore its state.  This models an approved subscription and should
         # result in the user getting subscribed.
@@ -394,6 +414,9 @@ class TestSubscriptionWorkflow(unittest.TestCase):
         # Now the user is subscribed to the mailing list.
         member = self._mlist.regular_members.get_member(self._anne)
         self.assertEqual(member.address, anne)
+        # No further token is needed.
+        self.assertIsNone(approved_workflow.token)
+        self.assertEqual(approved_workflow.token_owner, TokenOwner.no_one)
 
     def test_get_moderator_approval_log_on_hold(self):
         # When the subscription is held for moderator approval, a message is
@@ -530,6 +553,10 @@ approval:
         workflow = SubscriptionWorkflow(self._mlist, anne)
         list(workflow)
         self.assertIsNone(self._mlist.regular_members.get_member(self._anne))
+        # The token is owned by the subscriber.
+        self.assertIsNotNone(workflow.token)
+        self.assertEqual(workflow.token_owner, TokenOwner.subscriber)
+        # Confirm.
         confirm_workflow = SubscriptionWorkflow(self._mlist)
         confirm_workflow.token = workflow.token
         confirm_workflow.restore()
@@ -537,6 +564,9 @@ approval:
         self.assertIsNotNone(anne.verified_on)
         self.assertEqual(
             self._mlist.regular_members.get_member(self._anne).address, anne)
+        # No further token is needed.
+        self.assertIsNone(confirm_workflow.token)
+        self.assertEqual(confirm_workflow.token_owner, TokenOwner.no_one)
 
     def test_prevent_confirmation_replay_attacks(self):
         # Ensure that if the workflow requires two confirmations, e.g. first
@@ -553,11 +583,17 @@ approval:
         # Anne is not yet a member of the mailing list.
         member = self._mlist.regular_members.get_member(self._anne)
         self.assertIsNone(member)
+        # The token is owned by the subscriber.
+        self.assertIsNotNone(workflow.token)
+        self.assertEqual(workflow.token_owner, TokenOwner.subscriber)
         # The old token will not work for moderator approval.
         moderator_workflow = SubscriptionWorkflow(self._mlist)
         moderator_workflow.token = token
         moderator_workflow.restore()
         list(moderator_workflow)
+        # The token is owned by the moderator.
+        self.assertIsNotNone(moderator_workflow.token)
+        self.assertEqual(moderator_workflow.token_owner, TokenOwner.moderator)
         # While we wait for the moderator to approve the subscription, note
         # that there's a new token for the next steps.
         self.assertNotEqual(token, moderator_workflow.token)
@@ -578,3 +614,6 @@ approval:
         # And now Anne is a member.
         member = self._mlist.regular_members.get_member(self._anne)
         self.assertEqual(member.address.email, self._anne)
+        # No further token is needed.
+        self.assertIsNone(final_workflow.token)
+        self.assertEqual(final_workflow.token_owner, TokenOwner.no_one)