#! /usr/bin/env python3
-
+# -*- coding: utf-8 -*-
"""*********************************************************************
* Edward is free software: you can redistribute it and/or modify *
* it under the terms of the GNU Affero Public License as published by *
* *
************************************************************************
-Code used from:
+Code sourced from these projects:
* http://agpl.fsf.org/emailselfdefense.fsf.org/edward/CURRENT/edward.tar.gz
* https://git-tails.immerda.ch/whisperback/tree/whisperBack/encryption.py?h=feature/python3
* http://www.physics.drexel.edu/~wking/code/python/send_pgp_mime
-
"""
import sys
import gpgme
import re
import io
+import os
import email.parser
import email.message
from email.mime.application import MIMEApplication
from email.mime.nonmultipart import MIMENonMultipart
+import edward_config
+
+match_types = [('encrypted',
+ '-----BEGIN PGP MESSAGE-----.*?-----END PGP MESSAGE-----'),
+ ('pubkey',
+ '-----BEGIN PGP PUBLIC KEY BLOCK-----.*?-----END PGP PUBLIC KEY BLOCK-----'),
+ ('detachedsig',
+ '-----END PGP SIGNATURE-----.*?-----BEGIN PGP SIGNATURE-----'),
+ ('clearsign',
+ '-----BEGIN PGP SIGNED MESSAGE-----.*?-----BEGIN PGP SIGNATURE-----.*?-----END PGP SIGNATURE-----')]
+
+
+class EddyMsg (object):
+ def __init__(self):
+ self.multipart = False
+ self.subparts = []
+
+ self.charset = None
+ self.payload_bytes = None
+ self.payload_pieces = []
+
+ self.filename = None
+ self.content_type = None
+ self.description_list = None
+
+
+class PayloadPiece (object):
+ def __init__(self):
+ self.piece_type = None
+ self.string = None
+ self.gpg_data = None
+
def main ():
handle_args()
+ gpgme_ctx = get_gpg_context(edward_config.gnupghome,
+ edward_config.sign_with_key)
+
email_text = sys.stdin.read()
+ result = parse_pgp_mime(email_text)
email_from, email_subject = email_from_subject(email_text)
- plaintext, keys = email_decode_flatten(email_text)
- encrypt_to_key = choose_reply_encryption_key(keys)
+# plaintext, fingerprints = email_decode_flatten(email_text, gpgme_ctx, False)
+# encrypt_to_key = choose_reply_encryption_key(gpgme_ctx, fingerprints)
+#
+# reply_message = generate_reply(plaintext, email_from, \
+# email_subject, encrypt_to_key,
+# gpgme_ctx)
- reply_message = generate_reply(plaintext, email_from, \
- email_subject, encrypt_to_key,
- "DAB4F989E2788B8DF058E0EFEF1EC52039B36E58")
+ print(flatten_eddy(result))
- print(reply_message)
+def get_gpg_context (gnupghome, sign_with_key_fp):
-def email_decode_flatten (email_text):
+ os.environ['GNUPGHOME'] = gnupghome
- body = ""
- keys = []
+ gpgme_ctx = gpgme.Context()
+ gpgme_ctx.armor = True
+
+ try:
+ sign_with_key = gpgme_ctx.get_key(sign_with_key_fp)
+ except:
+ error("unable to load signing key. is the gnupghome "
+ + "and signing key properly set in the edward_config.py?")
+ exit(1)
+
+ gpgme_ctx.signers = [sign_with_key]
+
+ return gpgme_ctx
+
+
+def parse_pgp_mime (email_text):
email_struct = email.parser.Parser().parsestr(email_text)
- for subpart in email_struct.walk():
+ eddy_obj = parse_mime(email_struct)
+ eddy_obj = split_payloads(eddy_obj)
- payload, description, filename, content_type \
- = get_email_subpart_info(subpart)
+ return eddy_obj
- if payload == "":
- continue
- if content_type == "multipart":
- continue
+def parse_mime(msg_struct):
- if content_type == "application/pgp-encrypted":
- if description == "PGP/MIME version identification":
- if payload.strip() != "Version: 1":
- print(progname + ": Warning: unknown " \
- + description + ": " \
- + payload.strip(), file=sys.stderr)
- continue
+ eddy_obj = EddyMsg()
+
+ if msg_struct.is_multipart() == True:
+ payloads = msg_struct.get_payload()
+
+ eddy_obj.multipart = True
+ eddy_obj.subparts = list(map(parse_mime, payloads))
+
+ else:
+ eddy_obj = get_subpart_data(msg_struct)
+
+ return eddy_obj
+
+
+def split_payloads (eddy_obj):
+
+ if eddy_obj.multipart == True:
+ eddy_obj.subparts = list(map(split_payloads, eddy_obj.subparts))
+
+ else:
+ for (match_type, pattern) in match_types:
+ new_pieces_list = []
+ for payload_piece in eddy_obj.payload_pieces:
+ new_pieces_list += scan_and_split(payload_piece,
+ match_type, pattern)
+ eddy_obj.payload_pieces = new_pieces_list
- if (filename == "encrypted.asc") or (content_type == "pgp/mime"):
- plaintext, more_keys = decrypt_text(payload)
+ return eddy_obj
- body += plaintext
- keys += more_keys
- elif content_type == "text/plain":
- body += payload + "\n"
+def scan_and_split (payload_piece, match_type, pattern):
+ flags = re.DOTALL | re.MULTILINE
+ matches = re.search("(?P<beginning>.*?)(?P<match>" + pattern +
+ ")(?P<rest>.*)", payload_piece.string, flags=flags)
+
+ if matches == None:
+ pieces = [payload_piece]
+
+ else:
+
+ beginning = PayloadPiece()
+ beginning.string = matches.group('beginning')
+ beginning.piece_type = payload_piece.piece_type
+
+ match = PayloadPiece()
+ match.string = matches.group('match')
+ match.piece_type = match_type
+
+ rest = PayloadPiece()
+ rest.string = matches.group('rest')
+ rest.piece_type = payload_piece.piece_type
+
+ more_pieces = scan_and_split(rest, match_type, pattern)
+
+ if more_pieces == None:
+ pieces = [beginning, match, rest]
else:
- body += payload + "\n"
+ pieces = [beginning, match] + more_pieces
+
+ return pieces
+
- return body, keys
+def get_subpart_data (part):
+
+ obj = EddyMsg()
+
+ obj.charset = part.get_content_charset()
+ obj.payload_bytes = part.get_payload(decode=True)
+
+ obj.filename = part.get_filename()
+ obj.content_type = part.get_content_type()
+ obj.description_list = part['content-description']
+
+ # your guess is as good as a-myy-ee-ine...
+ if obj.charset == None:
+ obj.charset = 'utf-8'
+
+ if obj.payload_bytes != None:
+ try:
+ payload = PayloadPiece()
+ payload.string = obj.payload_bytes.decode(obj.charset)
+ payload.piece_type = 'text'
+
+ obj.payload_pieces = [payload]
+ except UnicodeDecodeError:
+ pass
+
+ return obj
+
+
+def do_to_eddys_pieces (function_to_do, eddy_obj, data):
+
+ if eddy_obj.multipart == True:
+ result_list = []
+ for sub in eddy_obj.subparts:
+ result_list += do_to_eddys_pieces(function_to_do, sub, data)
+ else:
+ result_list = [function_to_do(eddy_obj.payload_pieces, data)]
+
+ return result_list
+
+
+def flatten_eddy (eddy_obj):
+
+ string = "\n".join(do_to_eddys_pieces(flatten_payload_pieces, eddy_obj, None))
+
+ return string
+
+
+def flatten_payload_pieces (payload_pieces, _ignore):
+
+ string = ""
+ for piece in payload_pieces:
+ string += piece.string
+
+ return string
def email_from_subject (email_text):
return email_from, email_subject
-def get_email_subpart_info (part):
+def add_gpg_keys (text, gpgme_ctx):
- charset = part.get_content_charset()
- payload_bytes = part.get_payload(decode=True)
+ key_blocks = scan_and_grab(text,
+ '-----BEGIN PGP PUBLIC KEY BLOCK-----',
+ '-----END PGP PUBLIC KEY BLOCK-----')
- filename = part.get_filename()
- content_type = part.get_content_type()
- description_list = part.get_params(header='content-description')
+ fingerprints = []
+ for key_block in key_blocks:
+ fp = io.BytesIO(key_block.encode('ascii'))
- if charset == None:
- charset = 'utf-8'
+ result = gpgme_ctx.import_(fp)
+ imports = result.imports
- if payload_bytes != None:
- payload = payload_bytes.decode(charset)
- else:
- payload = ""
+ if imports != []:
+ fingerprint = imports[0][0]
+ fingerprints += [fingerprint]
- if description_list != None:
- description = description_list[0][0]
- else:
- description = ""
+ debug("added gpg key: " + fingerprint)
- return payload, description, filename, content_type
+ return fingerprints
-def decrypt_text (gpg_text):
+def decrypt_text (gpg_text, gpgme_ctx):
body = ""
- keys = []
+ fingerprints = []
- gpg_chunks = split_message(gpg_text)
+ msg_blocks = scan_and_grab(gpg_text,
+ '-----BEGIN PGP MESSAGE-----',
+ '-----END PGP MESSAGE-----')
- plaintext_and_sigs_chunks = decrypt_chunks(gpg_chunks)
+ plaintexts_and_sigs = decrypt_blocks(msg_blocks, gpgme_ctx)
- for chunk in plaintext_and_sigs_chunks:
- plaintext = chunk[0]
- sigs = chunk[1]
+ for pair in plaintexts_and_sigs:
+ plaintext = pair[0]
+ sigs = pair[1]
for sig in sigs:
- key = get_pub_key(sig)
- keys += [key]
+ fingerprints += [sig.fpr]
# recursive for nested layers of mime and/or gpg
- plaintext, more_keys = email_decode_flatten(plaintext)
+ plaintext, more_fps = email_decode_flatten(plaintext, gpgme_ctx, True)
body += plaintext
- keys += more_keys
+ fingerprints += more_fps
- return body, keys
+ return body, fingerprints
-def get_pub_key (sig):
+def verify_clear_signature (text, gpgme_ctx):
- gpgme_ctx = gpgme.Context()
+ sig_blocks = scan_and_grab(text,
+ '-----BEGIN PGP SIGNED MESSAGE-----',
+ '-----END PGP SIGNATURE-----')
- fingerprint = sig.fpr
- key = gpgme_ctx.get_key(fingerprint)
+ fingerprints = []
+ plaintext = ""
- return key
+ for sig_block in sig_blocks:
+ msg_fp = io.BytesIO(sig_block.encode('utf-8'))
+ ptxt_fp = io.BytesIO()
+ result = gpgme_ctx.verify(msg_fp, None, ptxt_fp)
-def split_message (text):
+ plaintext += ptxt_fp.getvalue().decode('utf-8')
+ fingerprint = result[0].fpr
- gpg_matches = re.search( \
- '(-----BEGIN PGP MESSAGE-----' + \
- '.*' + \
- '-----END PGP MESSAGE-----)', \
- text, \
- flags=re.DOTALL)
+ fingerprints += [fingerprint]
- if gpg_matches != None:
- gpg_chunks = gpg_matches.groups()
- else:
- gpg_chunks = ()
+ return plaintext, fingerprints
- return gpg_chunks
+def scan_and_grab (text, start_text, end_text):
-def decrypt_chunks (gpg_chunks):
+ matches = re.search('(' + start_text + '.*' + end_text + ')',
+ text, flags=re.DOTALL)
- return map(decrypt_chunk, gpg_chunks)
+ if matches != None:
+ match_tuple = matches.groups()
+ else:
+ match_tuple = ()
+ return match_tuple
-def decrypt_chunk (gpg_chunk):
- gpgme_ctx = gpgme.Context()
+def decrypt_blocks (msg_blocks, gpgme_ctx):
+
+ return [decrypt_block(block, gpgme_ctx) for block in msg_blocks]
- chunk_b = io.BytesIO(gpg_chunk.encode('ascii'))
+
+def decrypt_block (msg_block, gpgme_ctx):
+
+ block_b = io.BytesIO(msg_block.encode('ascii'))
plain_b = io.BytesIO()
- sigs = gpgme_ctx.decrypt_verify(chunk_b, plain_b)
+ try:
+ sigs = gpgme_ctx.decrypt_verify(block_b, plain_b)
+ except:
+ return ("",[])
plaintext = plain_b.getvalue().decode('utf-8')
return (plaintext, sigs)
-def choose_reply_encryption_key (keys):
+def choose_reply_encryption_key (gpgme_ctx, fingerprints):
reply_key = None
- for key in keys:
- if (key.can_encrypt == True):
- reply_key = key
- break
+ for fp in fingerprints:
+ try:
+ key = gpgme_ctx.get_key(fp)
+
+ if (key.can_encrypt == True):
+ reply_key = key
+ break
+ except:
+ continue
+
return reply_key
def generate_reply (plaintext, email_from, email_subject, encrypt_to_key,
- sign_with_fingerprint):
+ gpgme_ctx):
reply = "To: " + email_from + "\n"
encrypted_text = encrypt_sign_message(plaintext_mime.as_string(),
encrypt_to_key,
- sign_with_fingerprint)
+ gpgme_ctx)
control_mime = MIMEApplication("Version: 1",
_subtype='pgp-encrypted',
return quoted_message
-def encrypt_sign_message (plaintext, encrypt_to_key, sign_with_fingerprint):
-
- gpgme_ctx = gpgme.Context()
- gpgme_ctx.armor = True
-
- sign_with_key = gpgme_ctx.get_key(sign_with_fingerprint)
- gpgme_ctx.signers = [sign_with_key]
+def encrypt_sign_message (plaintext, encrypt_to_key, gpgme_ctx):
plaintext_bytes = io.BytesIO(plaintext.encode('ascii'))
encrypted_bytes = io.BytesIO()
return encrypted_txt
+def error (error_msg):
+
+ sys.stderr.write(progname + ": " + error_msg + "\n")
+
+
+def debug (debug_msg):
+
+ if edward_config.debug == True:
+ error(debug_msg)
+
+
def handle_args ():
if __name__ == "__main__":
projects / edward.git / blobdiff