diff options
| -rw-r--r-- | b4/__init__.py | 115 | ||||
| -rw-r--r-- | b4/attest.py | 2 | ||||
| -rw-r--r-- | b4/pr.py | 56 |
3 files changed, 122 insertions, 51 deletions
diff --git a/b4/__init__.py b/b4/__init__.py index b4ce2d4..41f8dbc 100644 --- a/b4/__init__.py +++ b/b4/__init__.py @@ -437,7 +437,7 @@ class LoreSeries: attpolicy = 'softfail' # Make sure it's not too old compared to the message date # Timezone doesn't matter as we calculate whole days - tdelta = lmsg.date.replace(tzinfo=None) - attdoc.sigdate + tdelta = lmsg.date.replace(tzinfo=None) - attdoc.lsig.sigdate if tdelta.days > attstaled: # Uh-oh, attestation is too old! logger.info(' %s %s', attfail, lmsg.full_subject) @@ -445,7 +445,7 @@ class LoreSeries: attstaled, tdelta.days)) else: logger.info(' %s %s', attpass, lmsg.full_subject) - attdata[at-1] = attdoc.attestor.get_trailer(lmsg.fromemail) + attdata[at-1] = attdoc.lsig.attestor.get_trailer(lmsg.fromemail) else: logger.info(' %s', lmsg.full_subject) @@ -876,7 +876,7 @@ class LoreMessage: # We return the first hit return attdoc # Does this doc have an exact match? - uid = attdoc.attestor.get_matching_uid(self.fromemail) + uid = attdoc.lsig.attestor.get_matching_uid(self.fromemail) if uid[1] == self.fromemail: return attdoc # stick an error in the first available attdoc saying @@ -1022,15 +1022,63 @@ class LoreAttestor: return '\n'.join(out) -class LoreAttestationDocument: - def __init__(self, source, sigdata): - self.source = source +class LoreAttestationSignature: + def __init__(self, output, trustmodel): self.good = False self.valid = False self.trusted = False self.sigdate = None - self.passing = False self.attestor = None + self.errors = set() + + gs_matches = re.search(r'^\[GNUPG:\] GOODSIG ([0-9A-F]+)\s+.*$', output, re.M) + if gs_matches: + logger.debug(' GOODSIG') + self.good = True + keyid = gs_matches.groups()[0] + self.attestor = LoreAttestor(keyid) + puid = '%s <%s>' % self.attestor.get_primary_uid() + vs_matches = re.search(r'^\[GNUPG:\] VALIDSIG ([0-9A-F]+) (\d{4}-\d{2}-\d{2}) (\d+)', output, re.M) + if vs_matches: + logger.debug(' VALIDSIG') + self.valid = True + ymd = vs_matches.groups()[1] + self.sigdate = datetime.datetime.strptime(ymd, '%Y-%m-%d') + # Do we have a TRUST_(FULLY|ULTIMATE)? + ts_matches = re.search(r'^\[GNUPG:\] TRUST_(FULLY|ULTIMATE)', output, re.M) + if ts_matches: + logger.debug(' TRUST_%s', ts_matches.groups()[0]) + self.trusted = True + else: + self.errors.add('Insufficient trust (model=%s): %s (%s)' + % (trustmodel, keyid, puid)) + else: + self.errors.add('Signature not valid from key: %s (%s)' % (keyid, puid)) + else: + # Are we missing a key? + matches = re.search(r'^\[GNUPG:\] NO_PUBKEY ([0-9A-F]+)$', output, re.M) + if matches: + self.errors.add('Missing public key: %s' % matches.groups()[0]) + + def __repr__(self): + out = list() + out.append(' good: %s' % self.good) + out.append(' valid: %s' % self.valid) + out.append(' trusted: %s' % self.trusted) + if self.attestor is not None: + out.append(' attestor: %s' % self.attestor.keyid) + + out.append(' --- validation errors ---') + for error in self.errors: + out.append(' | %s' % error) + return '\n'.join(out) + + +class LoreAttestationDocument: + def __init__(self, source, sigdata): + self.source = source + self.lsig = None + self.passing = False self.hashes = set() self.errors = set() @@ -1041,42 +1089,14 @@ class LoreAttestationDocument: logger.debug('Validating document obtained from %s', self.source) ecode, output = gpg_run_command(gpgargs, stdin=sigdata.encode('utf-8')) - if ecode == 0: - # We're looking for both GOODSIG and VALIDSIG - gs_matches = re.search(r'^\[GNUPG:\] GOODSIG ([0-9A-F]+)\s+.*$', output, re.M) - if gs_matches: - logger.debug(' GOODSIG') - self.good = True - keyid = gs_matches.groups()[0] - self.attestor = LoreAttestor(keyid) - puid = '%s <%s>' % self.attestor.get_primary_uid() - vs_matches = re.search(r'^\[GNUPG:\] VALIDSIG ([0-9A-F]+) (\d{4}-\d{2}-\d{2}) (\d+)', output, re.M) - if vs_matches: - logger.debug(' VALIDSIG') - self.valid = True - ymd = vs_matches.groups()[1] - self.sigdate = datetime.datetime.strptime(ymd, '%Y-%m-%d') - # Do we have a TRUST_(FULLY|ULTIMATE)? - ts_matches = re.search(r'^\[GNUPG:\] TRUST_(FULLY|ULTIMATE)', output, re.M) - if ts_matches: - logger.debug(' TRUST_%s', ts_matches.groups()[0]) - self.trusted = True - else: - self.errors.add('Insufficient trust (model=%s): %s (%s)' - % (config['attestation-trust-model'], keyid, puid)) - else: - self.errors.add('Signature not valid from key: %s (%s)' % (keyid, puid)) - else: - # Are we missing a key? - matches = re.search(r'^\[GNUPG:\] NO_PUBKEY ([0-9A-F]+)$', output, re.M) - if matches: - self.errors.add('Missing public key: %s' % matches.groups()[0]) - else: - logger.debug('NOGOOD: Signature on %s failed to verify', self.source) - return + self.lsig = LoreAttestationSignature(output, config['attestation-trust-model']) + self.errors.update(self.lsig.errors) - if self.good and self.valid and self.trusted: + if self.lsig.good and self.lsig.valid and self.lsig.trusted: self.passing = True + else: + # Not going any further + return if source.find('http') == 0: # We only cache known-good attestations obtained from remote @@ -1109,19 +1129,14 @@ class LoreAttestationDocument: def __repr__(self): out = list() out.append(' source: %s' % self.source) - out.append(' good: %s' % self.good) - out.append(' valid: %s' % self.valid) - out.append(' trusted: %s' % self.trusted) - if self.attestor is not None: - out.append(' attestor: %s' % self.attestor.keyid) - out.append(' --- validation errors ---') for error in self.errors: out.append(' | %s' % error) out.append(' --- hashes ---') for hg in self.hashes: out.append(' | %s-%s-%s' % (hg[0][:8], hg[1][:8], hg[2][:8])) - return '\n'.join(out) + ret = '\n'.join(out) + '\n' + str(self.lsig) + return ret @staticmethod def get_from_cache(attid): @@ -1249,7 +1264,9 @@ def _run_command(cmdargs, stdin=None, logstderr=False): output = output.decode('utf-8', errors='replace') if logstderr and len(error.strip()): - logger.debug('Stderr: %s', error.decode('utf-8', errors='replace')) + errout = error.decode('utf-8', errors='replace') + logger.debug('Stderr: %s', errout) + output += errout return sp.returncode, output diff --git a/b4/attest.py b/b4/attest.py index 35c062b..5f9a2b4 100644 --- a/b4/attest.py +++ b/b4/attest.py @@ -161,7 +161,7 @@ def verify_attestation(cmdargs): if ecode != 128: ecode = 0 logger.critical('%s %s', attpass, lmsg.full_subject) - attrailers.add(attdoc.attestor.get_trailer(lmsg.fromemail)) + attrailers.add(attdoc.lsig.attestor.get_trailer(lmsg.fromemail)) logger.critical('---') if ecode > 0: @@ -118,6 +118,55 @@ def parse_pr_data(msg): return lmsg +def attest_fetch_head(gitdir, lmsg): + config = b4.get_main_config() + attpolicy = config['attestation-policy'] + if config['attestation-checkmarks'] == 'fancy': + attpass = b4.PASS_FANCY + attfail = b4.FAIL_FANCY + else: + attpass = b4.PASS_SIMPLE + attfail = b4.FAIL_SIMPLE + # Is FETCH_HEAD a tag or a commit? + htype = b4.git_get_command_lines(gitdir, ['cat-file', '-t', 'FETCH_HEAD']) + lsig = None + passing = False + out = '' + otype = 'unknown' + if len(htype): + otype = htype[0] + if otype == 'tag': + ecode, out = b4.git_run_command(gitdir, ['verify-tag', '--raw', 'FETCH_HEAD'], logstderr=True) + elif otype == 'commit': + ecode, out = b4.git_run_command(gitdir, ['verify-commit', '--raw', 'FETCH_HEAD'], logstderr=True) + lsig = b4.LoreAttestationSignature(out, 'git') + if lsig.good and lsig.valid and lsig.trusted: + passing = True + + out = out.strip() + if not len(out) and attpolicy != 'check': + lsig.errors.add('Remote %s is not signed!' % otype) + + if passing: + trailer = lsig.attestor.get_trailer(lmsg.fromemail) + logger.info(' ---') + logger.info(' %s %s', attpass, trailer) + return + + if lsig.errors: + logger.critical(' ---') + if len(out): + logger.critical(' Pull request is signed, but verification did not succeed:') + else: + logger.critical(' Pull request verification did not succeed:') + for error in lsig.errors: + logger.critical(' %s %s', attfail, error) + + if attpolicy == 'hardfail': + import sys + sys.exit(128) + + def fetch_remote(gitdir, lmsg, branch=None): # Do we know anything about this base commit? if lmsg.pr_base_commit and not git_commit_exists(gitdir, lmsg.pr_base_commit): @@ -139,6 +188,10 @@ def fetch_remote(gitdir, lmsg, branch=None): logger.critical(out) return ecode + config = b4.get_main_config() + if config['attestation-policy'] != 'off': + attest_fetch_head(gitdir, lmsg) + logger.info('---') if branch: gitargs = ['checkout', '-b', branch, 'FETCH_HEAD'] @@ -244,7 +297,7 @@ def main(cmdargs): mbx.close() os.unlink(savefile) - if lmsg is None: + if lmsg is None or lmsg.pr_remote_tip_commit is None: logger.critical('ERROR: Could not find pull request info in %s', msgid) sys.exit(1) @@ -282,6 +335,7 @@ def main(cmdargs): if len(loglines) and loglines[0].find(lmsg.pr_tip_commit) == 0: logger.info('Pull request is at the tip of FETCH_HEAD') if cmdargs.check: + attest_fetch_head(gitdir, lmsg) sys.exit(0) elif cmdargs.check: |