3 files changed, 73 insertions, 17 deletions

diff --git a/src/mailman_pgp/commands/eml_key.py b/src/mailman_pgp/commands/eml_key.py
index 5bb2bd8..be0af44 100644
--- a/src/mailman_pgp/commands/eml_key.py
+++ b/src/mailman_pgp/commands/eml_key.py
@@ -67,6 +67,12 @@ def _cmd_set(pgp_list, mlist, msg, msgdata, arguments, results):
     if len(keys) != 1:
         print('More than one key! Send only one key.', file=results)
         return ContinueProcessing.no
+    key = keys.pop()
+
+    if not key.is_public:
+        print('You probably wanted to send your public key only.',
+              file=results)
+        return ContinueProcessing.no
 
     email = get_email(msg)
     if not email:
@@ -90,7 +96,7 @@ def _cmd_set(pgp_list, mlist, msg, msgdata, arguments, results):
         return ContinueProcessing.no
 
     with transaction():
-        pgp_address.key = keys.pop()
+        pgp_address.key = key
     ISubscriptionManager(mlist).confirm(token)
 
     print('Key succesfully set.', file=results)
@@ -206,8 +212,14 @@ def _cmd_change(pgp_list, mlist, msg, msgdata, arguments, results):
     if len(keys) != 1:
         print('More than one key! Send only one key.', file=results)
         return ContinueProcessing.no
+    key = keys.pop()
+
+    if not key.is_public:
+        print('You probably wanted to send your public key only.',
+              file=results)
+        return ContinueProcessing.no
 
-    workflow = KeyChangeWorkflow(mlist, pgp_address, keys.pop())
+    workflow = KeyChangeWorkflow(mlist, pgp_address, key)
     list(workflow)
     print('Key change request received.', file=results)
     return ContinueProcessing.no
diff --git a/src/mailman_pgp/commands/tests/test_key.py b/src/mailman_pgp/commands/tests/test_key.py
index af482b4..a9d021b 100644
--- a/src/mailman_pgp/commands/tests/test_key.py
+++ b/src/mailman_pgp/commands/tests/test_key.py
@@ -241,6 +241,21 @@ class TestPreSubscription(unittest.TestCase):
         self.assertIn('More than one key! Send only one key.',
                       results_msg.get_payload())
 
+    def test_set_private_key(self):
+        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(self.bart_key)
+
+        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('You probably wanted to send your public key only.',
+                      results_msg.get_payload())
+
     def test_set_no_email(self):
         message = _create_mixed('', 'test@example.com', 'key set token')
         wrapped_message = MIMEWrapper(message)
@@ -685,6 +700,34 @@ class TestAfterSubscription(unittest.TestCase):
         self.assertIn('Extraneous argument/s: extra,arguments',
                       results_msg.get_payload())
 
+    def test_change_no_email(self):
+        message = _create_mixed('', 'test@example.com', 'key change')
+        wrapped_message = MIMEWrapper(message)
+        message = wrapped_message.attach_key(self.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 change key of.', results_msg.get_payload())
+
+    def test_change_no_pgp_address(self):
+        message = _create_mixed('bart@example.com', 'test@example.com',
+                                'key change')
+        wrapped_message = MIMEWrapper(message)
+        message = wrapped_message.attach_key(self.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('A pgp enabled address not found.',
+                      results_msg.get_payload())
+
     def test_change_no_key(self):
         bart = getUtility(IUserManager).create_address('bart@example.com',
                                                        'Bart Person')
@@ -732,24 +775,19 @@ class TestAfterSubscription(unittest.TestCase):
         self.assertIn('More than one key! Send only one key.',
                       results_msg.get_payload())
 
-    def test_change_no_email(self):
-        message = _create_mixed('', 'test@example.com', 'key change')
-        wrapped_message = MIMEWrapper(message)
-        message = wrapped_message.attach_key(self.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 change key of.', results_msg.get_payload())
+    def test_change_private_key(self):
+        bart = getUtility(IUserManager).create_address('bart@example.com',
+                                                       'Bart Person')
+        with transaction() as t:
+            pgp_address = PGPAddress(bart)
+            pgp_address.key = self.bart_key.pubkey
+            pgp_address.key_confirmed = True
+            t.add(pgp_address)
 
-    def test_change_no_pgp_address(self):
         message = _create_mixed('bart@example.com', 'test@example.com',
                                 'key change')
         wrapped_message = MIMEWrapper(message)
-        message = wrapped_message.attach_key(self.bart_key.pubkey)
+        message = wrapped_message.attach_key(self.bart_key)
 
         mm_config.switchboards['command'].enqueue(message,
                                                   listid='test.example.com')
@@ -757,7 +795,7 @@ class TestAfterSubscription(unittest.TestCase):
         items = get_queue_messages('virgin', expected_count=1)
         results_msg = items[0].msg
 
-        self.assertIn('A pgp enabled address not found.',
+        self.assertIn('You probably wanted to send your public key only.',
                       results_msg.get_payload())
 
 
diff --git a/src/mailman_pgp/model/address.py b/src/mailman_pgp/model/address.py
index ee84d22..0cfefe2 100644
--- a/src/mailman_pgp/model/address.py
+++ b/src/mailman_pgp/model/address.py
@@ -123,6 +123,12 @@ class PGPAddress(Base):
 
     @staticmethod
     def for_email(email):
+        """
+
+        :param email:
+        :return:
+        :rtype: PGPAddress|None
+        """
         if email is None:
             return None
         return PGPAddress.query().filter_by(email=email).first()