Autocrypt header generator/importer

config

[bindings]
    [[thread]]
        A = call hooks.import_autocrypt(ui)

hooks.py

import gpg
import logging
import shutil
import subprocess

from alot.settings.const import settings
from base64 import b64encode, b64decode
from email.utils import parseaddr
from tempfile import mkdtemp


prefer_encrypt = "mutual"
num_cols = 78


class AutocryptError(Exception):
    pass


async def pre_envelope_send(ui, dbm, cmd):
    envelope = ui.current_buffer.envelope
    if envelope.get('Autocrypt', "") != "":
        return

    frm = envelope.get('From', "")
    address = parseaddr(frm)[1]
    acc = settings.account_matching_address(address)
    if not acc.gpg_key:
        return
    
    fpr = acc.gpg_key.fpr
    gpg = subprocess.Popen(['gpg', '--export', fpr],
                           stdout=subprocess.PIPE)
    header = subprocess.check_output(
        ['sq', 'autocrypt', 'encode-sender',
         '--address', address, '--prefer-encrypt', prefer_encrypt],
        stdin=gpg.stdout)
    envelope.add('Autocrypt', str(header)[len("Autocrypt: "):])


async def import_autocrypt(ui):
    mail = ui.current_buffer.get_selected_message().get_email()
    try:
        autocrypt = parse_autocrypt(mail)
        message = "Import autocrypt key: %(addr)s (%(fpr)s)?" % autocrypt
        if (await ui.choice(message, cancel='no', msg_position='left')
                == 'no'):
            return

        c = gpg.Context()
        c.op_import(autocrypt["bytes"])
        logging.debug("key imported for: %s", autocrypt["addr"])
    except AutocryptError as e:
        logging.debug(str(e))
        ui.notify(str(e), 'error')


def parse_autocrypt(mail):
    autocrypt = {}

    aheader = mail.get('Autocrypt', "")
    frm = mail.get('From', "")
    address = parseaddr(frm)[1]
    if aheader == "":
        raise AutocryptError("No autocrypt header found")

    for i in aheader.split(";"):
        attr = i.split("=", 1)
        if len(attr) != 2:
            logging.debug("Can parse autocrypt attribute: %s", i)
            continue
        autocrypt[attr[0].strip()] = attr[1].strip()

    if "addr" not in autocrypt or "keydata" not in autocrypt:
        raise AutocryptError(
            "Address or keydata not in the autocrypt header: %s"
            % str(autocrypt.keys()))
    if address != autocrypt["addr"]:
        raise AutocryptError("The from doesn't match the autocrypt header")

    fpr, uids, kbytes = parse_keydata(autocrypt["keydata"])
    if autocrypt["addr"] not in uids:
        raise AutocryptError("%s not present in uids: %s"
                             % (autocrypt["addr"], str(uids)))

    autocrypt["fpr"] = fpr
    autocrypt["uids"] = uids
    autocrypt["bytes"] = kbytes
    return autocrypt


def parse_keydata(keydata):
    gpghome = mkdtemp("autocrypt")
    c = gpg.Context(home_dir=gpghome)

    keybytes = b64decode(keydata)
    c.op_import(keybytes)
    k = next(c.keylist())
    uids = [u.email for u in k.uids]

    shutil.rmtree(gpghome)
    return (k.fpr, uids, keybytes)