test coverage complete

apprise/plugins/email.py

@@ -900,25 +900,29 @@ def send(self, body, title='', notify_type=NotifyType.INFO, attach=None,
 
                 # Apply our encryption
                 encrypted_content = self.pgp_encrypt_message(base.as_string())
-                if encrypted_content:
-                    # prepare our messsage
-                    base = MIMEMultipart(
-                        "encrypted", protocol="application/pgp-encrypted")
-
-                    # Store Autocrypt header (DeltaChat Support)
-                    base.add_header(
-                        "Autocrypt",
-                        "addr=%s; prefer-encrypt=mutual" % formataddr(
-                            (False, to_addr), charset='utf-8'))
-
-                    # Set Encryption Info Part
-                    enc_payload = MIMEText("Version: 1", "plain")
-                    enc_payload.set_type("application/pgp-encrypted")
-                    base.attach(enc_payload)
-
-                    enc_payload = MIMEBase("application", "octet-stream")
-                    enc_payload.set_payload(encrypted_content)
-                    base.attach(enc_payload)
+                if not encrypted_content:
+                    self.logger.warning('Unable to PGP encrypt email')
+                    # Unable to send notification
+                    return False
+
+                # prepare our messsage
+                base = MIMEMultipart(
+                    "encrypted", protocol="application/pgp-encrypted")
+
+                # Store Autocrypt header (DeltaChat Support)
+                base.add_header(
+                    "Autocrypt",
+                    "addr=%s; prefer-encrypt=mutual" % formataddr(
+                        (False, to_addr), charset='utf-8'))
+
+                # Set Encryption Info Part
+                enc_payload = MIMEText("Version: 1", "plain")
+                enc_payload.set_type("application/pgp-encrypted")
+                base.attach(enc_payload)
+
+                enc_payload = MIMEBase("application", "octet-stream")
+                enc_payload.set_payload(encrypted_content)
+                base.attach(enc_payload)
 
             # Apply any provided custom headers
             for k, v in self.headers.items():
@@ -1147,13 +1151,11 @@ def pgp_pubkey(self, email=None):
 
         for email in emails:
             _entry = email.split('@')[0].lower()
-            if _entry not in fnames:
-                fnames.insert(0, f'{_entry}-pub.asc')
+            fnames.insert(0, f'{_entry}-pub.asc')
 
             # Lowercase email (Highest Priority)
             _entry = email.lower()
-            if _entry not in fnames:
-                fnames.insert(0, f'{_entry}-pub.asc')
+            fnames.insert(0, f'{_entry}-pub.asc')
 
         return next(
             (os.path.join(self.store.path, fname)
@@ -1179,7 +1181,7 @@ def pgp_public_key(self, path=None, email=None):
                 self.logger.warning('No PGP Public Key could be loaded')
                 return None
 
-        # Persistent storage key:
+        # Persistent storage key
         ps_key = hashlib.sha1(
             os.path.abspath(path).encode('utf-8')).hexdigest()
         if ps_key in self.pgp_public_keys:
@@ -1230,6 +1232,11 @@ def pgp_encrypt_message(self, message, path=None):
             encrypted_message = public_key.encrypt(message_object)
             return str(encrypted_message)
 
+        except pgpy.errors.PGPError:
+            # Encryption not Possible
+            self.logger.debug(
+                'PGP Public Key Corruption; encryption not possible')
+
         except NameError:
             # PGPy not installed
             self.logger.debug('PGPy not installed; Skipping PGP encryption')

test/test_plugin_email.py

@@ -2147,6 +2147,12 @@ def test_plugin_email_pgp(mock_smtp, mock_smtpssl, tmpdir):
     # However explicitly setting a path works
     assert obj.pgp_generate_keys(str(tmpdir1)) is True
 
+    # Prepare Invalid PGP Key
+    obj = Apprise.instantiate(
+        'mailto://pgp:[email protected]?pgp=yes&pgpkey=invalid',
+        asset=asset)
+    assert obj.pgp_pubkey() is False
+
     tmpdir2 = tmpdir.mkdir('tmp02')
     asset = AppriseAsset(
         storage_mode=PersistentStoreMode.FLUSH,
@@ -2204,6 +2210,10 @@ def test_plugin_email_pgp(mock_smtp, mock_smtpssl, tmpdir):
         os.path.join(obj.store.path, '[email protected]'))
 
     assert obj.pgp_pubkey() is None
+    assert obj.pgp_pubkey(email="[email protected]") is None
+    assert obj.pgp_pubkey(email="[email protected]")\
+        .endswith('[email protected]')
+
     assert obj.pgp_pubkey(email="[email protected]")\
         .endswith('[email protected]')
     assert obj.pgp_pubkey(email="[email protected]")\
@@ -2214,6 +2224,7 @@ def test_plugin_email_pgp(mock_smtp, mock_smtpssl, tmpdir):
         os.path.join(obj.store.path, '[email protected]'),
         os.path.join(obj.store.path, 'user-pub.asc'),
     )
+
     assert obj.pgp_pubkey(email="[email protected]")\
         .endswith('[email protected]')
     assert obj.pgp_pubkey(email="[email protected]")\
@@ -2255,13 +2266,20 @@ def test_plugin_email_pgp(mock_smtp, mock_smtpssl, tmpdir):
     obj = Apprise.instantiate(
         'mailto://chris:[email protected]/[email protected]?pgp=yes', asset=asset)
 
-    with mock.patch('builtins.open', side_effect=OSError()):
+    with mock.patch('builtins.open', side_effect=FileNotFoundError):
+        # can't open key
+        assert obj.pgp_public_key(path=obj.store.path) is None
+
+    with mock.patch('builtins.open', side_effect=OSError):
         # can't open key
         assert obj.pgp_public_key(path=obj.store.path) is None
         # Test unlink
-        with mock.patch('os.unlink', side_effect=OSError()):
+        with mock.patch('os.unlink', side_effect=OSError):
             assert obj.pgp_public_key(path=obj.store.path) is None
 
+        # Key Generation will fail
+        assert obj.pgp_generate_keys() is False
+
     with mock.patch('pgpy.PGPKey.new', side_effect=NameError):
         # Can't Generate keys
         assert obj.pgp_generate_keys() is False
@@ -2282,3 +2300,67 @@ def test_plugin_email_pgp(mock_smtp, mock_smtpssl, tmpdir):
 
     with mock.patch('pgpy.PGPMessage.new', side_effect=NameError):
         assert obj.pgp_encrypt_message("message") is None
+        # Attempts to encrypt a message
+        assert obj.notify('test') is False
+
+    # Create new keys
+    assert obj.pgp_generate_keys() is True
+    with mock.patch('os.path.isfile', return_value=False):
+        with mock.patch('builtins.open', side_effect=OSError):
+            with mock.patch('os.unlink', return_value=None):
+                assert obj.pgp_generate_keys() is False
+
+    # Testing again
+    tmpdir5 = tmpdir.mkdir('tmp05')
+    asset = AppriseAsset(
+        storage_mode=PersistentStoreMode.FLUSH,
+        storage_path=str(tmpdir5),
+    )
+    obj = Apprise.instantiate(
+        'mailto://chris:[email protected]/[email protected]?pgp=yes', asset=asset)
+
+    # Catch edge case where we just can't generate the the key
+    with mock.patch('os.path.isfile', side_effect=(
+            # 5x False to skip through pgp_pubkey()
+            False, False, False, False, False, False,
+            # 1x True to pass pgp_generate_keys()
+            True,
+            # 5x False to skip through pgp_pubkey() second call
+            False, False, False, False, False, False)):
+        with mock.patch('pgpy.PGPKey.from_blob',
+                        side_effect=FileNotFoundError):
+            assert obj.pgp_public_key() is None
+
+    # Corrupt Data
+    tmpdir6 = tmpdir.mkdir('tmp06')
+    asset = AppriseAsset(
+        storage_mode=PersistentStoreMode.FLUSH,
+        storage_path=str(tmpdir6),
+    )
+    obj = Apprise.instantiate(
+        'mailto://chris:[email protected]/[email protected]?pgp=yes', asset=asset)
+
+    shutil.copyfile(
+        os.path.join(TEST_VAR_DIR, 'pgp', 'corrupt-pub.asc'),
+        os.path.join(obj.store.path, 'chris-pub.asc'),
+    )
+
+    # Key is corrupted
+    obj.notify('test') is False
+
+    shutil.copyfile(
+        os.path.join(TEST_VAR_DIR, 'apprise-test.jpeg'),
+        os.path.join(obj.store.path, 'chris-pub.asc'),
+    )
+
+    # Key is a binary image; definitely not a valid key
+    obj.notify('test') is False
+
+    # Using a public key
+    shutil.copyfile(
+        os.path.join(TEST_VAR_DIR, 'pgp', 'valid-pub.asc'),
+        os.path.join(obj.store.path, 'chris-pub.asc'),
+    )
+
+    # Notification goes through
+    obj.notify('test') is True