Cleanups and extra checks Signed-off-by: Konstantin Ryabitsev <konstantin@linuxfoundation.org>
diff --git a/.gitignore b/.gitignore index 9769964..74574e2 100644 --- a/.gitignore +++ b/.gitignore
@@ -1,3 +1,4 @@ *.pyc __pycache__ .idea +*~
diff --git a/.devkey/kernel.org/dev/patches.txt b/.keys/devkey/kernel.org/dev/patches.txt similarity index 100% rename from .devkey/kernel.org/dev/patches.txt rename to .keys/devkey/kernel.org/dev/patches.txt
diff --git a/.keys/openpgp/graphs/E63EDCA9329DD07E.svg b/.keys/openpgp/graphs/E63EDCA9329DD07E.svg new file mode 100644 index 0000000..1c50c80 --- /dev/null +++ b/.keys/openpgp/graphs/E63EDCA9329DD07E.svg
@@ -0,0 +1,45 @@ +<?xml version="1.0" encoding="UTF-8" standalone="no"?> +<!DOCTYPE svg PUBLIC "-//W3C//DTD SVG 1.1//EN" + "http://www.w3.org/Graphics/SVG/1.1/DTD/svg11.dtd"> +<!-- Generated by graphviz version 2.40.1 (0) + --> +<!-- Title: G Pages: 1 --> +<svg width="216pt" height="166pt" + viewBox="0.00 0.00 216.00 166.00" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink"> +<g id="graph0" class="graph" transform="scale(1 1) rotate(0) translate(4 162)"> +<title>G</title> +<polygon fill="#ffffff" stroke="transparent" points="-4,4 -4,-162 212,-162 212,4 -4,4"/> +<g id="clust1" class="cluster"> +<title>cluster_toplevel</title> +<polygon fill="none" stroke="#ffffff" points="8,-81 8,-150 200,-150 200,-81 8,-81"/> +</g> +<!-- a_387 --> +<g id="node1" class="node"> +<title>a_387</title> +<path fill="none" stroke="#c0c0c0" d="M25,-.5C25,-.5 183,-.5 183,-.5 189,-.5 195,-6.5 195,-12.5 195,-12.5 195,-40.5 195,-40.5 195,-46.5 189,-52.5 183,-52.5 183,-52.5 25,-52.5 25,-52.5 19,-52.5 13,-46.5 13,-40.5 13,-40.5 13,-12.5 13,-12.5 13,-6.5 19,-.5 25,-.5"/> +<text text-anchor="middle" x="104" y="-39.7" font-family="droid sans,dejavu sans,helvetica" font-size="11.00" fill="#000000">Konstantin Ryabitsev</text> +<text text-anchor="middle" x="104" y="-27.7" font-family="droid sans,dejavu sans,helvetica" font-size="11.00" fill="#000000">linuxfoundation.org</text> +<polyline fill="none" stroke="#c0c0c0" points="13,-20.5 195,-20.5 "/> +<text text-anchor="middle" x="44.5" y="-7.7" font-family="droid sans,dejavu sans,helvetica" font-size="11.00" fill="#000000">RSA 4096</text> +<polyline fill="none" stroke="#c0c0c0" points="76,-.5 76,-20.5 "/> +<text text-anchor="middle" x="135.5" y="-7.7" font-family="droid sans,dejavu sans,helvetica" font-size="11.00" fill="#000000">E63EDCA9329DD07E</text> +</g> +<!-- a_211 --> +<g id="node2" class="node"> +<title>a_211</title> +<path fill="none" stroke="#c0c0c0" d="M28,-89.5C28,-89.5 180,-89.5 180,-89.5 186,-89.5 192,-95.5 192,-101.5 192,-101.5 192,-129.5 192,-129.5 192,-135.5 186,-141.5 180,-141.5 180,-141.5 28,-141.5 28,-141.5 22,-141.5 16,-135.5 16,-129.5 16,-129.5 16,-101.5 16,-101.5 16,-95.5 22,-89.5 28,-89.5"/> +<text text-anchor="middle" x="104" y="-128.7" font-family="droid sans,dejavu sans,helvetica" font-size="11.00" fill="#000000">Linus Torvalds</text> +<text text-anchor="middle" x="104" y="-116.7" font-family="droid sans,dejavu sans,helvetica" font-size="11.00" fill="#000000">kernel.org</text> +<polyline fill="none" stroke="#c0c0c0" points="16,-109.5 192,-109.5 "/> +<text text-anchor="middle" x="47.5" y="-96.7" font-family="droid sans,dejavu sans,helvetica" font-size="11.00" fill="#000000">RSA 2048</text> +<polyline fill="none" stroke="#c0c0c0" points="79,-89.5 79,-109.5 "/> +<text text-anchor="middle" x="135.5" y="-96.7" font-family="droid sans,dejavu sans,helvetica" font-size="11.00" fill="#000000">79BE3E4300411886</text> +</g> +<!-- a_211->a_387 --> +<g id="edge1" class="edge"> +<title>a_211->a_387</title> +<path fill="none" stroke="#000000" d="M104,-89.2784C104,-80.9495 104,-71.5865 104,-62.6839"/> +<polygon fill="#000000" stroke="#000000" points="107.5001,-62.5341 104,-52.5341 100.5001,-62.5342 107.5001,-62.5341"/> +</g> +</g> +</svg>
diff --git a/mricon.asc b/.keys/openpgp/keys/E63EDCA9329DD07E.asc similarity index 100% rename from mricon.asc rename to .keys/openpgp/keys/E63EDCA9329DD07E.asc
diff --git a/.keys/openpgp/pubring.kbx b/.keys/openpgp/pubring.kbx new file mode 100644 index 0000000..2a7c5ef --- /dev/null +++ b/.keys/openpgp/pubring.kbx Binary files differ
diff --git a/emails/mricon-signed-pgp-commit-message-modified.eml b/emails/mricon-signed-pgp-commit-message-modified.eml new file mode 100644 index 0000000..7a5c824 --- /dev/null +++ b/emails/mricon-signed-pgp-commit-message-modified.eml
@@ -0,0 +1,55 @@ +Return-Path: <SRS0=87nF=CT=vger.kernel.org=linux-kernel-owner@kernel.org> +From: Konstantin Ryabitsev <mricon@kernel.org> +To: <linux-kernel@vger.kernel.org> +Subject: [PATCH] drm: xlnx: remove defined but not used 'scaling_factors_666' +Date: Thu, 10 Sep 2020 22:06:30 +0800 +Message-ID: <patch-attestation-examples@kernel.org> +X-Mailer: git-send-email 2.25.4 +X-Patch-Hashes: v=1; h=sha256; + i=pkD5Pg8+cndZAzQQzo3RBSOOUzZM3GYWxiFIKFGIKe0=; + m=yW4TvC/DGWCUJTa11Aw1b/2ZAXobsLD45aLA/440yQI=; + p=iJdYN6+isP/3HmQaf1IiG7OfA1vzRxXlPGZtvecS484= +X-Patch-Sig: m=pgp; i=mricon@kernel.org; s=0xE63EDCA9329DD07E; + b=iHUEABYIAB0WIQR2vl2yUnHhSB5njDW2xBzjVmSZbAUCX1+/nQAKCRC2xBzjVmSZbFiQAQD42c + l5It3AVJbtkwbY5XZxb9I9YuvvX3L3buU+EwjumwD9HBH8t6xcavIKQF6dwKjsmhwJnDj1tCfaxg + 3WRdUllgM= +MIME-Version: 1.0 +Content-Type: text/plain; charset="UTF-8" +Content-Transfer-Encoding: 8bit + +From: Dev Eloper <dev@kernel.org> + +This addresses the following gcc warning with "make W=1": +xxx + +drivers/gpu/drm/xlnx/zynqmp_disp.c:245:18: warning: +‘scaling_factors_666’ defined but not used [-Wunused-const-variable=] + 245 | static const u32 scaling_factors_666[] = { + | ^~~~~~~~~~~~~~~~~~~ + +Signed-off-by: Dev Eloper <dev@kernel.org> +--- + drivers/gpu/drm/xlnx/zynqmp_disp.c | 6 ------ + 1 file changed, 6 deletions(-) + +diff --git a/drivers/gpu/drm/xlnx/zynqmp_disp.c b/drivers/gpu/drm/xlnx/zynqmp_disp.c +index a455cfc1bee5..98bd48f13fd1 100644 +--- a/drivers/gpu/drm/xlnx/zynqmp_disp.c ++++ b/drivers/gpu/drm/xlnx/zynqmp_disp.c +@@ -242,12 +242,6 @@ static const u32 scaling_factors_565[] = { + ZYNQMP_DISP_AV_BUF_5BIT_SF, + }; + +-static const u32 scaling_factors_666[] = { +- ZYNQMP_DISP_AV_BUF_6BIT_SF, +- ZYNQMP_DISP_AV_BUF_6BIT_SF, +- ZYNQMP_DISP_AV_BUF_6BIT_SF, +-}; +- + static const u32 scaling_factors_888[] = { + ZYNQMP_DISP_AV_BUF_8BIT_SF, + ZYNQMP_DISP_AV_BUF_8BIT_SF, +-- +2.25.4 + +
diff --git a/emails/mricon-signed-pgp-patch-modified.eml b/emails/mricon-signed-pgp-patch-modified.eml new file mode 100644 index 0000000..e3a4c19 --- /dev/null +++ b/emails/mricon-signed-pgp-patch-modified.eml
@@ -0,0 +1,54 @@ +Return-Path: <SRS0=87nF=CT=vger.kernel.org=linux-kernel-owner@kernel.org> +From: Konstantin Ryabitsev <mricon@kernel.org> +To: <linux-kernel@vger.kernel.org> +Subject: [PATCH] drm: xlnx: remove defined but not used 'scaling_factors_666' +Date: Thu, 10 Sep 2020 22:06:30 +0800 +Message-ID: <patch-attestation-examples@kernel.org> +X-Mailer: git-send-email 2.25.4 +X-Patch-Hashes: v=1; h=sha256; + i=pkD5Pg8+cndZAzQQzo3RBSOOUzZM3GYWxiFIKFGIKe0=; + m=yW4TvC/DGWCUJTa11Aw1b/2ZAXobsLD45aLA/440yQI=; + p=iJdYN6+isP/3HmQaf1IiG7OfA1vzRxXlPGZtvecS484= +X-Patch-Sig: m=pgp; i=mricon@kernel.org; s=0xE63EDCA9329DD07E; + b=iHUEABYIAB0WIQR2vl2yUnHhSB5njDW2xBzjVmSZbAUCX1+/nQAKCRC2xBzjVmSZbFiQAQD42c + l5It3AVJbtkwbY5XZxb9I9YuvvX3L3buU+EwjumwD9HBH8t6xcavIKQF6dwKjsmhwJnDj1tCfaxg + 3WRdUllgM= +MIME-Version: 1.0 +Content-Type: text/plain; charset="UTF-8" +Content-Transfer-Encoding: 8bit + +From: Dev Eloper <dev@kernel.org> + +This addresses the following gcc warning with "make W=1": + +drivers/gpu/drm/xlnx/zynqmp_disp.c:245:18: warning: +‘scaling_factors_666’ defined but not used [-Wunused-const-variable=] + 245 | static const u32 scaling_factors_666[] = { + | ^~~~~~~~~~~~~~~~~~~ + +Signed-off-by: Dev Eloper <dev@kernel.org> +--- + drivers/gpu/drm/xlnx/zynqmp_disp.c | 6 ------ + 1 file changed, 6 deletions(-) + +diff --git a/drivers/gpu/drm/xlnx/zynqmp_disp.c b/drivers/gpu/drm/xlnx/zynqmp_disp.c +index a455cfc1bee5..98bd48f13fd1 100644 +--- a/drivers/gpu/drm/xlnx/zynqmp_disp.c ++++ b/drivers/gpu/drm/xlnx/zynqmp_disp.c +@@ -242,12 +242,6 @@ static const u32 scaling_factors_565[] = { + ZYNQMP_DISP_AV_BUF_5BIT_SF, xxx + }; + +-static const u32 scaling_factors_666[] = { +- ZYNQMP_DISP_AV_BUF_6BIT_SF, +- ZYNQMP_DISP_AV_BUF_6BIT_SF, +- ZYNQMP_DISP_AV_BUF_6BIT_SF, +-}; +- + static const u32 scaling_factors_888[] = { + ZYNQMP_DISP_AV_BUF_8BIT_SF, + ZYNQMP_DISP_AV_BUF_8BIT_SF, +-- +2.25.4 + +
diff --git a/main.py b/main.py index 4f84adb..2807be0 100755 --- a/main.py +++ b/main.py
@@ -4,6 +4,7 @@ import base64 import email import email.header +import email.utils import b4 import time import dkim @@ -15,6 +16,7 @@ import requests import anybase32 import urllib.parse +import logging from nacl.signing import SigningKey, VerifyKey from nacl.encoding import Base64Encoder @@ -26,25 +28,68 @@ from Cryptodome.Hash import SHA256 from Cryptodome.PublicKey import RSA +from typing import Tuple + XPH_HDR = 'X-Patch-Hashes' XPS_HDR = 'X-Patch-Sig' +logger = logging.getLogger(__name__) -def _run_command(cmdargs, stdin=None): + +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, stdin=None): +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 get_dkim_key(domain, selector, wantkey='rsa', timeout=5): +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 get_dkim_key(domain: str, selector: str, wantkey: str = 'rsa', timeout: int = 5) -> str: name = f'{selector}._domainkey.{domain}.' - print(f'DNS-lookup: {name}') + logger.info('DNS-lookup: %s', name) keydata = None try: a = dns.resolver.resolve(name, dns.rdatatype.TXT, raise_on_no_answer=False, lifetime=timeout) @@ -62,69 +107,81 @@ if keydata: break except dns.resolver.NXDOMAIN: - print(f'Domain {name} does not exist') + logger.critical('Domain %s does not exist', name) sys.exit(1) pass if not keydata: - print(f'Domain {name} does not contain a DKIM record') + 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: - print(f'Domain {name} does not contain a DKIM key') + logger.critical('Domain %s does not contain a DKIM key', name) sys.exit(1) return parts['p'] -def get_wk_key(domain, selector, timeout=5): +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' - print(f'Retrieving: {wkurl}') + logger.info('Retrieving: %s', wkurl) res = requests.get(wkurl, timeout=timeout) if res.status_code != 200: - print(f'Could not get {wkurl}') + 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, identity, selector, timeout=5): +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 - cmdargs = ['git', 'rev-parse', '--show-toplevel'] - ecode, out, err = _run_command(cmdargs) - if ecode == 0: + gittop = get_git_toplevel() + if gittop: # urlencode domain/identity/selector to make sure nobody tries path-based badness - subpath = os.path.join(out.decode().strip(), '.devkey', + 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: - print(f'Loading WKD key from {subpath}') + 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 + # 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' - print(f'Retrieving: {wkurl}') + logger.info('Retrieving: %s', wkurl) res = requests.get(wkurl, timeout=timeout) if res.status_code != 200: - print(f'Could not get {wkurl}') + 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): +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 @@ -158,7 +215,7 @@ return hhdr -def get_parts_from_header(hstr): +def get_parts_from_header(hstr: str) -> dict: hstr = re.sub(r'\s*', '', hstr) hdata = dict() for chunk in hstr.split(';'): @@ -169,7 +226,7 @@ return hdata -def dkim_canonicalize_header(hname, hval): +def dkim_canonicalize_header(hname: str, hval: str) -> Tuple[str, str]: hname = hname.lower() hval = hval.strip() hval = re.sub(r'\n', '', hval) @@ -177,7 +234,7 @@ return hname, hval -def verify_attestation_hashes(msg): +def verify_attestation_hashes(msg) -> bool: hhdr = msg.get(XPH_HDR) hdata = get_parts_from_header(str(hhdr)) adata = dict() @@ -187,20 +244,92 @@ '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' - print(f'{status} | {desc[part]}') + 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 cmd_hashes_hdr(msg): +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) - print(hhdr.encode()) + logger.info('--- HEADER STARTS ---') + sys.stdout.buffer.write(XPH_HDR.encode() + b': ' + hhdr.encode().encode() + b'\r\n') -def cmd_dkim_sign(msg): +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: @@ -210,21 +339,25 @@ # 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' + domain = cmdargs.domain.encode() + selector = cmdargs.selector.encode() + identity = cmdargs.identity.encode() include_headers = [b'from', b'date', b'x-patch-hashes'] - with open('rsa.key', 'rb') as fh: + 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, include_headers=include_headers) + bhdr = dk.sign(selector, domain, privkey, identity=identity, include_headers=include_headers) dhdr = bhdr.decode() msg['DKIM-Signature'] = dhdr.split(':', 1)[1] - print(msg.as_string()) + logger.info('--- MESSAGE STARTS ---') + sys.stdout.buffer.write(msg.as_bytes()) -def cmd_dkim_verify(msg): +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 @@ -233,6 +366,11 @@ # 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() @@ -256,13 +394,17 @@ # noinspection PyTypeChecker pkcs1_15.new(key).verify(hashed, sig) except (ValueError, TypeError): - print('The DKIM signature did NOT verify!') + logger.critical('FAIL: The DKIM signature did NOT verify!') sys.exit(1) - print(f'DKIM signature verified for d={ddata["i"]} s={ddata["s"]}') + if not verify_identity_domain(msg, ddata['i'], ddata['d']): + sys.exit(1) + logger.info('PASS : DKIM signature for d=%s, s=%s', ddata['d'], ddata['s']) verify_attestation_hashes(msg) -def cmd_pgp_sign(msg, identity='mricon@kernel.org', selector='E63EDCA9329DD07E'): +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 @@ -270,33 +412,37 @@ # 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={identity}', - f's=0x{selector}', + 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() - gpgargs = ['-b', '-u', selector] + 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: - print('Running gpg failed') - print(err.decode()) + logger.critical('Running gpg failed') + logger.critical(err.decode()) sys.exit(ecode) bdata = base64.b64encode(out) - shval += bdata.decode() + shval += splitter(bdata.decode()) shdr = email.header.make_header([(shval.encode(), 'us-ascii')], maxlinelen=78) msg[XPS_HDR] = shdr - print(msg.as_string()) + logger.info('--- MESSAGE STARTS ---') + sys.stdout.buffer.write(msg.as_bytes()) 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']) @@ -308,32 +454,14 @@ 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) + keyid = re.sub(r'^0x', '', sdata['s']) + if gpg_verify(payload, sig, keyid): + verify_attestation_hashes(msg) -def cmd_dk_sign(msg, domain='kernel.org', identity='@kernel.org', selector='patches', keyfile='dk.key', mode='dk'): +def cmd_sign_dk(cmdargs, mode='dk'): + logger.info('Signing: plain DKIM') + 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() @@ -345,9 +473,9 @@ hparts = [ f'm={mode}', - f'd={domain}', - f'i={identity}', - f's={selector}', + f'd={cmdargs.domain}', + f'i={cmdargs.identity}', + f's={cmdargs.selector}', f't={signtime}', 'a=ed25519-sha256', 'b=', @@ -358,20 +486,22 @@ hashed = hashlib.sha256() hashed.update(payload) try: - with open(keyfile, 'r') as fh: + with open(cmdargs.privkey, 'r') as fh: sk = SigningKey(fh.read(), encoder=Base64Encoder) except IOError: - print('Could not open %s' % keyfile) + logger.critical('Could not open %s', cmdargs.privkey) sys.exit(1) bdata = sk.sign(hashed.digest(), encoder=Base64Encoder) - shval += bdata.decode() + shval += splitter(bdata.decode()) shdr = email.header.make_header([(shval.encode(), 'us-ascii')], maxlinelen=78) msg[XPS_HDR] = shdr - print(msg.as_string()) + logger.info('--- MESSAGE STARTS ---') + sys.stdout.buffer.write(msg.as_bytes()) -def cmd_dk_verify(msg, mode='dk'): +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() @@ -391,101 +521,153 @@ elif mode == 'wkd': pk = get_wkd_key(sdata['d'], sdata['i'], sdata['s']) else: - print(f'Unknown mode: {mode}') + logger.critical('Unknown mode: %s', 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"]}') + 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(): - print(f'{mode} signature verification FAILED for: d={sdata["d"]}, i={sdata["i"]}, s={sdata["s"]}') + logger.critical('FAIL : mode=%s signature verification for: d=%s, i=%s, s=%s', mode, + sdata['d'], sdata['i'], sdata['s']) sys.exit(1) - print(f'{mode} signature verified for: d={sdata["d"]}, i={sdata["i"]}, s={sdata["s"]}') + if not verify_identity_domain(msg, sdata['i'], sdata['d']): + 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_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_sign_wk(cmdargs): + cmd_sign_dk(cmdargs, mode='wk') -def cmd_wk_verify(msg): - cmd_dk_verify(msg, mode='wk') +def cmd_verify_wk(msg): + cmd_verify_dk(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_sign_wkd(cmdargs): + cmd_sign_dk(cmdargs, mode='wkd') -def cmd_wkd_verify(msg): - cmd_dk_verify(msg, mode='wkd') +def cmd_verify_wkd(msg): + cmd_verify_dk(msg, mode='wkd') -def cmd_verify(msg): +def cmd_verify(cmdargs): + msg = load_message(cmdargs.message) # do we have a hashes header? if not msg.get(XPH_HDR): - print(f'Message does not contain {XPH_HDR}, nothing to verify') + 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'): - print(f'Message contains unsigned hashes, cannot verify') + logger.critical('Message contains unsigned hashes, cannot verify') sys.exit(1) # Run as a plain dkim verification - cmd_dkim_verify(msg) + 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_dk_verify(msg) + cmd_verify_dk(msg) sys.exit(0) if sdata['m'] == 'wk': - cmd_wk_verify(msg) + cmd_verify_wk(msg) sys.exit(0) if sdata['m'] == 'wkd': - cmd_wkd_verify(msg) + cmd_verify_wkd(msg) sys.exit(0) - print(f'Unknown mode: {sdata["m"]}') + logger.critical('Unknown mode: %s', sdata['m']) sys.exit(1) if __name__ == '__main__': - message = sys.stdin.buffer.read() - if not message: - print('Pass message on stdin') + 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) - 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) + args.func(args)
diff --git a/verify-all.sh b/verify-all.sh index 3e8595a..cbd3acf 100755 --- a/verify-all.sh +++ b/verify-all.sh
@@ -5,23 +5,38 @@ # source .venv/bin/activate # pip install -r requirements.txt echo "--------- UNHASHED (expected to fail) ---------" -./main.py verify < emails/unsigned.eml +./main.py -m emails/unsigned.eml verify +echo echo "--------- UNSIGNED (expected to fail) ---------" -./main.py verify < emails/unsigned-with-hashes.eml +./main.py -m emails/unsigned-with-hashes.eml verify +echo echo "--------- DKIM-SIGNED (all PASS) ---------" -./main.py verify < emails/korg-signed-dkim.eml -echo "--------- DKIM-SIGNED (with ML junk) ---------" -./main.py verify < emails/korg-signed-dkim-with-ml-junk.eml +./main.py -m emails/korg-signed-dkim.eml verify +echo +echo "--------- DKIM-SIGNED (with ML junk, all PASS) ---------" +./main.py -m emails/korg-signed-dkim-with-ml-junk.eml verify +echo echo "--------- DKIM-SIGNED (with edited subject, metadata will FAIL) ---------" -./main.py verify < emails/korg-signed-dkim-changed-subject.eml +./main.py -m emails/korg-signed-dkim-changed-subject.eml verify +echo echo "--------- DK-SIGNED (all PASS) ---------" -./main.py verify < emails/korg-signed-dk.eml +./main.py -m emails/korg-signed-dk.eml verify +echo echo "--------- WK-SIGNED (all PASS) ---------" -./main.py verify < emails/korg-signed-wk.eml +./main.py -m emails/korg-signed-wk.eml verify +echo echo "--------- PGP-SIGNED (by mricon, all PASS) ---------" -# This assumes you've imported mricon.asc into your keyring -./main.py verify < emails/mricon-signed-pgp.eml +./main.py -m emails/mricon-signed-pgp.eml verify +echo +echo "--------- PGP-SIGNED (by mricon, commit message will FAIL) ---------" +./main.py -m emails/mricon-signed-pgp-commit-message-modified.eml verify +echo +echo "--------- PGP-SIGNED (by mricon, diff content will FAIL) ---------" +./main.py -m emails/mricon-signed-pgp-patch-modified.eml verify +echo echo "--------- WKD-SIGNED (by mricon, web lookup, all PASS) ---------" -./main.py verify < emails/mricon-signed-wkd.eml +./main.py -m emails/mricon-signed-wkd.eml verify +echo echo "--------- WKD-SIGNED (by dev, stored in git, all PASS) ---------" -./main.py verify < emails/dev-signed-wkd-ingit.eml +./main.py -m emails/dev-signed-wkd-ingit.eml verify +echo
