interfaces/email/interactive/send_pgp_mime.py

   1 #!/usr/bin/python
   2 #
   3 # Copyright (C) 2009 W. Trevor King <wking@drexel.edu>
   4 #
   5 # This program is free software; you can redistribute it and/or modify
   6 # it under the terms of the GNU General Public License as published by
   7 # the Free Software Foundation; either version 2 of the License, or
   8 # (at your option) any later version.
   9 #
  10 # This program is distributed in the hope that it will be useful,
  11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 # GNU General Public License for more details.
  14 #
  15 # You should have received a copy of the GNU General Public License along
  16 # with this program; if not, write to the Free Software Foundation, Inc.,
  17 # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  18 """
  19 Python module and command line tool for sending pgp/mime email.
  20
  21 Mostly uses subprocess to call gpg and a sendmail-compatible mailer.
  22 If you lack gpg, either don't use the encryption functions or adjust
  23 the pgp_* commands.  You may need to adjust the sendmail command to
  24 point to whichever sendmail-compatible mailer you have on your system.
  25 """
  26
  27 from cStringIO import StringIO
  28 import os
  29 import re
  30 #import GnuPGInterface # Maybe should use this instead of subprocess
  31 import smtplib
  32 import subprocess
  33 import sys
  34 import tempfile
  35 import types
  36
  37 try:
  38     from email import Message
  39     from email.mime.text import MIMEText
  40     from email.mime.multipart import MIMEMultipart
  41     from email.mime.application import MIMEApplication
  42     from email.encoders import encode_7or8bit
  43     from email.generator import Generator
  44     from email.parser import Parser
  45     from email.utils import getaddress
  46 except ImportError:
  47     # adjust to old python 2.4
  48     from email import Message
  49     from email.MIMEText import MIMEText
  50     from email.MIMEMultipart import MIMEMultipart
  51     from email.MIMENonMultipart import MIMENonMultipart
  52     from email.Encoders import encode_7or8bit
  53     from email.Generator import Generator
  54     from email.parser import Parser
  55     from email.Utils import getaddresses
  56
  57     getaddress = getaddresses
  58     class MIMEApplication (MIMENonMultipart):
  59         def __init__(self, _data, _subtype, _encoder, **params):
  60             MIMENonMultipart.__init__(self, 'application', _subtype, **params)
  61             self.set_payload(_data)
  62             _encoder(self)
  63
  64 usage="""usage: %prog [options]
  65
  66 Scriptable PGP MIME email using gpg.
  67
  68 You can use gpg-agent for passphrase caching if your key requires a
  69 passphrase (it better!).  Example usage would be to install gpg-agent,
  70 and then run
  71   export GPG_TTY=`tty`
  72   eval $(gpg-agent --daemon)
  73 in your shell before invoking this script.  See gpg-agent(1) for more
  74 details.  Alternatively, you can send your passphrase in on stdin
  75   echo 'passphrase' | %prog [options]
  76 or use the --passphrase-file option
  77   %prog [options] --passphrase-file FILE [more options]
  78 Both of these alternatives are much less secure than gpg-agent.  You
  79 have been warned.
  80 """
  81
  82 verboseInvoke = False
  83 PGP_SIGN_AS = None
  84 PASSPHRASE = None
  85
  86 # The following commands are adapted from my .mutt/pgp configuration
  87 #
  88 # Printf-like sequences:
  89 #   %a The value of PGP_SIGN_AS.
  90 #   %f Expands to the name of a file with text to be signed/encrypted.
  91 #   %p Expands to the passphrase argument.
  92 #   %R A string with some number (0 on up) of pgp_reciepient_arg
  93 #      strings.
  94 #   %r One key ID (e.g. recipient email address) to build a
  95 #      pgp_reciepient_arg string.
  96 #
  97 # The above sequences can be used to optionally print a string if
  98 # their length is nonzero. For example, you may only want to pass the
  99 # -u/--local-user argument to gpg if PGP_SIGN_AS is defined.  To
 100 # optionally print a string based upon one of the above sequences, the
 101 # following construct is used
 102 #   %?<sequence_char>?<optional_string>?
 103 # where sequence_char is a character from the table above, and
 104 # optional_string is the string you would like printed if status_char
 105 # is nonzero. optional_string may contain other sequence as well as
 106 # normal text, but it may not contain any question marks.
 107 #
 108 # see http://codesorcery.net/old/mutt/mutt-gnupg-howto
 109 #     http://www.mutt.org/doc/manual/manual-6.html#pgp_autosign
 110 #     http://tldp.org/HOWTO/Mutt-GnuPG-PGP-HOWTO-8.html
 111 # for more details
 112
 113 pgp_recipient_arg='-r "%r"'
 114 pgp_stdin_passphrase_arg='--passphrase-fd 0'
 115 pgp_sign_command='/usr/bin/gpg --no-verbose --quiet --batch %p --output - --detach-sign --armor --textmode %?a?-u "%a"? %f'
 116 pgp_encrypt_only_command='/usr/bin/gpg --no-verbose --quiet --batch --output - --encrypt --armor --textmode --always-trust --encrypt-to "%a" %R -- %f'
 117 pgp_encrypt_sign_command='/usr/bin/gpg --no-verbose --quiet --batch %p --output - --encrypt --sign %?a?-u "%a"? --armor --textmode --always-trust --encrypt-to "%a" %R -- %f'
 118 sendmail='/usr/sbin/sendmail -t'
 119
 120 def mail(msg, sendmail=None):
 121     """
 122     Send an email Message instance on its merry way.
 123
 124     We can shell out to the user specified sendmail in case
 125     the local host doesn't have an SMTP server set up
 126     for easy smtplib usage.
 127     """
 128     if sendmail != None:
 129         execute(sendmail, stdin=flatten(msg))
 130         return None
 131     s = smtplib.SMTP()
 132     s.connect()
 133     s.sendmail(from_addr=source_email(msg),
 134                to_addrs=target_emails(msg),
 135                msg=flatten(msg))
 136     s.close()
 137
 138 def header_from_text(text, encoding="us-ascii"):
 139     """
 140     Simple wrapper for instantiating an email.Message from text.
 141     >>> header = header_from_text('\\n'.join(['From: me@big.edu','To: you@big.edu','Subject: testing']))
 142     >>> print flatten(header)
 143     From: me@big.edu
 144     To: you@big.edu
 145     Subject: testing
 146     <BLANKLINE>
 147     <BLANKLINE>
 148     """
 149     text = text.strip()
 150     if type(text) == types.UnicodeType:
 151         text = text.encode(encoding)
 152     # assume StringType arguments are already encoded
 153     p = Parser()
 154     return p.parsestr(text, headersonly=True)
 155
 156 def encodedMIMEText(body, encoding=None):
 157     if encoding == None:
 158         if type(body) == types.StringType:
 159             encoding = "us-ascii"
 160         elif type(body) == types.UnicodeType:
 161             for encoding in ["us-ascii", "iso-8859-1", "utf-8"]:
 162                 try:
 163                     body.encode(encoding)
 164                 except UnicodeError:
 165                     pass
 166                 else:
 167                     break
 168             assert encoding != None
 169     # Create the message ('plain' stands for Content-Type: text/plain)
 170     if encoding == "us-ascii":
 171         return MIMEText(body)
 172     else:
 173         return MIMEText(body.encode(encoding), 'plain', encoding)
 174
 175 def attach_root(header, root_part):
 176     """
 177     Attach the email.Message root_part to the email.Message header
 178     without generating a multi-part message.
 179     """
 180     for k,v in header.items():
 181         root_part[k] = v
 182     return root_part
 183
 184 def execute(args, stdin=None, expect=(0,)):
 185     """
 186     Execute a command (allows us to drive gpg).
 187     """
 188     if verboseInvoke == True:
 189         print >> sys.stderr, '$ '+args
 190     try:
 191         p = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, close_fds=True)
 192     except OSError, e:
 193         strerror = '%s\nwhile executing %s' % (e.args[1], args)
 194         raise Exception, strerror
 195     output, error = p.communicate(input=stdin)
 196     status = p.wait()
 197     if verboseInvoke == True:
 198         print >> sys.stderr, '(status: %d)\n%s%s' % (status, output, error)
 199     if status not in expect:
 200         strerror = '%s\nwhile executing %s\n%s\n%d' % (args[1], args, error, status)
 201         raise Exception, strerror
 202     return status, output, error
 203
 204 def replace(template, format_char, replacement_text):
 205     """
 206     >>> replace('--textmode %?a?-u %a? %f', 'f', 'file.in')
 207     '--textmode %?a?-u %a? file.in'
 208     >>> replace('--textmode %?a?-u %a? %f', 'a', '0xHEXKEY')
 209     '--textmode -u 0xHEXKEY %f'
 210     >>> replace('--textmode %?a?-u %a? %f', 'a', '')
 211     '--textmode  %f'
 212     """
 213     if replacement_text == None:
 214         replacement_text = ""
 215     regexp = re.compile('%[?]'+format_char+'[?]([^?]*)[?]')
 216     if len(replacement_text) > 0:
 217         str = regexp.sub('\g<1>', template)
 218     else:
 219         str = regexp.sub('', template)
 220     regexp = re.compile('%'+format_char)
 221     str = regexp.sub(replacement_text, str)
 222     return str
 223
 224 def flatten(msg, to_unicode=False):
 225     """
 226     Produce flat text output from an email Message instance.
 227     """
 228     assert msg != None
 229     fp = StringIO()
 230     g = Generator(fp, mangle_from_=False)
 231     g.flatten(msg)
 232     text = fp.getvalue()
 233     if to_unicode == True:
 234         encoding = msg.get_content_charset() or "utf-8"
 235         text = unicode(text, encoding=encoding)
 236     return text
 237
 238 def source_email(msg, return_realname=False):
 239     """
 240     Search the header of an email Message instance to find the
 241     sender's email address.
 242     """
 243     froms = msg.get_all('from', [])
 244     from_tuples = getaddresses(froms) # [(realname, email_address), ...]
 245     assert len(from_tuples) == 1
 246     if return_realname == True:
 247         return from_tuples[0] # (realname, email_address)
 248     return from_tuples[0][1]  # email_address
 249
 250 def target_emails(msg):
 251     """
 252     Search the header of an email Message instance to find a
 253     list of recipient's email addresses.
 254     """
 255     tos = msg.get_all('to', [])
 256     ccs = msg.get_all('cc', [])
 257     bccs = msg.get_all('bcc', [])
 258     resent_tos = msg.get_all('resent-to', [])
 259     resent_ccs = msg.get_all('resent-cc', [])
 260     resent_bccs = msg.get_all('resent-bcc', [])
 261     all_recipients = getaddresses(tos + ccs + bccs + resent_tos
 262                                   + resent_ccs + resent_bccs)
 263     return [addr[1] for addr in all_recipients]
 264
 265 class PGPMimeMessageFactory (object):
 266     """
 267     See http://www.ietf.org/rfc/rfc3156.txt for specification details.
 268     >>> from_addr = "me@big.edu"
 269     >>> to_addr = "you@you.edu"
 270     >>> header = header_from_text('\\n'.join(['From: %s'%from_addr,'To: %s'%to_addr,'Subject: testing']))
 271     >>> source_email(header) == from_addr
 272     True
 273     >>> target_emails(header) == [to_addr]
 274     True
 275     >>> m = EncryptedMessageFactory('check 1 2\\ncheck 1 2\\n')
 276     >>> print flatten(m.clearBodyPart())
 277     Content-Type: text/plain; charset="us-ascii"
 278     MIME-Version: 1.0
 279     Content-Transfer-Encoding: 7bit
 280     Content-Disposition: inline
 281     <BLANKLINE>
 282     check 1 2
 283     check 1 2
 284     <BLANKLINE>
 285     >>> print flatten(m.plain())
 286     Content-Type: text/plain; charset="us-ascii"
 287     MIME-Version: 1.0
 288     Content-Transfer-Encoding: 7bit
 289     <BLANKLINE>
 290     check 1 2
 291     check 1 2
 292     <BLANKLINE>
 293     >>> signed = m.sign(header)
 294     >>> signed.set_boundary('boundsep')
 295     >>> print flatten(signed).replace('\\t', ' '*4) # doctest: +ELLIPSIS, +NORMALIZE_WHITESPACE
 296     Content-Type: multipart/signed; protocol="application/pgp-signature";
 297         micalg="pgp-sha1"; boundary="boundsep"
 298     MIME-Version: 1.0
 299     Content-Disposition: inline
 300     <BLANKLINE>
 301     --boundsep
 302     Content-Type: text/plain; charset="us-ascii"
 303     MIME-Version: 1.0
 304     Content-Transfer-Encoding: 7bit
 305     Content-Disposition: inline
 306     <BLANKLINE>
 307     check 1 2
 308     check 1 2
 309     <BLANKLINE>
 310     --boundsep
 311     MIME-Version: 1.0
 312     Content-Transfer-Encoding: 7bit
 313     Content-Description: signature
 314     Content-Type: application/pgp-signature; name="signature.asc";
 315         charset="us-ascii"
 316     <BLANKLINE>
 317     -----BEGIN PGP SIGNATURE-----
 318     ...
 319     -----END PGP SIGNATURE-----
 320     <BLANKLINE>
 321     --boundsep--
 322     >>> encrypted = m.encrypt(header)
 323     >>> encrypted.set_boundary('boundsep')
 324     >>> print flatten(encrypted).replace('\\t', ' '*4) # doctest: +ELLIPSIS, +NORMALIZE_WHITESPACE
 325     Content-Type: multipart/encrypted;
 326         protocol="application/pgp-encrypted";
 327         micalg="pgp-sha1"; boundary="boundsep"
 328     MIME-Version: 1.0
 329     Content-Disposition: inline
 330     <BLANKLINE>
 331     --boundsep
 332     Content-Type: application/pgp-encrypted
 333     MIME-Version: 1.0
 334     Content-Transfer-Encoding: 7bit
 335     <BLANKLINE>
 336     Version: 1
 337     <BLANKLINE>
 338     --boundsep
 339     MIME-Version: 1.0
 340     Content-Transfer-Encoding: 7bit
 341     Content-Type: application/octet-stream; charset="us-ascii"
 342     <BLANKLINE>
 343     -----BEGIN PGP MESSAGE-----
 344     ...
 345     -----END PGP MESSAGE-----
 346     <BLANKLINE>
 347     --boundsep--
 348     >>> signedAndEncrypted = m.signAndEncrypt(header)
 349     >>> signedAndEncrypted.set_boundary('boundsep')
 350     >>> print flatten(signedAndEncrypted).replace('\\t', ' '*4) # doctest: +ELLIPSIS, +NORMALIZE_WHITESPACE
 351     Content-Type: multipart/encrypted;
 352         protocol="application/pgp-encrypted";
 353         micalg="pgp-sha1"; boundary="boundsep"
 354     MIME-Version: 1.0
 355     Content-Disposition: inline
 356     <BLANKLINE>
 357     --boundsep
 358     Content-Type: application/pgp-encrypted
 359     MIME-Version: 1.0
 360     Content-Transfer-Encoding: 7bit
 361     <BLANKLINE>
 362     Version: 1
 363     <BLANKLINE>
 364     --boundsep
 365     MIME-Version: 1.0
 366     Content-Transfer-Encoding: 7bit
 367     Content-Type: application/octet-stream; charset="us-ascii"
 368     <BLANKLINE>
 369     -----BEGIN PGP MESSAGE-----
 370     ...
 371     -----END PGP MESSAGE-----
 372     <BLANKLINE>
 373     --boundsep--
 374     """
 375     def __init__(self, body):
 376         self.body = body
 377     def clearBodyPart(self):
 378         body = encodedMIMEText(self.body)
 379         body.add_header('Content-Disposition', 'inline')
 380         return body
 381     def passphrase_arg(self, passphrase=None):
 382         if passphrase == None and PASSPHRASE != None:
 383             passphrase = PASSPHRASE
 384         if passphrase == None:
 385             return (None,'')
 386         return (passphrase, pgp_stdin_passphrase_arg)
 387     def plain(self):
 388         """
 389         text/plain
 390         """
 391         return encodedMIMEText(self.body)
 392     def sign(self, header, passphrase=None):
 393         """
 394         multipart/signed
 395           +-> text/plain                 (body)
 396           +-> application/pgp-signature  (signature)
 397         """
 398         passphrase,pass_arg = self.passphrase_arg(passphrase)
 399         body = self.clearBodyPart()
 400         bfile = tempfile.NamedTemporaryFile()
 401         bfile.write(flatten(body))
 402         bfile.flush()
 403
 404         args = replace(pgp_sign_command, 'f', bfile.name)
 405         if PGP_SIGN_AS == None:
 406             pgp_sign_as = '<%s>' % source_email(header)
 407         else:
 408             pgp_sign_as = PGP_SIGN_AS
 409         args = replace(args, 'a', pgp_sign_as)
 410         args = replace(args, 'p', pass_arg)
 411         status,output,error = execute(args, stdin=passphrase)
 412         signature = output
 413
 414         sig = MIMEApplication(_data=signature,
 415                               _subtype='pgp-signature; name="signature.asc"',
 416                               _encoder=encode_7or8bit)
 417         sig['Content-Description'] = 'signature'
 418         sig.set_charset('us-ascii')
 419
 420         msg = MIMEMultipart('signed', micalg='pgp-sha1',
 421                             protocol='application/pgp-signature')
 422         msg.attach(body)
 423         msg.attach(sig)
 424
 425         msg['Content-Disposition'] = 'inline'
 426         return msg
 427     def encrypt(self, header, passphrase=None):
 428         """
 429         multipart/encrypted
 430          +-> application/pgp-encrypted  (control information)
 431          +-> application/octet-stream   (body)
 432         """
 433         body = self.clearBodyPart()
 434         bfile = tempfile.NamedTemporaryFile()
 435         bfile.write(flatten(body))
 436         bfile.flush()
 437
 438         recipients = [replace(pgp_recipient_arg, 'r', recipient)
 439                       for recipient in target_emails(header)]
 440         recipient_string = ' '.join(recipients)
 441         args = replace(pgp_encrypt_only_command, 'R', recipient_string)
 442         args = replace(args, 'f', bfile.name)
 443         if PGP_SIGN_AS == None:
 444             pgp_sign_as = '<%s>' % source_email(header)
 445         else:
 446             pgp_sign_as = PGP_SIGN_AS
 447         args = replace(args, 'a', pgp_sign_as)
 448         status,output,error = execute(args)
 449         encrypted = output
 450
 451         enc = MIMEApplication(_data=encrypted, _subtype='octet-stream',
 452                               _encoder=encode_7or8bit)
 453         enc.set_charset('us-ascii')
 454
 455         control = MIMEApplication(_data='Version: 1\n', _subtype='pgp-encrypted',
 456                                   _encoder=encode_7or8bit)
 457
 458         msg = MIMEMultipart('encrypted', micalg='pgp-sha1',
 459                             protocol='application/pgp-encrypted')
 460         msg.attach(control)
 461         msg.attach(enc)
 462
 463         msg['Content-Disposition'] = 'inline'
 464         return msg
 465     def signAndEncrypt(self, header, passphrase=None):
 466         """
 467         multipart/encrypted
 468          +-> application/pgp-encrypted  (control information)
 469          +-> application/octet-stream   (body)
 470         """
 471         passphrase,pass_arg = self.passphrase_arg(passphrase)
 472         body = self.sign(header, passphrase)
 473         body.__delitem__('Bcc')
 474         bfile = tempfile.NamedTemporaryFile()
 475         bfile.write(flatten(body))
 476         bfile.flush()
 477
 478         recipients = [replace(pgp_recipient_arg, 'r', recipient)
 479                       for recipient in target_emails(header)]
 480         recipient_string = ' '.join(recipients)
 481         args = replace(pgp_encrypt_only_command, 'R', recipient_string)
 482         args = replace(args, 'f', bfile.name)
 483         if PGP_SIGN_AS == None:
 484             pgp_sign_as = '<%s>' % source_email(header)
 485         else:
 486             pgp_sign_as = PGP_SIGN_AS
 487         args = replace(args, 'a', pgp_sign_as)
 488         args = replace(args, 'p', pass_arg)
 489         status,output,error = execute(args, stdin=passphrase)
 490         encrypted = output
 491
 492         enc = MIMEApplication(_data=encrypted, _subtype='octet-stream',
 493                               _encoder=encode_7or8bit)
 494         enc.set_charset('us-ascii')
 495
 496         control = MIMEApplication(_data='Version: 1\n',
 497                                   _subtype='pgp-encrypted',
 498                                   _encoder=encode_7or8bit)
 499
 500         msg = MIMEMultipart('encrypted', micalg='pgp-sha1',
 501                             protocol='application/pgp-encrypted')
 502         msg.attach(control)
 503         msg.attach(enc)
 504
 505         msg['Content-Disposition'] = 'inline'
 506         return msg
 507
 508 def test():
 509     import doctest
 510     doctest.testmod()
 511
 512
 513 if __name__ == '__main__':
 514     from optparse import OptionParser
 515
 516     parser = OptionParser(usage=usage)
 517     parser.add_option('-t', '--test', dest='test', action='store_true',
 518                       help='Run doctests and exit')
 519
 520     parser.add_option('-H', '--header-file', dest='header_filename',
 521                       help='file containing email header', metavar='FILE')
 522     parser.add_option('-B', '--body-file', dest='body_filename',
 523                       help='file containing email body', metavar='FILE')
 524
 525     parser.add_option('-P', '--passphrase-file', dest='passphrase_file',
 526                       help='file containing gpg passphrase', metavar='FILE')
 527     parser.add_option('-p', '--passphrase-fd', dest='passphrase_fd',
 528                       help='file descriptor from which to read gpg passphrase (0 for stdin)',
 529                       type="int", metavar='DESCRIPTOR')
 530
 531     parser.add_option('--mode', dest='mode', default='sign',
 532                       help="One of 'sign', 'encrypt', 'sign-encrypt', or 'plain'.  Defaults to %default.",
 533                       metavar='MODE')
 534
 535     parser.add_option('-a', '--sign-as', dest='sign_as',
 536                       help="The gpg key to sign with (gpg's -u/--local-user)",
 537                       metavar='KEY')
 538
 539     parser.add_option('--output', dest='output', action='store_true',
 540                       help="Don't mail the generated message, print it to stdout instead.")
 541
 542     (options, args) = parser.parse_args()
 543
 544     stdin_used = False
 545
 546     if options.passphrase_file != None:
 547         PASSPHRASE = file(options.passphrase_file, 'r').read()
 548     elif options.passphrase_fd != None:
 549         if options.passphrase_fd == 0:
 550             stdin_used = True
 551             PASSPHRASE = sys.stdin.read()
 552         else:
 553             PASSPHRASE = os.read(options.passphrase_fd)
 554
 555     if options.sign_as:
 556         PGP_SIGN_AS = options.sign_as
 557
 558     if options.test == True:
 559         test()
 560         sys.exit(0)
 561
 562     header = None
 563     if options.header_filename != None:
 564         if options.header_filename == '-':
 565             assert stdin_used == False
 566             stdin_used = True
 567             header = sys.stdin.read()
 568         else:
 569             header = file(options.header_filename, 'r').read()
 570     if header == None:
 571         raise Exception, "missing header"
 572     headermsg = header_from_text(header)
 573     body = None
 574     if options.body_filename != None:
 575         if options.body_filename == '-':
 576             assert stdin_used == False
 577             stdin_used = True
 578             body = sys.stdin.read()
 579         else:
 580             body = file(options.body_filename, 'r').read()
 581     if body == None:
 582         raise Exception, "missing body"
 583
 584     m = EncryptedMessageFactory(body)
 585     if options.mode == "sign":
 586         bodymsg = m.sign(header)
 587     elif options.mode == "encrypt":
 588         bodymsg = m.encrypt(header)
 589     elif options.mode == "sign-encrypt":
 590         bodymsg = m.signAndEncrypt(header)
 591     elif options.mode == "plain":
 592         bodymsg = m.plain()
 593     else:
 594         print "Unrecognized mode '%s'" % options.mode
 595
 596     message = attach_root(headermsg, bodymsg)
 597     if options.output == True:
 598         message = flatten(message)
 599         print message
 600     else:
 601         mail(message, sendmail)