#!/usr/bin/env python3
# Copyright (C) 2020 by the Linux Foundation
# SPDX-License-Identifier: MIT-0
import sys
import os
import base64
import email
import email.header
import email.utils
import b4
import time
import dkim
import re
import dns.resolver
import binascii
import subprocess
import hashlib
import requests
import anybase32
import urllib.parse
import logging
import datetime

from nacl.signing import SigningKey, VerifyKey
from nacl.encoding import Base64Encoder
from nacl.exceptions import BadSignatureError

from tempfile import mkstemp

from Cryptodome.Signature import pkcs1_15
from Cryptodome.Hash import SHA256
from Cryptodome.PublicKey import RSA

from typing import Tuple

XPH_HDR = 'X-Patch-Hashes'
XPS_HDR = 'X-Patch-Sig'

# For POC purposes, we hardcode maximum drift to 30 days
MAXDRIFT = datetime.timedelta(days=30)

logger = logging.getLogger(__name__)


def _run_command(cmdargs: list, stdin: bytes = None) -> Tuple[int, bytes, bytes]:
    sp = subprocess.Popen(cmdargs, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.PIPE)
    (output, error) = sp.communicate(input=stdin)
    return sp.returncode, output, error


def gpg_run_command(cmdargs: list, stdin: bytes = None) -> Tuple[int, bytes, bytes]:
    gpgbin = 'gpg'
    cmdargs = [gpgbin, '--batch', '--no-auto-key-retrieve', '--no-auto-check-trustdb'] + cmdargs
    return _run_command(cmdargs, stdin)


def load_message(msgfile: str):
    with open(msgfile, 'rb') as fh:
        logger.info('Using %s as message source', msgfile)
        contents = fh.read()
        return email.message_from_bytes(contents)


def splitter(longstr: str, limit: int = 77) -> str:
    splitstr = list()
    first = True
    while len(longstr) > limit:
        at = limit
        if first:
            first = False
            at -= 2
        splitstr.append(longstr[:at])
        longstr = longstr[at:]
    splitstr.append(longstr)
    return ' '.join(splitstr)


def verify_identity_domain(msg, identity: str, domain: str) -> bool:
    # Domain is supposed to be present in identity
    if not identity.endswith(domain):
        logger.critical('FAIL : domain (d=%s) is not in identity (i=%s)', domain, identity)
        return False
    fromeml = email.utils.getaddresses(msg.get_all('from', []))[0][1]
    if identity.find('@') < 0:
        logger.critical('FAIL : identity must contain @ (i=%s)', identity)
        return False
    ilocal, idomain = identity.split('@')
    # identity is supposed to be present in from
    if not fromeml.endswith(f'@{idomain}'):
        logger.critical('FAIL : identity (i=%s) does not match from (from=%s)', identity, fromeml)
        return False
    logger.info('PASS : identity and domain match From header')
    return True


def verify_time_drift(msg, timestamp: str) -> bool:
    msgdt = email.utils.parsedate_to_datetime(str(msg['Date']))
    sigdt = datetime.datetime.utcfromtimestamp(int(timestamp)).replace(tzinfo=datetime.timezone.utc)
    sdrift = sigdt - msgdt
    if sdrift > MAXDRIFT:
        logger.critical('FAIL : time drift between Date and t too great (%s)', sdrift)
        return False
    logger.info('PASS : time drift between Date and t (%s)', sdrift)
    return True


def get_dkim_key(domain: str, selector: str, wantkey: str = 'rsa', timeout: int = 5) -> str:
    name = f'{selector}._domainkey.{domain}.'
    logger.info('DNS-lookup: %s', name)
    keydata = None
    try:
        a = dns.resolver.resolve(name, dns.rdatatype.TXT, raise_on_no_answer=False, lifetime=timeout)
        # Find v=DKIM1
        for r in a.response.answer:
            for item in r.items:
                for s in item.strings:
                    if s.find(b'v=DKIM1') >= 0:
                        keydata = s.decode()
                        if keydata.find(wantkey) >= 0:
                            break
                        keydata = None
                if keydata:
                    break
            if keydata:
                break
    except dns.resolver.NXDOMAIN:
        logger.critical('Domain %s does not exist', name)
        sys.exit(1)
        pass

    if not keydata:
        logger.critical('Domain %s does not contain a DKIM record', name)
        sys.exit(1)

    parts = get_parts_from_header(keydata)
    if 'p' not in parts:
        logger.critical('Domain %s does not contain a DKIM key', name)
        sys.exit(1)

    return parts['p']


def get_git_toplevel(gitdir: str = None) -> str:
    cmdargs = ['git']
    if gitdir:
        cmdargs += ['--git-dir', gitdir]
    cmdargs += ['rev-parse', '--show-toplevel']
    ecode, out, err = _run_command(cmdargs)
    if ecode == 0:
        return out.decode().strip()
    return ''


def get_wk_key(domain: str, selector: str, timeout: int = 5) -> str:
    wkurl = f'https://{domain}/.well-known/_domainkey/{selector}.txt'
    logger.info('Retrieving: %s', wkurl)
    res = requests.get(wkurl, timeout=timeout)
    if res.status_code != 200:
        logger.info('Could not retrieve %s: %s', wkurl, res.status_code)
        sys.exit(1)
    keydata = res.content.decode().strip()
    logger.debug('keydata: %s', keydata)
    parts = get_parts_from_header(keydata)
    return parts['p']


def get_wkd_key(domain: str, identity: str, selector: str, timeout: int = 5) -> str:
    identity = identity.split('@')[0]
    # Attempt to load from local git dir if we are in a git dir
    gittop = get_git_toplevel()
    if gittop:
        # urlencode domain/identity/selector to make sure nobody tries path-based badness
        subpath = os.path.join(gittop, '.keys', 'devkey',
                               urllib.parse.quote_plus(domain),
                               urllib.parse.quote_plus(identity),
                               urllib.parse.quote_plus(selector) + '.txt',
                               )
        try:
            with open(subpath) as fh:
                logger.info('Loading: WKD key from %s', subpath)
                return fh.readline().strip()
        except IOError:
            pass

    # We use zbase32 because this is what OpenPGP uses
    # I'm not convinced this is a sane choice, if only because
    # it has a swear word to my account record.
    i = hashlib.sha1(identity.lower().encode()).digest()
    zdir = anybase32.encode(i, anybase32.ZBASE32).decode()
    wkurl = f'https://{domain}/.well-known/devkey/{zdir}/{selector}.txt'
    logger.info('Retrieving: %s', wkurl)
    res = requests.get(wkurl, timeout=timeout)
    if res.status_code != 200:
        logger.critical('Could not get %s: %s', wkurl, res.status_code)
        sys.exit(1)
    # For POC purposes, we are not doing caching or TOFU management,
    # but this would be a required part of actual implementation
    return res.content.decode().strip()


def get_b64_attestation(msg) -> Tuple[str, str, str]:
    # b4 stores these as hexdigests, but it's more natural
    # for email headers to use b64 encoded values, if only
    # to save a few bytes of space
    lmsg = b4.LoreMessage(msg)
    lmsg.load_hashes()
    att = lmsg.attestation
    i = base64.b64encode(binascii.unhexlify(att.i)).decode()
    m = base64.b64encode(binascii.unhexlify(att.m)).decode()
    p = base64.b64encode(binascii.unhexlify(att.p)).decode()
    return i, m, p


def add_hashes_header(msg):
    hhdr = gen_hashes_header(msg)
    msg[XPH_HDR] = hhdr
    return msg


def gen_hashes_header(msg):
    i, m, p = get_b64_attestation(msg)
    # Hardcode to sha256 for the purposes of the POC
    hparts = [
        'v=1',
        'h=sha256',
        f'i={i}',
        f'm={m}',
        f'p={p}',
    ]
    hval = '; '.join(hparts)
    hhdr = email.header.make_header([(hval.encode(), 'us-ascii')], maxlinelen=78)
    return hhdr


def get_parts_from_header(hstr: str) -> dict:
    hstr = re.sub(r'\s*', '', hstr)
    hdata = dict()
    for chunk in hstr.split(';'):
        parts = chunk.split('=', 1)
        if len(parts) < 2:
            continue
        hdata[parts[0]] = parts[1]
    return hdata


def dkim_canonicalize_header(hname: str, hval: str) -> Tuple[str, str]:
    hname = hname.lower()
    hval = hval.strip()
    hval = re.sub(r'\n', '', hval)
    hval = re.sub(r'\s+', ' ', hval)
    return hname, hval


def verify_attestation_hashes(msg) -> bool:
    hhdr = msg.get(XPH_HDR)
    hdata = get_parts_from_header(str(hhdr))
    adata = dict()
    desc = {
        'i': 'metadata',
        'm': 'commit message',
        'p': 'diff content',
    }
    adata['i'], adata['m'], adata['p'] = get_b64_attestation(msg)
    verified = True
    logger.info('----- ---------------')
    for part, what in desc.items():
        if hdata[part] == adata[part]:
            status = 'PASS'
        else:
            verified = False
            status = 'FAIL'
        logger.info('%s : %s', status, desc[part])
    logger.info('----- ---------------')
    if verified:
        logger.info('PASS : All hashes verified')
    else:
        logger.info('FAIL : Some or all hashes failed verification')

    return verified


def gpg_verify(smsg: bytes, dsig: bytes, keyid: str) -> bool:
    # We can't pass both the detached sig and the content on stdin, so
    # use a temporary file
    savefile = mkstemp('attpoc-pgp-verify')[1]
    with open(savefile, 'wb') as fh:
        fh.write(dsig)
    vrfyargs = ['--verify', '--status-fd=1', savefile, '-']
    # Do we have this key in our default keyring?
    pubring = None
    ecode, out, err = gpg_run_command(['--list-key', keyid])
    if ecode > 0:
        # See if it's in git itself
        gittop = get_git_toplevel()
        if gittop:
            # Do we have that key in git?
            # URLencode keyid to avoid path-based badness
            if os.path.exists(
                    os.path.join(gittop, '.keys', 'openpgp', 'keys', urllib.parse.quote_plus(keyid) + '.asc')):
                pubring = os.path.join(gittop, '.keys', 'openpgp', 'pubring.kbx')
        if pubring and os.path.exists(pubring):
            logger.info('Loading: in-git pubring: %s', pubring)
            gpgargs = ['--no-default-keyring', '--keyring', pubring] + vrfyargs
        else:
            logger.critical('Unable to find key %s', keyid)
            sys.exit(1)
    else:
        gpgargs = vrfyargs
    ecode, out, err = gpg_run_command(gpgargs, stdin=smsg)
    os.unlink(savefile)
    if ecode > 0:
        logger.critical('PGP signature failed to verify')
        logger.critical(err.decode())
        return False
    output = out.decode()
    # We're looking for GOODSIG and VALIDSIG
    # For the purposes of this POC, we're not doing the following, but normally would:
    #   - check UIDs to make sure that they match From:
    #   - check signature date for its drift from Date:
    gs_matches = re.search(r'^\[GNUPG:] GOODSIG ([0-9A-F]+)\s+(.*)$', output, re.M)
    vs_matches = re.search(r'^\[GNUPG:] VALIDSIG ([0-9A-F]+) (\d{4}-\d{2}-\d{2}) (\d+)', output, re.M)
    if gs_matches and vs_matches:
        signer = gs_matches.groups()[1]
        if not pubring:
            # If we're not using an in-git pubring, also check TRUST_* output
            ts_matches = re.search(r'^\[GNUPG:] TRUST_(FULLY|ULTIMATE)', output, re.M)
            if not ts_matches:
                logger.critical('Insufficient TRUST on: %s', signer)
                return False

        logger.info('PGP signature verified: %s', signer)
        return True

    # Theoretically, this would have resulted in ecode > 0
    logger.critical('PGP signature failed to verify')
    logger.critical(err.decode())
    return False


def cmd_hashes_hdr(cmdargs) -> None:
    msg = load_message(cmdargs.message)
    hhdr = gen_hashes_header(msg)
    logger.info('--- HEADER STARTS ---')
    sys.stdout.buffer.write(XPH_HDR.encode() + b': ' + hhdr.encode().encode() + b'\r\n')


def cmd_sign_dkim(cmdargs) -> None:
    logger.info('Signing: plain DKIM')
    msg = load_message(cmdargs.message)
    # We use dkimpy here as a demonstration of an external DKIM compliant
    # implementation generating the DKIM-Signature tag, complete with bh=
    # body hash that we don't consider for our purposes:
    # - it will most certainly get mangled by mailing-list software
    # - if it's canonicalized with "relaxed", we can no longer consider
    #   patch content to be trusted, as "relaxed" canonicalization modifies
    #   whitespace, which allows sneaking in maliciously modified patches
    #   in languages with syntactic whitespace.
    msg = add_hashes_header(msg)
    domain = cmdargs.domain.encode()
    selector = cmdargs.selector.encode()
    identity = cmdargs.identity.encode()
    include_headers = [b'from', b'date', b'x-patch-hashes']

    with open(cmdargs.privkey, 'rb') as fh:
        logger.info('Using %s to sign', cmdargs.privkey)
        privkey = fh.read()

    dk = dkim.DKIM(msg.as_bytes())
    bhdr = dk.sign(selector, domain, privkey, identity=identity, include_headers=include_headers)
    dhdr = bhdr.decode()
    msg['DKIM-Signature'] = dhdr.split(':', 1)[1]
    logger.info('--- MESSAGE STARTS ---')
    sys.stdout.buffer.write(msg.as_bytes())


def cmd_verify_dkim(msg) -> None:
    logger.info('Verifying: Plain DKIM')
    # We don't use dkimpy to verify, as it will force verification of bh=, which
    # we intentionally choose not to use for reasons listed above.
    # However, our implementation is simplistic and doesn't cover many aspects
    # of the DKIM standard, so real-life implementations need to consider using
    # feature-complete DKIM verification tools, assuming they allow ignoring the
    # bh= field.
    dks = msg.get('dkim-signature')
    ddata = get_parts_from_header(dks)
    # Make sure the x-patch-hashes header is included
    included = ddata['h'].split(':')
    if XPH_HDR.lower() not in included:
        logger.critical('%s is not included in signed headers, unable to verify!', XPH_HDR)
        sys.exit(1)
    pk = base64.b64decode(get_dkim_key(ddata['d'], ddata['s']))
    sig = base64.b64decode(ddata['b'])
    headers = list()

    for header in ddata['h'].split(':'):
        # For the POC, we assume 'relaxed/'
        hval = msg.get(header)
        if hval is None:
            # Missing headers are omitted by the DKIM RFC
            continue
        hname, hval = dkim_canonicalize_header(header, str(msg.get(header)))
        headers.append(f'{hname}:{hval}')
    # Now we add the dkim-signature header itself, without b= content
    dname, dval = dkim_canonicalize_header('dkim-signature', dks)
    dval = dval.rsplit('; b=')[0] + '; b='
    headers.append(f'{dname}:{dval}')
    payload = ('\r\n'.join(headers)).encode()
    key = RSA.import_key(pk)
    hashed = SHA256.new(payload)
    try:
        # noinspection PyTypeChecker
        pkcs1_15.new(key).verify(hashed, sig)
    except (ValueError, TypeError):
        logger.critical('FAIL: The DKIM signature did NOT verify!')
        sys.exit(1)
    if not verify_identity_domain(msg, ddata['i'], ddata['d']):
        sys.exit(1)
    if not verify_time_drift(msg, ddata['t']):
        sys.exit(1)
    logger.info('PASS : DKIM signature for d=%s, s=%s', ddata['d'], ddata['s'])
    verify_attestation_hashes(msg)


def cmd_sign_pgp(cmdargs) -> None:
    logger.info('Signing: PGP')
    msg = load_message(cmdargs.message)
    # selector is the key id of the [C] public key, e.g. 0xE63EDCA9329DD07E.
    # identity is the UID we should look for on the key
    # The signature can be made with a subkey, so we embed the key ID into
    # the header for lookup convenience.
    # We don't embed signing time, as it's part of the PGP signature.
    headers = list()
    hhdr = gen_hashes_header(msg)
    msg[XPH_HDR] = hhdr
    hhname, hhval = dkim_canonicalize_header(XPH_HDR, hhdr.encode())
    headers.append(f'{hhname}:{hhval}')

    hparts = [
        'm=pgp',
        f'i={cmdargs.identity}',
        f's=0x{cmdargs.keyid}',
        'b=',
    ]
    shname, shval = dkim_canonicalize_header(XPS_HDR, '; '.join(hparts))
    headers.append(f'{shname}:{shval}')
    payload = '\r\n'.join(headers).encode()
    if cmdargs.subkeyid:
        gpgargs = ['-b', '-u', f'{cmdargs.subkeyid}!']
    else:
        gpgargs = ['-b', '-u', cmdargs.keyid]
    ecode, out, err = gpg_run_command(gpgargs, payload)
    if ecode > 0:
        logger.critical('Running gpg failed')
        logger.critical(err.decode())
        sys.exit(ecode)
    bdata = base64.b64encode(out)
    shval += splitter(bdata.decode())
    shdr = email.header.make_header([(shval.encode(), 'us-ascii')], maxlinelen=78)
    msg[XPS_HDR] = shdr
    logger.info('--- MESSAGE STARTS ---')
    sys.stdout.buffer.write(msg.as_bytes())


def cmd_pgp_verify(msg):
    shdr = msg.get(XPS_HDR)
    sdata = get_parts_from_header(shdr)
    sig = base64.b64decode(sdata['b'])
    headers = list()
    hhname, hhval = dkim_canonicalize_header(XPH_HDR, str(msg.get(XPH_HDR)))
    headers.append(f'{hhname}:{hhval}')
    # Now we add the sig header itself, without b= content
    shname, shval = dkim_canonicalize_header(XPS_HDR, shdr)
    shval = shval.rsplit('; b=')[0] + '; b='
    headers.append(f'{shname}:{shval}')
    payload = ('\r\n'.join(headers)).encode()
    keyid = re.sub(r'^0x', '', sdata['s'])
    if gpg_verify(payload, sig, keyid):
        verify_attestation_hashes(msg)


def cmd_sign_dk(cmdargs, mode='dk'):
    logger.info('Signing: %s header using %s mode', XPS_HDR, mode)
    msg = load_message(cmdargs.message)
    # selector is the same as in DKIM, the leftmost part of foo._domainkey.example.org
    # identity should match domain in from
    headers = list()
    hhdr = gen_hashes_header(msg)
    msg[XPH_HDR] = hhdr
    hhname, hhval = dkim_canonicalize_header(XPH_HDR, hhdr.encode())
    headers.append(f'{hhname}:{hhval}')
    signtime = str(int(time.time()))

    hparts = [
        f'm={mode}',
        f'd={cmdargs.domain}',
        f'i={cmdargs.identity}',
        f's={cmdargs.selector}',
        f't={signtime}',
        'a=ed25519-sha256',
        'b=',
    ]
    shname, shval = dkim_canonicalize_header(XPS_HDR, '; '.join(hparts))
    headers.append(f'{shname}:{shval}')
    payload = '\r\n'.join(headers).encode()
    hashed = hashlib.sha256()
    hashed.update(payload)
    try:
        with open(cmdargs.privkey, 'r') as fh:
            sk = SigningKey(fh.read(), encoder=Base64Encoder)
    except IOError:
        logger.critical('Could not open %s', cmdargs.privkey)
        sys.exit(1)

    bdata = sk.sign(hashed.digest(), encoder=Base64Encoder)
    shval += splitter(bdata.decode())
    shdr = email.header.make_header([(shval.encode(), 'us-ascii')], maxlinelen=78)
    msg[XPS_HDR] = shdr
    logger.info('--- MESSAGE STARTS ---')
    sys.stdout.buffer.write(msg.as_bytes())


def cmd_verify_dk(msg, mode='dk'):
    logger.info('Verifying: %s (mode=%s)', XPS_HDR, mode)
    shdr = msg.get(XPS_HDR)
    sdata = get_parts_from_header(shdr)
    headers = list()
    hhname, hhval = dkim_canonicalize_header(XPH_HDR, str(msg.get(XPH_HDR)))
    headers.append(f'{hhname}:{hhval}')
    # Now we add the sig header itself, without b= content
    shname, shval = dkim_canonicalize_header(XPS_HDR, shdr)
    shval = shval.rsplit('; b=')[0] + '; b='
    headers.append(f'{shname}:{shval}')
    payload = ('\r\n'.join(headers)).encode()
    hashed = hashlib.sha256()
    hashed.update(payload)
    if mode == 'dk':
        pk = get_dkim_key(sdata['d'], sdata['s'], wantkey='ed25519')
    elif mode == 'wk':
        pk = get_wk_key(sdata['d'], sdata['s'])
    elif mode == 'wkd':
        pk = get_wkd_key(sdata['d'], sdata['i'], sdata['s'])
    else:
        logger.critical('Unknown mode: %s', mode)
        sys.exit(1)

    vk = VerifyKey(pk, encoder=Base64Encoder)
    try:
        foo = vk.verify(sdata['b'].encode(), encoder=Base64Encoder)
    except BadSignatureError:
        logger.critical('FAIL : mode=%s signature verification for: d=%s, i=%s, s=%s', mode,
                        sdata['d'], sdata['i'], sdata['s'])
        sys.exit(1)
    if foo != hashed.digest():
        logger.critical('FAIL : mode=%s signature verification for: d=%s, i=%s, s=%s', mode,
                        sdata['d'], sdata['i'], sdata['s'])
        sys.exit(1)

    if not verify_identity_domain(msg, sdata['i'], sdata['d']):
        sys.exit(1)
    if not verify_time_drift(msg, sdata['t']):
        sys.exit(1)
    logger.info('PASS : mode=%s signature verified for: d=%s, i=%s, s=%s', mode,
                sdata['d'], sdata['i'], sdata['s'])
    verify_attestation_hashes(msg)


def cmd_sign_wk(cmdargs):
    cmd_sign_dk(cmdargs, mode='wk')


def cmd_verify_wk(msg):
    cmd_verify_dk(msg, mode='wk')


def cmd_sign_wkd(cmdargs):
    cmd_sign_dk(cmdargs, mode='wkd')


def cmd_verify_wkd(msg):
    cmd_verify_dk(msg, mode='wkd')


def cmd_verify(cmdargs):
    msg = load_message(cmdargs.message)
    # do we have a hashes header?
    if not msg.get(XPH_HDR):
        logger.critical('Message does not contain %s, nothing to verify', XPH_HDR)
        sys.exit(1)
    # do we have a sig header?
    shdr = msg.get(XPS_HDR)
    if not shdr:
        if not msg.get('dkim-signature'):
            logger.critical('Message contains unsigned hashes, cannot verify')
            sys.exit(1)
        # Run as a plain dkim verification
        cmd_verify_dkim(msg)
        sys.exit(0)
    sdata = get_parts_from_header(shdr)
    if sdata['m'] == 'pgp':
        cmd_pgp_verify(msg)
        sys.exit(0)
    if sdata['m'] == 'dk':
        cmd_verify_dk(msg)
        sys.exit(0)
    if sdata['m'] == 'wk':
        cmd_verify_wk(msg)
        sys.exit(0)
    if sdata['m'] == 'wkd':
        cmd_verify_wkd(msg)
        sys.exit(0)

    logger.critical('Unknown mode: %s', sdata['m'])
    sys.exit(1)


if __name__ == '__main__':
    import argparse

    # noinspection PyTypeChecker
    parser = argparse.ArgumentParser(
        prog='main',
        description='A proof of concept tool for header-based email patch attestation',
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    parser.add_argument('-m', '--message', default='emails/unsigned.eml',
                        help='File with the message to work with')
    parser.add_argument('-v', '--verbose', default=False,
                        help='Print extra debugging output')
    subparsers = parser.add_subparsers(help='sub-command help', dest='subcmd')

    # hashes-hdr
    sp_hhdr = subparsers.add_parser('hashes-hdr', help='Create a hashes header')
    sp_hhdr.set_defaults(func=cmd_hashes_hdr)

    # sign-dkim
    sp_sdkim = subparsers.add_parser('sign-dkim', help='Create a DKIM signature')
    sp_sdkim.add_argument('-i', '--identity', default='@example.org')
    sp_sdkim.add_argument('-d', '--domain', default='example.org')
    sp_sdkim.add_argument('-s', '--selector', default='patches')
    sp_sdkim.add_argument('-k', '--privkey', default='rsa.key')
    sp_sdkim.set_defaults(func=cmd_sign_dkim)

    # sign-dk
    sp_sdk = subparsers.add_parser('sign-dk', help='Create a mode=dk signature')
    sp_sdk.add_argument('-i', '--identity', default='@example.org')
    sp_sdk.add_argument('-d', '--domain', default='example.org')
    sp_sdk.add_argument('-s', '--selector', default='patches')
    sp_sdk.add_argument('-k', '--privkey', default='dk.key')
    sp_sdk.set_defaults(func=cmd_sign_dk)

    # sign-wk
    sp_swk = subparsers.add_parser('sign-wk', help='Create a mode=wk signature')
    sp_swk.add_argument('-i', '--identity', default='@example.org')
    sp_swk.add_argument('-d', '--domain', default='example.org')
    sp_swk.add_argument('-s', '--selector', default='patches')
    sp_swk.add_argument('-k', '--privkey', default='dk.key')
    sp_swk.set_defaults(func=cmd_sign_wk)

    # sign-wk
    sp_swkd = subparsers.add_parser('sign-wkd', help='Create a mode=wkd signature')
    sp_swkd.add_argument('-i', '--identity', default='dev@kernel.org')
    sp_swkd.add_argument('-d', '--domain', default='example.org')
    sp_swkd.add_argument('-s', '--selector', default='patches')
    sp_swkd.add_argument('-k', '--privkey', default='ingit.key')
    sp_swkd.set_defaults(func=cmd_sign_wkd)

    # sign-pgp
    sp_spgp = subparsers.add_parser('sign-pgp', help='Create a PGP signature')
    sp_spgp.add_argument('-i', '--identity', default='dev@kernel.org')
    sp_spgp.add_argument('-k', '--keyid', default='AAAABBBBCCCCDDDD')
    sp_spgp.add_argument('-u', '--subkeyid')
    sp_spgp.set_defaults(func=cmd_sign_pgp)

    # verify
    sp_verify = subparsers.add_parser('verify', help='Verify a signed message')
    sp_verify.set_defaults(func=cmd_verify)

    args = parser.parse_args()

    logger.setLevel(logging.DEBUG)
    ch = logging.StreamHandler()
    formatter = logging.Formatter('%(message)s')
    ch.setFormatter(formatter)
    if args.verbose:
        ch.setLevel(logging.DEBUG)
    else:
        ch.setLevel(logging.INFO)
    logger.addHandler(ch)

    if 'func' not in args:
        parser.print_help()
        sys.exit(1)

    args.func(args)