52 files changed, 2205 insertions, 759 deletions

diff --git a/src/mailman/app/docs/moderator.rst b/src/mailman/app/docs/moderator.rst
index 82d29074a..43dc7688f 100644
--- a/src/mailman/app/docs/moderator.rst
+++ b/src/mailman/app/docs/moderator.rst
@@ -424,7 +424,6 @@ There's now a message in the virgin queue, destined for the list owner.
     <BLANKLINE>
         For:  iris@example.org
         List: ant@example.com
-    ...
 
 Similarly, the administrator gets notifications on unsubscription requests.
 Jeff is a member of the mailing list, and chooses to unsubscribe.
diff --git a/src/mailman/app/registrar.py b/src/mailman/app/registrar.py
index 9272b95d7..ae4322d22 100644
--- a/src/mailman/app/registrar.py
+++ b/src/mailman/app/registrar.py
@@ -25,18 +25,15 @@ __all__ = [
 
 import logging
 
+from mailman.app.subscriptions import SubscriptionWorkflow
 from mailman.core.i18n import _
+from mailman.database.transaction import flush
 from mailman.email.message import UserNotification
-from mailman.interfaces.address import IEmailValidator
-from mailman.interfaces.listmanager import IListManager
-from mailman.interfaces.member import DeliveryMode, MemberRole
 from mailman.interfaces.pending import IPendable, IPendings
 from mailman.interfaces.registrar import ConfirmationNeededEvent, IRegistrar
 from mailman.interfaces.templates import ITemplateLoader
-from mailman.interfaces.usermanager import IUserManager
-from mailman.utilities.datetime import now
+from mailman.interfaces.workflow import IWorkflowStateManager
 from zope.component import getUtility
-from zope.event import notify
 from zope.interface import implementer
 
 
@@ -54,96 +51,34 @@ class PendableRegistration(dict):
 class Registrar:
     """Handle registrations and confirmations for subscriptions."""
 
-    def register(self, mlist, email, display_name=None, delivery_mode=None):
+    def __init__(self, mlist):
+        self._mlist = mlist
+
+    def register(self, subscriber=None, *,
+                 pre_verified=False, pre_confirmed=False, pre_approved=False):
         """See `IRegistrar`."""
-        if delivery_mode is None:
-            delivery_mode = DeliveryMode.regular
-        # First, do validation on the email address.  If the address is
-        # invalid, it will raise an exception, otherwise it just returns.
-        getUtility(IEmailValidator).validate(email)
-        # Create a pendable for the registration.
-        pendable = PendableRegistration(
-            type=PendableRegistration.PEND_KEY,
-            email=email,
-            display_name=display_name,
-            delivery_mode=delivery_mode.name,
-            list_id=mlist.list_id)
-        token = getUtility(IPendings).add(pendable)
-        # We now have everything we need to begin the confirmation dance.
-        # Trigger the event to start the ball rolling, and return the
-        # generated token.
-        notify(ConfirmationNeededEvent(mlist, pendable, token))
-        return token
+        workflow = SubscriptionWorkflow(
+            self._mlist, subscriber,
+            pre_verified=pre_verified,
+            pre_confirmed=pre_confirmed,
+            pre_approved=pre_approved)
+        list(workflow)
+        return workflow.token
 
     def confirm(self, token):
         """See `IRegistrar`."""
-        # For convenience
-        pendable = getUtility(IPendings).confirm(token)
-        if pendable is None:
-            return False
-        missing = object()
-        email = pendable.get('email', missing)
-        display_name = pendable.get('display_name', missing)
-        pended_delivery_mode = pendable.get('delivery_mode', 'regular')
-        try:
-            delivery_mode = DeliveryMode[pended_delivery_mode]
-        except ValueError:
-            log.error('Invalid pended delivery_mode for {0}: {1}',
-                      email, pended_delivery_mode)
-            delivery_mode = DeliveryMode.regular
-        if pendable.get('type') != PendableRegistration.PEND_KEY:
-            # It seems like it would be very difficult to accurately guess
-            # tokens, or brute force an attack on the SHA1 hash, so we'll just
-            # throw the pendable away in that case.  It's possible we'll need
-            # to repend the event or adjust the API to handle this case
-            # better, but for now, the simpler the better.
-            return False
-        # We are going to end up with an IAddress for the verified address
-        # and an IUser linked to this IAddress.  See if any of these objects
-        # currently exist in our database.
-        user_manager = getUtility(IUserManager)
-        address = (user_manager.get_address(email)
-                   if email is not missing else None)
-        user = (user_manager.get_user(email)
-                if email is not missing else None)
-        # If there is neither an address nor a user matching the confirmed
-        # record, then create the user, which will in turn create the address
-        # and link the two together
-        if address is None:
-            assert user is None, 'How did we get a user but not an address?'
-            user = user_manager.create_user(email, display_name)
-            # Because the database changes haven't been flushed, we can't use
-            # IUserManager.get_address() to find the IAddress just created
-            # under the hood.  Instead, iterate through the IUser's addresses,
-            # of which really there should be only one.
-            for address in user.addresses:
-                if address.email == email:
-                    break
-            else:
-                raise AssertionError('Could not find expected IAddress')
-        elif user is None:
-            user = user_manager.create_user()
-            user.display_name = display_name
-            user.link(address)
-        else:
-            # The IAddress and linked IUser already exist, so all we need to
-            # do is verify the address.
-            pass
-        address.verified_on = now()
-        # If this registration is tied to a mailing list, subscribe the person
-        # to the list right now.  That will generate a SubscriptionEvent,
-        # which can be used to send a welcome message.
-        list_id = pendable.get('list_id')
-        if list_id is not None:
-            mlist = getUtility(IListManager).get_by_list_id(list_id)
-            if mlist is not None:
-                member = mlist.subscribe(address, MemberRole.member)
-                member.preferences.delivery_mode = delivery_mode
-        return True
+        workflow = SubscriptionWorkflow(self._mlist)
+        workflow.token = token
+        workflow.restore()
+        list(workflow)
+        return workflow.token
 
     def discard(self, token):
-        # Throw the record away.
-        getUtility(IPendings).confirm(token)
+        """See `IRegistrar`."""
+        with flush():
+            getUtility(IPendings).confirm(token)
+            getUtility(IWorkflowStateManager).discard(
+                SubscriptionWorkflow.__name__, token)
 
 
 
@@ -156,18 +91,17 @@ def handle_ConfirmationNeededEvent(event):
     # the Subject header, or they can click on the URL in the body of the
     # message and confirm through the web.
     subject = 'confirm ' + event.token
-    mlist = getUtility(IListManager).get_by_list_id(event.pendable['list_id'])
-    confirm_address = mlist.confirm_address(event.token)
+    confirm_address = event.mlist.confirm_address(event.token)
     # For i18n interpolation.
-    confirm_url = mlist.domain.confirm_url(event.token)
-    email_address = event.pendable['email']
-    domain_name = mlist.domain.mail_host
-    contact_address = mlist.owner_address
+    confirm_url = event.mlist.domain.confirm_url(event.token)
+    email_address = event.email
+    domain_name = event.mlist.domain.mail_host
+    contact_address = event.mlist.owner_address
     # Send a verification email to the address.
     template = getUtility(ITemplateLoader).get(
         'mailman:///{0}/{1}/confirm.txt'.format(
-            mlist.fqdn_listname,
-            mlist.preferred_language.code))
+            event.mlist.fqdn_listname,
+            event.mlist.preferred_language.code))
     text = _(template)
     msg = UserNotification(email_address, confirm_address, subject, text)
-    msg.send(mlist)
+    msg.send(event.mlist)
diff --git a/src/mailman/app/subscriptions.py b/src/mailman/app/subscriptions.py
index cc363b30d..3138c513b 100644
--- a/src/mailman/app/subscriptions.py
+++ b/src/mailman/app/subscriptions.py
@@ -19,26 +19,50 @@
 
 __all__ = [
     'SubscriptionService',
+    'SubscriptionWorkflow',
     'handle_ListDeletingEvent',
     ]
 
 
-from operator import attrgetter
-from sqlalchemy import and_, or_
-from uuid import UUID
-from zope.component import getUtility
-from zope.interface import implementer
 
+import uuid
+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.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
+from mailman.interfaces.address import IAddress
+from mailman.interfaces.bans import IBanManager
 from mailman.interfaces.listmanager import (
     IListManager, ListDeletingEvent, NoSuchListError)
-from mailman.interfaces.member import DeliveryMode, MemberRole
+from mailman.interfaces.mailinglist import SubscriptionPolicy
+from mailman.interfaces.member import (
+    DeliveryMode, MemberRole, 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.user import IUser
 from mailman.interfaces.usermanager import IUserManager
+from mailman.interfaces.workflow import IWorkflowStateManager
 from mailman.model.member import Member
+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
+
+
+log = logging.getLogger('mailman.subscribe')
 
 
 
@@ -51,7 +75,245 @@ def _membership_sort_key(member):
     return (member.list_id, member.address.email, member.role.value)
 
 
+class WhichSubscriber(Enum):
+    address = 1
+    user = 2
+
+
+@implementer(IPendable)
+class Pendable(dict):
+    pass
+
+
 
+class SubscriptionWorkflow(Workflow):
+    """Workflow of a subscription request."""
+
+    INITIAL_STATE = 'sanity_checks'
+    SAVE_ATTRIBUTES = (
+        'pre_approved',
+        'pre_confirmed',
+        'pre_verified',
+        'address_key',
+        'subscriber_key',
+        'user_key',
+        )
+
+    def __init__(self, mlist, subscriber=None, *,
+                 pre_verified=False, pre_confirmed=False, pre_approved=False):
+        super().__init__()
+        self.mlist = mlist
+        self.address = None
+        self.user = None
+        self.which = None
+        # The subscriber must be either an IUser or IAddress.
+        if IAddress.providedBy(subscriber):
+            self.address = subscriber
+            self.user = self.address.user
+            self.which = WhichSubscriber.address
+        elif IUser.providedBy(subscriber):
+            self.address = subscriber.preferred_address
+            self.user = subscriber
+            self.which = WhichSubscriber.user
+        self.subscriber = subscriber
+        self.pre_verified = pre_verified
+        self.pre_confirmed = pre_confirmed
+        self.pre_approved = pre_approved
+
+    @property
+    def user_key(self):
+        # For save.
+        return self.user.user_id.hex
+
+    @user_key.setter
+    def user_key(self, hex_key):
+        # For restore.
+        uid = uuid.UUID(hex_key)
+        self.user = getUtility(IUserManager).get_user_by_id(uid)
+        assert self.user is not None
+
+    @property
+    def address_key(self):
+        # For save.
+        return self.address.email
+
+    @address_key.setter
+    def address_key(self, email):
+        # For restore.
+        self.address = getUtility(IUserManager).get_address(email)
+        assert self.address is not None
+
+    @property
+    def subscriber_key(self):
+        return self.which.value
+
+    @subscriber_key.setter
+    def subscriber_key(self, key):
+        self.which = WhichSubscriber(key)
+
+    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
+        # verified.
+        if self.user is None:
+            # The address has no linked user so create one, link it, and set
+            # the user's preferred address.
+            assert self.address is not None, 'No address or user'
+            self.user = getUtility(IUserManager).make_user(self.address.email)
+        if self.address is None:
+            assert self.user.preferred_address is None, (
+                "Preferred address exists, but wasn't used in constructor")
+            addresses = list(self.user.addresses)
+            if len(addresses) == 0:
+                raise AssertionError('User has no addresses: {}'.format(
+                    self.user))
+            # This is rather arbitrary, but we have no choice.
+            self.address = addresses[0]
+        assert self.user is not None and self.address is not None, (
+            'Insane sanity check results')
+        # 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))
+        self.push('verification_checks')
+
+    def _step_verification_checks(self):
+        # Is the address already verified, or is the pre-verified flag set?
+        if self.address.verified_on is None:
+            if self.pre_verified:
+                self.address.verified_on = now()
+            else:
+                # The address being subscribed is not yet verified, so we need
+                # to send a validation email that will also confirm that the
+                # user wants to be subscribed to this mailing list.
+                self.push('send_confirmation')
+                return
+        self.push('confirmation_checks')
+
+    def _step_confirmation_checks(self):
+        # If the list's subscription policy is open, then the user can be
+        # subscribed right here and now.
+        if self.mlist.subscription_policy is SubscriptionPolicy.open:
+            self.push('do_subscription')
+            return
+        # If we do not need the user's confirmation, then skip to the
+        # moderation checks.
+        if self.mlist.subscription_policy is SubscriptionPolicy.moderate:
+            self.push('moderation_checks')
+            return
+        # If the subscription has been pre-confirmed, then we can skip to the
+        # moderation checks.
+        if self.pre_confirmed:
+            self.push('moderation_checks')
+            return
+        # The user must confirm their subscription.
+        self.push('send_confirmation')
+
+    def _step_moderation_checks(self):
+        # Does the moderator need to approve the subscription request?
+        assert self.mlist.subscription_policy in (
+            SubscriptionPolicy.moderate,
+            SubscriptionPolicy.confirm_then_moderate)
+        if self.pre_approved:
+            self.push('do_subscription')
+        else:
+            self.push('get_moderator_approval')
+
+    def _step_get_moderator_approval(self):
+        # 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.push('subscribe_from_restored')
+        self.save()
+        log.info('{}: held subscription request from {}'.format(
+            self.mlist.fqdn_listname, self.address.email))
+        # Possibly send a notification to the list moderators.
+        if self.mlist.admin_immed_notify:
+            subject = _(
+                'New subscription request to $self.mlist.display_name '
+                'from $self.address.email')
+            username = formataddr(
+                (self.subscriber.display_name, self.address.email))
+            text = make('subauth.txt',
+                        mailing_list=self.mlist,
+                        username=username,
+                        listname=self.mlist.fqdn_listname,
+                        )
+            # This message should appear to come from the <list>-owner so as
+            # to avoid any useless bounce processing.
+            msg = UserNotification(
+                self.mlist.owner_address, self.mlist.owner_address,
+                subject, text, self.mlist.preferred_language)
+            msg.send(self.mlist, tomoderators=True)
+        # The workflow must stop running here.
+        raise StopIteration
+
+    def _step_subscribe_from_restored(self):
+        # 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.
+        if self.which is WhichSubscriber.address:
+            self.subscriber = self.address
+        else:
+            assert self.which is WhichSubscriber.user
+            self.subscriber = self.user
+        self.push('do_subscription')
+
+    def _step_do_subscription(self):
+        # We can immediately subscribe the user to the mailing list.
+        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.push('do_confirm_verify')
+        self.save()
+        # Triggering this event causes the confirmation message to be sent.
+        notify(ConfirmationNeededEvent(
+            self.mlist, self.token, self.address.email))
+        # Now we wait for the confirmation.
+        raise StopIteration
+
+    def _step_do_confirm_verify(self):
+        # Restore a little extra state that can't be stored in the database
+        # (because the order of setattr() on restore is indeterminate), then
+        # continue with the confirmation/verification step.
+        if self.which is WhichSubscriber.address:
+            self.subscriber = self.address
+        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))
+        # 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.
+        # We just continue along with the workflow.
+        if self.address.verified_on is None:
+            self.address.verified_on = now()
+        # The next step depends on the mailing list's subscription policy.
+        next_step = ('moderation_checks'
+                     if self.mlist.subscription_policy in (
+                         SubscriptionPolicy.moderate,
+                         SubscriptionPolicy.confirm_then_moderate,
+                         )
+                    else 'do_subscription')
+        self.push(next_step)
+
+
 @implementer(ISubscriptionService)
 class SubscriptionService:
     """Subscription services for the REST API."""
diff --git a/src/mailman/app/tests/test_registrar.py b/src/mailman/app/tests/test_registrar.py
new file mode 100644
index 000000000..4f5e1e3f9
--- /dev/null
+++ b/src/mailman/app/tests/test_registrar.py
@@ -0,0 +1,207 @@
+# Copyright (C) 2012-2015 by the Free Software Foundation, Inc.
+#
+# This file is part of GNU Mailman.
+#
+# GNU Mailman is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option)
+# any later version.
+#
+# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
+# more details.
+#
+# You should have received a copy of the GNU General Public License along with
+# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Test email address registration."""
+
+__all__ = [
+    'TestRegistrar',
+    ]
+
+
+import unittest
+
+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.usermanager import IUserManager
+from mailman.testing.layers import ConfigLayer
+from mailman.utilities.datetime import now
+from zope.component import getUtility
+
+
+
+class TestRegistrar(unittest.TestCase):
+    """Test registration."""
+
+    layer = ConfigLayer
+
+    def setUp(self):
+        self._mlist = create_list('ant@example.com')
+        self._registrar = IRegistrar(self._mlist)
+        self._pendings = getUtility(IPendings)
+        self._anne = getUtility(IUserManager).create_address(
+            'anne@example.com')
+
+    def test_unique_token(self):
+        # Registering a subscription request provides a unique token associated
+        # with a pendable.
+        self.assertEqual(self._pendings.count, 0)
+        token = self._registrar.register(self._anne)
+        self.assertIsNotNone(token)
+        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):
+        # Registering a subscription request where no confirmation or
+        # moderation steps are needed, leaves us with no token, since there's
+        # nothing more to do.
+        self._mlist.subscription_policy = SubscriptionPolicy.open
+        self._anne.verified_on = now()
+        token = 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)
+        member = self._mlist.regular_members.get_member('anne@example.com')
+        self.assertEqual(member.address, self._anne)
+
+    def test_no_such_token(self):
+        # Given a token which is not in the database, a LookupError is raised.
+        self._registrar.register(self._anne)
+        self.assertRaises(LookupError, self._registrar.confirm, 'not-a-token')
+
+    def test_confirm_because_verify(self):
+        # We have a subscription request which requires the user to confirm
+        # (because she does not have a verified address), but not the moderator
+        # 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)
+        self.assertIsNotNone(token)
+        member = self._mlist.regular_members.get_member('anne@example.com')
+        self.assertIsNone(member)
+        # Now confirm the subscription.
+        self._registrar.confirm(token)
+        member = self._mlist.regular_members.get_member('anne@example.com')
+        self.assertEqual(member.address, self._anne)
+
+    def test_confirm_because_confirm(self):
+        # We have a subscription request which requires the user to confirm
+        # (because of list policy), but not the moderator to approve.  Running
+        # the workflow gives us a token.  Confirming the token subscribes the
+        # user.
+        self._mlist.subscription_policy = SubscriptionPolicy.confirm
+        self._anne.verified_on = now()
+        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.
+        self._registrar.confirm(token)
+        member = self._mlist.regular_members.get_member('anne@example.com')
+        self.assertEqual(member.address, self._anne)
+
+    def test_confirm_because_moderation(self):
+        # We have a subscription request which requires the moderator to
+        # approve.  Running the workflow gives us a token.  Confirming the
+        # token subscribes the user.
+        self._mlist.subscription_policy = SubscriptionPolicy.moderate
+        self._anne.verified_on = now()
+        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.
+        self._registrar.confirm(token)
+        member = self._mlist.regular_members.get_member('anne@example.com')
+        self.assertEqual(member.address, self._anne)
+
+    def test_confirm_because_confirm_then_moderation(self):
+        # We have a subscription request which requires the user to confirm
+        # (because she does not have a verified address) and the moderator to
+        # approve.  Running the workflow gives us a token.  Confirming the
+        # token runs the workflow a little farther, but still gives us a
+        # token.  Confirming again subscribes the user.
+        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 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(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)
+
+    def test_discard_waiting_for_confirmation(self):
+        # While waiting for a user to confirm their subscription, we discard
+        # the workflow.
+        self._mlist.subscription_policy = SubscriptionPolicy.confirm
+        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 discard the subscription request.
+        self._registrar.discard(token)
+        # Trying to confirm the token now results in an exception.
+        self.assertRaises(LookupError, self._registrar.confirm, token)
diff --git a/src/mailman/app/tests/test_registration.py b/src/mailman/app/tests/test_registration.py
deleted file mode 100644
index ccc485492..000000000
--- a/src/mailman/app/tests/test_registration.py
+++ /dev/null
@@ -1,128 +0,0 @@
-# Copyright (C) 2012-2015 by the Free Software Foundation, Inc.
-#
-# This file is part of GNU Mailman.
-#
-# GNU Mailman is free software: you can redistribute it and/or modify it under
-# the terms of the GNU General Public License as published by the Free
-# Software Foundation, either version 3 of the License, or (at your option)
-# any later version.
-#
-# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
-# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
-# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
-# more details.
-#
-# You should have received a copy of the GNU General Public License along with
-# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
-
-"""Test email address registration."""
-
-__all__ = [
-    'TestEmailValidation',
-    'TestRegistration',
-    ]
-
-
-import unittest
-
-from mailman.app.lifecycle import create_list
-from mailman.interfaces.address import InvalidEmailAddressError
-from mailman.interfaces.pending import IPendings
-from mailman.interfaces.registrar import ConfirmationNeededEvent, IRegistrar
-from mailman.testing.helpers import event_subscribers
-from mailman.testing.layers import ConfigLayer
-from zope.component import getUtility
-
-
-
-class TestEmailValidation(unittest.TestCase):
-    """Test basic email validation."""
-
-    layer = ConfigLayer
-
-    def setUp(self):
-        self.registrar = getUtility(IRegistrar)
-        self.mlist = create_list('alpha@example.com')
-
-    def test_empty_string_is_invalid(self):
-        self.assertRaises(InvalidEmailAddressError,
-                          self.registrar.register, self.mlist,
-                          '')
-
-    def test_no_spaces_allowed(self):
-        self.assertRaises(InvalidEmailAddressError,
-                          self.registrar.register, self.mlist,
-                          'some name@example.com')
-
-    def test_no_angle_brackets(self):
-        self.assertRaises(InvalidEmailAddressError,
-                          self.registrar.register, self.mlist,
-                          '<script>@example.com')
-
-    def test_ascii_only(self):
-        self.assertRaises(InvalidEmailAddressError,
-                          self.registrar.register, self.mlist,
-                          '\xa0@example.com')
-
-    def test_domain_required(self):
-        self.assertRaises(InvalidEmailAddressError,
-                          self.registrar.register, self.mlist,
-                          'noatsign')
-
-    def test_full_domain_required(self):
-        self.assertRaises(InvalidEmailAddressError,
-                          self.registrar.register, self.mlist,
-                          'nodom@ain')
-
-
-
-class TestRegistration(unittest.TestCase):
-    """Test registration."""
-
-    layer = ConfigLayer
-
-    def setUp(self):
-        self.registrar = getUtility(IRegistrar)
-        self.mlist = create_list('alpha@example.com')
-
-    def test_confirmation_event_received(self):
-        # Registering an email address generates an event.
-        def capture_event(event):
-            self.assertIsInstance(event, ConfirmationNeededEvent)
-        with event_subscribers(capture_event):
-            self.registrar.register(self.mlist, 'anne@example.com')
-
-    def test_event_mlist(self):
-        # The event has a reference to the mailing list being subscribed to.
-        def capture_event(event):
-            self.assertIs(event.mlist, self.mlist)
-        with event_subscribers(capture_event):
-            self.registrar.register(self.mlist, 'anne@example.com')
-
-    def test_event_pendable(self):
-        # The event has an IPendable which contains additional information.
-        def capture_event(event):
-            pendable = event.pendable
-            self.assertEqual(pendable['type'], 'registration')
-            self.assertEqual(pendable['email'], 'anne@example.com')
-            # The key is present, but the value is None.
-            self.assertIsNone(pendable['display_name'])
-            # The default is regular delivery.
-            self.assertEqual(pendable['delivery_mode'], 'regular')
-            self.assertEqual(pendable['list_id'], 'alpha.example.com')
-        with event_subscribers(capture_event):
-            self.registrar.register(self.mlist, 'anne@example.com')
-
-    def test_token(self):
-        # Registering the email address returns a token, and this token links
-        # back to the pendable.
-        captured_events = []
-        def capture_event(event):
-            captured_events.append(event)
-        with event_subscribers(capture_event):
-            token = self.registrar.register(self.mlist, 'anne@example.com')
-        self.assertEqual(len(captured_events), 1)
-        event = captured_events[0]
-        self.assertEqual(event.token, token)
-        pending = getUtility(IPendings).confirm(token)
-        self.assertEqual(pending, event.pendable)
diff --git a/src/mailman/app/tests/test_subscriptions.py b/src/mailman/app/tests/test_subscriptions.py
index 8ba5f52ff..a4971d793 100644
--- a/src/mailman/app/tests/test_subscriptions.py
+++ b/src/mailman/app/tests/test_subscriptions.py
@@ -18,7 +18,8 @@
 """Tests for the subscription service."""
 
 __all__ = [
-    'TestJoin'
+    'TestJoin',
+    'TestSubscriptionWorkflow',
     ]
 
 
@@ -26,11 +27,20 @@ import uuid
 import unittest
 
 from mailman.app.lifecycle import create_list
+from mailman.app.subscriptions import SubscriptionWorkflow
 from mailman.interfaces.address import InvalidEmailAddressError
-from mailman.interfaces.member import MemberRole, MissingPreferredAddressError
+from mailman.interfaces.bans import IBanManager
+from mailman.interfaces.member import (
+    MemberRole, MembershipIsBannedError, MissingPreferredAddressError)
+from mailman.interfaces.pending import IPendings
 from mailman.interfaces.subscriptions import (
     MissingUserError, ISubscriptionService)
+from mailman.testing.helpers import LogFileMark, get_queue_messages
 from mailman.testing.layers import ConfigLayer
+from mailman.interfaces.mailinglist import SubscriptionPolicy
+from mailman.interfaces.usermanager import IUserManager
+from mailman.utilities.datetime import now
+from unittest.mock import patch
 from zope.component import getUtility
 
 
@@ -65,3 +75,506 @@ class TestJoin(unittest.TestCase):
                           self._service.join,
                           'test.example.com', anne.user.user_id,
                           role=MemberRole.owner)
+
+
+
+class TestSubscriptionWorkflow(unittest.TestCase):
+    layer = ConfigLayer
+    maxDiff = None
+
+    def setUp(self):
+        self._mlist = create_list('test@example.com')
+        self._mlist.admin_immed_notify = False
+        self._anne = 'anne@example.com'
+        self._user_manager = getUtility(IUserManager)
+
+    def test_user_or_address_required(self):
+        # The `subscriber` attribute must be a user or address.
+        workflow = SubscriptionWorkflow(self._mlist)
+        self.assertRaises(AssertionError, list, workflow)
+
+    def test_sanity_checks_address(self):
+        # Ensure that the sanity check phase, when given an IAddress, ends up
+        # with a linked user.
+        anne = self._user_manager.create_address(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne)
+        self.assertIsNotNone(workflow.address)
+        self.assertIsNone(workflow.user)
+        workflow.run_thru('sanity_checks')
+        self.assertIsNotNone(workflow.address)
+        self.assertIsNotNone(workflow.user)
+        self.assertEqual(list(workflow.user.addresses)[0].email, self._anne)
+
+    def test_sanity_checks_user_with_preferred_address(self):
+        # Ensure that the sanity check phase, when given an IUser with a
+        # preferred address, ends up with an address.
+        anne = self._user_manager.make_user(self._anne)
+        address = list(anne.addresses)[0]
+        address.verified_on = now()
+        anne.preferred_address = address
+        workflow = SubscriptionWorkflow(self._mlist, anne)
+        # The constructor sets workflow.address because the user has a
+        # preferred address.
+        self.assertEqual(workflow.address, address)
+        self.assertEqual(workflow.user, anne)
+        workflow.run_thru('sanity_checks')
+        self.assertEqual(workflow.address, address)
+        self.assertEqual(workflow.user, anne)
+
+    def test_sanity_checks_user_without_preferred_address(self):
+        # Ensure that the sanity check phase, when given a user without a
+        # preferred address, but with at least one linked address, gets an
+        # address.
+        anne = self._user_manager.make_user(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne)
+        self.assertIsNone(workflow.address)
+        self.assertEqual(workflow.user, anne)
+        workflow.run_thru('sanity_checks')
+        self.assertIsNotNone(workflow.address)
+        self.assertEqual(workflow.user, anne)
+
+    def test_sanity_checks_user_with_multiple_linked_addresses(self):
+        # Ensure that the santiy check phase, when given a user without a
+        # preferred address, but with multiple linked addresses, gets of of
+        # those addresses (exactly which one is undefined).
+        anne = self._user_manager.make_user(self._anne)
+        anne.link(self._user_manager.create_address('anne@example.net'))
+        anne.link(self._user_manager.create_address('anne@example.org'))
+        workflow = SubscriptionWorkflow(self._mlist, anne)
+        self.assertIsNone(workflow.address)
+        self.assertEqual(workflow.user, anne)
+        workflow.run_thru('sanity_checks')
+        self.assertIn(workflow.address.email, ['anne@example.com',
+                                               'anne@example.net',
+                                               'anne@example.org'])
+        self.assertEqual(workflow.user, anne)
+
+    def test_sanity_checks_user_without_addresses(self):
+        # It is an error to try to subscribe a user with no linked addresses.
+        user = self._user_manager.create_user()
+        workflow = SubscriptionWorkflow(self._mlist, user)
+        self.assertRaises(AssertionError, workflow.run_thru, 'sanity_checks')
+
+    def test_sanity_checks_globally_banned_address(self):
+        # An exception is raised if the address is globally banned.
+        anne = self._user_manager.create_address(self._anne)
+        IBanManager(None).ban(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne)
+        self.assertRaises(MembershipIsBannedError, list, workflow)
+
+    def test_sanity_checks_banned_address(self):
+        # An exception is raised if the address is banned by the mailing list.
+        anne = self._user_manager.create_address(self._anne)
+        IBanManager(self._mlist).ban(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne)
+        self.assertRaises(MembershipIsBannedError, list, workflow)
+
+    def test_verification_checks_with_verified_address(self):
+        # When the address is already verified, we skip straight to the
+        # confirmation checks.
+        anne = self._user_manager.create_address(self._anne)
+        anne.verified_on = now()
+        workflow = SubscriptionWorkflow(self._mlist, anne)
+        workflow.run_thru('verification_checks')
+        with patch.object(workflow, '_step_confirmation_checks') as step:
+            next(workflow)
+        step.assert_called_once_with()
+
+    def test_verification_checks_with_pre_verified_address(self):
+        # When the address is not yet verified, but the pre-verified flag is
+        # passed to the workflow, we skip to the confirmation checks.
+        anne = self._user_manager.create_address(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne, pre_verified=True)
+        workflow.run_thru('verification_checks')
+        with patch.object(workflow, '_step_confirmation_checks') as step:
+            next(workflow)
+        step.assert_called_once_with()
+        # And now the address is verified.
+        self.assertIsNotNone(anne.verified_on)
+
+    def test_verification_checks_confirmation_needed(self):
+        # The address is neither verified, nor is the pre-verified flag set.
+        # A confirmation message must be sent to the user which will also
+        # verify their address.
+        anne = self._user_manager.create_address(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne)
+        workflow.run_thru('verification_checks')
+        with patch.object(workflow, '_step_send_confirmation') as step:
+            next(workflow)
+        step.assert_called_once_with()
+        # The address still hasn't been verified.
+        self.assertIsNone(anne.verified_on)
+
+    def test_confirmation_checks_open_list(self):
+        # A subscription to an open list does not need to be confirmed or
+        # moderated.
+        self._mlist.subscription_policy = SubscriptionPolicy.open
+        anne = self._user_manager.create_address(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne, pre_verified=True)
+        workflow.run_thru('confirmation_checks')
+        with patch.object(workflow, '_step_do_subscription') as step:
+            next(workflow)
+        step.assert_called_once_with()
+
+    def test_confirmation_checks_no_user_confirmation_needed(self):
+        # A subscription to a list which does not need user confirmation skips
+        # to the moderation checks.
+        self._mlist.subscription_policy = SubscriptionPolicy.moderate
+        anne = self._user_manager.create_address(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne, pre_verified=True)
+        workflow.run_thru('confirmation_checks')
+        with patch.object(workflow, '_step_moderation_checks') as step:
+            next(workflow)
+        step.assert_called_once_with()
+
+    def test_confirmation_checks_confirm_pre_confirmed(self):
+        # The subscription policy requires user confirmation, but their
+        # subscription is pre-confirmed.
+        self._mlist.subscription_policy = SubscriptionPolicy.confirm
+        anne = self._user_manager.create_address(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne,
+                                        pre_verified=True,
+                                        pre_confirmed=True)
+        workflow.run_thru('confirmation_checks')
+        with patch.object(workflow, '_step_moderation_checks') as step:
+            next(workflow)
+        step.assert_called_once_with()
+
+    def test_confirmation_checks_confirm_and_moderate_pre_confirmed(self):
+        # The subscription policy requires user confirmation and moderation,
+        # but their subscription is pre-confirmed.
+        self._mlist.subscription_policy = \
+          SubscriptionPolicy.confirm_then_moderate
+        anne = self._user_manager.create_address(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne,
+                                        pre_verified=True,
+                                        pre_confirmed=True)
+        workflow.run_thru('confirmation_checks')
+        with patch.object(workflow, '_step_moderation_checks') as step:
+            next(workflow)
+        step.assert_called_once_with()
+
+    def test_confirmation_checks_confirmation_needed(self):
+        # The subscription policy requires confirmation and the subscription
+        # is not pre-confirmed.
+        self._mlist.subscription_policy = SubscriptionPolicy.confirm
+        anne = self._user_manager.create_address(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne, pre_verified=True)
+        workflow.run_thru('confirmation_checks')
+        with patch.object(workflow, '_step_send_confirmation') as step:
+            next(workflow)
+        step.assert_called_once_with()
+
+    def test_confirmation_checks_moderate_confirmation_needed(self):
+        # The subscription policy requires confirmation and moderation, and the
+        # subscription is not pre-confirmed.
+        self._mlist.subscription_policy = \
+          SubscriptionPolicy.confirm_then_moderate
+        anne = self._user_manager.create_address(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne, pre_verified=True)
+        workflow.run_thru('confirmation_checks')
+        with patch.object(workflow, '_step_send_confirmation') as step:
+            next(workflow)
+        step.assert_called_once_with()
+
+    def test_moderation_checks_pre_approved(self):
+        # The subscription is pre-approved by the moderator.
+        self._mlist.subscription_policy = SubscriptionPolicy.moderate
+        anne = self._user_manager.create_address(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne,
+                                        pre_verified=True,
+                                        pre_approved=True)
+        workflow.run_thru('moderation_checks')
+        with patch.object(workflow, '_step_do_subscription') as step:
+            next(workflow)
+        step.assert_called_once_with()
+
+    def test_moderation_checks_approval_required(self):
+        # The moderator must approve the subscription.
+        self._mlist.subscription_policy = SubscriptionPolicy.moderate
+        anne = self._user_manager.create_address(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne, pre_verified=True)
+        workflow.run_thru('moderation_checks')
+        with patch.object(workflow, '_step_get_moderator_approval') as step:
+            next(workflow)
+        step.assert_called_once_with()
+
+    def test_do_subscription(self):
+        # An open subscription policy plus a pre-verified address means the
+        # user gets subscribed to the mailing list without any further
+        # confirmations or approvals.
+        self._mlist.subscription_policy = SubscriptionPolicy.open
+        anne = self._user_manager.create_address(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne, pre_verified=True)
+        # Consume the entire state machine.
+        list(workflow)
+        # Anne is now a member of the mailing list.
+        member = self._mlist.regular_members.get_member(self._anne)
+        self.assertEqual(member.address, anne)
+
+    def test_do_subscription_pre_approved(self):
+        # An moderation-requiring subscription policy plus a pre-verified and
+        # pre-approved address means the user gets subscribed to the mailing
+        # list without any further confirmations or approvals.
+        self._mlist.subscription_policy = SubscriptionPolicy.moderate
+        anne = self._user_manager.create_address(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne,
+                                        pre_verified=True,
+                                        pre_approved=True)
+        # Consume the entire state machine.
+        list(workflow)
+        # Anne is now a member of the mailing list.
+        member = self._mlist.regular_members.get_member(self._anne)
+        self.assertEqual(member.address, anne)
+
+    def test_do_subscription_pre_approved_pre_confirmed(self):
+        # An moderation-requiring subscription policy plus a pre-verified and
+        # pre-approved address means the user gets subscribed to the mailing
+        # list without any further confirmations or approvals.
+        self._mlist.subscription_policy = \
+          SubscriptionPolicy.confirm_then_moderate
+        anne = self._user_manager.create_address(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne,
+                                        pre_verified=True,
+                                        pre_confirmed=True,
+                                        pre_approved=True)
+        # Consume the entire state machine.
+        list(workflow)
+        # Anne is now a member of the mailing list.
+        member = self._mlist.regular_members.get_member(self._anne)
+        self.assertEqual(member.address, anne)
+
+    def test_do_subscription_cleanups(self):
+        # Once the user is subscribed, the token, and its associated pending
+        # database record will be removed from the database.
+        self._mlist.subscription_policy = SubscriptionPolicy.open
+        anne = self._user_manager.create_address(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne,
+                                        pre_verified=True,
+                                        pre_confirmed=True,
+                                        pre_approved=True)
+        # Cache the token.
+        token = workflow.token
+        # Consume the entire state machine.
+        list(workflow)
+        # Anne is now a member of the mailing list.
+        member = self._mlist.regular_members.get_member(self._anne)
+        self.assertEqual(member.address, anne)
+        # The workflow is done, so it has no token.
+        self.assertIsNone(workflow.token)
+        # 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
+        # as an exception when we try to restore the workflow.
+        new_workflow = SubscriptionWorkflow(self._mlist)
+        new_workflow.token = token
+        self.assertRaises(LookupError, new_workflow.restore)
+
+    def test_moderator_approves(self):
+        # The workflow runs until moderator approval is required, at which
+        # point the workflow is saved.  Once the moderator approves, the
+        # workflow resumes and the user is subscribed.
+        self._mlist.subscription_policy = SubscriptionPolicy.moderate
+        anne = self._user_manager.create_address(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne,
+                                        pre_verified=True,
+                                        pre_confirmed=True)
+        # Consume the entire state machine.
+        list(workflow)
+        # The user is not currently subscribed to the mailing list.
+        member = self._mlist.regular_members.get_member(self._anne)
+        self.assertIsNone(member)
+        # 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.
+        approved_workflow = SubscriptionWorkflow(self._mlist)
+        approved_workflow.token = workflow.token
+        approved_workflow.restore()
+        list(approved_workflow)
+        # Now the user is subscribed to the mailing list.
+        member = self._mlist.regular_members.get_member(self._anne)
+        self.assertEqual(member.address, anne)
+
+    def test_get_moderator_approval_log_on_hold(self):
+        # When the subscription is held for moderator approval, a message is
+        # logged.
+        mark = LogFileMark('mailman.subscribe')
+        self._mlist.subscription_policy = SubscriptionPolicy.moderate
+        anne = self._user_manager.create_address(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne,
+                                        pre_verified=True,
+                                        pre_confirmed=True)
+        # Consume the entire state machine.
+        list(workflow)
+        line = mark.readline()
+        self.assertEqual(
+            line[29:-1],
+            'test@example.com: held subscription request from anne@example.com'
+            )
+
+    def test_get_moderator_approval_notifies_moderators(self):
+        # When the subscription is held for moderator approval, and the list
+        # is so configured, a notification is sent to the list moderators.
+        self._mlist.admin_immed_notify = True
+        self._mlist.subscription_policy = SubscriptionPolicy.moderate
+        anne = self._user_manager.create_address(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne,
+                                        pre_verified=True,
+                                        pre_confirmed=True)
+        # Consume the entire state machine.
+        list(workflow)
+        items = get_queue_messages('virgin')
+        self.assertEqual(len(items), 1)
+        message = items[0].msg
+        self.assertEqual(message['From'], 'test-owner@example.com')
+        self.assertEqual(message['To'], 'test-owner@example.com')
+        self.assertEqual(
+            message['Subject'],
+            'New subscription request to Test from anne@example.com')
+        self.assertEqual(message.get_payload(), """\
+Your authorization is required for a mailing list subscription request
+approval:
+
+    For:  anne@example.com
+    List: test@example.com""")
+
+    def test_get_moderator_approval_no_notifications(self):
+        # When the subscription is held for moderator approval, and the list
+        # is so configured, a notification is sent to the list moderators.
+        self._mlist.admin_immed_notify = False
+        self._mlist.subscription_policy = SubscriptionPolicy.moderate
+        anne = self._user_manager.create_address(self._anne)
+        workflow = SubscriptionWorkflow(self._mlist, anne,
+                                        pre_verified=True,
+                                        pre_confirmed=True)
+        # Consume the entire state machine.
+        list(workflow)
+        items = get_queue_messages('virgin')
+        self.assertEqual(len(items), 0)
+
+    def test_send_confirmation(self):
+        # A confirmation message gets sent when the address is not verified.
+        anne = self._user_manager.create_address(self._anne)
+        self.assertIsNone(anne.verified_on)
+        # Run the workflow to model the confirmation step.
+        workflow = SubscriptionWorkflow(self._mlist, anne)
+        list(workflow)
+        items = get_queue_messages('virgin')
+        self.assertEqual(len(items), 1)
+        message = items[0].msg
+        token = workflow.token
+        self.assertEqual(message['Subject'], 'confirm {}'.format(token))
+        self.assertEqual(
+            message['From'], 'test-confirm+{}@example.com'.format(token))
+
+    def test_send_confirmation_pre_confirmed(self):
+        # A confirmation message gets sent when the address is not verified
+        # but the subscription is pre-confirmed.
+        anne = self._user_manager.create_address(self._anne)
+        self.assertIsNone(anne.verified_on)
+        # Run the workflow to model the confirmation step.
+        workflow = SubscriptionWorkflow(self._mlist, anne, pre_confirmed=True)
+        list(workflow)
+        items = get_queue_messages('virgin')
+        self.assertEqual(len(items), 1)
+        message = items[0].msg
+        token = workflow.token
+        self.assertEqual(
+            message['Subject'], 'confirm {}'.format(workflow.token))
+        self.assertEqual(
+            message['From'], 'test-confirm+{}@example.com'.format(token))
+
+    def test_send_confirmation_pre_verified(self):
+        # A confirmation message gets sent even when the address is verified
+        # when the subscription must be confirmed.
+        self._mlist.subscription_policy = SubscriptionPolicy.confirm
+        anne = self._user_manager.create_address(self._anne)
+        self.assertIsNone(anne.verified_on)
+        # Run the workflow to model the confirmation step.
+        workflow = SubscriptionWorkflow(self._mlist, anne, pre_verified=True)
+        list(workflow)
+        items = get_queue_messages('virgin')
+        self.assertEqual(len(items), 1)
+        message = items[0].msg
+        token = workflow.token
+        self.assertEqual(
+            message['Subject'], 'confirm {}'.format(workflow.token))
+        self.assertEqual(
+            message['From'], 'test-confirm+{}@example.com'.format(token))
+
+    def test_do_confirm_verify_address(self):
+        # The address is not yet verified, nor are we pre-verifying.  A
+        # confirmation message will be sent.  When the user confirms their
+        # subscription request, the address will end up being verified.
+        anne = self._user_manager.create_address(self._anne)
+        self.assertIsNone(anne.verified_on)
+        # Run the workflow to model the confirmation step.
+        workflow = SubscriptionWorkflow(self._mlist, anne)
+        list(workflow)
+        # The address is still not verified.
+        self.assertIsNone(anne.verified_on)
+        confirm_workflow = SubscriptionWorkflow(self._mlist)
+        confirm_workflow.token = workflow.token
+        confirm_workflow.restore()
+        confirm_workflow.run_thru('do_confirm_verify')
+        # The address is now verified.
+        self.assertIsNotNone(anne.verified_on)
+
+    def test_do_confirmation_subscribes_user(self):
+        # Subscriptions to the mailing list must be confirmed.  Once that's
+        # done, the user's address (which is not initially verified) gets
+        # subscribed to the mailing list.
+        self._mlist.subscription_policy = SubscriptionPolicy.confirm
+        anne = self._user_manager.create_address(self._anne)
+        self.assertIsNone(anne.verified_on)
+        workflow = SubscriptionWorkflow(self._mlist, anne)
+        list(workflow)
+        self.assertIsNone(self._mlist.regular_members.get_member(self._anne))
+        confirm_workflow = SubscriptionWorkflow(self._mlist)
+        confirm_workflow.token = workflow.token
+        confirm_workflow.restore()
+        list(confirm_workflow)
+        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)
diff --git a/src/mailman/app/tests/test_workflow.py b/src/mailman/app/tests/test_workflow.py
new file mode 100644
index 000000000..51beceb86
--- /dev/null
+++ b/src/mailman/app/tests/test_workflow.py
@@ -0,0 +1,128 @@
+# Copyright (C) 2015 by the Free Software Foundation, Inc.
+#
+# This file is part of GNU Mailman.
+#
+# GNU Mailman is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option)
+# any later version.
+#
+# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
+# more details.
+#
+# You should have received a copy of the GNU General Public License along with
+# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
+
+"""App-level workflow tests."""
+
+__all__ = [
+    'TestWorkflow',
+    ]
+
+
+import unittest
+
+from mailman.app.workflow import Workflow
+from mailman.testing.layers import ConfigLayer
+
+
+class MyWorkflow(Workflow):
+    INITIAL_STATE = 'first'
+    SAVE_ATTRIBUTES = ('ant', 'bee', 'cat')
+
+    def __init__(self):
+        super().__init__()
+        self.token = 'test-workflow'
+        self.ant = 1
+        self.bee = 2
+        self.cat = 3
+        self.dog = 4
+
+    def _step_first(self):
+        self.push('second')
+        return 'one'
+
+    def _step_second(self):
+        self.push('third')
+        return 'two'
+
+    def _step_third(self):
+        return 'three'
+
+
+
+class TestWorkflow(unittest.TestCase):
+    layer = ConfigLayer
+
+    def setUp(self):
+        self._workflow = iter(MyWorkflow())
+
+    def test_basic_workflow(self):
+        # The work flows from one state to the next.
+        results = list(self._workflow)
+        self.assertEqual(results, ['one', 'two', 'three'])
+
+    def test_partial_workflow(self):
+        # You don't have to flow through every step.
+        results = next(self._workflow)
+        self.assertEqual(results, 'one')
+
+    def test_exhaust_workflow(self):
+        # Manually flow through a few steps, then consume the whole thing.
+        results = [next(self._workflow)]
+        results.extend(self._workflow)
+        self.assertEqual(results, ['one', 'two', 'three'])
+
+    def test_save_and_restore_workflow(self):
+        # Without running any steps, save and restore the workflow.  Then
+        # consume the restored workflow.
+        self._workflow.save()
+        new_workflow = MyWorkflow()
+        new_workflow.restore()
+        results = list(new_workflow)
+        self.assertEqual(results, ['one', 'two', 'three'])
+
+    def test_save_and_restore_partial_workflow(self):
+        # After running a few steps, save and restore the workflow.  Then
+        # consume the restored workflow.
+        next(self._workflow)
+        self._workflow.save()
+        new_workflow = MyWorkflow()
+        new_workflow.restore()
+        results = list(new_workflow)
+        self.assertEqual(results, ['two', 'three'])
+
+    def test_save_and_restore_exhausted_workflow(self):
+        # After consuming the entire workflow, save and restore it.
+        list(self._workflow)
+        self._workflow.save()
+        new_workflow = MyWorkflow()
+        new_workflow.restore()
+        results = list(new_workflow)
+        self.assertEqual(len(results), 0)
+
+    def test_save_and_restore_attributes(self):
+        # Saved attributes are restored.
+        self._workflow.ant = 9
+        self._workflow.bee = 8
+        self._workflow.cat = 7
+        # Don't save .dog.
+        self._workflow.save()
+        new_workflow = MyWorkflow()
+        new_workflow.restore()
+        self.assertEqual(new_workflow.ant, 9)
+        self.assertEqual(new_workflow.bee, 8)
+        self.assertEqual(new_workflow.cat, 7)
+        self.assertEqual(new_workflow.dog, 4)
+
+    def test_run_thru(self):
+        # Run all steps through the given one.
+        results = self._workflow.run_thru('second')
+        self.assertEqual(results, ['one', 'two'])
+
+    def test_run_until(self):
+        # Run until (but not including) the given step.
+        results = self._workflow.run_until('second')
+        self.assertEqual(results, ['one'])
diff --git a/src/mailman/app/workflow.py b/src/mailman/app/workflow.py
new file mode 100644
index 000000000..b83d1c3aa
--- /dev/null
+++ b/src/mailman/app/workflow.py
@@ -0,0 +1,156 @@
+# Copyright (C) 2015 by the Free Software Foundation, Inc.
+#
+# This file is part of GNU Mailman.
+#
+# GNU Mailman is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option)
+# any later version.
+#
+# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
+# more details.
+#
+# You should have received a copy of the GNU General Public License along with
+# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Generic workflow."""
+
+__all__ = [
+    'Workflow',
+    ]
+
+
+import sys
+import json
+import logging
+
+from collections import deque
+from mailman.interfaces.workflow import IWorkflowStateManager
+from zope.component import getUtility
+
+
+COMMASPACE = ', '
+log = logging.getLogger('mailman.error')
+
+
+
+class Workflow:
+    """Generic workflow."""
+
+    SAVE_ATTRIBUTES = ()
+    INITIAL_STATE = None
+
+    def __init__(self):
+        self.token = None
+        self._next = deque()
+        self.push(self.INITIAL_STATE)
+        self.debug = False
+        self._count = 0
+
+    @property
+    def name(self):
+        return self.__class__.__name__
+
+    def __iter__(self):
+        return self
+
+    def push(self, step):
+        self._next.append(step)
+
+    def _pop(self):
+        name = self._next.popleft()
+        step = getattr(self, '_step_{}'.format(name))
+        self._count += 1
+        if self.debug:
+            print('[{:02d}] -> {}'.format(self._count, name), file=sys.stderr)
+        return name, step
+
+    def __next__(self):
+        try:
+            name, step = self._pop()
+            return step()
+        except IndexError:
+            raise StopIteration
+        except:
+            log.exception('deque: {}'.format(COMMASPACE.join(self._next)))
+            raise
+
+    def run_thru(self, stop_after):
+        """Run the state machine through and including the given step.
+
+        :param stop_after: Name of method, sans prefix to run the
+            state machine through.  In other words, the state machine runs
+            until the named method completes.
+        """
+        results = []
+        while True:
+            try:
+                name, step = self._pop()
+            except (StopIteration, IndexError):
+                # We're done.
+                break
+            results.append(step())
+            if name == stop_after:
+                break
+        return results
+
+    def run_until(self, stop_before):
+        """Trun the state machine until (not including) the given step.
+
+        :param stop_before: Name of method, sans prefix that the
+            state machine is run until the method is reached.  Unlike
+            `run_thru()` the named method is not run.
+        """
+        results = []
+        while True:
+            try:
+                name, step = self._pop()
+            except (StopIteration, IndexError):
+                # We're done.
+                break
+            if name == stop_before:
+                # Stop executing, but not before we push the last state back
+                # onto the deque.  Otherwise, resuming the state machine would
+                # skip this step.
+                self._next.appendleft(step)
+                break
+            results.append(step())
+        return results
+
+    def save(self):
+        assert self.token, 'Workflow token must be set'
+        state_manager = getUtility(IWorkflowStateManager)
+        data = {attr: getattr(self, attr) for attr in self.SAVE_ATTRIBUTES}
+        # Note: only the next step is saved, not the whole stack.  This is not
+        # an issue in practice, since there's never more than a single step in
+        # the queue anyway.  If we want to support more than a single step in
+        # the queue *and* want to support state saving/restoring, change this
+        # method and the restore() method.
+        if len(self._next) == 0:
+            step = None
+        elif len(self._next) == 1:
+            step = self._next[0]
+        else:
+            raise AssertionError(
+                "Can't save a workflow state with more than one step "
+                "in the queue")
+        state_manager.save(
+            self.__class__.__name__,
+            self.token,
+            step,
+            json.dumps(data))
+
+    def restore(self):
+        state_manager = getUtility(IWorkflowStateManager)
+        state = state_manager.restore(self.__class__.__name__, self.token)
+        if state is None:
+            # The token doesn't exist in the database.
+            raise LookupError(self.token)
+        self._next.clear()
+        if state.step:
+            self._next.append(state.step)
+        if state.data is not None:
+            for attr, value in json.loads(state.data).items():
+                setattr(self, attr, value)
diff --git a/src/mailman/commands/docs/membership.rst b/src/mailman/commands/docs/membership.rst
index bdebba8f7..49e80511d 100644
--- a/src/mailman/commands/docs/membership.rst
+++ b/src/mailman/commands/docs/membership.rst
@@ -70,7 +70,7 @@ The ``subscribe`` command is an alias.
 Joining the sender
 ------------------
 
-When the message has a From field, that address will be subscribed.
+When the message has a ``From`` field, that address will be subscribed.
 
     >>> msg = message_from_string("""\
     ... From: Anne Person <anne@example.com>
@@ -85,13 +85,10 @@ When the message has a From field, that address will be subscribed.
     Confirmation email sent to Anne Person <anne@example.com>
     <BLANKLINE>
 
-Anne is not yet a member because she must confirm her subscription request
-first.
+Anne is not yet a member of the mailing list because she must confirm her
+subscription request first.
 
-    >>> from mailman.interfaces.usermanager import IUserManager
-    >>> from zope.component import getUtility
-    >>> user_manager = getUtility(IUserManager)
-    >>> print(user_manager.get_user('anne@example.com'))
+    >>> print(mlist.members.get_member('anne@example.com'))
     None
 
 Mailman has sent her the confirmation message.
@@ -118,10 +115,7 @@ Mailman has sent her the confirmation message.
     <BLANKLINE>
     Before you can start using GNU Mailman at this site, you must first
     confirm that this is your email address.  You can do this by replying to
-    this message, keeping the Subject header intact.  Or you can visit this
-    web page
-    <BLANKLINE>
-        http://lists.example.com/confirm/...
+    this message, keeping the Subject header intact.
     <BLANKLINE>
     If you do not wish to register this email address simply disregard this
     message.  If you think you are being maliciously subscribed to the list, or
@@ -130,8 +124,7 @@ Mailman has sent her the confirmation message.
         alpha-owner@example.com
     <BLANKLINE>
 
-Once Anne confirms her registration, she will be made a member of the mailing
-list.
+Anne confirms her registration.
 ::
 
     >>> def extract_token(message):
@@ -156,13 +149,7 @@ list.
     Confirmed
     <BLANKLINE>
 
-    >>> user = user_manager.get_user('anne@example.com')
-    >>> print(user.display_name)
-    Anne Person
-    >>> list(user.addresses)
-    [<Address: Anne Person <anne@example.com> [verified] at ...>]
-
-Anne is also now a member of the mailing list.
+Anne is now a member of the mailing list.
 
     >>> mlist.members.get_member('anne@example.com')
     <Member: Anne Person <anne@example.com>
@@ -180,12 +167,7 @@ Joining a second list
     >>> print(join.process(mlist_2, msg, {}, (), Results()))
     ContinueProcessing.yes
 
-Anne of course, is still registered.
-
-    >>> print(user_manager.get_user('anne@example.com'))
-    <User "Anne Person" (...) at ...>
-
-But she is not a member of the mailing list.
+Anne is not a member of the mailing list.
 
     >>> print(mlist_2.members.get_member('anne@example.com'))
     None
@@ -257,7 +239,9 @@ subscribe with.  Any of her registered, linked, and validated email addresses
 will do.
 ::
 
-    >>> anne = user_manager.get_user('anne@example.com')
+    >>> from mailman.interfaces.usermanager import IUserManager
+    >>> from zope.component import getUtility
+    >>> anne = getUtility(IUserManager).get_user('anne@example.com')
     >>> address = anne.register('anne.person@example.org')
 
     >>> results = Results()
@@ -333,11 +317,6 @@ message.
     ...     raise AssertionError('No confirmation message')
     >>> token = extract_token(item.msg)
 
-Bart is still not a user.
-
-    >>> print(user_manager.get_user('bart@example.com'))
-    None
-
 Bart replies to the original message, specifically keeping the Subject header
 intact except for any prefix.  Mailman matches the token and confirms Bart as
 a user of the system.
@@ -360,12 +339,7 @@ a user of the system.
     Confirmed
     <BLANKLINE>
 
-Now Bart is a user...
-
-    >>> print(user_manager.get_user('bart@example.com'))
-    <User "Bart Person" (...) at ...>
-
-...and a member of the mailing list.
+Now Bart is now a member of the mailing list.
 
     >>> print(mlist.members.get_member('bart@example.com'))
     <Member: Bart Person <bart@example.com>
diff --git a/src/mailman/commands/eml_confirm.py b/src/mailman/commands/eml_confirm.py
index a28a3f728..077fab9a6 100644
--- a/src/mailman/commands/eml_confirm.py
+++ b/src/mailman/commands/eml_confirm.py
@@ -25,7 +25,6 @@ __all__ = [
 from mailman.core.i18n import _
 from mailman.interfaces.command import ContinueProcessing, IEmailCommand
 from mailman.interfaces.registrar import IRegistrar
-from zope.component import getUtility
 from zope.interface import implementer
 
 
@@ -53,7 +52,11 @@ class Confirm:
             return ContinueProcessing.yes
         tokens.add(token)
         results.confirms = tokens
-        succeeded = getUtility(IRegistrar).confirm(token)
+        try:
+            succeeded = (IRegistrar(mlist).confirm(token) is None)
+        except LookupError:
+            # The token must not exist in the database.
+            succeeded = False
         if succeeded:
             print(_('Confirmed'), file=results)
             return ContinueProcessing.yes
diff --git a/src/mailman/commands/eml_membership.py b/src/mailman/commands/eml_membership.py
index 059b9b634..970fd4429 100644
--- a/src/mailman/commands/eml_membership.py
+++ b/src/mailman/commands/eml_membership.py
@@ -37,6 +37,28 @@ from zope.interface import implementer
 
 
 
+def match_subscriber(email, display_name):
+    # Return something matching the email which should be used as the
+    # subscriber by the IRegistrar interface.
+    manager = getUtility(IUserManager)
+    # Is there a user with a preferred address matching the email?
+    user = manager.get_user(email)
+    if user is not None:
+        preferred = user.preferred_address
+        if preferred is not None and preferred.email == email.lower():
+            return user
+    # Is there an address matching the email?
+    address = manager.get_address(email)
+    if address is not None:
+        return address
+    # Make a new user and subscribe their first (and only) address.  We can't
+    # make the first address their preferred address because it hasn't been
+    # verified yet.
+    user = manager.make_user(email, display_name)
+    return list(user.addresses)[0]
+
+
+
 @implementer(IEmailCommand)
 class Join:
     """The email 'join' command."""
@@ -60,35 +82,35 @@ used.
         delivery_mode = self._parse_arguments(arguments, results)
         if delivery_mode is ContinueProcessing.no:
             return ContinueProcessing.no
-        display_name, address = parseaddr(msg['from'])
+        display_name, email = parseaddr(msg['from'])
         # Address could be None or the empty string.
-        if not address:
-            address = msg.sender
-        if not address:
+        if not email:
+            email = msg.sender
+        if not email:
             print(_('$self.name: No valid address found to subscribe'),
                   file=results)
             return ContinueProcessing.no
-        if isinstance(address, bytes):
-            address = address.decode('ascii')
+        if isinstance(email, bytes):
+            email = email.decode('ascii')
         # Have we already seen one join request from this user during the
         # processing of this email?
         joins = getattr(results, 'joins', set())
-        if address in joins:
+        if email in joins:
             # Do not register this join.
             return ContinueProcessing.yes
-        joins.add(address)
+        joins.add(email)
         results.joins = joins
-        person = formataddr((display_name, address))
+        person = formataddr((display_name, email))
         # Is this person already a member of the list?  Search for all
         # matching memberships.
         members = getUtility(ISubscriptionService).find_members(
-            address, mlist.list_id, MemberRole.member)
+            email, mlist.list_id, MemberRole.member)
         if len(members) > 0:
             print(_('$person is already a member'), file=results)
-        else:
-            getUtility(IRegistrar).register(mlist, address,
-                                            display_name, delivery_mode)
-            print(_('Confirmation email sent to $person'), file=results)
+            return ContinueProcessing.yes
+        subscriber = match_subscriber(email, display_name)
+        IRegistrar(mlist).register(subscriber)
+        print(_('Confirmation email sent to $person'), file=results)
         return ContinueProcessing.yes
 
     def _parse_arguments(self, arguments, results):
diff --git a/src/mailman/commands/tests/test_confirm.py b/src/mailman/commands/tests/test_confirm.py
index dd168454f..98a26bf7d 100644
--- a/src/mailman/commands/tests/test_confirm.py
+++ b/src/mailman/commands/tests/test_confirm.py
@@ -29,8 +29,9 @@ from mailman.commands.eml_confirm import Confirm
 from mailman.email.message import Message
 from mailman.interfaces.command import ContinueProcessing
 from mailman.interfaces.registrar import IRegistrar
+from mailman.interfaces.usermanager import IUserManager
 from mailman.runners.command import Results
-from mailman.testing.helpers import get_queue_messages, reset_the_world
+from mailman.testing.helpers import get_queue_messages
 from mailman.testing.layers import ConfigLayer
 from zope.component import getUtility
 
@@ -43,15 +44,13 @@ class TestConfirm(unittest.TestCase):
 
     def setUp(self):
         self._mlist = create_list('test@example.com')
-        self._token = getUtility(IRegistrar).register(
-            self._mlist, 'anne@example.com', 'Anne Person')
+        anne = getUtility(IUserManager).create_address(
+            'anne@example.com', 'Anne Person')
+        self._token = IRegistrar(self._mlist).register(anne)
         self._command = Confirm()
         # Clear the virgin queue.
         get_queue_messages('virgin')
 
-    def tearDown(self):
-        reset_the_world()
-
     def test_welcome_message(self):
         # A confirmation causes a welcome message to be sent to the member, if
         # enabled by the mailing list.
diff --git a/src/mailman/config/configure.zcml b/src/mailman/config/configure.zcml
index 24061f0f0..632771d42 100644
--- a/src/mailman/config/configure.zcml
+++ b/src/mailman/config/configure.zcml
@@ -40,6 +40,12 @@
     factory="mailman.model.requests.ListRequests"
     />
 
+  <adapter
+    for="mailman.interfaces.mailinglist.IMailingList"
+    provides="mailman.interfaces.registrar.IRegistrar"
+    factory="mailman.app.registrar.Registrar"
+    />
+
   <utility
     provides="mailman.interfaces.bounce.IBounceProcessor"
     factory="mailman.model.bounce.BounceProcessor"
@@ -88,11 +94,6 @@
     />
 
   <utility
-    provides="mailman.interfaces.registrar.IRegistrar"
-    factory="mailman.app.registrar.Registrar"
-    />
-
-  <utility
    provides="mailman.interfaces.styles.IStyleManager"
    factory="mailman.styles.manager.StyleManager"
    />
@@ -117,4 +118,9 @@
    factory="mailman.app.templates.TemplateLoader"
    />
 
+  <utility
+    provides="mailman.interfaces.workflow.IWorkflowStateManager"
+    factory="mailman.model.workflow.WorkflowStateManager"
+    />
+
 </configure>
diff --git a/src/mailman/database/alembic/versions/16c2b25c7b_list_subscription_policy.py b/src/mailman/database/alembic/versions/16c2b25c7b_list_subscription_policy.py
new file mode 100644
index 000000000..8534f4b73
--- /dev/null
+++ b/src/mailman/database/alembic/versions/16c2b25c7b_list_subscription_policy.py
@@ -0,0 +1,41 @@
+"""List subscription policy
+
+Revision ID: 16c2b25c7b
+Revises: 46e92facee7
+Create Date: 2015-03-21 11:00:44.634883
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '16c2b25c7b'
+down_revision = '46e92facee7'
+
+from alembic import op
+import sqlalchemy as sa
+
+from mailman.database.types import Enum
+from mailman.interfaces.mailinglist import SubscriptionPolicy
+
+
+def upgrade():
+
+    ### Update the schema
+    op.add_column('mailinglist', sa.Column(
+        'subscription_policy', Enum(SubscriptionPolicy), nullable=True))
+
+    ### Now migrate the data
+    # don't import the table definition from the models, it may break this
+    # migration when the model is updated in the future (see the Alembic doc)
+    mlist = sa.sql.table('mailinglist',
+        sa.sql.column('subscription_policy', Enum(SubscriptionPolicy))
+        )
+    # there were no enforced subscription policy before, so all lists are
+    # considered open
+    op.execute(mlist.update().values(
+        {'subscription_policy': op.inline_literal(SubscriptionPolicy.open)}))
+
+
+def downgrade():
+    if op.get_bind().dialect.name != 'sqlite':
+        # SQLite does not support dropping columns.
+        op.drop_column('mailinglist', 'subscription_policy')
diff --git a/src/mailman/database/alembic/versions/2bb9b382198_workflow_state_table.py b/src/mailman/database/alembic/versions/2bb9b382198_workflow_state_table.py
new file mode 100644
index 000000000..59cb1121e
--- /dev/null
+++ b/src/mailman/database/alembic/versions/2bb9b382198_workflow_state_table.py
@@ -0,0 +1,28 @@
+"""Workflow state table
+
+Revision ID: 2bb9b382198
+Revises: 16c2b25c7b
+Create Date: 2015-03-25 18:09:18.338790
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '2bb9b382198'
+down_revision = '16c2b25c7b'
+
+from alembic import op
+import sqlalchemy as sa
+
+
+def upgrade():
+    op.create_table('workflowstate',
+        sa.Column('name', sa.Unicode(), nullable=False),
+        sa.Column('key', sa.Unicode(), nullable=False),
+        sa.Column('step', sa.Unicode(), nullable=True),
+        sa.Column('data', sa.Unicode(), nullable=True),
+        sa.PrimaryKeyConstraint('name', 'key')
+        )
+
+
+def downgrade():
+    op.drop_table('workflowstate')
diff --git a/src/mailman/database/transaction.py b/src/mailman/database/transaction.py
index b96562d3f..fef76b73c 100644
--- a/src/mailman/database/transaction.py
+++ b/src/mailman/database/transaction.py
@@ -19,6 +19,7 @@
 
 __all__ = [
     'dbconnection',
+    'flush',
     'transaction',
     'transactional',
     ]
@@ -63,6 +64,22 @@ def transactional(function):
 
 
 
+@contextmanager
+def flush():
+    """Context manager for flushing SQLAlchemy.
+
+    We need this for SA whereas we didn't need it for Storm because the latter
+    did auto-reloads.  However, in SA this is needed when we add or delete
+    objects from the database.  Use it when you need the id after adding, or
+    when you want to be sure the object won't be found after a delete.
+
+    This is lighter weight than committing the transaction.
+    """
+    yield
+    config.db.store.flush()
+
+
+
 def dbconnection(function):
     """Decorator for getting at the database connection.
 
diff --git a/src/mailman/docs/NEWS.rst b/src/mailman/docs/NEWS.rst
index 68ebe6143..ebe90a6df 100644
--- a/src/mailman/docs/NEWS.rst
+++ b/src/mailman/docs/NEWS.rst
@@ -19,6 +19,10 @@ Architecture
    also have a ``is_server_owner`` flag (defaulting to False) to indicate
    whether they have superuser privileges.  Give by Abhliash Raj, with fixes
    and refinements by Barry Warsaw.  (LP: #1423756)
+ * Mailing list subscription policy work flow has been completely rewritten.
+   It now properly supports email verification and subscription confirmation
+   by the user, and approval by the moderator using unique tokens.
+   ``IMailingList`` objects now have a ``subscription_policy`` attribute.
 
 Bugs
 ----
diff --git a/src/mailman/interfaces/domain.py b/src/mailman/interfaces/domain.py
index 33d835325..d5cbf8e2b 100644
--- a/src/mailman/interfaces/domain.py
+++ b/src/mailman/interfaces/domain.py
@@ -89,7 +89,7 @@ class IDomain(Interface):
         'The human readable description of the domain name.')
 
     owners = Attribute("""\
-        The relationship with the user database representing domain owners""")
+        The relationship with the user database representing domain owners.""")
 
     mailing_lists = Attribute(
         """All mailing lists for this domain.
diff --git a/src/mailman/interfaces/mailinglist.py b/src/mailman/interfaces/mailinglist.py
index 23d2fadf4..f112b2a11 100644
--- a/src/mailman/interfaces/mailinglist.py
+++ b/src/mailman/interfaces/mailinglist.py
@@ -25,6 +25,7 @@ __all__ = [
     'IMailingList',
     'Personalization',
     'ReplyToMunging',
+    'SubscriptionPolicy',
     ]
 
 
@@ -53,6 +54,18 @@ class ReplyToMunging(Enum):
     explicit_header = 2
 
 
+class SubscriptionPolicy(Enum):
+    # Neither confirmation, nor moderator approval is required.
+    open = 0
+    # The user must confirm the subscription.
+    confirm = 1
+    # The moderator must approve the subscription.
+    moderate = 2
+    # The user must first confirm their subscription, and then if that is
+    # successful, the moderator must also approve it.
+    confirm_then_moderate = 3
+
+
 
 class IMailingList(Interface):
     """A mailing list."""
@@ -234,6 +247,9 @@ class IMailingList(Interface):
         deliver disabled or not, or of the type of digest they are to
         receive.""")
 
+    subscription_policy = Attribute(
+        """The policy for subscribing new members to the list.""")
+
     subscribers = Attribute(
         """An iterator over all IMembers subscribed to this list, with any
         role.
diff --git a/src/mailman/interfaces/member.py b/src/mailman/interfaces/member.py
index c06cc95b1..d863e1ef1 100644
--- a/src/mailman/interfaces/member.py
+++ b/src/mailman/interfaces/member.py
@@ -123,7 +123,7 @@ class MembershipIsBannedError(MembershipError):
     """The address is not allowed to subscribe to the mailing list."""
 
     def __init__(self, mlist, address):
-        super(MembershipIsBannedError, self).__init__()
+        super().__init__()
         self._mlist = mlist
         self._address = address
 
@@ -175,6 +175,14 @@ class IMember(Interface):
     user = Attribute(
         """The user associated with this member.""")
 
+    subscriber = Attribute(
+        """The object representing how this member is subscribed.
+
+        This will be an ``IAddress`` if the user is subscribed via an explicit
+        address, otherwise if the the user is subscribed via their preferred
+        address, it will be an ``IUser``.
+        """)
+
     preferences = Attribute(
         """This member's preferences.""")
 
diff --git a/src/mailman/interfaces/pending.py b/src/mailman/interfaces/pending.py
index 9907aa779..222f0dfbf 100644
--- a/src/mailman/interfaces/pending.py
+++ b/src/mailman/interfaces/pending.py
@@ -82,11 +82,11 @@ class IPendings(Interface):
         :return: A token string for inclusion in urls and email confirmations.
         """
 
-    def confirm(token, expunge=True):
+    def confirm(token, *, expunge=True):
         """Return the IPendable matching the token.
 
         :param token: The token string for the IPendable given by the `.add()`
-            method.
+            method, or None if there is no record associated with the token.
         :param expunge: A flag indicating whether the pendable record should
             also be removed from the database or not.
         :return: The matching IPendable or None if no match was found.
@@ -94,3 +94,5 @@ class IPendings(Interface):
 
     def evict():
         """Remove all pended items whose lifetime has expired."""
+
+    count = Attribute('The number of pendables in the pendings database.')
diff --git a/src/mailman/interfaces/registrar.py b/src/mailman/interfaces/registrar.py
index 7d3cf9c25..ff3f26898 100644
--- a/src/mailman/interfaces/registrar.py
+++ b/src/mailman/interfaces/registrar.py
@@ -35,79 +35,75 @@ from zope.interface import Interface
 class ConfirmationNeededEvent:
     """Triggered when an address needs confirmation.
 
-    Addresses must be verified before they can receive messages or post to
-    mailing list.  When an address is registered with Mailman, via the
-    `IRegistrar` interface, an `IPendable` is created which represents the
-    pending registration.  This pending registration is stored in the
-    database, keyed by a token.  Then this event is triggered.
-
-    There may be several ways to confirm an email address.  On some sites,
-    registration may immediately produce a verification, e.g. because it is on
-    a known intranet.  Or verification may occur via external database lookup
-    (e.g. LDAP).  On most public mailing lists, a mail-back confirmation is
-    sent to the address, and only if they reply to the mail-back, or click on
-    an embedded link, is the registered address confirmed.
+    Addresses must be verified before they can receive messages or post
+    to mailing list.  The confirmation message is sent to the user when
+    this event is triggered.
     """
-    def __init__(self, mlist, pendable, token):
+    def __init__(self, mlist, token, email):
         self.mlist = mlist
-        self.pendable = pendable
         self.token = token
+        self.email = email
 
 
 
 class IRegistrar(Interface):
-    """Interface for registering and verifying email addresses and users.
+    """Interface for subscribing addresses and users.
 
     This is a higher level interface to user registration, email address
     confirmation, etc. than the IUserManager.  The latter does no validation,
     syntax checking, or confirmation, while this interface does.
     """
 
-    def register(mlist, email, display_name=None, delivery_mode=None):
-        """Register the email address, requesting verification.
+    def register(mlist, subscriber=None, *,
+                 pre_verified=False, pre_confirmed=False, pre_approved=False):
+        """Subscribe an address or user according to subscription policies.
 
-        No `IAddress` or `IUser` is created during this step, but after
-        successful confirmation, it is guaranteed that an `IAddress` with a
-        linked `IUser` will exist.  When a verified `IAddress` matching
-        `email` already exists, this method will do nothing, except link a new
-        `IUser` to the `IAddress` if one is not yet associated with the
-        email address.
+        The mailing list's subscription policy is used to subscribe
+        `subscriber` to the given mailing list.  The subscriber can be
+        an ``IUser``, in which case the user must have a preferred
+        address, and that preferred address will be subscribed.  The
+        subscriber can also be an ``IAddress``, in which case the
+        address will be subscribed.
 
-        In all cases, the email address is sanity checked for validity first.
+        The workflow may pause (i.e. be serialized, saved, and
+        suspended) when some out-of-band confirmation step is required.
+        For example, if the user must confirm, or the moderator must
+        approve the subscription.  Use the ``confirm(token)`` method to
+        resume the workflow.
 
-        :param mlist: The mailing list that is the focus of this registration.
+        :param mlist: The mailing list to subscribe to.
         :type mlist: `IMailingList`
-        :param email: The email address to register.
-        :type email: str
-        :param display_name: The optional display name of the user.
-        :type display_name: str
-        :param delivery_mode: The optional delivery mode for this
-            registration.  If not given, regular delivery is used.
-        :type delivery_mode: `DeliveryMode`
-        :return: The confirmation token string.
-        :rtype: str
-        :raises InvalidEmailAddressError: if the address is not allowed.
+        :param subscriber: The user or address to subscribe.
+        :type email: ``IUser`` or ``IAddress``
+        :return: The confirmation token string, or None if the workflow
+            completes (i.e. the member has been subscribed).
+        :rtype: str or None
+        :raises MembershipIsBannedError: when the address being subscribed
+            appears in the global or list-centric bans.
         """
 
     def confirm(token):
-        """Confirm the pending registration matched to the given `token`.
+        """Continue any paused workflow.
 
-        Confirmation ensures that the IAddress exists and is linked to an
-        IUser, with the latter being created and linked if necessary.
+        Confirmation may occur after the user confirms their
+        subscription request, or their email address must be verified,
+        or the moderator must approve the subscription request.
 
-        :param token: A token matching a pending event with a type of
-            'registration'.
-        :return: Boolean indicating whether the confirmation succeeded or
-            not.  It may fail if the token is no longer in the database, or if
-            the token did not match a registration event.
+        :param token: A token matching a workflow.
+        :type token: string
+        :return: The new token for any follow up confirmation, or None if the
+            user was subscribed.
+        :rtype: str or None
+        :raises LookupError: when no workflow is associated with the token.
         """
 
     def discard(token):
-        """Discard the pending registration matched to the given `token`.
-
-        The event record is discarded and the IAddress is not verified.  No
-        IUser is created.
+        """Discard the workflow matched to the given `token`.
 
         :param token: A token matching a pending event with a type of
             'registration'.
+        :raises LookupError: when no workflow is associated with the token.
         """
+
+    def evict():
+        """Evict all saved workflows which have expired."""
diff --git a/src/mailman/interfaces/roster.py b/src/mailman/interfaces/roster.py
index 5d0b9d6c2..af473a553 100644
--- a/src/mailman/interfaces/roster.py
+++ b/src/mailman/interfaces/roster.py
@@ -53,11 +53,26 @@ class IRoster(Interface):
         managed by this roster.
         """)
 
-    def get_member(address):
+    def get_member(email):
         """Get the member for the given address.
 
-        :param address: The email address to search for.
-        :type address: text
+        *Note* that it is possible for an email to be subscribed to a
+        mailing list twice, once through its explicit address and once
+        indirectly through a user's preferred address.  In this case,
+        this API always returns the explicit address.  Use
+        ``get_memberships()`` to return them all.
+
+        :param email: The email address to search for.
+        :type email: string
         :return: The member if found, otherwise None
         :rtype: `IMember` or None
         """
+
+    def get_memberships(email):
+        """Get the memberships for the given address.
+
+        :param email: The email address to search for.
+        :type email: string
+        :return: All the memberships associated with this email address.
+        :rtype: sequence of length 0, 1, or 2 of ``IMember``
+        """
diff --git a/src/mailman/interfaces/workflow.py b/src/mailman/interfaces/workflow.py
new file mode 100644
index 000000000..f80e38547
--- /dev/null
+++ b/src/mailman/interfaces/workflow.py
@@ -0,0 +1,80 @@
+# Copyright (C) 2015 by the Free Software Foundation, Inc.
+#
+# This file is part of GNU Mailman.
+#
+# GNU Mailman is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option)
+# any later version.
+#
+# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
+# more details.
+#
+# You should have received a copy of the GNU General Public License along with
+# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Interfaces describing the state of a workflow."""
+
+__all__ = [
+    'IWorkflowState',
+    'IWorkflowStateManager',
+    ]
+
+
+from zope.interface import Attribute, Interface
+
+
+
+class IWorkflowState(Interface):
+    """The state of a workflow."""
+
+    name = Attribute('The name of the workflow.')
+
+    token = Attribute('A unique key identifying the workflow instance.')
+
+    step = Attribute("This workflow's next step.")
+
+    data = Attribute('Additional data (may be JSON-encoded).')
+
+
+
+class IWorkflowStateManager(Interface):
+    """The workflow states manager."""
+
+    def save(name, token, step, data=None):
+        """Save the state of a workflow.
+
+        :param name: The name of the workflow.
+        :type name: str
+        :param token: A unique token identifying this workflow instance.
+        :type token: str
+        :param step: The next step for this workflow.
+        :type step: str
+        :param data: Additional data (workflow-specific).
+        :type data: str
+        """
+
+    def restore(name, token):
+        """Get the saved state for a workflow or None if nothing was saved.
+
+        :param name: The name of the workflow.
+        :type name: str
+        :param token: A unique token identifying this workflow instance.
+        :type token: str
+        :return: The saved state associated with this name/token pair, or None
+            if the pair isn't in the database.
+        :rtype: ``IWorkflowState``
+        """
+
+    def discard(name, token):
+        """Throw away the saved state for a workflow.
+
+        :param name: The name of the workflow.
+        :type name: str
+        :param token: A unique token identifying this workflow instance.
+        :type token: str
+        """
+
+    count = Attribute('The number of saved workflows in the database.')
diff --git a/src/mailman/model/docs/membership.rst b/src/mailman/model/docs/membership.rst
index 60ccd1ac1..0fd748d6a 100644
--- a/src/mailman/model/docs/membership.rst
+++ b/src/mailman/model/docs/membership.rst
@@ -228,6 +228,38 @@ regardless of their role.
     fperson@example.com MemberRole.nonmember
 
 
+Subscriber type
+===============
+
+Members can be subscribed to a mailing list either via an explicit address, or
+indirectly through a user's preferred address.  Sometimes you want to know
+which one it is.
+
+Herb subscribes to the mailing list via an explicit address.
+
+    >>> herb = user_manager.create_address(
+    ...     'hperson@example.com', 'Herb Person')
+    >>> herb_member = mlist.subscribe(herb)
+
+Iris subscribes to the mailing list via her preferred address.
+
+    >>> iris = user_manager.make_user(
+    ...     'iperson@example.com', 'Iris Person')
+    >>> preferred = list(iris.addresses)[0]
+    >>> from mailman.utilities.datetime import now
+    >>> preferred.verified_on = now()
+    >>> iris.preferred_address = preferred
+    >>> iris_member = mlist.subscribe(iris)
+
+When we need to know which way a member is subscribed, we can look at the this
+attribute.
+
+    >>> herb_member.subscriber
+    <Address: Herb Person <hperson@example.com> [not verified] at ...>
+    >>> iris_member.subscriber
+    <User "Iris Person" (5) at ...>
+
+
 Moderation actions
 ==================
 
@@ -250,6 +282,8 @@ should go through the normal moderation checks.
     aperson@example.com MemberRole.member Action.defer
     bperson@example.com MemberRole.member Action.defer
     cperson@example.com MemberRole.member Action.defer
+    hperson@example.com MemberRole.member Action.defer
+    iperson@example.com MemberRole.member Action.defer
 
 Postings by nonmembers are held for moderator approval by default.
 
@@ -272,7 +306,7 @@ though that the address they're changing to must be verified.
     >>> gwen_member = bee.subscribe(gwen_address)
     >>> for m in bee.members.members:
     ...     print(m.member_id.int, m.mailing_list.list_id, m.address.email)
-    7 bee.example.com gwen@example.com
+    9 bee.example.com gwen@example.com
 
 Gwen gets a email address.
 
@@ -288,7 +322,7 @@ Now her membership reflects the new address.
 
     >>> for m in bee.members.members:
     ...     print(m.member_id.int, m.mailing_list.list_id, m.address.email)
-    7 bee.example.com gperson@example.com
+    9 bee.example.com gperson@example.com
 
 
 Events
diff --git a/src/mailman/model/docs/pending.rst b/src/mailman/model/docs/pending.rst
index a634322a1..4a14edb2a 100644
--- a/src/mailman/model/docs/pending.rst
+++ b/src/mailman/model/docs/pending.rst
@@ -13,6 +13,11 @@ In order to pend an event, you first need a pending database.
     >>> from zope.component import getUtility
     >>> pendingdb = getUtility(IPendings)
 
+There are nothing in the pendings database.
+
+    >>> pendingdb.count
+    0
+
 The pending database can add any ``IPendable`` to the database, returning a
 token that can be used in urls and such.
 ::
@@ -33,6 +38,11 @@ token that can be used in urls and such.
     >>> len(token)
     40
 
+There's exactly one entry in the pendings database now.
+
+    >>> pendingdb.count
+    1
+
 There's not much you can do with tokens except to *confirm* them, which
 basically means returning the `IPendable` structure (as a dictionary) from the
 database that matches the token.  If the token isn't in the database, None is
diff --git a/src/mailman/model/docs/registration.rst b/src/mailman/model/docs/registration.rst
index 2d2aa8ec7..fc7ad6f1a 100644
--- a/src/mailman/model/docs/registration.rst
+++ b/src/mailman/model/docs/registration.rst
@@ -1,333 +1,90 @@
-====================
-Address registration
-====================
-
-Before users can join a mailing list, they must first register with Mailman.
-The only thing they must supply is an email address, although there is
-additional information they may supply.  All registered email addresses must
-be verified before Mailman will send them any list traffic.
+============
+Registration
+============
 
-The ``IUserManager`` manages users, but it does so at a fairly low level.
-Specifically, it does not handle verification, email address syntax validity
-checks, etc.  The ``IRegistrar`` is the interface to the object handling all
-this stuff.
+When a user wants to join a mailing list, they must register and verify their
+email address.  Then depending on how the mailing list is configured, they may
+need to confirm their subscription and have it approved by the list
+moderator.  The ``IRegistrar`` interface manages this work flow.
 
     >>> from mailman.interfaces.registrar import IRegistrar
-    >>> from zope.component import getUtility
-    >>> registrar = getUtility(IRegistrar)
-
-Here is a helper function to check the token strings.
-
-    >>> def check_token(token):
-    ...     assert isinstance(token, str), 'Not a string'
-    ...     assert len(token) == 40, 'Unexpected length: %d' % len(token)
-    ...     assert token.isalnum(), 'Not alphanumeric'
-    ...     print('ok')
-
-Here is a helper function to extract tokens from confirmation messages.
-
-    >>> import re
-    >>> cre = re.compile('http://lists.example.com/confirm/(.*)')
-    >>> def extract_token(msg):
-    ...     mo = cre.search(msg.get_payload())
-    ...     return mo.group(1)
 
+Registrars adapt mailing lists.
 
-Invalid email addresses
-=======================
-
-Addresses are registered within the context of a mailing list, mostly so that
-confirmation emails can come from some place.  You also need the email
-address of the user who is registering.
-
-    >>> mlist = create_list('alpha@example.com')
+    >>> from mailman.interfaces.mailinglist import SubscriptionPolicy
+    >>> mlist = create_list('ant@example.com')
     >>> mlist.send_welcome_message = False
+    >>> mlist.subscription_policy = SubscriptionPolicy.open
+    >>> registrar = IRegistrar(mlist)
 
-Some amount of sanity checks are performed on the email address, although
-honestly, not as much as probably should be done.  Still, some patently bad
-addresses are rejected outright.
-
-
-Register an email address
-=========================
-
-Registration of an unknown address creates nothing until the confirmation step
-is complete.  No ``IUser`` or ``IAddress`` is created at registration time,
-but a record is added to the pending database, and the token for that record
-is returned.
-
-    >>> token = registrar.register(mlist, 'aperson@example.com', 'Anne Person')
-    >>> check_token(token)
-    ok
-
-There should be no records in the user manager for this address yet.
+Usually, addresses are registered, but users with preferred addresses can be
+registered too.
 
     >>> from mailman.interfaces.usermanager import IUserManager
     >>> from zope.component import getUtility
-    >>> user_manager = getUtility(IUserManager)
-    >>> print(user_manager.get_user('aperson@example.com'))
-    None
-    >>> print(user_manager.get_address('aperson@example.com'))
-    None
-
-But this address is waiting for confirmation.
-
-    >>> from mailman.interfaces.pending import IPendings
-    >>> pendingdb = getUtility(IPendings)
-
-    >>> dump_msgdata(pendingdb.confirm(token, expunge=False))
-    delivery_mode: regular
-    display_name : Anne Person
-    email        : aperson@example.com
-    list_id      : alpha.example.com
-    type         : registration
-
-
-Verification by email
-=====================
-
-There is also a verification email sitting in the virgin queue now.  This
-message is sent to the user in order to verify the registered address.
-
-    >>> from mailman.testing.helpers import get_queue_messages
-    >>> items = get_queue_messages('virgin')
-    >>> len(items)
-    1
-    >>> print(items[0].msg.as_string())
-    MIME-Version: 1.0
-    ...
-    Subject: confirm ...
-    From: alpha-confirm+...@example.com
-    To: aperson@example.com
-    ...
-    <BLANKLINE>
-    Email Address Registration Confirmation
-    <BLANKLINE>
-    Hello, this is the GNU Mailman server at example.com.
-    <BLANKLINE>
-    We have received a registration request for the email address
-    <BLANKLINE>
-        aperson@example.com
-    <BLANKLINE>
-    Before you can start using GNU Mailman at this site, you must first
-    confirm that this is your email address.  You can do this by replying to
-    this message, keeping the Subject header intact.  Or you can visit this
-    web page
-    <BLANKLINE>
-        http://lists.example.com/confirm/...
-    <BLANKLINE>
-    If you do not wish to register this email address simply disregard this
-    message.  If you think you are being maliciously subscribed to the list,
-    or have any other questions, you may contact
-    <BLANKLINE>
-        alpha-owner@example.com
-    <BLANKLINE>
-    >>> dump_msgdata(items[0].msgdata)
-    _parsemsg           : False
-    listid              : alpha.example.com
-    nodecorate          : True
-    recipients          : {'aperson@example.com'}
-    reduced_list_headers: True
-    version             : 3
-
-The confirmation token shows up in several places, each of which provides an
-easy way for the user to complete the confirmation.  The token will always
-appear in a URL in the body of the message.
-
-    >>> sent_token = extract_token(items[0].msg)
-    >>> sent_token == token
-    True
-
-The same token will appear in the ``From`` header.
-
-    >>> items[0].msg['from'] == 'alpha-confirm+' + token + '@example.com'
-    True
-
-It will also appear in the ``Subject`` header.
+    >>> anne = getUtility(IUserManager).create_address(
+    ...     'anne@example.com', 'Anne Person')
 
-    >>> items[0].msg['subject'] == 'confirm ' + token
-    True
-
-The user would then validate their registered address by clicking on a url or
-responding to the message.  Either way, the confirmation process extracts the
-token and uses that to confirm the pending registration.
-
-    >>> registrar.confirm(token)
-    True
-
-Now, there is an `IAddress` in the database matching the address, as well as
-an `IUser` linked to this address.  The `IAddress` is verified.
-
-    >>> found_address = user_manager.get_address('aperson@example.com')
-    >>> found_address
-    <Address: Anne Person <aperson@example.com> [verified] at ...>
-    >>> found_user = user_manager.get_user('aperson@example.com')
-    >>> found_user
-    <User "Anne Person" (...) at ...>
-    >>> found_user.controls(found_address.email)
-    True
-    >>> from datetime import datetime
-    >>> isinstance(found_address.verified_on, datetime)
-    True
 
+Register an email address
+=========================
 
-Non-standard registrations
-==========================
+When the registration steps involve confirmation or moderator approval, the
+process will pause until these steps are completed.  A unique token is created
+which represents this work flow.
 
-If you try to confirm a registration token twice, of course only the first one
-will work.  The second one is ignored.
+Anne attempts to join the mailing list.
 
-    >>> token = registrar.register(mlist, 'bperson@example.com')
-    >>> check_token(token)
-    ok
-    >>> items = get_queue_messages('virgin')
-    >>> len(items)
-    1
-    >>> sent_token = extract_token(items[0].msg)
-    >>> token == sent_token
-    True
-    >>> registrar.confirm(token)
-    True
-    >>> registrar.confirm(token)
-    False
+    >>> token = registrar.register(anne)
 
-If an address is in the system, but that address is not linked to a user yet
-and the address is not yet validated, then no user is created until the
-confirmation step is completed.
+Because her email address has not yet been verified, she has not yet become a
+member of the mailing list.
 
-    >>> user_manager.create_address('cperson@example.com')
-    <Address: cperson@example.com [not verified] at ...>
-    >>> token = registrar.register(
-    ...     mlist, 'cperson@example.com', 'Claire Person')
-    >>> print(user_manager.get_user('cperson@example.com'))
+    >>> print(mlist.members.get_member('anne@example.com'))
     None
-    >>> items = get_queue_messages('virgin')
-    >>> len(items)
-    1
-    >>> sent_token = extract_token(items[0].msg)
-    >>> registrar.confirm(sent_token)
-    True
-    >>> user_manager.get_user('cperson@example.com')
-    <User "Claire Person" (...) at ...>
-    >>> user_manager.get_address('cperson@example.com')
-    <Address: cperson@example.com [verified] at ...>
-
-Even if the address being registered has already been verified, the
-registration sends a confirmation.
-
-    >>> token = registrar.register(mlist, 'cperson@example.com')
-    >>> token is not None
-    True
-
 
-Discarding
-==========
+Once she verifies her email address, she will become a member of the mailing
+list.  In this case, verifying implies that she also confirms her wish to join
+the mailing list.
 
-A confirmation token can also be discarded, say if the user changes his or her
-mind about registering.  When discarded, no `IAddress` or `IUser` is created.
-::
+    >>> registrar.confirm(token)
+    >>> mlist.members.get_member('anne@example.com')
+    <Member: Anne Person <anne@example.com> on ant@example.com
+        as MemberRole.member>
 
-    >>> token = registrar.register(mlist, 'eperson@example.com', 'Elly Person')
-    >>> check_token(token)
-    ok
-    >>> registrar.discard(token)
-    >>> print(pendingdb.confirm(token))
-    None
-    >>> print(user_manager.get_address('eperson@example.com'))
-    None
-    >>> print(user_manager.get_user('eperson@example.com'))
-    None
 
-    # Clear the virgin queue of all the preceding confirmation messages.
-    >>> ignore = get_queue_messages('virgin')
+Register a user
+===============
 
+Users can also register, but they must have a preferred address.  The mailing
+list will deliver messages to this preferred address.
 
-Registering a new address for an existing user
-==============================================
+    >>> bart = getUtility(IUserManager).make_user(
+    ...     'bart@example.com', 'Bart Person')
 
-When a new address for an existing user is registered, there isn't too much
-different except that the new address will still need to be verified before it
-can be used.
-::
+Bart verifies his address and makes it his preferred address.
 
     >>> from mailman.utilities.datetime import now
-    >>> dperson = user_manager.create_user(
-    ...     'dperson@example.com', 'Dave Person')
-    >>> dperson
-    <User "Dave Person" (...) at ...>
-    >>> address = user_manager.get_address('dperson@example.com')
-    >>> address.verified_on = now()
-
-    >>> from operator import attrgetter
-    >>> dump_list(repr(address) for address in dperson.addresses)
-    <Address: Dave Person <dperson@example.com> [verified] at ...>
-    >>> dperson.register('david.person@example.com', 'David Person')
-    <Address: David Person <david.person@example.com> [not verified] at ...>
-    >>> token = registrar.register(mlist, 'david.person@example.com')
-
-    >>> items = get_queue_messages('virgin')
-    >>> len(items)
-    1
-    >>> sent_token = extract_token(items[0].msg)
-    >>> registrar.confirm(sent_token)
-    True
-    >>> user = user_manager.get_user('david.person@example.com')
-    >>> user is dperson
-    True
-    >>> user
-    <User "Dave Person" (...) at ...>
-    >>> dump_list(repr(address) for address in user.addresses)
-    <Address: Dave Person <dperson@example.com> [verified] at ...>
-    <Address: David Person <david.person@example.com> [verified] at ...>
-
-
-Corner cases
-============
-
-If you try to confirm a token that doesn't exist in the pending database, the
-confirm method will just return False.
-
-    >>> registrar.confirm(bytes(b'no token'))
-    False
-
-Likewise, if you try to confirm, through the `IRegistrar` interface, a token
-that doesn't match a registration event, you will get ``None``.  However, the
-pending event matched with that token will still be removed.
-::
-
-    >>> from mailman.interfaces.pending import IPendable
-    >>> from zope.interface import implementer
-
-    >>> @implementer(IPendable)
-    ... class SimplePendable(dict):
-    ...     pass
-
-    >>> pendable = SimplePendable(type='foo', bar='baz')
-    >>> token = pendingdb.add(pendable)
-    >>> registrar.confirm(token)
-    False
-    >>> print(pendingdb.confirm(token))
-    None
-
-
-Registration and subscription
-=============================
+    >>> preferred = list(bart.addresses)[0]
+    >>> preferred.verified_on = now()
+    >>> bart.preferred_address = preferred
 
-Fred registers with Mailman at the same time that he subscribes to a mailing
-list.
+The mailing list's subscription policy does not require Bart to confirm his
+subscription, but the moderate does want to approve all subscriptions.
 
-    >>> token = registrar.register(
-    ...     mlist, 'fred.person@example.com', 'Fred Person')
+    >>> mlist.subscription_policy = SubscriptionPolicy.moderate
 
-Before confirmation, Fred is not a member of the mailing list.
+Now when Bart registers as a user for the mailing list, a token will still be
+generated, but this is only used by the moderator.  At first, Bart is not
+subscribed to the mailing list.
 
-    >>> print(mlist.members.get_member('fred.person@example.com'))
+    >>> token = registrar.register(bart)
+    >>> print(mlist.members.get_member('bart@example.com'))
     None
 
-But after confirmation, he is.
+When the moderator confirms Bart's subscription, he joins the mailing list.
 
     >>> registrar.confirm(token)
-    True
-    >>> print(mlist.members.get_member('fred.person@example.com'))
-    <Member: Fred Person <fred.person@example.com>
-             on alpha@example.com as MemberRole.member>
+    >>> mlist.members.get_member('bart@example.com')
+    <Member: Bart Person <bart@example.com> on ant@example.com
+        as MemberRole.member>
diff --git a/src/mailman/model/mailinglist.py b/src/mailman/model/mailinglist.py
index cef272437..f04c534e1 100644
--- a/src/mailman/model/mailinglist.py
+++ b/src/mailman/model/mailinglist.py
@@ -38,7 +38,7 @@ from mailman.interfaces.domain import IDomainManager
 from mailman.interfaces.languages import ILanguageManager
 from mailman.interfaces.mailinglist import (
     IAcceptableAlias, IAcceptableAliasSet, IListArchiver, IListArchiverSet,
-    IMailingList, Personalization, ReplyToMunging)
+    IMailingList, Personalization, ReplyToMunging, SubscriptionPolicy)
 from mailman.interfaces.member import (
     AlreadySubscribedError, MemberRole, MissingPreferredAddressError,
     SubscriptionEvent)
@@ -183,6 +183,7 @@ class MailingList(Model):
     send_goodbye_message = Column(Boolean)
     send_welcome_message = Column(Boolean)
     subject_prefix = Column(Unicode)
+    subscription_policy = Column(Enum(SubscriptionPolicy))
     topics = Column(PickleType)
     topics_bodylines_limit = Column(Integer)
     topics_enabled = Column(Boolean)
diff --git a/src/mailman/model/member.py b/src/mailman/model/member.py
index ee6d246f5..e6e4933f9 100644
--- a/src/mailman/model/member.py
+++ b/src/mailman/model/member.py
@@ -135,6 +135,10 @@ class Member(Model):
                 if self._address is None
                 else getUtility(IUserManager).get_user(self._address.email))
 
+    @property
+    def subscriber(self):
+        return (self._user if self._address is None else self._address)
+
     def _lookup(self, preference, default=None):
         pref = getattr(self.preferences, preference)
         if pref is not None:
diff --git a/src/mailman/model/pending.py b/src/mailman/model/pending.py
index 77b68bd2d..bbe95d5f0 100644
--- a/src/mailman/model/pending.py
+++ b/src/mailman/model/pending.py
@@ -128,7 +128,7 @@ class Pendings:
         return token
 
     @dbconnection
-    def confirm(self, store, token, expunge=True):
+    def confirm(self, store, token, *, expunge=True):
         # Token can come in as a unicode, but it's stored in the database as
         # bytes.  They must be ascii.
         pendings = store.query(Pended).filter_by(token=str(token))
@@ -165,3 +165,8 @@ class Pendings:
                 for keyvalue in q:
                     store.delete(keyvalue)
                 store.delete(pending)
+
+    @property
+    @dbconnection
+    def count(self, store):
+        return store.query(Pended).count()
diff --git a/src/mailman/model/roster.py b/src/mailman/model/roster.py
index 91211c665..da2ed4582 100644
--- a/src/mailman/model/roster.py
+++ b/src/mailman/model/roster.py
@@ -97,21 +97,48 @@ class AbstractRoster:
             yield member.address
 
     @dbconnection
-    def get_member(self, store, address):
-        """See `IRoster`."""
-        results = store.query(Member).filter(
+    def _get_all_memberships(self, store, email):
+        # Avoid circular imports.
+        from mailman.model.user import User
+        # Here's a query that finds all members subscribed with an explicit
+        # email address.
+        members_a = store.query(Member).filter(
             Member.list_id == self._mlist.list_id,
             Member.role == self.role,
-            Address.email == address,
+            Address.email == email,
             Member.address_id == Address.id)
-        if results.count() == 0:
+        # Here's a query that finds all members subscribed with their
+        # preferred address.
+        members_u = store.query(Member).filter(
+            Member.list_id == self._mlist.list_id,
+            Member.role == self.role,
+            Address.email==email,
+            Member.user_id == User.id)
+        return members_a.union(members_u).all()
+
+    def get_member(self, email):
+        """See ``IRoster``."""
+        memberships = self._get_all_memberships(email)
+        count = len(memberships)
+        if count == 0:
             return None
-        elif results.count() == 1:
-            return results[0]
-        else:
-            raise AssertionError(
-                'Too many matching member results: {0}'.format(
-                    results.count()))
+        elif count == 1:
+            return memberships[0]
+        assert count == 2, 'Unexpected membership count: {}'.format(count)
+        # This is the case where the email address is subscribed both
+        # explicitly and indirectly through the preferred address.  By
+        # definition, we return the explicit address membership only.
+        return (memberships[0]
+                if memberships[0]._address is not None
+                else memberships[1])
+
+    def get_memberships(self, email):
+        """See ``IRoster``."""
+        memberships = self._get_all_memberships(email)
+        count = len(memberships)
+        assert 0 <= count <= 2, 'Unexpected membership count: {}'.format(
+            count)
+        return memberships
 
 
 
@@ -160,13 +187,13 @@ class AdministratorRoster(AbstractRoster):
                 Member.role == MemberRole.moderator))
 
     @dbconnection
-    def get_member(self, store, address):
+    def get_member(self, store, email):
         """See `IRoster`."""
         results = store.query(Member).filter(
             Member.list_id == self._mlist.list_id,
             or_(Member.role == MemberRole.moderator,
                 Member.role == MemberRole.owner),
-            Address.email == address,
+            Address.email == email,
             Member.address_id == Address.id)
         if results.count() == 0:
             return None
@@ -181,6 +208,8 @@ class AdministratorRoster(AbstractRoster):
 class DeliveryMemberRoster(AbstractRoster):
     """Return all the members having a particular kind of delivery."""
 
+    role = MemberRole.member
+
     @property
     def member_count(self):
         """See `IRoster`."""
@@ -285,7 +314,7 @@ class Memberships:
             yield address
 
     @dbconnection
-    def get_member(self, store, address):
+    def get_member(self, store, email):
         """See `IRoster`."""
         results = store.query(Member).filter(
             Member.address_id == Address.id,
@@ -298,3 +327,10 @@ class Memberships:
             raise AssertionError(
                 'Too many matching member results: {0}'.format(
                     results.count()))
+
+    @dbconnection
+    def get_memberships(self, store, address):
+        """See `IRoster`."""
+        # 2015-04-14 BAW: See LP: #1444055 -- this currently exists just to
+        # pass a test.
+        raise NotImplementedError
diff --git a/src/mailman/model/tests/test_registrar.py b/src/mailman/model/tests/test_registrar.py
deleted file mode 100644
index e6df7f0d1..000000000
--- a/src/mailman/model/tests/test_registrar.py
+++ /dev/null
@@ -1,64 +0,0 @@
-# Copyright (C) 2014-2015 by the Free Software Foundation, Inc.
-#
-# This file is part of GNU Mailman.
-#
-# GNU Mailman is free software: you can redistribute it and/or modify it under
-# the terms of the GNU General Public License as published by the Free
-# Software Foundation, either version 3 of the License, or (at your option)
-# any later version.
-#
-# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
-# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
-# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
-# more details.
-#
-# You should have received a copy of the GNU General Public License along with
-# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
-
-"""Test `IRegistrar`."""
-
-__all__ = [
-    'TestRegistrar',
-    ]
-
-
-import unittest
-
-from functools import partial
-from mailman.app.lifecycle import create_list
-from mailman.interfaces.address import InvalidEmailAddressError
-from mailman.interfaces.registrar import IRegistrar
-from mailman.testing.layers import ConfigLayer
-from zope.component import getUtility
-
-
-
-class TestRegistrar(unittest.TestCase):
-    layer = ConfigLayer
-
-    def setUp(self):
-        mlist = create_list('test@example.com')
-        self._register = partial(getUtility(IRegistrar).register, mlist)
-
-    def test_invalid_empty_string(self):
-        self.assertRaises(InvalidEmailAddressError, self._register, '')
-
-    def test_invalid_space_in_name(self):
-        self.assertRaises(InvalidEmailAddressError, self._register,
-                          'some name@example.com')
-
-    def test_invalid_funky_characters(self):
-        self.assertRaises(InvalidEmailAddressError, self._register,
-                          '<script>@example.com')
-
-    def test_invalid_nonascii(self):
-        self.assertRaises(InvalidEmailAddressError, self._register,
-                          '\xa0@example.com')
-
-    def test_invalid_no_at_sign(self):
-        self.assertRaises(InvalidEmailAddressError, self._register,
-                          'noatsign')
-
-    def test_invalid_no_domain(self):
-        self.assertRaises(InvalidEmailAddressError, self._register,
-                          'nodom@ain')
diff --git a/src/mailman/model/tests/test_roster.py b/src/mailman/model/tests/test_roster.py
index 44735cf4b..ca950cfc5 100644
--- a/src/mailman/model/tests/test_roster.py
+++ b/src/mailman/model/tests/test_roster.py
@@ -26,7 +26,9 @@ __all__ = [
 import unittest
 
 from mailman.app.lifecycle import create_list
+from mailman.interfaces.address import IAddress
 from mailman.interfaces.member import DeliveryMode, MemberRole
+from mailman.interfaces.user import IUser
 from mailman.interfaces.usermanager import IUserManager
 from mailman.testing.layers import ConfigLayer
 from mailman.utilities.datetime import now
@@ -136,7 +138,8 @@ class TestMembershipsRoster(unittest.TestCase):
         self._ant = create_list('ant@example.com')
         self._bee = create_list('bee@example.com')
         user_manager = getUtility(IUserManager)
-        self._anne = user_manager.create_user('anne@example.com')
+        self._anne = user_manager.make_user(
+            'anne@example.com', 'Anne Person')
         preferred = list(self._anne.addresses)[0]
         preferred.verified_on = now()
         self._anne.preferred_address = preferred
@@ -144,9 +147,56 @@ class TestMembershipsRoster(unittest.TestCase):
     def test_no_memberships(self):
         # An unsubscribed user has no memberships.
         self.assertEqual(self._anne.memberships.member_count, 0)
+        self.assertIsNone(self._ant.members.get_member('anne@example.com'))
+        self.assertEqual(
+            self._ant.members.get_memberships('anne@example.com'),
+            [])
 
     def test_subscriptions(self):
         # Anne subscribes to a couple of mailing lists.
         self._ant.subscribe(self._anne)
         self._bee.subscribe(self._anne)
         self.assertEqual(self._anne.memberships.member_count, 2)
+
+    def test_subscribed_as_user(self):
+        # Anne subscribes to a mailing list as a user and the member roster
+        # contains her membership.
+        self._ant.subscribe(self._anne)
+        self.assertEqual(
+            self._ant.members.get_member('anne@example.com').user,
+            self._anne)
+        memberships = self._ant.members.get_memberships('anne@example.com')
+        self.assertEqual(
+            [member.address.email for member in memberships],
+            ['anne@example.com'])
+
+    def test_subscribed_as_user_and_address(self):
+        # Anne subscribes to a mailing list twice, once as a user and once
+        # with an explicit address.  She has two memberships.
+        self._ant.subscribe(self._anne)
+        self._ant.subscribe(self._anne.preferred_address)
+        self.assertEqual(self._anne.memberships.member_count, 2)
+        self.assertEqual(self._ant.members.member_count, 2)
+        self.assertEqual(
+            [member.address.email for member in self._ant.members.members],
+            ['anne@example.com', 'anne@example.com'])
+        # get_member() is defined to return the explicit address.
+        member = self._ant.members.get_member('anne@example.com')
+        subscriber = member.subscriber
+        self.assertTrue(IAddress.providedBy(subscriber))
+        self.assertFalse(IUser.providedBy(subscriber))
+        # get_memberships() returns them all.
+        memberships = self._ant.members.get_memberships('anne@example.com')
+        self.assertEqual(len(memberships), 2)
+        as_address = (memberships[0]
+                      if IAddress.providedBy(memberships[0].subscriber)
+                      else memberships[1])
+        as_user = (memberships[1]
+                   if IUser.providedBy(memberships[1].subscriber)
+                   else memberships[0])
+        self.assertEqual(as_address.subscriber, self._anne.preferred_address)
+        self.assertEqual(as_user.subscriber, self._anne)
+        # All the email addresses match.
+        self.assertEqual(
+            [record.address.email for record in memberships],
+            ['anne@example.com', 'anne@example.com'])
diff --git a/src/mailman/model/tests/test_usermanager.py b/src/mailman/model/tests/test_usermanager.py
index 31f1a7275..f4643f031 100644
--- a/src/mailman/model/tests/test_usermanager.py
+++ b/src/mailman/model/tests/test_usermanager.py
@@ -30,6 +30,7 @@ from mailman.testing.layers import ConfigLayer
 from zope.component import getUtility
 
 
+
 class TestUserManager(unittest.TestCase):
     layer = ConfigLayer
 
@@ -80,3 +81,8 @@ class TestUserManager(unittest.TestCase):
         user = self._usermanager.create_user('anne@example.com', 'Anne Person')
         other_user = self._usermanager.make_user('anne@example.com')
         self.assertIs(user, other_user)
+
+    def test_get_user_by_id(self):
+        original = self._usermanager.make_user('anne@example.com')
+        copy = self._usermanager.get_user_by_id(original.user_id)
+        self.assertEqual(original, copy)
diff --git a/src/mailman/model/tests/test_workflow.py b/src/mailman/model/tests/test_workflow.py
new file mode 100644
index 000000000..88ed506bd
--- /dev/null
+++ b/src/mailman/model/tests/test_workflow.py
@@ -0,0 +1,148 @@
+# Copyright (C) 2015 by the Free Software Foundation, Inc.
+#
+# This file is part of GNU Mailman.
+#
+# GNU Mailman is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option)
+# any later version.
+#
+# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
+# more details.
+#
+# You should have received a copy of the GNU General Public License along with
+# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Test the workflow model."""
+
+__all__ = [
+    'TestWorkflow',
+    ]
+
+
+import unittest
+
+from mailman.interfaces.workflow import IWorkflowStateManager
+from mailman.testing.layers import ConfigLayer
+from zope.component import getUtility
+
+
+
+class TestWorkflow(unittest.TestCase):
+    layer = ConfigLayer
+
+    def setUp(self):
+        self._manager = getUtility(IWorkflowStateManager)
+
+    def test_save_restore_workflow(self):
+        # Save and restore a workflow.
+        name = 'ant'
+        token = 'bee'
+        step = 'cat'
+        data = 'dog'
+        self._manager.save(name, token, step, data)
+        state = self._manager.restore(name, token)
+        self.assertEqual(state.name, name)
+        self.assertEqual(state.token, token)
+        self.assertEqual(state.step, step)
+        self.assertEqual(state.data, data)
+
+    def test_save_restore_workflow_without_step(self):
+        # Save and restore a workflow that contains no step.
+        name = 'ant'
+        token = 'bee'
+        data = 'dog'
+        self._manager.save(name, token, data=data)
+        state = self._manager.restore(name, token)
+        self.assertEqual(state.name, name)
+        self.assertEqual(state.token, token)
+        self.assertIsNone(state.step)
+        self.assertEqual(state.data, data)
+
+    def test_save_restore_workflow_without_data(self):
+        # Save and restore a workflow that contains no data.
+        name = 'ant'
+        token = 'bee'
+        step = 'cat'
+        self._manager.save(name, token, step)
+        state = self._manager.restore(name, token)
+        self.assertEqual(state.name, name)
+        self.assertEqual(state.token, token)
+        self.assertEqual(state.step, step)
+        self.assertIsNone(state.data)
+
+    def test_save_restore_workflow_without_step_or_data(self):
+        # Save and restore a workflow that contains no step or data.
+        name = 'ant'
+        token = 'bee'
+        self._manager.save(name, token)
+        state = self._manager.restore(name, token)
+        self.assertEqual(state.name, name)
+        self.assertEqual(state.token, token)
+        self.assertIsNone(state.step)
+        self.assertIsNone(state.data)
+
+    def test_restore_workflow_with_no_matching_name(self):
+        # Try to restore a workflow that has no matching name in the database.
+        name = 'ant'
+        token = 'bee'
+        self._manager.save(name, token)
+        state = self._manager.restore('ewe', token)
+        self.assertIsNone(state)
+
+    def test_restore_workflow_with_no_matching_token(self):
+        # Try to restore a workflow that has no matching token in the database.
+        name = 'ant'
+        token = 'bee'
+        self._manager.save(name, token)
+        state = self._manager.restore(name, 'fly')
+        self.assertIsNone(state)
+
+    def test_restore_workflow_with_no_matching_token_or_name(self):
+        # Try to restore a workflow that has no matching token or name in the
+        # database.
+        name = 'ant'
+        token = 'bee'
+        self._manager.save(name, token)
+        state = self._manager.restore('ewe', 'fly')
+        self.assertIsNone(state)
+
+    def test_restore_removes_record(self):
+        name = 'ant'
+        token = 'bee'
+        self.assertEqual(self._manager.count, 0)
+        self._manager.save(name, token)
+        self.assertEqual(self._manager.count, 1)
+        self._manager.restore(name, token)
+        self.assertEqual(self._manager.count, 0)
+
+    def test_save_after_restore(self):
+        name = 'ant'
+        token = 'bee'
+        self.assertEqual(self._manager.count, 0)
+        self._manager.save(name, token)
+        self.assertEqual(self._manager.count, 1)
+        self._manager.restore(name, token)
+        self.assertEqual(self._manager.count, 0)
+        self._manager.save(name, token)
+        self.assertEqual(self._manager.count, 1)
+
+    def test_discard(self):
+        # Discard some workflow state.  This is use by IRegistrar.discard().
+        self._manager.save('ant', 'token', 'one')
+        self._manager.save('bee', 'token', 'two')
+        self._manager.save('ant', 'nekot', 'three')
+        self._manager.save('bee', 'nekot', 'four')
+        self.assertEqual(self._manager.count, 4)
+        self._manager.discard('bee', 'token')
+        self.assertEqual(self._manager.count, 3)
+        state = self._manager.restore('ant', 'token')
+        self.assertEqual(state.step, 'one')
+        state = self._manager.restore('bee', 'token')
+        self.assertIsNone(state)
+        state = self._manager.restore('ant', 'nekot')
+        self.assertEqual(state.step, 'three')
+        state = self._manager.restore('bee', 'nekot')
+        self.assertEqual(state.step, 'four')
diff --git a/src/mailman/model/usermanager.py b/src/mailman/model/usermanager.py
index ae499452b..3d7777099 100644
--- a/src/mailman/model/usermanager.py
+++ b/src/mailman/model/usermanager.py
@@ -64,7 +64,6 @@ class UserManager:
             user.display_name = (
                 display_name if display_name else address.display_name)
             user.link(address)
-            return user
         return user
 
     @dbconnection
diff --git a/src/mailman/model/workflow.py b/src/mailman/model/workflow.py
new file mode 100644
index 000000000..392ab0798
--- /dev/null
+++ b/src/mailman/model/workflow.py
@@ -0,0 +1,76 @@
+# Copyright (C) 2015 by the Free Software Foundation, Inc.
+#
+# This file is part of GNU Mailman.
+#
+# GNU Mailman is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option)
+# any later version.
+#
+# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
+# more details.
+#
+# You should have received a copy of the GNU General Public License along with
+# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Model for workflow states."""
+
+__all__ = [
+    'WorkflowState',
+    'WorkflowStateManager',
+    ]
+
+
+from mailman.database.model import Model
+from mailman.database.transaction import dbconnection
+from mailman.interfaces.workflow import IWorkflowState, IWorkflowStateManager
+from sqlalchemy import Column, Unicode
+from zope.interface import implementer
+
+
+
+@implementer(IWorkflowState)
+class WorkflowState(Model):
+    """Workflow states."""
+
+    __tablename__ = 'workflowstate'
+
+    name = Column(Unicode, primary_key=True)
+    token = Column(Unicode, primary_key=True)
+    step = Column(Unicode)
+    data = Column(Unicode)
+
+
+
+@implementer(IWorkflowStateManager)
+class WorkflowStateManager:
+    """See `IWorkflowStateManager`."""
+
+    @dbconnection
+    def save(self, store, name, token, step=None, data=None):
+        """See `IWorkflowStateManager`."""
+        state = WorkflowState(name=name, token=token, step=step, data=data)
+        store.add(state)
+
+    @dbconnection
+    def restore(self, store, name, token):
+        """See `IWorkflowStateManager`."""
+        state = store.query(WorkflowState).get((name, token))
+        if state is not None:
+            store.delete(state)
+        return state
+
+    @dbconnection
+    def discard(self, store, name, token):
+        """See `IWorkflowStateManager`."""
+        state = store.query(WorkflowState).get((name, token))
+        if state is not None:
+            store.delete(state)
+
+    @property
+    @dbconnection
+    def count(self, store):
+        """See `IWorkflowStateManager`."""
+        return store.query(WorkflowState).count()
diff --git a/src/mailman/rest/docs/listconf.rst b/src/mailman/rest/docs/listconf.rst
index 841ab3c27..bcf4f856e 100644
--- a/src/mailman/rest/docs/listconf.rst
+++ b/src/mailman/rest/docs/listconf.rst
@@ -61,6 +61,7 @@ All readable attributes for a list are available on a sub-resource.
     scheme: http
     send_welcome_message: True
     subject_prefix: [Ant]
+    subscription_policy: confirm
     volume: 1
     web_host: lists.example.com
     welcome_message_uri: mailman:///welcome.txt
@@ -106,6 +107,7 @@ When using ``PUT``, all writable attributes must be included.
     ...             reply_to_address='bee@example.com',
     ...             send_welcome_message=False,
     ...             subject_prefix='[ant]',
+    ...             subscription_policy='moderate',
     ...             welcome_message_uri='mailman:///welcome.txt',
     ...             default_member_action='hold',
     ...             default_nonmember_action='discard',
@@ -156,6 +158,7 @@ These values are changed permanently.
     ...
     send_welcome_message: False
     subject_prefix: [ant]
+    subscription_policy: moderate
     ...
     welcome_message_uri: mailman:///welcome.txt
 
diff --git a/src/mailman/rest/listconf.py b/src/mailman/rest/listconf.py
index d618ce116..04dea996f 100644
--- a/src/mailman/rest/listconf.py
+++ b/src/mailman/rest/listconf.py
@@ -29,7 +29,8 @@ from mailman.core.errors import (
 from mailman.interfaces.action import Action
 from mailman.interfaces.archiver import ArchivePolicy
 from mailman.interfaces.autorespond import ResponseAction
-from mailman.interfaces.mailinglist import IAcceptableAliasSet, ReplyToMunging
+from mailman.interfaces.mailinglist import (
+    IAcceptableAliasSet, ReplyToMunging, SubscriptionPolicy)
 from mailman.rest.helpers import (
     GetterSetter, bad_request, etag, no_content, okay)
 from mailman.rest.validator import (
@@ -135,6 +136,7 @@ ATTRIBUTES = dict(
     scheme=GetterSetter(None),
     send_welcome_message=GetterSetter(as_boolean),
     subject_prefix=GetterSetter(str),
+    subscription_policy=GetterSetter(enum_validator(SubscriptionPolicy)),
     volume=GetterSetter(None),
     web_host=GetterSetter(None),
     welcome_message_uri=GetterSetter(str),
diff --git a/src/mailman/rest/tests/test_domains.py b/src/mailman/rest/tests/test_domains.py
index 716ded580..9d03859ef 100644
--- a/src/mailman/rest/tests/test_domains.py
+++ b/src/mailman/rest/tests/test_domains.py
@@ -27,6 +27,7 @@ import unittest
 
 from mailman.app.lifecycle import create_list
 from mailman.database.transaction import transaction
+from mailman.interfaces.domain import IDomainManager
 from mailman.interfaces.listmanager import IListManager
 from mailman.testing.helpers import call_api
 from mailman.testing.layers import RESTLayer
@@ -54,6 +55,21 @@ class TestDomains(unittest.TestCase):
             'http://localhost:9001/3.0/domains', data, method="POST")
         self.assertEqual(response.status, 201)
 
+    def test_domain_create_with_single_owner(self):
+        # Creating domain with single owner should not raise InvalidEmailError.
+        content, response = call_api(
+            'http://localhost:9001/3.0/domains', dict(
+                mail_host='example.net',
+                owner='anne@example.com',
+                ),
+            method='POST')
+        self.assertEqual(response.status, 201)
+        # The domain has the expected owner.
+        domain = getUtility(IDomainManager).get('example.net')
+        self.assertEqual(
+            [list(owner.addresses)[0].email for owner in domain.owners],
+            ['anne@example.com'])
+
     def test_bogus_endpoint_extension(self):
         # /domains/<domain>/lists/<anything> is not a valid endpoint.
         with self.assertRaises(HTTPError) as cm:
@@ -102,6 +118,7 @@ class TestDomains(unittest.TestCase):
         self.assertEqual(cm.exception.code, 404)
 
 
+
 
 class TestDomainOwners(unittest.TestCase):
     layer = RESTLayer
diff --git a/src/mailman/rest/tests/test_listconf.py b/src/mailman/rest/tests/test_listconf.py
index b0107b199..ddb43a8ea 100644
--- a/src/mailman/rest/tests/test_listconf.py
+++ b/src/mailman/rest/tests/test_listconf.py
@@ -79,6 +79,7 @@ class TestConfiguration(unittest.TestCase):
                 reply_to_address='bee@example.com',
                 send_welcome_message=False,
                 subject_prefix='[ant]',
+                subscription_policy='confirm_then_moderate',
                 welcome_message_uri='mailman:///welcome.txt',
                 default_member_action='hold',
                 default_nonmember_action='discard',
diff --git a/src/mailman/rest/tests/test_validator.py b/src/mailman/rest/tests/test_validator.py
new file mode 100644
index 000000000..c670fc77c
--- /dev/null
+++ b/src/mailman/rest/tests/test_validator.py
@@ -0,0 +1,48 @@
+# Copyright (C) 2015 by the Free Software Foundation, Inc.
+#
+# This file is part of GNU Mailman.
+#
+# GNU Mailman is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option)
+# any later version.
+#
+# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
+# more details.
+#
+# You should have received a copy of the GNU General Public License along with
+# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Test REST validators."""
+
+__all__ = [
+    'TestValidators',
+    ]
+
+
+import unittest
+
+from mailman.rest.validator import list_of_strings_validator
+from mailman.testing.layers import RESTLayer
+
+
+
+class TestValidators(unittest.TestCase):
+    layer = RESTLayer
+
+    def test_list_of_strings_validator_single(self):
+        # This validator should turn a single key into a list of keys.
+        self.assertEqual(list_of_strings_validator('ant'), ['ant'])
+
+    def test_list_of_strings_validator_multiple(self):
+        # This validator should turn a single key into a list of keys.
+        self.assertEqual(
+            list_of_strings_validator(['ant', 'bee', 'cat']),
+            ['ant', 'bee', 'cat'])
+
+    def test_list_of_strings_validator_invalid(self):
+        # Strings are required.
+        self.assertRaises(ValueError, list_of_strings_validator, 7)
+        self.assertRaises(ValueError, list_of_strings_validator, ['ant', 7])
diff --git a/src/mailman/rest/validator.py b/src/mailman/rest/validator.py
index d09886e36..720d7adc1 100644
--- a/src/mailman/rest/validator.py
+++ b/src/mailman/rest/validator.py
@@ -68,7 +68,9 @@ def language_validator(code):
 
 
 def list_of_strings_validator(values):
-    """Turn a list of things into a list of unicodes."""
+    """Turn a list of things, or a single thing, into a list of unicodes."""
+    if not isinstance(values, (list, tuple)):
+        values = [values]
     for value in values:
         if not isinstance(value, str):
             raise ValueError('Expected str, got {!r}'.format(value))
diff --git a/src/mailman/runners/docs/command.rst b/src/mailman/runners/docs/command.rst
index 82ee33fbc..19d772b00 100644
--- a/src/mailman/runners/docs/command.rst
+++ b/src/mailman/runners/docs/command.rst
@@ -141,15 +141,14 @@ address, and the other is the results of his email command.
     2
 
     >>> from mailman.interfaces.registrar import IRegistrar
-    >>> from zope.component import getUtility
-    >>> registrar = getUtility(IRegistrar)
+    >>> registrar = IRegistrar(mlist)
     >>> for item in messages:
     ...     subject = item.msg['subject']
     ...     print('Subject:', subject)
     ...     if 'confirm' in str(subject):
     ...         token = str(subject).split()[1].strip()
-    ...         status = registrar.confirm(token)
-    ...         assert status, 'Confirmation failed'
+    ...         new_token = registrar.confirm(token)
+    ...         assert new_token is None, 'Confirmation failed'
     Subject: The results of your email commands
     Subject: confirm ...
 
diff --git a/src/mailman/runners/tests/test_confirm.py b/src/mailman/runners/tests/test_confirm.py
index 090451ce7..71ec5988d 100644
--- a/src/mailman/runners/tests/test_confirm.py
+++ b/src/mailman/runners/tests/test_confirm.py
@@ -46,14 +46,14 @@ class TestConfirm(unittest.TestCase):
     layer = ConfigLayer
 
     def setUp(self):
-        registrar = getUtility(IRegistrar)
         self._commandq = config.switchboards['command']
         self._runner = make_testable_runner(CommandRunner, 'command')
         with transaction():
             # Register a subscription requiring confirmation.
             self._mlist = create_list('test@example.com')
             self._mlist.send_welcome_message = False
-            self._token = registrar.register(self._mlist, 'anne@example.org')
+            anne = getUtility(IUserManager).create_address('anne@example.org')
+            self._token = IRegistrar(self._mlist).register(anne)
 
     def test_confirm_with_re_prefix(self):
         subject = 'Re: confirm {0}'.format(self._token)
diff --git a/src/mailman/runners/tests/test_join.py b/src/mailman/runners/tests/test_join.py
index 4006675e4..c3b52cf0c 100644
--- a/src/mailman/runners/tests/test_join.py
+++ b/src/mailman/runners/tests/test_join.py
@@ -160,8 +160,8 @@ class TestJoinWithDigests(unittest.TestCase):
         subject_words = str(messages[1].msg['subject']).split()
         self.assertEqual(subject_words[0], 'confirm')
         token = subject_words[1]
-        status = getUtility(IRegistrar).confirm(token)
-        self.assertTrue(status, 'Confirmation failed')
+        status = IRegistrar(self._mlist).confirm(token)
+        self.assertIsNone(status, 'Confirmation failed')
         # Now, make sure that Anne is a member of the list and is receiving
         # digest deliveries.
         members = getUtility(ISubscriptionService).find_members(
@@ -197,6 +197,8 @@ join digest=no
         self.assertEqual(anne.address.email, 'anne@example.org')
         self.assertEqual(anne.delivery_mode, DeliveryMode.regular)
 
+    # LP: #1444184 - digest=mime is not currently supported.
+    @unittest.expectedFailure
     def test_join_with_mime_digests(self):
         # Test the digest=mime argument to the join command.
         msg = mfs("""\
@@ -211,6 +213,8 @@ join digest=mime
         self.assertEqual(anne.address.email, 'anne@example.org')
         self.assertEqual(anne.delivery_mode, DeliveryMode.mime_digests)
 
+    # LP: #1444184 - digest=mime is not currently supported.
+    @unittest.expectedFailure
     def test_join_with_plain_digests(self):
         # Test the digest=mime argument to the join command.
         msg = mfs("""\
diff --git a/src/mailman/styles/base.py b/src/mailman/styles/base.py
index 50cddbc32..7a77af609 100644
--- a/src/mailman/styles/base.py
+++ b/src/mailman/styles/base.py
@@ -41,7 +41,8 @@ from mailman.interfaces.archiver import ArchivePolicy
 from mailman.interfaces.autorespond import ResponseAction
 from mailman.interfaces.bounce import UnrecognizedBounceDisposition
 from mailman.interfaces.digests import DigestFrequency
-from mailman.interfaces.mailinglist import Personalization, ReplyToMunging
+from mailman.interfaces.mailinglist import (
+    Personalization, ReplyToMunging, SubscriptionPolicy)
 from mailman.interfaces.nntp import NewsgroupModeration
 
 
@@ -75,6 +76,7 @@ class BasicOperation:
         mlist.personalize = Personalization.none
         mlist.default_member_action = Action.defer
         mlist.default_nonmember_action = Action.hold
+        mlist.subscription_policy = SubscriptionPolicy.confirm
         # Notify the administrator of pending requests and membership changes.
         mlist.admin_immed_notify = True
         mlist.admin_notify_mchanges = False
diff --git a/src/mailman/templates/en/confirm.txt b/src/mailman/templates/en/confirm.txt
index d02cb462b..7c8bee75f 100644
--- a/src/mailman/templates/en/confirm.txt
+++ b/src/mailman/templates/en/confirm.txt
@@ -8,9 +8,7 @@ We have received a registration request for the email address
 
 Before you can start using GNU Mailman at this site, you must first confirm
 that this is your email address.  You can do this by replying to this message,
-keeping the Subject header intact.  Or you can visit this web page
-
-    $confirm_url
+keeping the Subject header intact.
 
 If you do not wish to register this email address simply disregard this
 message.  If you think you are being maliciously subscribed to the list, or
diff --git a/src/mailman/templates/en/subauth.txt b/src/mailman/templates/en/subauth.txt
index 1b13ebaeb..041be5e55 100644
--- a/src/mailman/templates/en/subauth.txt
+++ b/src/mailman/templates/en/subauth.txt
@@ -3,9 +3,3 @@ approval:
 
     For:  $username
     List: $listname
-
-At your convenience, visit:
-
-    $admindb_url
-
-to process the request.
diff --git a/src/mailman/utilities/i18n.py b/src/mailman/utilities/i18n.py
index 1d3fbca6f..ba3923e06 100644
--- a/src/mailman/utilities/i18n.py
+++ b/src/mailman/utilities/i18n.py
@@ -150,7 +150,7 @@ def find(template_file, mlist=None, language=None, _trace=False):
         try:
             if _trace:
                 print('@@@', path, end='', file=sys.stderr)
-            fp = open(path)
+            fp = open(path, 'r', encoding='utf-8')
         except IOError as error:
             if error.errno == errno.ENOENT:
                 if _trace:
diff --git a/src/mailman/utilities/importer.py b/src/mailman/utilities/importer.py
index bb4273b12..66a23123c 100644
--- a/src/mailman/utilities/importer.py
+++ b/src/mailman/utilities/importer.py
@@ -41,6 +41,7 @@ from mailman.interfaces.digests import DigestFrequency
 from mailman.interfaces.languages import ILanguageManager
 from mailman.interfaces.mailinglist import IAcceptableAliasSet
 from mailman.interfaces.mailinglist import Personalization, ReplyToMunging
+from mailman.interfaces.mailinglist import SubscriptionPolicy
 from mailman.interfaces.member import DeliveryMode, DeliveryStatus, MemberRole
 from mailman.interfaces.nntp import NewsgroupModeration
 from mailman.interfaces.usermanager import IUserManager
@@ -178,6 +179,7 @@ TYPES = dict(
     personalize=Personalization,
     preferred_language=check_language_code,
     reply_goes_to_list=ReplyToMunging,
+    subscription_policy=SubscriptionPolicy,
     topics_enabled=bool,
     )
 
@@ -202,6 +204,7 @@ NAME_MAPPINGS = dict(
     real_name='display_name',
     send_goodbye_msg='send_goodbye_message',
     send_welcome_msg='send_welcome_message',
+    subscribe_policy='subscription_policy',
     )
 
 # These DateTime fields of the mailinglist table need a type conversion to
diff --git a/src/mailman/utilities/tests/test_import.py b/src/mailman/utilities/tests/test_import.py
index 938ef7d2e..9f3d59d5a 100644
--- a/src/mailman/utilities/tests/test_import.py
+++ b/src/mailman/utilities/tests/test_import.py
@@ -44,7 +44,8 @@ from mailman.interfaces.autorespond import ResponseAction
 from mailman.interfaces.bans import IBanManager
 from mailman.interfaces.bounce import UnrecognizedBounceDisposition
 from mailman.interfaces.languages import ILanguageManager
-from mailman.interfaces.mailinglist import IAcceptableAliasSet
+from mailman.interfaces.mailinglist import (
+    IAcceptableAliasSet, SubscriptionPolicy)
 from mailman.interfaces.member import DeliveryMode, DeliveryStatus
 from mailman.interfaces.nntp import NewsgroupModeration
 from mailman.interfaces.templates import ITemplateLoader
@@ -301,6 +302,34 @@ class TestBasicImport(unittest.TestCase):
         self._import()
         self.assertEqual(self._mlist.encode_ascii_prefixes, True)
 
+    def test_subscription_policy_open(self):
+        self._mlist.subscription_policy = SubscriptionPolicy.confirm
+        self._pckdict['subscribe_policy'] = 0
+        self._import()
+        self.assertEqual(self._mlist.subscription_policy,
+                         SubscriptionPolicy.open)
+
+    def test_subscription_policy_confirm(self):
+        self._mlist.subscription_policy = SubscriptionPolicy.open
+        self._pckdict['subscribe_policy'] = 1
+        self._import()
+        self.assertEqual(self._mlist.subscription_policy,
+                         SubscriptionPolicy.confirm)
+
+    def test_subscription_policy_moderate(self):
+        self._mlist.subscription_policy = SubscriptionPolicy.open
+        self._pckdict['subscribe_policy'] = 2
+        self._import()
+        self.assertEqual(self._mlist.subscription_policy,
+                         SubscriptionPolicy.moderate)
+
+    def test_subscription_policy_confirm_then_moderate(self):
+        self._mlist.subscription_policy = SubscriptionPolicy.open
+        self._pckdict['subscribe_policy'] = 3
+        self._import()
+        self.assertEqual(self._mlist.subscription_policy,
+                         SubscriptionPolicy.confirm_then_moderate)
+
 
 
 class TestArchiveImport(unittest.TestCase):