| #!/usr/bin/env python3 |
| import sys |
| import os |
| import base64 |
| import email |
| import email.header |
| import b4 |
| import time |
| import dkim |
| import re |
| import dns.resolver |
| import binascii |
| import subprocess |
| import hashlib |
| import requests |
| import anybase32 |
| import urllib.parse |
| |
| 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 |
| |
| XPH_HDR = 'X-Patch-Hashes' |
| XPS_HDR = 'X-Patch-Sig' |
| |
| |
| def _run_command(cmdargs, stdin=None): |
| 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, stdin=None): |
| gpgbin = 'gpg' |
| cmdargs = [gpgbin, '--batch', '--no-auto-key-retrieve', '--no-auto-check-trustdb'] + cmdargs |
| return _run_command(cmdargs, stdin) |
| |
| |
| def get_dkim_key(domain, selector, wantkey='rsa', timeout=5): |
| name = f'{selector}._domainkey.{domain}.' |
| print(f'DNS-lookup: {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: |
| print(f'Domain {name} does not exist') |
| sys.exit(1) |
| pass |
| |
| if not keydata: |
| print(f'Domain {name} does not contain a DKIM record') |
| sys.exit(1) |
| |
| parts = get_parts_from_header(keydata) |
| if 'p' not in parts: |
| print(f'Domain {name} does not contain a DKIM key') |
| sys.exit(1) |
| |
| return parts['p'] |
| |
| |
| def get_wk_key(domain, selector, timeout=5): |
| wkurl = f'https://{domain}/.well-known/_domainkey/{selector}.txt' |
| print(f'Retrieving: {wkurl}') |
| res = requests.get(wkurl, timeout=timeout) |
| if res.status_code != 200: |
| print(f'Could not get {wkurl}') |
| sys.exit(1) |
| keydata = res.content.decode().strip() |
| parts = get_parts_from_header(keydata) |
| return parts['p'] |
| |
| |
| def get_wkd_key(domain, identity, selector, timeout=5): |
| identity = identity.split('@')[0] |
| # Attempt to load from local git dir if we are in a git dir |
| cmdargs = ['git', 'rev-parse', '--show-toplevel'] |
| ecode, out, err = _run_command(cmdargs) |
| if ecode == 0: |
| # urlencode domain/identity/selector to make sure nobody tries path-based badness |
| subpath = os.path.join(out.decode().strip(), '.devkey', |
| urllib.parse.quote_plus(domain), |
| urllib.parse.quote_plus(identity), |
| urllib.parse.quote_plus(selector) + '.txt', |
| ) |
| try: |
| with open(subpath) as fh: |
| print(f'Loading WKD key from {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 |
| i = hashlib.sha1(identity.lower().encode()).digest() |
| zdir = anybase32.encode(i, anybase32.ZBASE32).decode() |
| wkurl = f'https://{domain}/.well-known/devkey/{zdir}/{selector}.txt' |
| print(f'Retrieving: {wkurl}') |
| res = requests.get(wkurl, timeout=timeout) |
| if res.status_code != 200: |
| print(f'Could not get {wkurl}') |
| 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): |
| # 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): |
| 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, hval): |
| 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): |
| 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) |
| for part, what in desc.items(): |
| if hdata[part] == adata[part]: |
| status = 'PASS' |
| else: |
| status = 'FAIL' |
| print(f'{status} | {desc[part]}') |
| |
| |
| def cmd_hashes_hdr(msg): |
| hhdr = gen_hashes_header(msg) |
| print(hhdr.encode()) |
| |
| |
| def cmd_dkim_sign(msg): |
| # 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) |
| selector = b'example' |
| domain = b'example.org' |
| include_headers = [b'from', b'date', b'x-patch-hashes'] |
| |
| with open('rsa.key', 'rb') as fh: |
| privkey = fh.read() |
| |
| dk = dkim.DKIM(msg.as_bytes()) |
| bhdr = dk.sign(selector, domain, privkey, include_headers=include_headers) |
| dhdr = bhdr.decode() |
| msg['DKIM-Signature'] = dhdr.split(':', 1)[1] |
| print(msg.as_string()) |
| |
| |
| def cmd_dkim_verify(msg): |
| # 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) |
| 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): |
| print('The DKIM signature did NOT verify!') |
| sys.exit(1) |
| print(f'DKIM signature verified for d={ddata["i"]} s={ddata["s"]}') |
| verify_attestation_hashes(msg) |
| |
| |
| def cmd_pgp_sign(msg, identity='mricon@kernel.org', selector='E63EDCA9329DD07E'): |
| # 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) |
| hhname, hhval = dkim_canonicalize_header(XPH_HDR, hhdr.encode()) |
| headers.append(f'{hhname}:{hhval}') |
| |
| hparts = [ |
| 'm=pgp', |
| f'i={identity}', |
| f's=0x{selector}', |
| 'b=', |
| ] |
| shname, shval = dkim_canonicalize_header(XPS_HDR, '; '.join(hparts)) |
| headers.append(f'{shname}:{shval}') |
| payload = '\r\n'.join(headers).encode() |
| gpgargs = ['-b', '-u', selector] |
| ecode, out, err = gpg_run_command(gpgargs, payload) |
| if ecode > 0: |
| print('Running gpg failed') |
| print(err.decode()) |
| sys.exit(ecode) |
| bdata = base64.b64encode(out) |
| shval += bdata.decode() |
| shdr = email.header.make_header([(shval.encode(), 'us-ascii')], maxlinelen=78) |
| msg[XPS_HDR] = shdr |
| print(msg.as_string()) |
| |
| |
| def cmd_pgp_verify(msg): |
| # For the POC purposes, we ignore key lookup and assume it's in the keyring |
| 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() |
| # 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(sig) |
| gpgargs = ['--verify', '--status-fd=1', savefile, '-'] |
| ecode, out, err = gpg_run_command(gpgargs, stdin=payload) |
| os.unlink(savefile) |
| if ecode > 0: |
| print('PGP signature failed to verify') |
| print(err.decode()) |
| sys.exit(1) |
| 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 to make sure its drift from Date: |
| # - check TRUST_ to make sure it's sufficiently trusted key |
| 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: |
| print('PGP signature verified: %s' % gs_matches.groups()[1]) |
| verify_attestation_hashes(msg) |
| |
| |
| def cmd_dk_sign(msg, domain='kernel.org', identity='@kernel.org', selector='patches', keyfile='dk.key', mode='dk'): |
| # 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={domain}', |
| f'i={identity}', |
| f's={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(keyfile, 'r') as fh: |
| sk = SigningKey(fh.read(), encoder=Base64Encoder) |
| except IOError: |
| print('Could not open %s' % keyfile) |
| sys.exit(1) |
| |
| bdata = sk.sign(hashed.digest(), encoder=Base64Encoder) |
| shval += bdata.decode() |
| shdr = email.header.make_header([(shval.encode(), 'us-ascii')], maxlinelen=78) |
| msg[XPS_HDR] = shdr |
| print(msg.as_string()) |
| |
| |
| def cmd_dk_verify(msg, mode='dk'): |
| 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: |
| print(f'Unknown mode: {mode}') |
| sys.exit(1) |
| |
| mode = mode.upper() |
| |
| vk = VerifyKey(pk, encoder=Base64Encoder) |
| try: |
| foo = vk.verify(sdata['b'].encode(), encoder=Base64Encoder) |
| except BadSignatureError: |
| print(f'{mode} signature verification FAILED for: d={sdata["d"]}, i={sdata["i"]}, s={sdata["s"]}') |
| sys.exit(1) |
| if foo != hashed.digest(): |
| print(f'{mode} signature verification FAILED for: d={sdata["d"]}, i={sdata["i"]}, s={sdata["s"]}') |
| sys.exit(1) |
| |
| print(f'{mode} signature verified for: d={sdata["d"]}, i={sdata["i"]}, s={sdata["s"]}') |
| verify_attestation_hashes(msg) |
| |
| |
| def cmd_wk_sign(msg, domain='kernel.org', identity='@kernel.org', selector='patches', keyfile='dk.key'): |
| cmd_dk_sign(msg, domain=domain, identity=identity, selector=selector, keyfile=keyfile, mode='wk') |
| |
| |
| def cmd_wk_verify(msg): |
| cmd_dk_verify(msg, mode='wk') |
| |
| |
| def cmd_wkd_sign(msg, domain='kernel.org', identity='dev@kernel.org', selector='patches', keyfile='ingit.key'): |
| cmd_dk_sign(msg, domain=domain, identity=identity, selector=selector, keyfile=keyfile, mode='wkd') |
| |
| |
| def cmd_wkd_verify(msg): |
| cmd_dk_verify(msg, mode='wkd') |
| |
| |
| def cmd_verify(msg): |
| # do we have a hashes header? |
| if not msg.get(XPH_HDR): |
| print(f'Message does not contain {XPH_HDR}, nothing to verify') |
| sys.exit(1) |
| # do we have a sig header? |
| shdr = msg.get(XPS_HDR) |
| if not shdr: |
| if not msg.get('dkim-signature'): |
| print(f'Message contains unsigned hashes, cannot verify') |
| sys.exit(1) |
| # Run as a plain dkim verification |
| cmd_dkim_verify(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_dk_verify(msg) |
| sys.exit(0) |
| if sdata['m'] == 'wk': |
| cmd_wk_verify(msg) |
| sys.exit(0) |
| if sdata['m'] == 'wkd': |
| cmd_wkd_verify(msg) |
| sys.exit(0) |
| |
| print(f'Unknown mode: {sdata["m"]}') |
| sys.exit(1) |
| |
| |
| if __name__ == '__main__': |
| message = sys.stdin.buffer.read() |
| if not message: |
| print('Pass message on stdin') |
| sys.exit(1) |
| |
| orig_msg = email.message_from_bytes(message) |
| |
| commands = { |
| 'hashes-hdr': cmd_hashes_hdr, |
| 'dkim-sign': cmd_dkim_sign, |
| 'pgp-sign': cmd_pgp_sign, |
| 'dk-sign': cmd_dk_sign, |
| 'wk-sign': cmd_wk_sign, |
| 'wkd-sign': cmd_wkd_sign, |
| 'verify': cmd_verify, |
| } |
| |
| command = None |
| if len(sys.argv) > 1: |
| command = sys.argv[1] |
| |
| if command not in commands: |
| print(f'Unknown command: {command}') |
| print('Known commands:') |
| for goodcmd in commands.keys(): |
| print(f' {goodcmd}') |
| sys.exit(1) |
| |
| execfunc = commands[command] |
| execfunc(orig_msg) |
