Prevent replay attacks with the confirmation token.

2 files changed, 79 insertions, 5 deletions

diff --git a/src/mailman/app/tests/test_registrar.py b/src/mailman/app/tests/test_registrar.py
index 3e3c30590..06c22386a 100644
--- a/src/mailman/app/tests/test_registrar.py
+++ b/src/mailman/app/tests/test_registrar.py
@@ -146,15 +146,48 @@ class TestRegistrar(unittest.TestCase):
         self.assertIsNone(member)
         # Now confirm the subscription, and wait for the moderator to approve
         # the subscription.  She is still not subscribed.
-        status = self._registrar.confirm(token)
-        # The status is not true because the user has not yet been subscribed
-        # to the mailing list.
-        self.assertFalse(status)
+        new_token = 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)
         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(token)
+        self._registrar.confirm(new_token)
+        member = self._mlist.regular_members.get_member('anne@example.com')
+        self.assertEqual(member.address, self._anne)
+
+    def test_confirm_then_moderate_with_different_tokens(self):
+        # Ensure that the confirmation token the user sees when they have to
+        # confirm their subscription is different than the token the moderator
+        # sees when they approve the subscription.  This prevents the user
+        # from using a replay attack to subvert moderator approval.
+        self._mlist.subscription_policy = \
+          SubscriptionPolicy.confirm_then_moderate
+        self._anne.verified_on = now()
+        # Runs until subscription confirmation.
+        token = self._registrar.register(self._anne)
+        self.assertIsNotNone(token)
+        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)
+        # The status is not true because the user has not yet been subscribed
+        # to the mailing list.
+        self.assertIsNotNone(new_token)
+        member = self._mlist.regular_members.get_member('anne@example.com')
+        self.assertIsNone(member)
+        # The new token is different than the old token.
+        self.assertNotEqual(token, new_token)
+        # Trying to confirm with the old token does not work.
+        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)
+        # The token is None, signifying that the member has been subscribed.
+        self.assertIsNone(done_token)
         member = self._mlist.regular_members.get_member('anne@example.com')
         self.assertEqual(member.address, self._anne)
 
diff --git a/src/mailman/app/tests/test_subscriptions.py b/src/mailman/app/tests/test_subscriptions.py
index 478c7e33b..a4971d793 100644
--- a/src/mailman/app/tests/test_subscriptions.py
+++ b/src/mailman/app/tests/test_subscriptions.py
@@ -537,3 +537,44 @@ approval:
         self.assertIsNotNone(anne.verified_on)
         self.assertEqual(
             self._mlist.regular_members.get_member(self._anne).address, anne)
+
+    def test_prevent_confirmation_replay_attacks(self):
+        # Ensure that if the workflow requires two confirmations, e.g. first
+        # the user confirming their subscription, and then the moderator
+        # approving it, that different tokens are used in these two cases.
+        self._mlist.subscription_policy = \
+          SubscriptionPolicy.confirm_then_moderate
+        anne = self._user_manager.create_address(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne, pre_verified=True)
+        # Run the state machine up to the first confirmation, and cache the
+        # confirmation token.
+        list(workflow)
+        token = workflow.token
+        # Anne is not yet a member of the mailing list.
+        member = self._mlist.regular_members.get_member(self._anne)
+        self.assertIsNone(member)
+        # The old token will not work for moderator approval.
+        moderator_workflow = SubscriptionWorkflow(self._mlist)
+        moderator_workflow.token = token
+        moderator_workflow.restore()
+        list(moderator_workflow)
+        # 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)
+        # The old token won't work.
+        final_workflow = SubscriptionWorkflow(self._mlist)
+        final_workflow.token = token
+        self.assertRaises(LookupError, final_workflow.restore)
+        # Running this workflow will fail.
+        self.assertRaises(AssertionError, list, final_workflow)
+        # Anne is still not subscribed.
+        member = self._mlist.regular_members.get_member(self._anne)
+        self.assertIsNone(member)
+        # However, if we use the new token, her subscription request will be
+        # approved by the moderator.
+        final_workflow.token = moderator_workflow.token
+        final_workflow.restore()
+        list(final_workflow)
+        # And now Anne is a member.
+        member = self._mlist.regular_members.get_member(self._anne)
+        self.assertEqual(member.address.email, self._anne)