#!/usr/bin/env python # # Copyright (C) 2009 W. Trevor King # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License along # with this program; if not, write to the Free Software Foundation, Inc., # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. """Provide and email interface to the distributed bugtracker Bugs Everywhere. Recieves incoming email via procmail and allows users to select actions with their subject lines. Subject lines follow the format [be-bug] command (options) (args) With the body of the email being used as the final argument for the commands "new" and "comment", and ignored otherwise. The options and arguments are split on whitespace, so don't use whitespace inside a single argument. Eventually we'll commit after every message. """ import cStringIO as StringIO import email import email.utils import libbe.cmdutil, libbe.utility import os import os.path import send_pgp_mime import sys import time SUBJECT_COMMENT = "[be-bug]" HANDLER_ADDRESS = "BE Bugs " LOGFILE = os.path.join(os.path.dirname(__file__), "be-handle-mail.log") BE_DIR = os.path.expanduser("~/src/fun/be/be.email") ALLOWED_COMMANDS = ["new", "comment", "list", "show", "help"] class InvalidEmail (ValueError): def __init__(self, msg, info, message): ValueError.__init__(self, message) self.msg = msg self.info = info def response(self): ret = 1 out_text = None return (ret, out_text, self.stderr_msg(), self.info) def stderr_msg(self): err_text = [u"Invalid email (particular type unknown):\n", unicode(self), u"", send_pgp_mime.flatten(self.msg)] return u"\n".join(err_text) class InvalidSubject (InvalidEmail): def stderr_msg(self): err_text = u"\n".join([u"InvalidSubject:\n", unicode(e), u"", u"full subject was:", e.msg["subject"]]) return err_text class InvalidCommand (InvalidEmail): def __init__(self, msg, info, command): message = "Invalid command '%s'" % command InvalidEmail.__init__(self, msg, info, message) self.command = command def stderr_msg(self): err_text = u"\n".join([u"InvalidCommand:\n", unicode(self), u"", u"full subject was:", self.msg["subject"]]) return err_text def get_body_type(msg): for part in msg.walk(): if part.is_multipart(): continue return (part.get_payload(decode=1), part.get_content_type()) def run_message(msg_text): """ Attempt to execute the email given in the email string msg_text. Raises assorted subclasses of InvalidEmail in the case of invalid messages, otherwise return the exit code, stdout, and stderr produced by the command, as well as a dictionary of information gleaned from the email. """ encoding = libbe.encoding.get_encoding() libbe.encoding.set_IO_stream_encodings(encoding) p=email.Parser.Parser() msg=p.parsestr(msg_text) info = {} info["encoding"] = encoding author = send_pgp_mime.source_email(msg, return_realname=True) info["author_name"] = author[0] info["author_email"] = author[1] info["author_addr"] = email.utils.formataddr( (info["author_name"], info["author_email"])) info["message-id"] = msg["message-id"] if LOGFILE != None: f = file(LOGFILE, "w+") f.write("handling %s\n" % (info["author_addr"])) f.write("\n%s\n\n" % msg_text) f.close() if "subject" not in msg: raise InvalidSubject(msg, info, "Email must contain a subject") args = msg["subject"].split() if len(args) < 1 or args[0] != SUBJECT_COMMENT: raise InvalidSubject( msg, info, "Subject must start with '%s '" % SUBJECT_COMMENT) elif len(args) < 2: raise InvalidCommand(msg, info, "") # don't accept blank commands command = args[1] info["command"] = command if command not in ALLOWED_COMMANDS: raise InvalidCommand(msg, command) if len(args) > 2: command_args = args[2:] else: command_args = [] if command in ["new", "comment"]: body,type = get_body_type(msg) if command == "new": if "--reporter" not in args and "-r" not in args: command_args = ["--reporter", author_addr] + command_args body = body.strip().split("\n", 1)[0] # only take first line elif command == "comment": if "--author" not in args and "-a" not in args: command_args = ["--author", author_addr] + command_args if "--content-type" not in args and "-c" not in args: command_args = ["--content-type", type] + command_args if "--alt-id" not in args: command_args = ["--alt-id", msg["message-id"]] + command_args command_args.append(body) info["command-args"] = command_args # catch stdout and stderr new_stdout = StringIO.StringIO() new_stderr = StringIO.StringIO() orig_stdout = sys.stdout orig_stderr = sys.stderr sys.stdout = new_stdout sys.stderr = new_stderr # run the command err = None os.chdir(BE_DIR) try: ret = libbe.cmdutil.execute(command, command_args, manipulate_encodings=False) except libbe.cmdutil.GetHelp: print libbe.cmdutil.help(command) except libbe.cmdutil.GetCompletions: err = InvalidCommand(msg, info, "invalid option '--complete'") except libbe.cmdutil.UsageError, e: err = InvalidCommand(msg, info, e) except libbe.cmdutil.UserError, e: err = InvalidCommand(msg, info, e) # restore stdout and stderr sys.stdout.flush() sys.stderr.flush() sys.stdout = orig_stdout sys.stderr = orig_stderr out_text = new_stdout.getvalue() err_text = new_stderr.getvalue() if err != None: raise err return (ret, out_text, err_text, info) def compose_response(ret, out_text, err_text, info): assert "encoding" in info if "author_addr" not in info: return None if "command" not in info: info["command"] = u"-BLANK-" if "command_args" not in info: info["command_args"] = [] response_header = [u"From: %s" % HANDLER_ADDRESS, u"To: %s" % info["author_addr"], u"Date: %s" % libbe.utility.time_to_str(time.time()), u"Content-Type: text/plain; charset=%s"%info["encoding"], u"Content-Transfer-Encoding: 8bit", u"Subject: %s Re: %s"%(SUBJECT_COMMENT,info["command"]), ] if "message-id" in info: response_header.append(u"In-reply-to: %s" % info["message-id"]) response_body = [u"Results of running: (exit code %d)" % ret, u" %s %s" % (info["command"], u" ".join(info["command_args"]))] if out_text != None and len(out_text) > 0: response_body.extend([u"", u"stdout:", u"", out_text]) if err_text != None and len(err_text) > 0: response_body.extend([u"", u"stderr:", u"", err_text]) response_body.append(u"") # trailing endline response_email = send_pgp_mime.Mail(u"\n".join(response_header), u"\n".join(response_body)) if LOGFILE != None: f = file(LOGFILE, "a+") f.write("responding to %s: %s\n" % (info["author_addr"], info["command"])) f.write("\n%s\n\n" % send_pgp_mime.flatten(response_email.plain())) f.close() return response_email def main(): msg_text = sys.stdin.read() try: ret,out_text,err_text,info = run_message(msg_text) except InvalidEmail, e: ret,out_text,err_text,info = e.response() except Exception, e: if LOGFILE != None: f = file(LOGFILE, "a+") f.write("Uncaught exception:\n%s\n" % (e,)) f.close() sys.exit(1) response_email = compose_response(ret, out_text, err_text, info) send_pgp_mime.mail(response_email.plain(), send_pgp_mime.sendmail) if __name__ == "__main__": main()