src/mailman/commands/tests/test_eml_confirm.py

   1 # Copyright (C) 2012-2023 by the Free Software Foundation, Inc.
   2 #
   3 # This file is part of GNU Mailman.
   4 #
   5 # GNU Mailman is free software: you can redistribute it and/or modify it under
   6 # the terms of the GNU General Public License as published by the Free
   7 # Software Foundation, either version 3 of the License, or (at your option)
   8 # any later version.
   9 #
  10 # GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
  11 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  12 # FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
  13 # more details.
  14 #
  15 # You should have received a copy of the GNU General Public License along with
  16 # GNU Mailman.  If not, see <https://www.gnu.org/licenses/>.
  17
  18 """Test the `confirm` command."""
  19
  20 import re
  21 import unittest
  22
  23 from mailman.app.lifecycle import create_list
  24 from mailman.chains.hold import HoldChain
  25 from mailman.commands.eml_confirm import Confirm
  26 from mailman.config import config
  27 from mailman.email.message import Message
  28 from mailman.interfaces.bans import IBanManager
  29 from mailman.interfaces.command import ContinueProcessing
  30 from mailman.interfaces.mailinglist import SubscriptionPolicy
  31 from mailman.interfaces.requests import IListRequests, RequestType
  32 from mailman.interfaces.subscriptions import ISubscriptionManager
  33 from mailman.interfaces.usermanager import IUserManager
  34 from mailman.runners.command import CommandRunner, Results
  35 from mailman.testing.helpers import (
  36     get_queue_messages,
  37     make_testable_runner,
  38     specialized_message_from_string as mfs,
  39     subscribe,
  40 )
  41 from mailman.testing.layers import ConfigLayer
  42 from zope.component import getUtility
  43
  44
  45 class TestConfirmJoin(unittest.TestCase):
  46     """Test the `confirm` command when joining a mailing list."""
  47
  48     layer = ConfigLayer
  49
  50     def setUp(self):
  51         self._mlist = create_list('test@example.com')
  52         anne = getUtility(IUserManager).create_address(
  53             'anne@example.com', 'Anne Person')
  54         self._address = anne
  55         self._token, token_owner, member = ISubscriptionManager(
  56             self._mlist).register(anne)
  57         self._command = Confirm()
  58         # Clear the virgin queue.
  59         get_queue_messages('virgin')
  60
  61     def test_welcome_message(self):
  62         # A confirmation causes a welcome message to be sent to the member, if
  63         # enabled by the mailing list.
  64         status = self._command.process(
  65             self._mlist, Message(), {}, (self._token,), Results())
  66         self.assertEqual(status, ContinueProcessing.no)
  67         # There should be one messages in the queue; the welcome message.
  68         items = get_queue_messages('virgin', expected_count=1)
  69         # Grab the welcome message.
  70         welcome = items[0].msg
  71         self.assertEqual(welcome['subject'],
  72                          'Welcome to the "Test" mailing list')
  73         self.assertEqual(welcome['to'], 'Anne Person <anne@example.com>')
  74
  75     def test_no_welcome_message(self):
  76         # When configured not to send a welcome message, none is sent.
  77         self._mlist.send_welcome_message = False
  78         status = self._command.process(
  79             self._mlist, Message(), {}, (self._token,), Results())
  80         self.assertEqual(status, ContinueProcessing.no)
  81         # There will be no messages in the queue.
  82         get_queue_messages('virgin', expected_count=0)
  83
  84     def test_confim_token_twice(self):
  85         # Don't try to confirm the same token twice.
  86         # We test this by passing a result that already confirms the same
  87         # token and it doesn't try to look at the database.
  88         result = Results()
  89         result.confirms = self._token
  90         status = self._command.process(
  91             self._mlist, Message(), {}, (self._token,), result)
  92         self.assertEqual(status, ContinueProcessing.no)
  93
  94     def test_confirm_banned_address(self):
  95         # Confirmation of a banned address should return an appropriate error.
  96         IBanManager(self._mlist).ban('anne@example.com')
  97         result = Results()
  98         status = self._command.process(
  99             self._mlist, Message(), {}, (self._token,), result)
 100         self.assertEqual(status, ContinueProcessing.no)
 101         # Anne will not be subscribed.
 102         self.assertFalse(self._mlist.is_subscribed('anne@example.com'))
 103         # The result will contain an error message.
 104         self.assertIn('anne@example.com is not allowed to subscribe to '
 105                       'test@example.com', str(result))
 106
 107     def test_confirm_already_member(self):
 108         # Confirmation of an already subscribed address should return an
 109         # appropriate error.
 110         # Subscribe anne.
 111         self._mlist.subscribe(self._address)
 112         result = Results()
 113         status = self._command.process(
 114             self._mlist, Message(), {}, (self._token,), result)
 115         self.assertEqual(status, ContinueProcessing.no)
 116         # The result will contain an error message.
 117         self.assertIn('anne@example.com is already a member of '
 118                       'mailing list test@example.com', str(result))
 119
 120
 121 class TestConfirmLeave(unittest.TestCase):
 122     """Test the `confirm` command when leaving a mailing list."""
 123
 124     layer = ConfigLayer
 125
 126     def setUp(self):
 127         self._mlist = create_list('test@example.com')
 128         anne = subscribe(self._mlist, 'Anne', email='anne@example.com')
 129         self._token, token_owner, member = ISubscriptionManager(
 130             self._mlist).unregister(anne.address)
 131
 132     def test_confirm_leave(self):
 133         msg = mfs("""\
 134 From: Anne Person <anne@example.com>
 135 To: test-confirm+{token}@example.com
 136 Subject: Re: confirm {token}
 137
 138 """.format(token=self._token))
 139         Confirm().process(self._mlist, msg, {}, (self._token,), Results())
 140         # Anne is no longer a member of the mailing list.
 141         member = self._mlist.members.get_member('anne@example.com')
 142         self.assertIsNone(member)
 143
 144     def test_confirm_leave_moderate(self):
 145         msg = mfs("""\
 146 From: Anne Person <anne@example.com>
 147 To: test-confirm+{token}@example.com
 148 Subject: Re: confirm {token}
 149
 150 """.format(token=self._token))
 151         self._mlist.unsubscription_policy = (
 152             SubscriptionPolicy.confirm_then_moderate)
 153         # Clear any previously queued confirmation messages.
 154         get_queue_messages('virgin')
 155         Confirm().process(self._mlist, msg, {}, (self._token,), Results())
 156         # Anne is still a member of the mailing list.
 157         member = self._mlist.members.get_member('anne@example.com')
 158         self.assertIsNotNone(member)
 159         # There should be a notice to the list owners
 160         item = get_queue_messages('virgin', expected_count=1)[0]
 161         self.assertEqual(item.msg['to'], 'test-owner@example.com')
 162
 163
 164 class TestEmailResponses(unittest.TestCase):
 165     """Test the `confirm` command through the command runner."""
 166
 167     layer = ConfigLayer
 168
 169     def setUp(self):
 170         self._mlist = create_list('test@example.com')
 171
 172     def test_confirm_then_moderate_workflow(self):
 173         # Issue #114 describes a problem when confirming the moderation email.
 174         self._mlist.subscription_policy = (
 175             SubscriptionPolicy.confirm_then_moderate)
 176         bart = getUtility(IUserManager).create_address(
 177             'bart@example.com', 'Bart Person')
 178         # Clear any previously queued confirmation messages.
 179         get_queue_messages('virgin')
 180         self._token, token_owner, member = ISubscriptionManager(
 181             self._mlist).register(bart)
 182         # There should now be one email message in the virgin queue, i.e. the
 183         # confirmation message sent to Bart.
 184         items = get_queue_messages('virgin', expected_count=1)
 185         msg = items[0].msg
 186         # Confirmations come first, so this one goes to the subscriber.
 187         self.assertEqual(msg['to'], 'bart@example.com')
 188         confirm, token = re.sub(r'^.*(confirm)\+([^+@]*)@.*$', r'\1 \2',
 189                                 str(msg['from'])).split()
 190         self.assertEqual(confirm, 'confirm')
 191         self.assertEqual(token, self._token)
 192         # Craft a confirmation response with the expected tokens.
 193         user_response = Message()
 194         user_response['From'] = 'bart@example.com'
 195         user_response['To'] = 'test-confirm+{}@example.com'.format(token)
 196         user_response['Subject'] = 'Re: confirm {}'.format(token)
 197         user_response.set_payload('')
 198         # Process the message through the command runner.
 199         config.switchboards['command'].enqueue(
 200             user_response, listid='test.example.com')
 201         make_testable_runner(CommandRunner, 'command').run()
 202         # There are now two messages in the virgin queue.  One is going to the
 203         # subscriber containing the results of their confirmation message, and
 204         # the other is to the moderators informing them that they need to
 205         # handle the moderation queue.
 206         items = get_queue_messages('virgin', expected_count=2)
 207         if items[0].msg['to'] == 'bart@example.com':
 208             results = items[0].msg
 209             moderator_msg = items[1].msg
 210         else:
 211             results = items[1].msg
 212             moderator_msg = items[0].msg
 213         # Check the moderator message first.
 214         self.assertEqual(moderator_msg['to'], 'test-owner@example.com')
 215         self.assertEqual(
 216             moderator_msg['subject'],
 217             'New subscription request to Test from bart@example.com')
 218         lines = moderator_msg.get_payload().splitlines()
 219         self.assertEqual(
 220             lines[-2].strip(),
 221             'For:  Bart Person <bart@example.com>')
 222         self.assertEqual(lines[-1].strip(), 'List: test@example.com')
 223         # Now check the results message.
 224         self.assertEqual(
 225             str(results['subject']), 'The results of your email commands')
 226         self.assertMultiLineEqual(results.get_payload(), """\
 227 The results of your email command are provided below.
 228
 229 - Original message details:
 230 From: bart@example.com
 231 Subject: Re: confirm {}
 232 Date: n/a
 233 Message-ID: n/a
 234
 235 - Results:
 236 Confirmed
 237
 238 - Done.
 239 """.format(token))
 240
 241
 242 class TestConfirmMessage(unittest.TestCase):
 243     """Test the `confirm` command for held messages."""
 244
 245     layer = ConfigLayer
 246
 247     def setUp(self):
 248         self._mlist = create_list('test@example.com')
 249         self._mlist.moderator_password = config.password_context.encrypt(
 250             'super secret').encode('us-ascii')
 251         self._mlist.respond_to_post_requests = False
 252         self._mlist.admin_immed_notify = True
 253         # Create a message and hold it.
 254         msg = Message()
 255         msg['To'] = 'test@example.com'
 256         msg['From'] = 'nobody@example.com'
 257         msg['Subject'] = 'A message to be held'
 258         msg['Message-ID'] = '<alpha>'
 259         msg.set_payload('body')
 260         msgdata = dict()
 261         HoldChain()._process(self._mlist, msg, msgdata)
 262         # There should now be one message in the virgin queue.
 263         items = get_queue_messages('virgin', expected_count=1)
 264         # And the confirm command is the Subject of the third part.
 265         self._command = (
 266             items[0].msg.get_payload(2).get_payload(0).get('subject'))
 267         self._requestdb = IListRequests(self._mlist)
 268
 269     def test_discard_message(self):
 270         # Test that confirming the token discards the message.
 271         self.assertEqual(self._requestdb.count_of(RequestType.held_message), 1)
 272         # Craft a confirmation message.
 273         user_response = Message()
 274         user_response['From'] = 'bart@example.com'
 275         user_response['To'] = 'test-request@example.com'
 276         user_response['Subject'] = self._command
 277         user_response.set_payload('')
 278         # Process the message through the command runner.
 279         config.switchboards['command'].enqueue(
 280             user_response, listid='test.example.com')
 281         make_testable_runner(CommandRunner, 'command').run()
 282         # The held message should be disposed of.
 283         self.assertEqual(self._requestdb.count_of(RequestType.held_message), 0)
 284         # There should be one `results` message in the virgin queue.
 285         items = get_queue_messages('virgin', expected_count=1)
 286         response = items[0].msg.get_payload()
 287         self.assertIn('Message discarded', response)
 288
 289     def test_accept_message(self):
 290         # Test that confirming the token with Approved: accepts the message.
 291         self.assertEqual(self._requestdb.count_of(RequestType.held_message), 1)
 292         # Craft a confirmation message.
 293         user_response = Message()
 294         user_response['From'] = 'bart@example.com'
 295         user_response['To'] = 'test-request@example.com'
 296         user_response['Subject'] = self._command
 297         user_response['Approved'] = 'super secret'
 298         user_response.set_payload('')
 299         # Process the message through the command runner.
 300         config.switchboards['command'].enqueue(
 301             user_response, listid='test.example.com')
 302         make_testable_runner(CommandRunner, 'command').run()
 303         # The held message should be disposed of.
 304         self.assertEqual(self._requestdb.count_of(RequestType.held_message), 0)
 305         # There should be one `results` message in the virgin queue.
 306         items = get_queue_messages('virgin', expected_count=1)
 307         response = items[0].msg.get_payload()
 308         self.assertIn('Message accepted', response)
 309
 310     def test_accept_wrong_password(self):
 311         # Test that confirming the token with incorrest password leaves the
 312         # message held.
 313         self.assertEqual(self._requestdb.count_of(RequestType.held_message), 1)
 314         # Craft a confirmation message.
 315         user_response = Message()
 316         user_response['From'] = 'bart@example.com'
 317         user_response['To'] = 'test-request@example.com'
 318         user_response['Subject'] = self._command
 319         user_response['Approved'] = 'wrong'
 320         user_response.set_payload('')
 321         # Process the message through the command runner.
 322         config.switchboards['command'].enqueue(
 323             user_response, listid='test.example.com')
 324         make_testable_runner(CommandRunner, 'command').run()
 325         # The held message should still be held.
 326         self.assertEqual(self._requestdb.count_of(RequestType.held_message), 1)
 327         # There should be one `results` message in the virgin queue.
 328         items = get_queue_messages('virgin', expected_count=1)
 329         response = items[0].msg.get_payload()
 330         self.assertIn('Invalid Approved: password', response)