spacefreak/pyquarantine-milter
1
0
Fork 0
Code Issues Pull Requests Releases Activity

Improved CLI, quarantine- and error handling

Browse Source
This commit is contained in:
Thomas Oettli
2019-03-13 09:27:58 +01:00
parent da6f08ee25
commit 88cea0e127
10 changed files with 508 additions and 301 deletions
Download Patch File Download Diff File Expand all files Collapse all files

117
pyquarantine/notifications.py
View File

@@ -20,25 +20,25 @@ import mailer
import re
from bs4 import BeautifulSoup
from cgi import escape
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.image import MIMEImage
class BaseNotification(object):
"Notification base class"
def __init__(self, quarantine_name, config, configtest=False):
self.quarantine_name = quarantine_name
self.config = config[quarantine_name]
self.global_config = config["global"]
def __init__(self, global_config, config, configtest=False):
self.quarantine_name = config["name"]
self.global_config = global_config
self.config = config
self.logger = logging.getLogger(__name__)
def notify(self, queueid, quarantine_id, subject, mailfrom, recipients, fp):
fp.seek(0)
pass
class EMailNotification(BaseNotification):
"Notification class to send notifications via mail."
_html_text = "text/html"
@@ -84,10 +84,13 @@ class EMailNotification(BaseNotification):
"border",
"cellpadding",
"cellspacing",
"color",
"colspan",
"dir",
"face",
"headers",
"height",
"id",
"name",
"rowspan",
"size",
@@ -95,26 +98,33 @@ class EMailNotification(BaseNotification):
"style",
"title",
"type",
"valign",
"value",
"width"
]
def __init__(self, quarantine_name, config, configtest=False):
super(EMailNotification, self).__init__(quarantine_name, config, configtest)
def __init__(self, global_config, config, configtest=False):
super(EMailNotification, self).__init__(global_config, config, configtest)
# check if mandatory options are present in config
for option in ["smtp_host", "smtp_port", "notification_email_from", "notification_email_subject", "notification_email_template", "notification_email_replacement_img"]:
if option not in self.config.keys() and option in self.global_config.keys():
self.config[option] = self.global_config[option]
if option not in self.config.keys():
raise RuntimeError("mandatory option '{}' not present in config section '{}' or 'global'".format(option, self.quarantine_name))
self.smtp_host = self.config["smtp_host"]
self.smtp_port = self.config["smtp_port"]
self.mailfrom = self.config["notification_email_from"]
self.subject = self.config["notification_email_subject"]
# read email notification template
try:
self.template = open(self.config["notification_email_template"], "rb").read()
except IOError as e:
raise RuntimeError("error reading template: {}".format(e))
# read email replacement image
try:
self.replacement_img = MIMEImage(open(self.config["notification_email_replacement_img"], "rb").read())
except IOError as e:
@@ -122,95 +132,130 @@ class EMailNotification(BaseNotification):
else:
self.replacement_img.add_header("Content-ID", "<removed_for_security_reasons>")
def get_text(self, part):
def get_text(self, queueid, part):
"Get the mail text in html form from email part."
mimetype = part.get_content_type()
text = part.get_payload(decode=True)
if mimetype == EMailNotification._plain_text:
text = re.sub(r"^(.*)$", r"\1<br/>\n", text, flags=re.MULTILINE)
soup = BeautifulSoup(text, "lxml", from_encoding=part.get_content_charset())
return soup
def get_text_multipart(self, msg, preferred=_html_text):
self.logger.debug("{}: extracting content of email text part".format(queueid))
text = part.get_payload(decode=True)
if mimetype == EMailNotification._plain_text:
self.logger.debug("{}: content mimetype is {}, converting to {}".format(queueid, mimetype, self._html_text))
text = re.sub(r"^(.*)$", r"\1<br/>\n", text, flags=re.MULTILINE)
else:
self.logger.debug("{}: content mimetype is {}".format(queueid, mimetype))
return BeautifulSoup(text, "lxml", from_encoding=part.get_content_charset())
def get_text_multipart(self, queueid, msg, preferred=_html_text):
"Get the mail text of a multipart email in html form."
soup = None
for part in msg.get_payload():
mimetype = part.get_content_type()
if mimetype in [EMailNotification._plain_text, EMailNotification._html_text]:
soup = self.get_text(part)
soup = self.get_text(queueid, part)
elif mimetype.startswith("multipart"):
soup = self.get_text_multipart(part, preferred)
soup = self.get_text_multipart(queueid, part, preferred)
if soup != None and mimetype == preferred:
break
return soup
def sanitize(self, soup):
def sanitize(self, queueid, soup):
"Sanitize mail html text."
self.logger.debug("{}: sanitizing email text".format(queueid))
# completly remove bad elements
for element in soup(EMailNotification._bad_tags):
self.logger.debug("{}: removing dangerous tag '{}' and its content".format(queueid, element.name))
element.extract()
# remove not whitelisted elements, but keep their content
for element in soup.find_all(True):
if element.name not in EMailNotification._good_tags:
self.logger.debug("{}: removing tag '{}', keep its content".format(queueid, element.name))
element.replaceWithChildren()
# remove not whitelisted attributes
for element in soup.find_all(True):
for attribute in element.attrs.keys():
if attribute not in EMailNotification.good_attributes:
del(element.attrs[attribute])
# set href attribute for all a-tags to #
for element in soup("a"):
element["href"] = "#"
if element.name == "a" and attribute == "href":
self.logger.debug("{}: setting attribute href to '#' on tag '{}'".format(queueid, element.name))
element["href"] = "#"
else:
self.logger.debug("{}: removing attribute '{}' from tag '{}'".format(queueid, attribute, element.name))
del(element.attrs[attribute])
return soup
def get_html_text_part(self, msg):
def get_html_text_part(self, queueid, msg):
"Get the mail text of an email in html form."
soup = None
mimetype = msg.get_content_type()
self.logger.debug("{}: trying to find text part of email".format(queueid))
if mimetype in [EMailNotification._plain_text, EMailNotification._html_text]:
soup = self.get_text(msg)
soup = self.get_text(queueid, msg)
elif mimetype.startswith("multipart"):
soup = self.get_text_multipart(msg)
soup = self.get_text_multipart(queueid, msg)
if soup == None:
self.logger.error("{}: unable to extract text part of email".format(queueid))
text = "ERROR: unable to extract text from email body"
soup = BeautifulSoup(text, "lxml", "UTF-8")
return soup
def notify(self, queueid, quarantine_id, subject, mailfrom, recipients, fp):
"Notify recipients via email."
super(EMailNotification, self).notify(queueid, quarantine_id, subject, mailfrom, recipients, fp)
self.logger.debug("{}: generating notification email".format(queueid))
# extract html text from email
self.logger.debug("{}: extraction email text from original email".format(queueid))
soup = self.get_html_text_part(email.message_from_file(fp))
soup = self.get_html_text_part(queueid, email.message_from_file(fp))
# replace picture sources
picture_replaced = False
for element in soup("img"):
if "src" in element:
if "src" in element.attrs.keys():
self.logger.debug("{}: replacing image: {}".format(queueid, element["src"]))
element["src"] = "cid:removed_for_security_reasons"
picture_replaced = True
# sanitizing email text of original email
sanitized_text = self.sanitize(queueid, soup)
# escape possible html entities in subject
subject = escape(subject)
# sending email notifications
for recipient in recipients:
self.logger.debug("{}: sending notification to <{}>".format(queueid, recipient))
self.logger.debug("{}: generating notification email for '{}'".format(queueid, recipient))
self.logger.debug("{}: parsing email template".format(queueid))
htmltext = self.template.format( \
EMAIL_HTML_TEXT=self.sanitize(soup), \
EMAIL_FROM=mailfrom, \
EMAIL_TO=recipient, \
EMAIL_HTML_TEXT=sanitized_text, \
EMAIL_FROM=escape(mailfrom), \
EMAIL_TO=escape(recipient), \
EMAIL_SUBJECT=subject, \
EMAIL_QUARANTINE_ID=quarantine_id
)
msg = MIMEMultipart('alternative')
msg["Subject"] = self.subject
msg["From"] = "<{}>".format(self.mailfrom)
msg["To"] = "<{}>".format(recipient)
msg["Date"] = email.utils.formatdate()
msg.attach(MIMEText(htmltext, "html", 'UTF-8'))
if picture_replaced:
msg.attach(self.replacement_img)
mailer.sendmail(self.smtp_host, self.smtp_port, queueid, self.mailfrom, recipient, msg.as_string())
if picture_replaced:
self.logger.debug("{}: attaching notification_replacement_img".format(queueid))
msg.attach(self.replacement_img)
self.logger.debug("{}: sending notification email to: {}".format(queueid, recipient))
mailer.sendmail(self.smtp_host, self.smtp_port, queueid, self.mailfrom, recipient, msg.as_string(), "notification email")
# list of notification types and their related notification classes
notification_types = {"email": EMailNotification}
TYPES = {"email": EMailNotification}