import base64 import datetime import email.utils import html import json import logging import os import pathlib import re import subprocess import zipfile import bs4 import chardet import compressed_rtf import RTFDE from email.parser import Parser as EmailParser from typing import Callable, Dict, Optional, Tuple, Union from . import constants from .enums import DeencapType, RecipientType from .exceptions import ( DataNotFoundError, DeencapMalformedData, DeencapNotEncapsulated, IncompatibleOptionsError, WKError ) from .msg import MSGFile from .structures.report_tag import ReportTag from .recipient import Recipient from .utils import ( addNumToDir, addNumToZipDir, createZipOpen, findWk, htmlSanitize, inputToBytes, inputToString, isEncapsulatedRtf, prepareFilename, rtfSanitizeHtml, rtfSanitizePlain, validateHtml ) from imapclient.imapclient import decode_utf7 logger = logging.getLogger(__name__) logger.addHandler(logging.NullHandler()) class MessageBase(MSGFile): """ Base class for Message like msg files. """ def __init__(self, path, **kwargs): """ :param path: path to the msg file in the system or is the raw msg file. :param prefix: used for extracting embeded msg files inside the main one. Do not set manually unless you know what you are doing. :param parentMsg: Used for syncronizing named properties instances. Do not set this unless you know what you are doing. :param attachmentClass: Optional, the class the Message object will use for attachments. You probably should not change this value unless you know what you are doing. :param filename: Optional, the filename to be used by default when saving. :param delayAttachments: Optional, delays the initialization of attachments until the user attempts to retrieve them. Allows MSG files with bad attachments to be initialized so the other data can be retrieved. :param overrideEncoding: Optional, an encoding to use instead of the one specified by the msg file. Do not report encoding errors caused by this. :param attachmentErrorBehavior: Optional, the behavior to use in the event of an error when parsing the attachments. :param recipientSeparator: Optional, separator string to use between recipients. :param ignoreRtfDeErrors: Optional, specifies that any errors that occur from the usage of RTFDE should be ignored (default: False). :param deencapsulationFunc: Optional, if specified must be a callable that will override the way that HTML/text is deencapsulated from the RTF body. This function must take exactly 2 arguments, the first being the RTF body from the message and the second being an instance of the enum DeencapType that will tell the function what type of body is desired. The function should return a string for plain text and bytes for HTML. If any problems occur, the function *must* either return None or raise one of the appropriate functions from extract_msg.exceptions. All other exceptions must be handled internally or they will not be caught. The original deencapsulation method will not run if this is set. """ super().__init__(path, **kwargs) recipientSeparator = ';' self.__recipientSeparator = kwargs.get('recipientSeparator', ';') self.__ignoreRtfDeErrors = kwargs.get('ignoreRtfDeErrors', False) self.__deencap = kwargs.get('deencapsulationFunc') # Initialize properties in the order that is least likely to cause bugs. # TODO have each function check for initialization of needed data so # these lines will be unnecessary. self.props self.header self.recipients self.to self.cc self.sender self.date # This variable keeps track of what the new line character should be. self.__crlf = '\n' try: self.body except Exception as e: # Prevent an error in the body from preventing opening. logger.exception('Critical error accessing the body. File opened but accessing the body will throw an exception.') self.named self.namedProperties def _genRecipient(self, recipientType, recipientInt : RecipientType) -> Optional[str]: """ Returns the specified recipient field. """ private = '_' + recipientType recipientInt = RecipientType(recipientInt) try: return getattr(self, private) except AttributeError: value = None # Check header first. if self.headerInit(): value = self.header[recipientType] if value: value = value.replace(',', self.__recipientSeparator) # If the header had a blank field or didn't have the field, generate # it manually. if not value: # Check if the header has initialized. if self.headerInit(): logger.info(f'Header found, but "{recipientType}" is not included. Will be generated from other streams.') # Get a list of the recipients of the specified type. foundRecipients = tuple(recipient.formatted for recipient in self.recipients if recipient.type == recipientInt) # If we found recipients, join them with the recipient separator # and a space. if len(foundRecipients) > 0: value = (self.__recipientSeparator + ' ').join(foundRecipients) # Code to fix the formatting so it's all a single line. This allows # the user to format it themself if they want. This should probably # be redone to use re or something, but I can do that later. This # shouldn't be a huge problem for now. if value: value = value.replace(' \r\n\t', ' ').replace('\r\n\t ', ' ').replace('\r\n\t', ' ') value = value.replace('\r\n', ' ').replace('\r', ' ').replace('\n', ' ') while value.find(' ') != -1: value = value.replace(' ', ' ') # Set the field in the class. setattr(self, private, value) return value def deencapsulateBody(self, rtfBody : bytes, bodyType : DeencapType) -> Optional[Union[bytes, str]]: """ A function to deencapsulate the specified body from the rtfBody. Returns a string for plain text and bytes for HTML. If specified, uses the deencapsulation override function. Returns None if nothing could be deencapsulated. If you want to change the deencapsulation behaviour in a base class, simply override this function. """ if rtfBody: bodyType = DeencapType(bodyType) if bodyType == DeencapType.PLAIN: if self.__deencap: try: return self.__deencap(rtfBody, DeencapType.PLAIN) except DeencapMalformedData: logger.exception('Custom deencapsulation function reported encapsulated data was malformed.') except DeencapNotEncapsulated: logger.exception('Custom deencapsulation function reported data is not encapsulated.') else: if self.deencapsulatedRtf and self.deencapsulatedRtf.content_type == 'text': return self.deencapsulatedRtf.text else: if self.__deencap: try: return self.__deencap(rtfBody, DeencapType.HTML) except DeencapMalformedData: logger.exception('Custom deencapsulation function reported encapsulated data was malformed.') except DeencapNotEncapsulated: logger.exception('Custom deencapsulation function reported data is not encapsulated.') else: if self.deencapsulatedRtf and self.deencapsulatedRtf.content_type == 'html': return self.deencapsulatedRtf.html.encode('utf-8') if bodyType == DeencapType.PLAIN: logger.info('Could not deencapsulate plain text from RTF body.') else: logger.info('Could not deencapsulate HTML from RTF body.') else: logger.info('No RTF body to deencapsulate from.') return None def dump(self) -> None: """ Prints out a summary of the message. """ print('Message') print('Subject:', self.subject) print('Date:', self.date) print('Body:') print(self.body) def getInjectableHeader(self, prefix : str, joinStr : str, suffix : str, formatter : Callable[[str, str], str]) -> str: """ Using the specified prefix, suffix, formatter, and join string, generates the injectable header. Prefix is placed at the beginning, followed by a series of format strings joined together with the join string, with the suffix placed afterwards. Effectively makes this structure: {prefix}{formatter()}{joinStr}{formatter()}{joinStr}...{formatter()}{suffix} Formatter be a function that takes first a name variable then a value variable and formats the line. If self.headerFormatProperties is None, immediately returns an empty string. """ formattedProps = [] allProps = self.headerFormatProperties if allProps is None: return '' for entry in allProps: isGroup = False entryUsed = False # This is how we handle the groups. if isinstance(allProps[entry], dict): props = allProps[entry] isGroup = True else: props = {entry: allProps[entry]} for name in props: if props[name]: if isinstance(props[name], tuple): if props[name][1]: value = props[name][0] or '' elif props[name][0] is not None: value = props[name][0] else: continue else: value = props[name] entryUsed = True formattedProps.append(formatter(name, value)) # Now if we are working with a group, add an empty entry to get a # second join string between this section and the last, but *only* # if any of the entries were used. if isGroup and entryUsed: formattedProps.append('') # If the last entry is empty, remove it. We don't want extra spacing at # the end. if formattedProps[-1] == '': formattedProps.pop() return prefix + joinStr.join(formattedProps) + suffix def getJson(self) -> str: """ Returns the JSON representation of the Message. """ return json.dumps({ 'from': inputToString(self.sender, self.stringEncoding), 'to': inputToString(self.to, self.stringEncoding), 'cc': inputToString(self.cc, self.stringEncoding), 'bcc': inputToString(self.bcc, self.stringEncoding), 'subject': inputToString(self.subject, self.stringEncoding), 'date': inputToString(self.date, self.stringEncoding), 'body': decode_utf7(self.body), }) def getSaveBody(self, **kwargs) -> bytes: """ Returns the plain text body that will be used in saving based on the arguments. :param kwargs: Used to allow kwargs expansion in the save function. Arguments absorbed by this are simply ignored. """ # Get the type of line endings. crlf = inputToString(self.crlf, 'utf-8') prefix = '' suffix = crlf + '-----------------' + crlf + crlf joinStr = crlf formatter = (lambda name, value : f'{name}: {value}') header = self.getInjectableHeader(prefix, joinStr, suffix, formatter).encode('utf-8') return header + inputToBytes(self.body, 'utf-8') def getSaveHtmlBody(self, preparedHtml : bool = False, charset : str = 'utf-8', **kwargs) -> bytes: """ Returns the HTML body that will be used in saving based on the arguments. :param preparedHtml: Whether or not the HTML should be prepared for standalone use (add tags, inject images, etc.). :param charset: If the html is being prepared, the charset to use for the Content-Type meta tag to insert. This exists to ensure that something parsing the html can properly determine the encoding (as not having this tag can cause errors in some programs). Set this to `None` or an empty string to not insert the tag (Default: 'utf-8'). :param kwargs: Used to allow kwargs expansion in the save function. Arguments absorbed by this are simply ignored. :raises BadHtmlError: if :param preparedHtml: is False and the HTML fails to validate. """ if self.htmlBody: # Inject the header into the data. data = self.injectHtmlHeader(prepared = preparedHtml) # If we are preparing the HTML, then we should if preparedHtml and charset: bs = bs4.BeautifulSoup(data, features = 'html.parser') if not bs.find('meta', {'http-equiv': 'Content-Type'}): # Setup the attributes for the tag. tagAttrs = { 'http-equiv': 'Content-Type', 'content': f'text/html; charset={charset}', } # Create the tag. tag = bs4.Tag(parser = bs, name = 'meta', attrs = tagAttrs, can_be_empty_element = True) # Add the tag to the head section. if bs.find('head'): bs.find('head').insert(0, tag) else: # If we are here, the head doesn't exist, so let's add # it. if bs.find('html'): # This should always be true, but I want to be safe. head = bs4.Tag(parser = bs, name = 'head') head.insert(0, tag) bs.find('html').insert(0, head) data = bs.prettify('utf-8') return data else: return self.htmlBody def getSavePdfBody(self, **kwargs) -> bytes: """ Returns the PDF body that will be used in saving based on the arguments. :param wkPath: Used to manually specify the path of the wkhtmltopdf executable. If not specified, the function will try to find it. Useful if wkhtmltopdf is not on the path. If :param pdf: is False, this argument is ignored. :param wkOptions: Used to specify additional options to wkhtmltopdf. this must be a list or list-like object composed of strings and bytes. :param kwargs: Used to allow kwargs expansion in the save function. Arguments absorbed by this are simply ignored. :raises ExecutableNotFound: The wkhtmltopdf executable could not be found. :raises WKError: Something went wrong in creating the PDF body. """ # Immediately try to find the executable. wkPath = findWk(kwargs.get('wkPath')) # First thing is first, we need to parse our wkOptions if they exist. wkOptions = kwargs.get('wkOptions') if wkOptions: try: # Try to convert to a list, whatever it is, and fail if it is # not possible. parsedWkOptions = [*wkOptions] except TypeError: raise TypeError(f':param wkOptions: must be an iterable, not {type(wkOptions)}.') else: parsedWkOptions = [] # Confirm that all of our options we now have are either strings or # bytes. if not all(isinstance(option, (str, bytes)) for option in parsedWkOptions): raise TypeError(':param wkOptions: must be an iterable of strings and bytes.') processArgs = [wkPath, *parsedWkOptions, '-', '-'] # Log the arguments. logger.info(f'Converting to PDF with the following arguments: {processArgs}') # Get the html body *before* calling Popen. htmlBody = self.getSaveHtmlBody(**kwargs) # We call the program to convert the html, but give tell it the data # will go in and come out through stdin and stdout, respectively. This # way we don't have to write temporary files to the disk. We also ask # that it be quiet about it. process = subprocess.run(processArgs, input = htmlBody, stdout = subprocess.PIPE, stderr = subprocess.PIPE) # Give the program the data and wait for the program to finish. #output = process.communicate(htmlBody) # If it errored, throw it as an exception. if process.returncode != 0: raise WKError(process.stderr.decode('utf-8')) return process.stdout def getSaveRtfBody(self, **kwargs) -> bytes: """ Returns the RTF body that will be used in saving based on the arguments. :param kwargs: Used to allow kwargs expansion in the save function. Arguments absorbed by this are simply ignored. """ # Inject the header into the data. return self.injectRtfHeader() def headerInit(self) -> bool: """ Checks whether the header has been initialized. """ try: self._header return True except AttributeError: return False def injectHtmlHeader(self, prepared : bool = False) -> bytes: """ Returns the HTML body from the MSG file (will check that it has one) with the HTML header injected into it. :param prepared: Determines whether to be using the standard HTML (False) or the prepared HTML (True) body (Default: False). :raises AttributeError: if the correct HTML body cannot be acquired. :raises BadHtmlError: if :param preparedHtml: is False and the HTML fails to validate. """ if not self.htmlBody: raise AttributeError('Cannot inject the HTML header without an HTML body attribute.') body = None # We don't do this all at once because the prepared body is not cached. if prepared: body = self.htmlBodyPrepared # If the body is not valid or not found, raise an AttributeError. if not body: raise AttributeError('Cannot find a prepared HTML body to inject into.') else: body = self.htmlBody # Validate the HTML. if not validateHtml(body): # If we are not preparing the HTML body, then raise an # exception. if not prepared: raise BadHtmlError('HTML body failed to pass validation.') # If we are here, then we need to do what we can to fix the HTML body. # Unfortunately this gets complicated because of the various ways the # body could be wrong. If only the tag is missing, then we just # need to insert it at the end and be done. If both the and # tag are missing, we determine where to put the body tag (around # everything if there is no tag, otherwise at the end) and then # wrap it all in the tag. parser = bs4.BeautifulSoup(body, features = 'html.parser') if not parser.find('html') and not parser.find('body'): if parser.find('head') or parser.find('footer'): # Create the parser we will be using for the corrections. correctedHtml = bs4.BeautifulSoup(b'', features = 'html.parser') htmlTag = correctedHtml.find('html') # Iterate over each of the direct descendents of the parser and # add each to a new tag if they are not the head or footer. bodyTag = parser.new_tag('body') # What we are going to be doing will be causing some of the tags # to be moved out of the parser, and so the iterator will end up # pointing to the wrong place after that. To compensate we first # create a tuple and iterate over that. for tag in tuple(parser.children): if tag.name.lower() in ('head', 'footer'): correctedHtml.append(tag) else: bodyTag.append(tag) # All the tags should now be properly in the body, so let's # insert it. if correctedHtml.find('head'): correctedHtml.find('head').insert_after(bodyTag) elif correctedHtml.find('footer'): correctedHtml.find('footer').insert_before(bodyTag) else: # Neither a head or a body are present, so just append it to # the main tag. htmlTag.append(bodyTag) else: # If there is no , ,