Coverage for lib/utils/mail.py: 65%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3 

4# Hermes : Change Data Capture (CDC) tool from any source(s) to any target 

5# Copyright (C) 2023, 2024 INSA Strasbourg 

6# 

7# This file is part of Hermes. 

8# 

9# Hermes is free software: you can redistribute it and/or modify 

10# it under the terms of the GNU General Public License as published by 

11# the Free Software Foundation, either version 3 of the License, or 

12# (at your option) any later version. 

13# 

14# Hermes is distributed in the hope that it will be useful, 

15# but WITHOUT ANY WARRANTY; without even the implied warranty of 

16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

17# GNU General Public License for more details. 

18# 

19# You should have received a copy of the GNU General Public License 

20# along with Hermes. If not, see <https://www.gnu.org/licenses/>. 

21 

22from typing import Any 

23 

24from lib.config import HermesConfig 

25 

26from email.message import EmailMessage 

27from dataclasses import dataclass 

28import difflib 

29import gzip 

30import smtplib 

31 

32 

33@dataclass 

34class Attachment: 

35 filename: str 

36 mimetype: str 

37 content: bytes 

38 

39 def __len__(self) -> int: 

40 return len(self.content) 

41 

42 @property 

43 def maintype(self) -> str: 

44 return self.mimetype.split("/", 1)[0] 

45 

46 @property 

47 def subtype(self) -> str: 

48 return self.mimetype.split("/", 1)[1] 

49 

50 

51class Email: 

52 """Helper class to send mails""" 

53 

54 @staticmethod 

55 def send( 

56 config: HermesConfig, 

57 subject: str, 

58 content: str, 

59 attachments: list[Attachment] = [], 

60 ): 

61 """Send a mail with specified subject and content, using "server", 

62 "from" and "to" set in specified config. 

63 Can attach files from 'attachments' list""" 

64 try: 

65 server = config["hermes"]["mail"]["server"] 

66 mailfrom = config["hermes"]["mail"]["from"] 

67 mailto = config["hermes"]["mail"]["to"] 

68 

69 # Create a text/plain message 

70 msg = EmailMessage() 

71 msg.set_content(content) 

72 

73 msg["Subject"] = subject 

74 msg["From"] = mailfrom 

75 msg["To"] = mailto 

76 

77 for attachment in attachments: 

78 msg.add_attachment( 

79 attachment.content, 

80 maintype=attachment.maintype, 

81 subtype=attachment.subtype, 

82 filename=attachment.filename, 

83 ) 

84 

85 s = smtplib.SMTP(server) 

86 s.send_message(msg) 

87 s.quit() 

88 except Exception as e: 

89 __hermes__.logger.warning(f"Fail to send mail {subject=}: {str(e)}") 

90 

91 @staticmethod 

92 def sendDiff( 

93 config: HermesConfig, 

94 contentdesc: str, 

95 previous: str, 

96 current: str, 

97 ): 

98 """Send a mail with a diff between two strings. 

99 

100 'contentdesc': string (first letter should be lowercase) that will be used 

101 in mail subject, and as prefix of mail content 

102 'previous': previous data used to compute diff 

103 'current': current data used to compute diff 

104 """ 

105 nl = "\n" 

106 

107 d = difflib.unified_diff( 

108 previous.splitlines(keepends=True), 

109 current.splitlines(keepends=True), 

110 "previous.txt", 

111 "current.txt", 

112 n=0, 

113 ) 

114 diff = "".join(d) 

115 

116 # Convert string to bytes 

117 previous = "".join(previous).encode() 

118 current = "".join(current).encode() 

119 difffile = diff.encode() 

120 

121 if config["hermes"]["mail"]["compress_attachments"]: 

122 mimetype = "application/gzip" 

123 ext = ".txt.gz" 

124 compress = gzip.compress 

125 else: 

126 mimetype = "text/plain" 

127 ext = ".txt" 

128 compress = Email._dontCompress # Keep data as is 

129 

130 tmpattachments = [ 

131 Attachment(f"previous{ext}", mimetype, compress(previous)), 

132 Attachment(f"current{ext}", mimetype, compress(current)), 

133 Attachment(f"diff{ext}", mimetype, compress(difffile)), 

134 ] 

135 

136 # Ensure attachments doesn't exceed attachment_maxsize 

137 attachments = [] 

138 toobig = [] 

139 errmsg = "" 

140 for a in tmpattachments: 

141 if len(a) <= config["hermes"]["mail"]["attachment_maxsize"]: 

142 attachments.append(a) 

143 else: 

144 toobig.append(a.filename) 

145 

146 if toobig: 

147 errmsg = ( 

148 f"Some files were too big to be attached to mail: {toobig}.{nl}{nl}" 

149 ) 

150 

151 if len(diff.encode()) < config["hermes"]["mail"]["mailtext_maxsize"]: 

152 content = f"{errmsg}{contentdesc.capitalize()}. Diff is:{nl}{nl}{diff}" 

153 else: 

154 content = ( 

155 f"{errmsg}{contentdesc.capitalize()}. " 

156 "Diff is too big to be displayed in mail content, " 

157 "please see attachments or log files." 

158 ) 

159 

160 Email.send( 

161 config=config, 

162 subject=f"[{config['appname']}] {contentdesc}", 

163 content=content, 

164 attachments=attachments, 

165 ) 

166 

167 @staticmethod 

168 def _dontCompress(data: Any) -> Any: 

169 return data