Add attestation checks for b4 pr

We now use similar subroutines for checking signatures on FETCH_HEAD as
we do for patch attestation, making it a convenient operation during the
fetch stage:

$ b4 pr 202003292114.2252CAEF7@keescook
Looking up https://lore.kernel.org/r/202003292114.2252CAEF7@keescook
Grabbing thread from lore.kernel.org
Looking at: [GIT PULL] seccomp updates for v5.7-rc1
  Fetching https://git.kernel.org/pub/scm/linux/kernel/git/kees/linux.git tags/seccomp-v5.7-rc1
  ---
  [✓] Attestation-by: Kees Cook <keescook@chromium.org> (pgp: 8972F4DFDC6DC026)
---
Successfully fetched into FETCH_HEAD

Hopefully, I didn't introduce too many bugs into patch attestation,
since I had to rewrite the backend a bit to work for both native git
operations and patch attestation calls.

Signed-off-by: Konstantin Ryabitsev <konstantin@linuxfoundation.org>

1 files changed, 55 insertions, 1 deletions

diff --git a/b4/pr.py b/b4/pr.py
index 55261f1..0ed9277 100644
--- a/b4/pr.py
+++ b/b4/pr.py
@@ -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: