1# 2# Copyright OpenEmbedded Contributors 3# 4# SPDX-License-Identifier: GPL-2.0-only 5# 6 7"""Helper module for GPG signing""" 8 9import bb 10import os 11import shlex 12import subprocess 13import tempfile 14 15class LocalSigner(object): 16 """Class for handling local (on the build host) signing""" 17 def __init__(self, d): 18 self.gpg_bin = d.getVar('GPG_BIN') or \ 19 bb.utils.which(os.getenv('PATH'), 'gpg') 20 self.gpg_cmd = [self.gpg_bin] 21 self.gpg_agent_bin = bb.utils.which(os.getenv('PATH'), "gpg-agent") 22 # Without this we see "Cannot allocate memory" errors when running processes in parallel 23 # It needs to be set for any gpg command since any agent launched can stick around in memory 24 # and this parameter must be set. 25 if self.gpg_agent_bin: 26 self.gpg_cmd += ["--agent-program=%s|--auto-expand-secmem" % (self.gpg_agent_bin)] 27 self.gpg_path = d.getVar('GPG_PATH') 28 self.rpm_bin = bb.utils.which(os.getenv('PATH'), "rpmsign") 29 self.gpg_version = self.get_gpg_version() 30 31 32 def export_pubkey(self, output_file, keyid, armor=True): 33 """Export GPG public key to a file""" 34 cmd = self.gpg_cmd + ["--no-permission-warning", "--batch", "--yes", "--export", "-o", output_file] 35 if self.gpg_path: 36 cmd += ["--homedir", self.gpg_path] 37 if armor: 38 cmd += ["--armor"] 39 cmd += [keyid] 40 subprocess.check_output(cmd, stderr=subprocess.STDOUT) 41 42 def sign_rpms(self, files, keyid, passphrase, digest, sign_chunk, fsk=None, fsk_password=None): 43 """Sign RPM files""" 44 45 cmd = self.rpm_bin + " --addsign --define '_gpg_name %s' " % keyid 46 gpg_args = '--no-permission-warning --batch --passphrase=%s --agent-program=%s|--auto-expand-secmem' % (passphrase, self.gpg_agent_bin) 47 if self.gpg_version > (2,1,): 48 gpg_args += ' --pinentry-mode=loopback' 49 cmd += "--define '_gpg_sign_cmd_extra_args %s' " % gpg_args 50 cmd += "--define '_binary_filedigest_algorithm %s' " % digest 51 if self.gpg_bin: 52 cmd += "--define '__gpg %s' " % self.gpg_bin 53 if self.gpg_path: 54 cmd += "--define '_gpg_path %s' " % self.gpg_path 55 if fsk: 56 cmd += "--signfiles --fskpath %s " % fsk 57 if fsk_password: 58 cmd += "--define '_file_signing_key_password %s' " % fsk_password 59 60 # Sign in chunks 61 for i in range(0, len(files), sign_chunk): 62 subprocess.check_output(shlex.split(cmd + ' '.join(files[i:i+sign_chunk])), stderr=subprocess.STDOUT) 63 64 def detach_sign(self, input_file, keyid, passphrase_file, passphrase=None, armor=True, output_suffix=None, use_sha256=False): 65 """Create a detached signature of a file""" 66 67 if passphrase_file and passphrase: 68 raise Exception("You should use either passphrase_file of passphrase, not both") 69 70 cmd = self.gpg_cmd + ['--detach-sign', '--no-permission-warning', '--batch', 71 '--no-tty', '--yes', '--passphrase-fd', '0', '-u', keyid] 72 73 if self.gpg_path: 74 cmd += ['--homedir', self.gpg_path] 75 if armor: 76 cmd += ['--armor'] 77 if use_sha256: 78 cmd += ['--digest-algo', "SHA256"] 79 80 #gpg > 2.1 supports password pipes only through the loopback interface 81 #gpg < 2.1 errors out if given unknown parameters 82 if self.gpg_version > (2,1,): 83 cmd += ['--pinentry-mode', 'loopback'] 84 85 try: 86 if passphrase_file: 87 with open(passphrase_file) as fobj: 88 passphrase = fobj.readline(); 89 90 if not output_suffix: 91 output_suffix = 'asc' if armor else 'sig' 92 output_file = input_file + "." + output_suffix 93 with tempfile.TemporaryDirectory(dir=os.path.dirname(output_file)) as tmp_dir: 94 tmp_file = os.path.join(tmp_dir, os.path.basename(output_file)) 95 cmd += ['-o', tmp_file] 96 97 cmd += [input_file] 98 99 job = subprocess.Popen(cmd, stdin=subprocess.PIPE, stderr=subprocess.PIPE) 100 (_, stderr) = job.communicate(passphrase.encode("utf-8")) 101 102 if job.returncode: 103 bb.fatal("GPG exited with code %d: %s" % (job.returncode, stderr.decode("utf-8"))) 104 105 os.rename(tmp_file, output_file) 106 except IOError as e: 107 bb.error("IO error (%s): %s" % (e.errno, e.strerror)) 108 raise Exception("Failed to sign '%s'" % input_file) 109 110 except OSError as e: 111 bb.error("OS error (%s): %s" % (e.errno, e.strerror)) 112 raise Exception("Failed to sign '%s" % input_file) 113 114 115 def get_gpg_version(self): 116 """Return the gpg version as a tuple of ints""" 117 try: 118 cmd = self.gpg_cmd + ["--version", "--no-permission-warning"] 119 ver_str = subprocess.check_output(cmd).split()[2].decode("utf-8") 120 return tuple([int(i) for i in ver_str.split("-")[0].split('.')]) 121 except subprocess.CalledProcessError as e: 122 bb.fatal("Could not get gpg version: %s" % e) 123 124 125 def verify(self, sig_file, valid_sigs = ''): 126 """Verify signature""" 127 cmd = self.gpg_cmd + ["--verify", "--no-permission-warning", "--status-fd", "1"] 128 if self.gpg_path: 129 cmd += ["--homedir", self.gpg_path] 130 131 cmd += [sig_file] 132 status = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) 133 # Valid if any key matches if unspecified 134 if not valid_sigs: 135 ret = False if status.returncode else True 136 return ret 137 138 import re 139 goodsigs = [] 140 sigre = re.compile(r'^\[GNUPG:\] GOODSIG (\S+)\s(.*)$') 141 for l in status.stdout.decode("utf-8").splitlines(): 142 s = sigre.match(l) 143 if s: 144 goodsigs += [s.group(1)] 145 146 for sig in valid_sigs.split(): 147 if sig in goodsigs: 148 return True 149 if len(goodsigs): 150 bb.warn('No accepted signatures found. Good signatures found: %s.' % ' '.join(goodsigs)) 151 return False 152 153 154def get_signer(d, backend): 155 """Get signer object for the specified backend""" 156 # Use local signing by default 157 if backend == 'local': 158 return LocalSigner(d) 159 else: 160 bb.fatal("Unsupported signing backend '%s'" % backend) 161