From bed181fdf8fd2174ca4740b0455a97081a10763f Mon Sep 17 00:00:00 2001 From: Patrick Uiterwijk Date: Aug 05 2021 09:23:53 +0000 Subject: Support packages that are head-signed This supports packages which do not have the RPMv3 signature scheme (over the full RPM header+payload), but instead only have signatures over the header (v4 scheme). For the v4 scheme, the signature is only stored in SIGTAG_RSA (or SIGTAG_DSA). Signed-off-by: Patrick Uiterwijk --- diff --git a/cli/koji_cli/commands.py b/cli/koji_cli/commands.py index 70af3a5..f95240c 100644 --- a/cli/koji_cli/commands.py +++ b/cli/koji_cli/commands.py @@ -1596,13 +1596,18 @@ def handle_import_sig(goptions, session, args): activate_session(session, goptions) for path in args: data = koji.get_header_fields(path, ('name', 'version', 'release', 'arch', 'siggpg', - 'sigpgp', 'sourcepackage')) + 'sigpgp', 'dsaheader', 'rsaheader', + 'sourcepackage')) if data['sourcepackage']: data['arch'] = 'src' sigkey = data['siggpg'] if not sigkey: sigkey = data['sigpgp'] if not sigkey: + sigkey = data['dsaheader'] + if not sigkey: + sigkey = data['rsaheader'] + if not sigkey: sigkey = "" if not options.with_unsigned: print(_("Skipping unsigned package: %s" % path)) @@ -1611,6 +1616,8 @@ def handle_import_sig(goptions, session, args): sigkey = koji.get_sigpacket_key_id(sigkey) del data['siggpg'] del data['sigpgp'] + del data['dsaheader'] + del data['rsaheader'] rinfo = session.getRPM(data) if not rinfo: print("No such rpm in system: %(name)s-%(version)s-%(release)s.%(arch)s" % data) diff --git a/hub/kojihub.py b/hub/kojihub.py index ca8977a..58e90af 100644 --- a/hub/kojihub.py +++ b/hub/kojihub.py @@ -7541,6 +7541,10 @@ def add_rpm_sig(an_rpm, sighdr): sigkey = rawhdr.get(koji.RPM_SIGTAG_GPG) if not sigkey: sigkey = rawhdr.get(koji.RPM_SIGTAG_PGP) + if not sigkey: + sigkey = rawhdr.get(koji.RPM_SIGTAG_DSA) + if not sigkey: + sigkey = rawhdr.get(koji.RPM_SIGTAG_RSA) else: # In older rpms, this field in the signature header does not actually match # sigmd5 (I think rpmlib pulls it from SIGTAG_GPG). Anyway, this @@ -7684,6 +7688,10 @@ def _scan_sighdr(sighdr, fn): sig = koji.get_header_field(hdr, 'siggpg') if not sig: sig = koji.get_header_field(hdr, 'sigpgp') + if not sig: + sig = koji.get_header_field(hdr, 'dsaheader') + if not sig: + sig = koji.get_header_field(hdr, 'rsaheader') return koji.get_header_field(hdr, 'sigmd5'), sig @@ -7715,6 +7723,10 @@ def check_rpm_sig(an_rpm, sigkey, sighdr): if not raw_key: raw_key = koji.get_header_field(hdr, 'sigpgp') if not raw_key: + raw_key = koji.get_header_field(hdr, 'dsaheader') + if not raw_key: + raw_key = koji.get_header_field(hdr, 'rsaheader') + if not raw_key: found_key = None else: found_key = koji.get_sigpacket_key_id(raw_key) diff --git a/koji/__init__.py b/koji/__init__.py index 360ab11..ffea6c7 100644 --- a/koji/__init__.py +++ b/koji/__init__.py @@ -100,6 +100,8 @@ def _(args): RPM_HEADER_MAGIC = six.b('\x8e\xad\xe8') RPM_TAG_HEADERSIGNATURES = 62 RPM_TAG_FILEDIGESTALGO = 5011 +RPM_SIGTAG_DSA = 267 +RPM_SIGTAG_RSA = 268 RPM_SIGTAG_PGP = 1002 RPM_SIGTAG_MD5 = 1004 RPM_SIGTAG_GPG = 1005 @@ -925,6 +927,10 @@ def get_sighdr_key(sighdr): if not sig: sig = rh.get(RPM_SIGTAG_PGP) if not sig: + sig = rh.GIT(RPM_SIGTAG_DSA) + if not sig: + sig = rh.GET(RPM_SIGTAG_RSA) + if not sig: return None else: return get_sigpacket_key_id(sig) @@ -1016,7 +1022,7 @@ def get_header_field(hdr, name, src_arch=False): pass # some string results are binary and should not be decoded - if name.startswith('SIG'): + if name.startswith('SIG') or name.endswith('HEADER'): return result # Some older versions of rpm return string header values as bytes. Newer diff --git a/tests/test_cli/test_import_sig.py b/tests/test_cli/test_import_sig.py index c617d36..477f45e 100644 --- a/tests/test_cli/test_import_sig.py +++ b/tests/test_cli/test_import_sig.py @@ -43,6 +43,8 @@ class TestImportSIG(utils.CliTestCase): 'arch': 'x86_64', 'siggpg': None, 'sigpgp': None, + 'dsaheader': None, + 'rsaheader': None, }, { 'sourcepackage': 1, @@ -52,6 +54,8 @@ class TestImportSIG(utils.CliTestCase): 'arch': 'x86_64', 'siggpg': None, 'sigpgp': None, + 'dsaheader': None, + 'rsaheader': None, }, { 'sourcepackage': 1, @@ -61,6 +65,8 @@ class TestImportSIG(utils.CliTestCase): 'arch': 'x86_64', 'siggpg': None, 'sigpgp': None, + 'dsaheader': None, + 'rsaheader': None, } ] @@ -121,6 +127,8 @@ class TestImportSIG(utils.CliTestCase): for data in self.rpm_headers: data['siggpg'] = fake_sigkey data['sigpgp'] = fake_sigkey + data['dsaheader'] = fake_sigkey + data['rsaheader'] = fake_sigkey tmp = data.copy() tmp['arch'] = 'src' if tmp['sourcepackage'] else tmp['arch'] expected += "No such rpm in system: %(name)s-%(version)s-%(release)s.%(arch)s" % \ @@ -240,6 +248,47 @@ class TestImportSIG(utils.CliTestCase): # restore os.path.exists patch os_path_exists_patch.stop() + @mock.patch('sys.stderr', new_callable=six.StringIO) + @mock.patch('sys.stdout', new_callable=six.StringIO) + @mock.patch('koji_cli.commands.activate_session') + def test_handle_import_sig_sigkey_from_header_signed( + self, + activate_session_mock, + stdout, stderr): + """Test sigkey computation from header-only signed rpm in handle_import_sig function""" + data_path = os.path.abspath("tests/test_hub/data/rpms") + arguments = [os.path.join(data_path, 'header-signed.rpm')] + sigkey = '15f712be' + + options = mock.MagicMock() + session = mock.MagicMock() + expected = '' + + for pkg in arguments: + expected += "Importing signature [key %s] from %s..." % (sigkey, pkg) + "\n" + expected += "Writing signed copy" + "\n" + + session.getRPM.side_effect = [ + { + 'sourcepackage': 0, + 'name': 'testpkg', + 'version': '1.0.0', + 'release': '1', + 'arch': 'x86_64', + 'external_repo_id': 0, + 'id': 1, + } + ] + session.queryRPMSigs.side_effect = None + session.queryRPMSigs.return_value = [] + + # Run + handle_import_sig(options, session, arguments + ['--test']) + + self.assert_console_message(stdout, expected) + session.addRPMSig.assert_not_called() + session.writeSignedRPM.assert_not_called() + def test_handle_import_sig_argument_test(self): """Test handle_import_sig function without arguments""" options = mock.MagicMock() diff --git a/tests/test_hub/data/rpms/header-signed.rpm b/tests/test_hub/data/rpms/header-signed.rpm new file mode 100644 index 0000000..c62c521 Binary files /dev/null and b/tests/test_hub/data/rpms/header-signed.rpm differ diff --git a/tests/test_hub/test_add_rpm_sig.py b/tests/test_hub/test_add_rpm_sig.py new file mode 100644 index 0000000..40d1325 --- /dev/null +++ b/tests/test_hub/test_add_rpm_sig.py @@ -0,0 +1,86 @@ +import base64 +import mock +import os +import unittest + +import koji +import kojihub + +IP = kojihub.InsertProcessor + + +class TestAddRPMSig(unittest.TestCase): + def getInsert(self, *args, **kwargs): + insert = IP(*args, **kwargs) + insert.execute = mock.MagicMock() + self.inserts.append(insert) + return insert + + def setUp(self): + self.InsertProcessor = mock.patch('kojihub.InsertProcessor', + side_effect=self.getInsert).start() + self.inserts = [] + self.context = mock.patch('kojihub.context').start() + # It seems MagicMock will not automatically handle attributes that + # start with "assert" + self.context.session.assertLogin = mock.MagicMock() + self.context.session.assertPerm = mock.MagicMock() + self.context.opts = {'HostPrincipalFormat': '-%s-'} + self.exports = kojihub.RootExports() + self.data_path = os.path.abspath("tests/test_hub/data/rpms") + + def tearDown(self): + mock.patch.stopall() + + @mock.patch('kojihub._fetchMulti') + @mock.patch('koji.plugin.run_callbacks') + @mock.patch('kojihub.get_rpm') + @mock.patch('kojihub.get_build') + @mock.patch('os.path.isdir') + @mock.patch('koji.ensuredir') + @mock.patch('kojihub.open') + def test_add_rpm_sig_header_signed( + self, + open, + ensuredir, + isdir, + get_build, + get_rpm, + run_callbacks, + _fetchMulti): + """Test addRPMSig with header-only signed RPM""" + _fetchMulti.side_effect = [[]] + isdir.side_effect = [True] + get_rpm.side_effect = [{ + 'id': 1, + 'name': 'testpkg', + 'version': '1.0.0', + 'release': '1', + 'arch': 'noarch', + 'epoch': None, + 'payloadhash': '1706d0174aa29a5a3e5c60855a778c35', + 'size': 123, + 'external_repo_id': None, + 'build_id': 1, + }] + open.side_effect = [mock.MagicMock()] + + rpm_path = os.path.join(self.data_path, 'header-signed.rpm') + sighdr = koji.rip_rpm_sighdr(rpm_path) + + self.exports.addRPMSig(1, base64.b64encode(sighdr)) + self.context.session.assertPerm.assert_called_once_with('sign') + self.assertEqual(len(self.inserts), 1) + insert = self.inserts[0] + self.assertEqual(insert.data['rpm_id'], 1) + self.assertEqual(insert.data['sigkey'], '15f712be') + + def test_scan_sighdr_header_signed(self): + """Test _scan_sighdr on a header-only signed package""" + rpm_path = os.path.join(self.data_path, 'header-signed.rpm') + sighdr = koji.rip_rpm_sighdr(rpm_path) + + sigmd5, sig = kojihub._scan_sighdr(sighdr, rpm_path) + self.assertEqual(koji.hex_string(sigmd5), '1706d0174aa29a5a3e5c60855a778c35') + sigkey = koji.get_sigpacket_key_id(sig) + self.assertEqual(sigkey, '15f712be')