Coverage for paperless_mail/preprocessor.py: 38%

1import abc 

2from email import message_from_bytes 

3from email import policy 

4from email.message import Message 

5from pathlib import Path 

6 

7from django.conf import settings 

8from gnupg import GPG 

9from imap_tools import MailMessage 

10 

11from documents.loggers import LoggingMixin 

12 

13 

14class MailMessagePreprocessor(abc.ABC): 

15 """ 

16 Defines the interface for preprocessors that alter messages before they are handled in MailAccountHandler 

17 """ 

18 

19 NAME: str = "MailMessagePreprocessor" 

20 

21 @staticmethod 

22 @abc.abstractmethod 

23 def able_to_run() -> bool: 

24 """ 

25 Return True if the conditions are met for the preprocessor to run, False otherwise 

26 

27 If False, run(message) will not be called 

28 """ 

29 

30 @abc.abstractmethod 

31 def run(self, message: MailMessage) -> MailMessage: 

32 """ 

33 Performs the actual preprocessing task 

34 """ 

35 

36 

37class MailMessageDecryptor(MailMessagePreprocessor, LoggingMixin): 

38 logging_name = "paperless_mail_message_decryptor" 

39 

40 NAME = "MailMessageDecryptor" 

41 

42 def __init__(self) -> None: 

43 super().__init__() 

44 self.renew_logging_group() 

45 self._gpg = GPG(gnupghome=settings.EMAIL_GNUPG_HOME) 

46 

47 @staticmethod 

48 def able_to_run() -> bool: 

49 if not settings.EMAIL_ENABLE_GPG_DECRYPTOR: 

50 return False 

51 if settings.EMAIL_GNUPG_HOME is None: 

52 return True 

53 return Path(settings.EMAIL_GNUPG_HOME).is_dir() 

54 

55 def run(self, message: MailMessage) -> MailMessage: 

56 if not hasattr(message, "obj"): 

57 self.log.debug("Message does not have 'obj' attribute") 

58 return message 

59 if message.obj.get_content_type() != "multipart/encrypted": 

60 self.log.debug("Message not encrypted. Keep unchanged") 

61 return message 

62 

63 self.log.debug("Message is encrypted.") 

64 email_message = self._to_email_message(message) 

65 decrypted_raw_message = self._gpg.decrypt(email_message.as_string()) 

66 

67 if not decrypted_raw_message.ok: 

68 self.log.debug( 

69 f"Message decryption failed with status message " 

70 f"{decrypted_raw_message.status}", 

71 ) 

72 raise Exception( 

73 f"Decryption failed: {decrypted_raw_message.status}, {decrypted_raw_message.stderr}", 

74 ) 

75 self.log.debug("Message decrypted successfully.") 

76 

77 decrypted_message = self._build_decrypted_message( 

78 decrypted_raw_message, 

79 email_message, 

80 ) 

81 

82 return MailMessage( 

83 [(f"UID {message.uid}".encode(), decrypted_message.as_bytes())], 

84 ) 

85 

86 @staticmethod 

87 def _to_email_message(message: MailMessage) -> Message: 

88 email_message = message_from_bytes( 

89 message.obj.as_bytes(), 

90 policy=policy.default, 

91 ) 

92 return email_message 

93 

94 @staticmethod 

95 def _build_decrypted_message(decrypted_raw_message, email_message): 

96 decrypted_message = message_from_bytes( 

97 decrypted_raw_message.data, 

98 policy=policy.default, 

99 ) 

100 for header, value in email_message.items(): 

101 if not decrypted_message.get(header): 

102 decrypted_message.add_header(header, value) 

103 return decrypted_message