pungi.spec

@@ -41,6 +41,7 @@ BuildRequires:  python3-dogpile-cache
 BuildRequires:  python3-parameterized
 BuildRequires:  python3-gobject-base
 BuildRequires:  python3-dataclasses
+BuildRequires:  python3-pgpy
 
 #deps for doc building
 BuildRequires:  python3-sphinx
@@ -67,6 +68,7 @@ Requires:       python3-PyYAML
 Requires:       python3-productmd >= 1.28R
 Requires:       python3-gobject-base
 Requires:       lorax
+Requires:  python3-pgpy
 
 # This package is not available on i686, hence we cannot require it
 # See https://bugzilla.redhat.com/show_bug.cgi?id=1743421

pungi/phases/pkgset/pkgsets.py

@@ -23,6 +23,8 @@ import itertools
 import json
 import os
 import time
+import pgpy
+import rpm
 from six.moves import cPickle as pickle
 
 import kobo.log
@@ -993,6 +995,24 @@ class KojiMockPackageSet(PackageSetBase):
 
         return response
 
+    def _is_rpm_signed(self, rpm_path) -> bool:
+        ts = rpm.TransactionSet()
+        ts.setVSFlags(rpm._RPMVSF_NOSIGNATURES)
+        sigkeys = [
+            sigkey.lower() for sigkey in self.sigkey_ordering
+            if sigkey is not None
+        ]
+        with open(rpm_path, 'rb') as fd:
+            header = ts.hdrFromFdno(fd)
+            signature = header[rpm.RPMTAG_SIGGPG] or header[rpm.RPMTAG_SIGPGP]
+            if signature is None:
+                return False
+            pgp_msg = pgpy.PGPMessage.from_blob(signature)
+            return any(
+                signature.signer.lower() in sigkeys
+                for signature in pgp_msg.signatures
+            )
+
     def get_package_path(self, queue_item):
         rpm_info, build_info = queue_item
 
@@ -1010,6 +1030,13 @@ class KojiMockPackageSet(PackageSetBase):
 
         rpm_path = os.path.join(pathinfo.topdir, pathinfo.rpm(rpm_info))
         if os.path.isfile(rpm_path):
+            if not self._is_rpm_signed(rpm_path):
+                self._invalid_sigkey_rpms.append(rpm_info)
+                self.log_error(
+                    'RPM "%s" not found for sigs: "%s". Path checked: "%s"',
+                    rpm_info, self.sigkey_ordering, rpm_path
+                )
+                return
             return rpm_path
         else:
             self.log_warning("RPM %s not found" % rpm_path)

tests/test_pkgset_pkgsets.py

@@ -137,6 +137,21 @@ class PkgsetCompareMixin(object):
 @mock.patch("pungi.phases.pkgset.pkgsets.ReaderPool", new=FakePool)
 @mock.patch("kobo.pkgset.FileCache", new=MockFileCache)
 class TestKojiPkgset(PkgsetCompareMixin, helpers.PungiTestCase):
+
+    @classmethod
+    def setUpClass(cls) -> None:
+
+        cls.patcher = mock.patch.object(
+            pkgsets.KojiMockPackageSet,
+            '_is_rpm_signed',
+            return_value=True,
+        )
+        cls.patcher.start()
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        cls.patcher.stop()
+
     def setUp(self):
         super(TestKojiPkgset, self).setUp()
         with open(os.path.join(helpers.FIXTURE_DIR, "tagged-rpms.json")) as f: