1 files changed, 151 insertions, 60 deletions

diff --git a/src/mailman_pgp/commands/tests/test_key.py b/src/mailman_pgp/commands/tests/test_key.py
index b3d8c95..48ca3c9 100644
--- a/src/mailman_pgp/commands/tests/test_key.py
+++ b/src/mailman_pgp/commands/tests/test_key.py
@@ -40,6 +40,24 @@ from mailman_pgp.workflows.base import CONFIRM_REQUEST
 from mailman_pgp.workflows.subscription import OpenSubscriptionPolicy
 
 
+def _create_plain(from_hdr, to_hdr, subject_hdr, payload):
+    message = Message()
+    message['From'] = from_hdr
+    message['To'] = to_hdr
+    message['Subject'] = subject_hdr
+    message.set_payload(payload)
+    return message
+
+
+def _create_mixed(from_hdr, to_hdr, subject_hdr):
+    message = Message()
+    message['From'] = from_hdr
+    message['To'] = to_hdr
+    message['Subject'] = subject_hdr
+    message.set_type('multipart/mixed')
+    return message
+
+
 class TestPreDispatch(unittest.TestCase):
     layer = PGPConfigLayer
 
@@ -47,11 +65,8 @@ class TestPreDispatch(unittest.TestCase):
         self.mlist = create_list('test@example.com')
 
     def test_no_arguments(self):
-        message = Message()
-        message['From'] = 'bart@example.com'
-        message['To'] = 'test@example.com'
-        message['Subject'] = 'key'
-        message.set_payload('')
+        message = _create_plain('bart@example.com', 'test@example.com',
+                                'key', '')
 
         mm_config.switchboards['command'].enqueue(message,
                                                   listid='test.example.com')
@@ -62,11 +77,8 @@ class TestPreDispatch(unittest.TestCase):
         self.assertIn('No sub-command specified', results_msg.get_payload())
 
     def test_wrong_subcommand(self):
-        message = Message()
-        message['From'] = 'bart@example.com'
-        message['To'] = 'test@example.com'
-        message['Subject'] = 'key wroooooong'
-        message.set_payload('')
+        message = _create_plain('bart@example.com', 'test@example.com',
+                                'key wrooooooong', '')
 
         mm_config.switchboards['command'].enqueue(message,
                                                   listid='test.example.com')
@@ -77,11 +89,8 @@ class TestPreDispatch(unittest.TestCase):
         self.assertIn('Wrong sub-command specified', results_msg.get_payload())
 
     def test_no_pgp_list(self):
-        message = Message()
-        message['From'] = 'bart@example.com'
-        message['To'] = 'test@example.com'
-        message['Subject'] = 'key set'
-        message.set_payload('')
+        message = _create_plain('bart@example.com', 'test@example.com',
+                                'key set', '')
 
         mm_config.switchboards['command'].enqueue(message,
                                                   listid='test.example.com')
@@ -99,7 +108,7 @@ class TestPreSubscription(unittest.TestCase):
     def setUp(self):
         self.mlist = create_list('test@example.com', style_name='pgp-default')
         self.pgp_list = PGPMailingList.for_list(self.mlist)
-        self.pgp_list.generate_key(True)
+        self.pgp_list.key = load_key('ecc_p256.priv.asc')
 
     def test_set(self):
         self.mlist.subscription_policy = OpenSubscriptionPolicy
@@ -113,11 +122,8 @@ class TestPreSubscription(unittest.TestCase):
 
         bart_key = load_key('rsa_1024.priv.asc')
 
-        set_message = Message()
-        set_message['From'] = 'bart@example.com'
-        set_message['To'] = 'test@example.com'
-        set_message['Subject'] = 'Re: key set {}'.format(token)
-        set_message.set_type('multipart/mixed')
+        set_message = _create_mixed('bart@example.com', 'test@example.com',
+                                    'Re: key set {}'.format(token))
         wrapped_set_message = MIMEWrapper(set_message)
         set_message = wrapped_set_message.attach_key(bart_key.pubkey)
 
@@ -133,19 +139,22 @@ class TestPreSubscription(unittest.TestCase):
 
         items = get_queue_messages('virgin', expected_count=2)
         if items[0].msg['Subject'] == 'The results of your email commands':
+            results = items[0].msg
             confirm_request = items[1].msg
         else:
+            results = items[1].msg
             confirm_request = items[0].msg
 
+        self.assertIn('Key succesfully set.', results.get_payload())
+        self.assertIn('Key fingerprint: {}'.format(bart_key.fingerprint),
+                      results.get_payload())
+
         confirm_wrapped = PGPWrapper(confirm_request)
         self.assertTrue(confirm_wrapped.is_encrypted())
 
     def test_set_no_token(self):
-        message = Message()
-        message['From'] = 'bart@example.com'
-        message['To'] = 'test@example.com'
-        message['Subject'] = 'key set'
-        message.set_payload('')
+        message = _create_plain('bart@example.com', 'test@example.com',
+                                'key set', '')
 
         mm_config.switchboards['command'].enqueue(message,
                                                   listid='test.example.com')
@@ -156,11 +165,8 @@ class TestPreSubscription(unittest.TestCase):
         self.assertIn('Missing token.', results_msg.get_payload())
 
     def test_set_no_key(self):
-        message = Message()
-        message['From'] = 'bart@example.com'
-        message['To'] = 'test@example.com'
-        message['Subject'] = 'key set token'
-        message.set_payload('')
+        message = _create_plain('bart@example.com', 'test@example.com',
+                                'key set token', '')
 
         mm_config.switchboards['command'].enqueue(message,
                                                   listid='test.example.com')
@@ -175,11 +181,8 @@ class TestPreSubscription(unittest.TestCase):
         bart_key = load_key('rsa_1024.priv.asc')
         anne_key = load_key('ecc_p256.priv.asc')
 
-        set_message = Message()
-        set_message['From'] = 'bart@example.com'
-        set_message['To'] = 'test@example.com'
-        set_message['Subject'] = 'Re: key set token'
-        set_message.set_type('multipart/mixed')
+        set_message = _create_mixed('bart@example.com', 'test@example.com',
+                                    'Re: key set token')
         wrapped_set_message = MIMEWrapper(set_message)
         set_message = wrapped_set_message.attach_key(bart_key.pubkey)
         wrapped_set_message = MIMEWrapper(set_message)
@@ -194,6 +197,38 @@ class TestPreSubscription(unittest.TestCase):
         self.assertIn('More than one key! Send only one key.',
                       results_msg.get_payload())
 
+    def test_set_no_email(self):
+        bart_key = load_key('rsa_1024.priv.asc')
+
+        message = _create_mixed('', 'test@example.com', 'key set token')
+        wrapped_message = MIMEWrapper(message)
+        message = wrapped_message.attach_key(bart_key.pubkey)
+
+        mm_config.switchboards['command'].enqueue(message,
+                                                  listid='test.example.com')
+        make_testable_runner(CommandRunner, 'command').run()
+        items = get_queue_messages('virgin', expected_count=1)
+        results_msg = items[0].msg
+
+        self.assertIn('No email to subscribe with.', results_msg.get_payload())
+
+    def test_set_no_address(self):
+        bart_key = load_key('rsa_1024.priv.asc')
+
+        set_message = _create_mixed('bart@example.com', 'test@example.com',
+                                    'key set token')
+        wrapped_set_message = MIMEWrapper(set_message)
+        set_message = wrapped_set_message.attach_key(bart_key.pubkey)
+
+        mm_config.switchboards['command'].enqueue(set_message,
+                                                  listid='test.example.com')
+        make_testable_runner(CommandRunner, 'command').run()
+        items = get_queue_messages('virgin', expected_count=1)
+        results_msg = items[0].msg
+
+        self.assertIn('No adddress to subscribe with.',
+                      results_msg.get_payload())
+
     def test_confirm(self):
         self.mlist.subscription_policy = OpenSubscriptionPolicy
         bart = getUtility(IUserManager).create_address('bart@example.com',
@@ -212,16 +247,48 @@ class TestPreSubscription(unittest.TestCase):
             pgp_address.key = bart_key.pubkey
             t.add(pgp_address)
 
-        confirm_message = Message()
-        confirm_message['From'] = 'bart@example.com'
-        confirm_message['To'] = 'test@example.com'
-        confirm_message['Subject'] = 'Re: key confirm {}'.format(token)
-        confirm_message.set_payload(
-                CONFIRM_REQUEST.format(bart_key.fingerprint, token))
-        wrapped_confirm_message = MIMEWrapper(confirm_message)
-        confirm_message = wrapped_confirm_message.sign(bart_key)
+        message = _create_plain('bart@example.com', 'test@example.com',
+                                'Re: key confirm {}'.format(token),
+                                CONFIRM_REQUEST.format(bart_key.fingerprint,
+                                                       token))
+        wrapped_message = MIMEWrapper(message)
+        message = wrapped_message.sign(bart_key)
 
-        mm_config.switchboards['command'].enqueue(confirm_message,
+        mm_config.switchboards['command'].enqueue(message,
+                                                  listid='test.example.com')
+        make_testable_runner(CommandRunner, 'command').run()
+
+        pgp_address = PGPAddress.for_address(bart)
+        self.assertTrue(pgp_address.key_confirmed)
+        self.assertTrue(self.mlist.is_subscribed(bart))
+
+    def test_confirm_encrypted(self):
+        self.mlist.subscription_policy = OpenSubscriptionPolicy
+        bart = getUtility(IUserManager).create_address('bart@example.com',
+                                                       'Bart Person')
+        bart.verified_on = now()
+
+        bart_key = load_key('rsa_1024.priv.asc')
+
+        token, token_owner, member = ISubscriptionManager(self.mlist).register(
+                bart, pubkey=bart_key.pubkey)
+
+        get_queue_messages('virgin')
+
+        with transaction() as t:
+            pgp_address = PGPAddress(bart)
+            pgp_address.key = bart_key.pubkey
+            t.add(pgp_address)
+
+        message = _create_plain('bart@example.com', 'test@example.com',
+                                'Re: key confirm {}'.format(token),
+                                CONFIRM_REQUEST.format(bart_key.fingerprint,
+                                                       token))
+        wrapped_message = MIMEWrapper(message)
+        message = wrapped_message.sign_encrypt(bart_key, self.pgp_list.pubkey,
+                                               bart_key.pubkey)
+
+        mm_config.switchboards['command'].enqueue(message,
                                                   listid='test.example.com')
         make_testable_runner(CommandRunner, 'command').run()
 
@@ -230,11 +297,8 @@ class TestPreSubscription(unittest.TestCase):
         self.assertTrue(self.mlist.is_subscribed(bart))
 
     def test_confirm_no_token(self):
-        message = Message()
-        message['From'] = 'bart@example.com'
-        message['To'] = 'test@example.com'
-        message['Subject'] = 'key confirm'
-        message.set_payload('')
+        message = _create_plain('bart@example.com', 'test@example.com',
+                                'key confirm', '')
 
         mm_config.switchboards['command'].enqueue(message,
                                                   listid='test.example.com')
@@ -245,11 +309,8 @@ class TestPreSubscription(unittest.TestCase):
         self.assertIn('Missing token.', results_msg.get_payload())
 
     def test_confirm_no_email(self):
-        message = Message()
-        message['From'] = ''
-        message['To'] = 'test@example.com'
-        message['Subject'] = 'key confirm token'
-        message.set_payload('')
+        message = _create_plain('', 'test@example.com',
+                                'key confirm token', '')
 
         mm_config.switchboards['command'].enqueue(message,
                                                   listid='test.example.com')
@@ -260,11 +321,8 @@ class TestPreSubscription(unittest.TestCase):
         self.assertIn('No email to subscribe with.', results_msg.get_payload())
 
     def test_confirm_no_pgp_address(self):
-        message = Message()
-        message['From'] = 'bart@example.com'
-        message['To'] = 'test@example.com'
-        message['Subject'] = 'key confirm token'
-        message.set_payload('')
+        message = _create_plain('bart@example.com', 'test@example.com',
+                                'key confirm token', '')
 
         mm_config.switchboards['command'].enqueue(message,
                                                   listid='test.example.com')
@@ -274,3 +332,36 @@ class TestPreSubscription(unittest.TestCase):
 
         self.assertIn('A pgp enabled address not found.',
                       results_msg.get_payload())
+
+    def test_confirm_not_signed(self):
+        self.mlist.subscription_policy = OpenSubscriptionPolicy
+        bart = getUtility(IUserManager).create_address('bart@example.com',
+                                                       'Bart Person')
+        bart.verified_on = now()
+
+        bart_key = load_key('rsa_1024.priv.asc')
+
+        token, token_owner, member = ISubscriptionManager(self.mlist).register(
+                bart, pubkey=bart_key.pubkey)
+
+        get_queue_messages('virgin')
+
+        with transaction() as t:
+            pgp_address = PGPAddress(bart)
+            pgp_address.key = bart_key.pubkey
+            t.add(pgp_address)
+
+        message = _create_plain('bart@example.com', 'test@example.com',
+                                'Re: key confirm {}'.format(token),
+                                CONFIRM_REQUEST.format(bart_key.fingerprint,
+                                                       token))
+
+        mm_config.switchboards['command'].enqueue(message,
+                                                  listid='test.example.com')
+        make_testable_runner(CommandRunner, 'command').run()
+
+        items = get_queue_messages('virgin', expected_count=1)
+        results_msg = items[0].msg
+
+        self.assertIn('Message not signed, ignoring.',
+                      results_msg.get_payload())