"""Email helpers for sending update-related emails.""" from __future__ import print_function from config import git_config from email.header import Header from email.mime.text import MIMEText from email.utils import getaddresses, parseaddr from errors import InvalidUpdate from git import get_module_name import os from string import strip from subprocess import Popen, PIPE, STDOUT from time import sleep from updates.sendmail import sendmail from utils import debug, get_user_name, get_user_full_name # The delay (in seconds) between each email being sent out. # The purpose of the delay is to help separate each email # in time, in order to increase our chances of having each # one of them delivered in order. EMAIL_DELAY_IN_SECONDS = 5 class EmailInfo(object): """Aggregates various pieces of info needed to send emails. ATTRIBUTES project_name: The name of the project (usually, the name of the directory holding the git repository). email_from: The email address to use in the From: field when sending the email notification. REMARKS This class assumes that the hooks.from-domain config parameter is set. Otherwise, an InvalidUpdate exception is raised when the object is initialized. """ def __init__(self, email_from): """The constructor. PARAMETERS email_from: If not None, a string that provides the email address of the sender. Eg: 'David Smith '. If None, this address is computed from the environment. """ self.project_name = get_module_name() from_domain = git_config('hooks.from-domain') if not from_domain: raise InvalidUpdate( 'Error: hooks.from-domain config variable not set.', 'Please contact your repository\'s administrator.') if email_from is None: self.email_from = '%s <%s@%s>' % (get_user_full_name(), get_user_name(), from_domain) else: self.email_from = email_from class EmailQueue(object): """An email queue (a singleton). ATTRIBUTES queue: A list of emails to be sent. """ def __new__(cls, *args, **kw): """The allocator.""" if not hasattr(cls, '_instance'): orig = super(EmailQueue, cls) cls._instance = orig.__new__(cls, *args, **kw) return cls._instance def __init__(self): """The constructor.""" # If the singleton has never been initialized, do it now. if not hasattr(self, 'queue'): self.queue = [] def enqueue(self, email): """Enqueue the given email. PARAMETERS email: An Email object. """ self.queue.append(email) def flush(self): """Send all enqueued emails... ... in the same order that they were enqueued. A delay of EMAIL_DELAY_IN_SECONDS is also introduced between emails. REMARKS If the GIT_HOOKS_TESTSUITE_MODE environment variable is set, then a trace of the delay is printed, instead of actually delaying the execution. Since emails are not actually sent when in GIT_HOOKS_TESTSUITE_MODE, there is no point in waiting for this delay. """ nb_emails_left = len(self.queue) for email in self.queue: email.send() nb_emails_left -= 1 if nb_emails_left > 0: # Need a small delay until we can send the next one. if 'GIT_HOOKS_TESTSUITE_MODE' in os.environ: # For the testsuite, print a debug trace in place # of delaying the execution. Use debug level 0 # to make sure it is always printed (to make sure # the testsuite always alerts us if there is any # change in the delay policy). debug('inter-email delay...', level=0) else: # pragma: no cover (do not want delays during testing) sleep(EMAIL_DELAY_IN_SECONDS) self.queue = [] class EmailCustomContents(object): """An object used to describe email customizations. This class is mostly a convenience class to store various attributes, and objects of this class are only really meaningful in conjunction with an Email object, as the attributes describe how certain parts of the email are to be adapted. ATTRIBUTES subject: If not None, the new email subject. body: If not None, the new email body (not including the "Diff:" section. appendix: If not None, some additional text to be added at the end of the email's body. diff: The contents of the "Diff:" section. If None, the "Diff:" section is omitted. """ def __init__(self, subject=None, body=None, appendix=None, diff=None): """Initialize an EmailCustomContents object. PARAMETERS subject: Same as the attribute. body: Same as the attribute. appendix: Same as the attribute. diff: Same as the attribute. """ self.subject = subject self.body = body self.appendix = appendix self.diff = diff class Email(object): """An email object. All emails to be sent by the git-hooks should be sent via this class to ensure consistency. ATTRIBUTES email_info: An EmailInfo object. email_to: A list of email addresses, in RFC 822 format, whom to send this email to. email_bcc: An iterable of email addresses, in RFC 822 format, whom to Bcc this email on. None indicates no Bcc needed. email_subject: The email's subject. email_body: The email's body, NOT including the diff. diff: A diff to be included at the end of the email being sent out. filer_cmd: If not None, sending this email also results in this command being called with the contents of the email_body parameter (and therefore, no diff). author: A string in "name " format, to be used as the X-Git-Author field in the email header. May be None, in which case we'll use email_info.email_from instead. ref_name: See AbstractUpdate.ref_name attribute. old_rev: See AbstractUpdate.old_rev attribute. new_rev: See AbstractUpdate.new_rev attribute. """ def __init__(self, email_info, email_to, email_bcc, email_subject, email_body, author, ref_name, old_rev, new_rev, diff=None, filer_cmd=None): """The constructor. PARAMETERS email_info: Same as the attribute. email_to: Same as the attribute. email_bcc: Same as the attribute. email_subject: Same as the attribute. email_body: Same as the attribute. author: Same as the attribute. ref_name: Same as the attribute. old_rev: Same as the attribute. new_rev: Same as the attribute. diff: A diff string, if applicable. Otherwise None. When not None, the diff is appended at the end of the email's body - truncated if necessary. """ self.email_info = email_info self.email_to = email_to self.email_bcc = email_bcc self.email_subject = email_subject self.email_body = email_body self.diff = diff self.filer_cmd = filer_cmd self.author = author self.ref_name = ref_name self.old_rev = old_rev self.new_rev = new_rev def enqueue(self): """Enqueue this email in the EmailQueue. REMARKS This is mostly a convenience method. """ EmailQueue().enqueue(self) def send(self): """Perform all send operations related to this email... These consists in: - send the notification email; - call self.filer_cmd if not None. REMARKS If the GIT_HOOKS_TESTSUITE_MODE environment variable is set, then a trace of the email is printed, instead of sending it. This is for testing purposes. """ e_msg_body = self.__email_body_with_diff e_msg_charset = guess_encoding(e_msg_body) e_msg = MIMEText(e_msg_body) if e_msg_charset is not None: e_msg.set_charset(e_msg_charset) # Create the email's header. e_msg['From'] = sanitized_email_address(self.email_info.email_from) e_msg['To'] = ', '.join(map(sanitized_email_address, self.email_to)) if self.email_bcc: e_msg['Bcc'] = ', '.join(map(sanitized_email_address, self.email_bcc)) e_msg['Subject'] = sanitized_email_header_field(self.email_subject) e_msg['X-Act-Checkin'] = self.email_info.project_name e_msg['X-Git-Author'] = sanitized_email_address( self.author or self.email_info.email_from) e_msg['X-Git-Refname'] = self.ref_name e_msg['X-Git-Oldrev'] = self.old_rev e_msg['X-Git-Newrev'] = self.new_rev # email_from = e_msg.get('From') email_recipients = [addr[1] for addr in getaddresses(e_msg.get_all('To', []) + e_msg.get_all('Cc', []) + e_msg.get_all('Bcc', []))] if 'GIT_HOOKS_TESTSUITE_MODE' in os.environ: if 'GIT_HOOKS_MINIMAL_EMAIL_DEBUG_TRACE' in os.environ: # This environment variable is set when the testcase # doesn't really care about the full contents of # the email. Just print a simple trace showing that # the email _is_ being sent, and that's it. # # Use debug level 0 to make sure that the trace is always # printed. debug("Sending email: {e_msg[Subject]}..." .format(e_msg=e_msg), level=0) else: # Use debug level 0 to make sure that the trace is always # printed. debug(e_msg.as_string(), level=0) else: # pragma: no cover (do not want real emails during testing) sendmail(self.email_info.email_from, email_recipients, e_msg.as_string(), 'localhost') if self.filer_cmd is not None: self.__call_filer_cmd() @property def __email_body_with_diff(self): """Return self.email_body with the diff at the end (if any). This attributes returns self.email_body augmentted with self.diff (if not None), possibly truncated to fit the hooks.max-email-diff-size limit, with a "diff marker" between email_body and diff. The diff marker is meant to be used by scripts processing the contents of those emails but not wanting to include the diff as part of their processing. """ email_body = self.email_body if self.diff is not None: # Append the "Diff:" marker to email_body, followed by # the diff. Truncate the patch if necessary. diff = self.diff max_diff_size = git_config('hooks.max-email-diff-size') if len(diff) > max_diff_size: diff = diff[:max_diff_size] diff += ('[...]\n\n[diff truncated at %d bytes]\n' % max_diff_size) email_body += '\nDiff:\n' email_body += diff return email_body def __call_filer_cmd(self): """Call self.filer_cmd to get self.email_body filed. The contents that gets filed is a slightly augmented version of self.email to provide a little context of what's being changed. Prints a message on stdout in case of error returned during the call. """ ref_name = self.ref_name if ref_name.startswith('refs/heads/'): # Replace the reference name by something a little more # intelligible for normal users. ref_name = 'The %s branch' % ref_name[11:] to_be_filed = ('%s has been updated by %s:' % (ref_name, self.email_info.email_from) + '\n\n' + self.email_body) p = Popen(self.filer_cmd, stdin=PIPE, stdout=PIPE, stderr=STDOUT) out, _ = p.communicate(to_be_filed) if p.returncode != 0: print(out) def guess_encoding(text): """Try to guess the given string's encoding. The guessing is done by simply trying to decode the string using the more popular encodings. PARAMETERS text: The string whose encoding we are trying to guess. RETURN VALUE The name of the encoding (e.g. 'ascii', or 'UTF-8') if that encoding is able to decode the given string. None otherwise. Note that one of the encodings that this function tries is the 'iso-8859-1' encoding which, as of this writing (using Python 2.7.10), appears to be an encoding accepting any sequence of bytes. So, as of now, the implementation of this function is such that it should never return None. However, users of this function should still be prepared for that as the above may not last forever, since some codes are not legal with that encoding. """ for potential_encoding in ('ascii', 'UTF-8', 'iso-8859-1'): # Note: It looks like iso-8859-1 accepts any sequence of bytes. # So, always place it last in the list above, so as to try all # the other encodings before defaulting to that one. We do try # it before using it, though, just in case it starts failing # for some reason. try: text.decode(potential_encoding) return potential_encoding except Exception: pass def sanitized_email_header_field(field_body): """Return an RFC2047-encoded version of field_body (if necessary) If field_body contains characters that are not printable ASCII, return an RFC2047-encoded version of this string. Otherwise, that string untouched. PARAMETERS field_body: A string, to be used as the value of a field in an email's header. """ encoding = guess_encoding(field_body) if encoding == 'ascii' and \ not any(c for c in field_body if ord(c) < 32 or ord(c) > 126): # The field body has only ASCII characters in the range 32-126. # So no encoding required. return field_body if encoding is None: # pragma: no cover (see explanation below) # Should never happen, since guess_encoding tries the iso-8859-1 # encoding which apparently accepts any byte stream. # Nevertheless, just in case, do the best we can in that situation, # and just send the header as is. return field_body return Header(field_body, encoding).encode() def sanitized_email_address(email_address): """Return an RFC2047-encoded version of the email_address (if necessary) This function splits the email_address into a (gecos, email_spec) tuple, and then RFC2047-encodes the gecos portion of the email when necessary (using sanitized_email_header_field). The sanitized version of the email address (gecos + email_spec) is then returned. PARAMETERS email_address: A string containing an email address. """ email_address = strip(email_address) gecos, email_spec = parseaddr(email_address) if not gecos: # There is no GECOS, so the email address is made of only # the email-spec portion, which should not require any encoding. # So return the email_address as is. return email_address if not email_spec: # pragma: no cover (see below) # Does not seem to be possible, because (1) Git seems to verify # the format of all email addresses before accepting them; and # (2) even when feeding an invalid email address such as # "invalid ' % (sanitized_email_header_field(gecos), email_spec)