From f8d9188bbc52d8f315ba2a2b659d99e18a866d79 Mon Sep 17 00:00:00 2001 From: Mike McLean Date: Jan 26 2026 19:18:42 +0000 Subject: [PATCH 1/2] option to query signed rpm headers --- diff --git a/kojihub/kojihub.py b/kojihub/kojihub.py index 9e9fb47..8ecac26 100644 --- a/kojihub/kojihub.py +++ b/kojihub/kojihub.py @@ -13198,7 +13198,8 @@ class RootExports(object): "No file: %s found in RPM: %s" % (filename, rpmID)) return {} - def getRPMHeaders(self, rpmID=None, taskID=None, filepath=None, headers=None, strict=False): + def getRPMHeaders(self, rpmID=None, taskID=None, filepath=None, headers=None, strict=False, + sigkey=None): """ Get the requested headers from the rpm, specified either by rpmID or taskID + filepath @@ -13234,7 +13235,16 @@ class RootExports(object): raise koji.GenericError('Build %(nvr)s is deleted' % build_info) else: return {} - rpm_path = joinpath(koji.pathinfo.build(build_info), koji.pathinfo.rpm(rpm_info)) + builddir = koji.pathinfo.build(build_info) + if sigkey is not None: + if not query_rpm_sigs(rpm_id=rpm_info['id'], sigkey=sigkey): + if strict: + raise koji.GenericError(f'No {sigkey} signature for rpm {rpm_info["id"]}') + else: + return {} + rpm_path = joinpath(builddir, koji.pathinfo.signed(rpm_info, sigkey)) + else: + rpm_path = joinpath(builddir, koji.pathinfo.rpm(rpm_info)) if not os.path.exists(rpm_path): if strict: raise koji.GenericError('Missing rpm file: %s' % rpm_path) @@ -13243,6 +13253,8 @@ class RootExports(object): logger.error('Missing rpm file: %s' % rpm_path) return {} elif taskID: + if sigkey is not None: + raise koji.GenericError('The sigkey option cannot be used with a task id') if not filepath: raise koji.GenericError('filepath must be specified with taskID') if filepath.startswith('/') or '../' in filepath: From 66116f5ba041c01f9536c7f894f16992b0ca7ee8 Mon Sep 17 00:00:00 2001 From: Mike McLean Date: Jan 26 2026 19:37:18 +0000 Subject: [PATCH 2/2] test coverage --- diff --git a/tests/test_hub/test_getRPM.py b/tests/test_hub/test_getRPM.py index 86a7ef1..5a0cd57 100644 --- a/tests/test_hub/test_getRPM.py +++ b/tests/test_hub/test_getRPM.py @@ -171,6 +171,7 @@ class TestGetRPMHeaders(unittest.TestCase): self.get_rpm = mock.patch('kojihub.kojihub.get_rpm').start() self.get_build = mock.patch('kojihub.kojihub.get_build').start() self.get_header_fields = mock.patch('koji.get_header_fields').start() + self.query_rpm_sigs = mock.patch('kojihub.kojihub.query_rpm_sigs').start() self.tempdir = tempfile.mkdtemp() self.pathinfo = koji.PathInfo(self.tempdir) mock.patch('koji.pathinfo', new=self.pathinfo).start() @@ -189,6 +190,7 @@ class TestGetRPMHeaders(unittest.TestCase): self.get_rpm.assert_not_called() self.get_build.assert_not_called() self.get_header_fields.assert_not_called() + self.query_rpm_sigs.assert_not_called() def test_taskid_without_filepath(self): self.cursor.fetchone.return_value = None @@ -199,6 +201,7 @@ class TestGetRPMHeaders(unittest.TestCase): self.get_rpm.assert_not_called() self.get_build.assert_not_called() self.get_header_fields.assert_not_called() + self.query_rpm_sigs.assert_not_called() def test_insufficient_args(self): with self.assertRaises(koji.GenericError) as cm: @@ -211,6 +214,13 @@ class TestGetRPMHeaders(unittest.TestCase): str(cm.exception)) # we already test taskid without filepath above + def test_sigkey_not_allowed(self): + filepath = 'pkg-1-2.noarch.rpm' + with self.assertRaises(koji.GenericError) as cm: + self.exports.getRPMHeaders(taskID=99, filepath=filepath, sigkey='c0ffee') + self.assertEqual("The sigkey option cannot be used with a task id", + str(cm.exception)) + def test_unknown_rpm(self): self.get_rpm.return_value = None result = self.exports.getRPMHeaders(rpmID='FOO-1-1.noarch', strict=False) @@ -225,6 +235,7 @@ class TestGetRPMHeaders(unittest.TestCase): self.get_rpm.assert_called_with('FOO-1-1.noarch', strict=True) self.get_build.assert_not_called() self.get_header_fields.assert_not_called() + self.query_rpm_sigs.assert_not_called() def test_external_rpm(self): rpm_info = {'external_repo_id': 1, 'id': 'RPMID'} @@ -238,6 +249,7 @@ class TestGetRPMHeaders(unittest.TestCase): self.assertEqual(f"External rpm: {rpm_info['id']}", str(cm.exception)) self.get_build.assert_not_called() self.get_header_fields.assert_not_called() + self.query_rpm_sigs.assert_not_called() def test_deleted_build(self): self.get_rpm.return_value = {'build_id': 'BUILDID', 'external_repo_id': 0} @@ -253,6 +265,7 @@ class TestGetRPMHeaders(unittest.TestCase): self.assertEqual(f"Build {build_info['nvr']} is deleted", str(cm.exception)) self.get_build.assert_called_with('BUILDID', strict=True) self.get_header_fields.assert_not_called() + self.query_rpm_sigs.assert_not_called() def test_missing_rpm(self): self.get_rpm.return_value = { @@ -281,6 +294,37 @@ class TestGetRPMHeaders(unittest.TestCase): self.assertEqual(f"Missing rpm file: {rpmpath}", str(e.exception)) self.get_build.assert_called_with('BUILDID', strict=True) self.get_header_fields.assert_not_called() + self.query_rpm_sigs.assert_not_called() + + def test_missing_sig(self): + self.get_rpm.return_value = { + 'id': 1234, + 'build_id': 'BUILDID', + 'external_repo_id': 0, + 'name': 'pkg', + 'version': '1', + 'release': '2', + 'arch': 'noarch'} + self.get_build.return_value = { + 'name': 'pkg', + 'version': '1', + 'release': '2', + 'nvr': 'pkg-1-2', + 'state': koji.BUILD_STATES['COMPLETE']} + self.query_rpm_sigs.return_value = [] + rpmpath = '%s/packages/pkg/1/2/noarch/pkg-1-2.noarch.rpm' % self.tempdir + + # rpm does not exist + result = self.exports.getRPMHeaders(rpmID='pkg-1-2.noarch', strict=False, sigkey='c0ffee') + self.assertEqual(result, {}) + self.get_build.assert_called_with('BUILDID', strict=True) + + # again with strict mode + with self.assertRaises(koji.GenericError) as e: + self.exports.getRPMHeaders(rpmID='FOO-1-1.noarch', strict=True, sigkey='c0ffee') + self.assertEqual(f"No c0ffee signature for rpm 1234", str(e.exception)) + self.get_build.assert_called_with('BUILDID', strict=True) + self.get_header_fields.assert_not_called() def test_rpm_exists(self): self.get_rpm.return_value = { @@ -307,6 +351,36 @@ class TestGetRPMHeaders(unittest.TestCase): self.assertEqual(result, fakeheaders) self.get_build.assert_called_with('BUILDID', strict=True) self.get_header_fields.assert_called_with(rpmpath, None) + self.query_rpm_sigs.assert_not_called() + + def test_sig_exists(self): + self.get_rpm.return_value = { + 'id': 1234, + 'build_id': 'BUILDID', + 'external_repo_id': 0, + 'name': 'pkg', + 'version': '1', + 'release': '2', + 'arch': 'noarch'} + self.get_build.return_value = { + 'name': 'pkg', + 'version': '1', + 'release': '2', + 'nvr': 'pkg-1-2', + 'state': koji.BUILD_STATES['COMPLETE']} + self.query_rpm_sigs.return_value = [{'rpm_id': 1234, 'sigkey': 'c0ffee', 'sighash': 'whatever'}] + rpmpath = '%s/packages/pkg/1/2/data/signed/c0ffee/noarch/pkg-1-2.noarch.rpm' % self.tempdir + koji.ensuredir(os.path.dirname(rpmpath)) + with open(rpmpath, 'w') as fo: + fo.write('hello world') + fakeheaders = {'HEADER': 'SOMETHING'} + self.get_header_fields.return_value = fakeheaders + + result = self.exports.getRPMHeaders(rpmID='pkg-1-2.noarch', strict=True, sigkey='c0ffee') + self.assertEqual(result, fakeheaders) + self.get_build.assert_called_with('BUILDID', strict=True) + self.get_header_fields.assert_called_with(rpmpath, None) + self.query_rpm_sigs.assert_called_once() def test_task_rpm_exists(self): taskid = 137 @@ -323,6 +397,7 @@ class TestGetRPMHeaders(unittest.TestCase): self.get_rpm.assert_not_called() self.get_build.assert_not_called() self.get_header_fields.assert_called_with(rpmpath, None) + self.query_rpm_sigs.assert_not_called() def test_task_rpm_missing(self): taskid = 137 @@ -338,3 +413,7 @@ class TestGetRPMHeaders(unittest.TestCase): self.get_rpm.assert_not_called() self.get_build.assert_not_called() self.get_header_fields.assert_not_called() + self.query_rpm_sigs.assert_not_called() + + +# the end