ALBS-732: pungi: do not build anything if unsigned packages found #8

Closed
soksanichenko wants to merge 3 commits from ALBS-732 into aln8_separate_sources_module
2 changed files with 30 additions and 5 deletions

View File

@ -65,6 +65,7 @@ Requires: python3-createrepo_c >= 0.20.1
Requires: python3-PyYAML Requires: python3-PyYAML
Requires: python3-gobject-base Requires: python3-gobject-base
Requires: lorax Requires: lorax
Requires: python3-pgpy
# This package is not available on i686, hence we cannot require it # This package is not available on i686, hence we cannot require it
# See https://bugzilla.redhat.com/show_bug.cgi?id=1743421 # See https://bugzilla.redhat.com/show_bug.cgi?id=1743421

View File

@ -22,6 +22,9 @@ It automatically finds a signed copies according to *sigkey_ordering*.
import itertools import itertools
import json import json
import os import os
import pgpy
import rpm
from six.moves import cPickle as pickle from six.moves import cPickle as pickle
import kobo.log import kobo.log
@ -493,8 +496,6 @@ class KojiPackageSet(PackageSetBase):
return response return response
def get_package_path(self, queue_item): def get_package_path(self, queue_item):
rpm_info, build_info = queue_item rpm_info, build_info = queue_item
@ -849,6 +850,8 @@ class KojiMockPackageSet(PackageSetBase):
self.extra_builds = extra_builds or [] self.extra_builds = extra_builds or []
self.extra_tasks = extra_tasks or [] self.extra_tasks = extra_tasks or []
self.reuse = None self.reuse = None
self.sigkey_ordering = [sigkey.lower() for sigkey in sigkey_ordering] \
or [None]
def __getstate__(self): def __getstate__(self):
result = self.__dict__.copy() result = self.__dict__.copy()
@ -965,11 +968,25 @@ class KojiMockPackageSet(PackageSetBase):
return response return response
def _is_rpm_signed(self, rpm_path) -> bool:
ts = rpm.TransactionSet()
ts.setVSFlags(rpm._RPMVSF_NOSIGNATURES)
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 self.sigkey_ordering
for signature in pgp_msg.signatures
)
def get_package_path(self, queue_item): def get_package_path(self, queue_item):
rpm_info, build_info = queue_item rpm_info, build_info = queue_item
# Check if this RPM is coming from scratch task. In this case, we already # Check if this RPM is coming from scratch task.
# know the path. # In this case, we already know the path.
if "path_from_task" in rpm_info: if "path_from_task" in rpm_info:
return rpm_info["path_from_task"] return rpm_info["path_from_task"]
@ -982,6 +999,13 @@ class KojiMockPackageSet(PackageSetBase):
rpm_path = os.path.join(pathinfo.topdir, pathinfo.rpm(rpm_info)) rpm_path = os.path.join(pathinfo.topdir, pathinfo.rpm(rpm_info))
if os.path.isfile(rpm_path): 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 return rpm_path
else: else:
self.log_warning("RPM %s not found" % rpm_path) self.log_warning("RPM %s not found" % rpm_path)