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