Require and verify file (IMA) signatures for selected platforms

node-config/sign_node.yml

@@ -19,3 +19,11 @@ is_community_sign_node: true
 #   bitwarden_password: "..."
 #   Optional: restrict the lookup to a single collection (real UUID only).
 #   bitwarden_collection_id: <uuid>
+
+# Platforms whose packages must carry file (IMA) signatures. For sign tasks
+# touching a listed platform, file signing is forced even when the task
+# payload says otherwise, and the task fails if any signed package with
+# regular files lacks file signatures. Requires 'files_sign_cert_path' to
+# point to a provisioned IMA signing key. Absent or empty -> nothing changes.
+#   require_files_signature_platforms:
+#     - AlmaLinux-10

sign_node/config.py

@@ -76,6 +76,7 @@ def __init__(self, config_file=None, **cmd_args):
             'immudb_address': None,
             'immudb_public_key_file': None,
             'files_sign_cert_path': '/etc/pki/ima/ima-sign.key',
+            'require_files_signature_platforms': [],
             'locks_dir_path': '/tmp/gpg_locks',
             'bitwarden_enabled': False,
             'bitwarden_username': None,
@@ -112,6 +113,10 @@ def __init__(self, config_file=None, **cmd_args):
                 'type': 'string', 'required': False,
                 'coerce': normalize_path,
             },
+            'require_files_signature_platforms': {
+                'type': 'list', 'required': False,
+                'schema': {'type': 'string'},
+            },
             'locks_dir_path': {'type': 'string', 'required': True},
             'bitwarden_enabled': {'type': 'boolean', 'default': False},
             'bitwarden_username': {'type': 'string', 'nullable': True},

sign_node/signer.py

@@ -6,6 +6,7 @@
 import enum
 import os
 import logging
+import stat
 import pprint
 import shutil
 import glob
@@ -51,6 +52,7 @@ class SignStatusEnum(enum.IntEnum):
     READ_ERROR = 2
     NO_SIGNATURE = 3
     WRONG_SIGNATURE = 4
+    NO_FILE_SIGNATURE = 5
 
 
 class Signer(object):
@@ -183,9 +185,67 @@ def sign_loop(self):
                             err,
                         )
 
-    def _check_signature(self, files, key_id):
+    @staticmethod
+    def _check_file_signatures(header) -> bool:
+        """
+        Checks that every regular packaged file in an RPM header carries
+        a file (IMA) signature. Returns True for packages that have no
+        regular files to sign (e.g. metapackages containing only
+        directories or symlinks).
+        """
+        modes = header[rpm.RPMTAG_FILEMODES] or []
+        flags = header[rpm.RPMTAG_FILEFLAGS] or []
+        signatures = header[rpm.RPMTAG_FILESIGNATURES] or []
+        for idx, mode in enumerate(modes):
+            if not stat.S_ISREG(mode):
+                continue
+            # %ghost files have no payload content, nothing to sign
+            if idx < len(flags) and flags[idx] & rpm.RPMFILE_GHOST:
+                continue
+            if idx >= len(signatures) or not signatures[idx]:
+                return False
+        return True
+
+    def _files_signature_required(self, task: typing.Dict) -> bool:
+        """
+        Checks whether the task packages must carry file (IMA) signatures
+        according to the 'require_files_signature_platforms' config option.
+
+        Raises
+        ------
+        SignError
+            If the option is set but the task payload carries no platform
+            information (the web server is too old to provide it).
+        """
+        required_platforms = (
+            self.__config.require_files_signature_platforms or []
+        )
+        if not required_platforms:
+            return False
+        rpm_packages = [
+            pkg for pkg in task['packages']
+            if pkg.get('type', 'rpm') == 'rpm'
+        ]
+        missing = [
+            pkg['name'] for pkg in rpm_packages
+            if not pkg.get('platform_name')
+        ]
+        if missing:
+            raise SignError(
+                'require_files_signature_platforms is enabled, but the sign '
+                'task payload carries no platform information for the '
+                'following packages (is the web server up to date?): '
+                '{}'.format(', '.join(missing))
+            )
+        return any(
+            pkg['platform_name'] in required_platforms
+            for pkg in rpm_packages
+        )
+
+    def _check_signature(self, files, key_id, files_require_signature=None):
         errors = []
         key_id_lower = key_id.lower()
+        files_require_signature = files_require_signature or frozenset()
         ts = rpm.TransactionSet()
         ts.setVSFlags(rpm._RPMVSF_NOSIGNATURES)
         subkeys = [i.lower() for i in self.__password_db.get_subkeys(key_id)]
@@ -201,6 +261,11 @@ def check(pkg_path: str) -> typing.Tuple[SignStatusEnum, str]:
                     signature = header[rpm.RPMTAG_SIGPGP]
                 if not signature:
                     return SignStatusEnum.NO_SIGNATURE, ''
+                if (
+                    pkg_path in files_require_signature
+                    and not self._check_file_signatures(header)
+                ):
+                    return SignStatusEnum.NO_FILE_SIGNATURE, ''
 
             pgp_msg = pgpy.PGPMessage.from_blob(signature)
             sig = ''
@@ -228,6 +293,9 @@ def check(pkg_path: str) -> typing.Tuple[SignStatusEnum, str]:
                 elif result == SignStatusEnum.WRONG_SIGNATURE:
                     errors.append(f'Package {pkg_path} is signed '
                                   f'with the wrong key: {signature}')
+                elif result == SignStatusEnum.NO_FILE_SIGNATURE:
+                    errors.append(f'Package {pkg_path} does not contain '
+                                  f'file (IMA) signatures')
 
         return errors
 
@@ -387,7 +455,17 @@ def download_package(pkg: dict):
 
         stats = {'sign_task_start_time': str(datetime.utcnow())}
         pgp_keyid = task['keyid']
-        sign_files = task.get('sign_files', False)
+        require_files_platforms = set(
+            self.__config.require_files_signature_platforms or []
+        )
+        files_signature_required = self._files_signature_required(task)
+        sign_files = task.get('sign_files', False) or files_signature_required
+        if files_signature_required and not task.get('sign_files', False):
+            logging.info(
+                'Forcing file signing for task %s: it contains packages of '
+                'platforms listed in require_files_signature_platforms',
+                task['id'],
+            )
         pgp_key_password = self.__password_db.get_password(pgp_keyid)
         fingerprint = self.__password_db.get_fingerprint(pgp_keyid)
         task_dir = self.__working_dir_path.joinpath(str(task['id']))
@@ -469,6 +547,8 @@ def download_package(pkg: dict):
             sequential_upload_files = {}
             packages_hrefs = {}
             files_to_check = list()
+            files_require_signature = set()
+            checked_paths_by_sha = {}
             for package_id, file_name, package_path in downloaded:
                 old_meta = pkg_verification_mapping.get(package_path)
                 if self.__notar_enabled and old_meta is not None:
@@ -487,14 +567,24 @@ def download_package(pkg: dict):
                             package_id, file_name, package_path)
                     files_to_upload.add(sha256)
                     files_to_check.append(package_path)
+                    checked_paths_by_sha[sha256] = package_path
                 packages[package_id]['sha256'] = sha256
+                pkg_info = packages[package_id]
+                if (pkg_info.get('type', 'rpm') == 'rpm'
+                        and pkg_info.get('platform_name')
+                        in require_files_platforms):
+                    files_require_signature.add(checked_paths_by_sha[sha256])
 
             finish_time = datetime.utcnow()
             stats['notarization_packages_time'] = self.timedelta_seconds(
                 start_time, finish_time)
             start_time = datetime.utcnow()
 
-            sign_errors = self._check_signature(files_to_check, pgp_keyid)
+            sign_errors = self._check_signature(
+                files_to_check,
+                pgp_keyid,
+                files_require_signature=files_require_signature,
+            )
             finish_time = datetime.utcnow()
             stats['signature_check_packages_time'] = self.timedelta_seconds(
                 start_time, finish_time)

tests/sign_node/test_signer.py

@@ -1,13 +1,21 @@
 import os
+import stat
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
+import pytest
+import rpm
 from pyfakefs.fake_filesystem_unittest import TestCase
 
 import sign_node
 from sign_node.config import SignNodeConfig
+from sign_node.errors import SignError
 from sign_node.signer import Signer
 
+REGULAR_FILE_MODE = stat.S_IFREG | 0o644
+DIRECTORY_MODE = stat.S_IFDIR | 0o755
+SYMLINK_MODE = stat.S_IFLNK | 0o777
+
 
 class TestSigner(TestCase):
 
@@ -114,3 +122,159 @@ def test_generate_sign_key(self):
         assert private_key.exists()
         assert public_key.open().read() == key
         assert private_key.open().read() == key
+
+
+def make_header(modes, signatures, flags=None):
+    if flags is None:
+        flags = [0] * len(modes)
+    return {
+        rpm.RPMTAG_FILEMODES: modes,
+        rpm.RPMTAG_FILEFLAGS: flags,
+        rpm.RPMTAG_FILESIGNATURES: signatures,
+    }
+
+
+class TestCheckFileSignatures:
+
+    def test_signed_regular_files(self):
+        header = make_header(
+            [REGULAR_FILE_MODE, REGULAR_FILE_MODE],
+            ['aabb', 'ccdd'],
+        )
+        assert Signer._check_file_signatures(header) is True
+
+    def test_unsigned_regular_file(self):
+        header = make_header(
+            [REGULAR_FILE_MODE, REGULAR_FILE_MODE],
+            ['aabb', ''],
+        )
+        assert Signer._check_file_signatures(header) is False
+
+    def test_no_signatures_at_all(self):
+        header = make_header([REGULAR_FILE_MODE], [])
+        assert Signer._check_file_signatures(header) is False
+        header = make_header([REGULAR_FILE_MODE], None)
+        assert Signer._check_file_signatures(header) is False
+
+    def test_metapackage_without_regular_files(self):
+        header = make_header(
+            [DIRECTORY_MODE, SYMLINK_MODE],
+            [],
+        )
+        assert Signer._check_file_signatures(header) is True
+
+    def test_empty_package(self):
+        header = make_header([], [])
+        assert Signer._check_file_signatures(header) is True
+
+    def test_unsigned_ghost_file_is_skipped(self):
+        header = make_header(
+            [REGULAR_FILE_MODE, REGULAR_FILE_MODE],
+            ['aabb', ''],
+            flags=[0, rpm.RPMFILE_GHOST],
+        )
+        assert Signer._check_file_signatures(header) is True
+
+
+class TestFilesSignatureRequired:
+
+    def make_signer(self, platforms):
+        config = SignNodeConfig(
+            require_files_signature_platforms=platforms,
+        )
+        return Signer(config, 'password', None)
+
+    def test_option_not_set(self):
+        signer = self.make_signer([])
+        task = {'packages': [{'name': 'pkg-1.rpm', 'type': 'rpm'}]}
+        assert signer._files_signature_required(task) is False
+
+    def test_platform_listed(self):
+        signer = self.make_signer(['AlmaLinux-10'])
+        task = {'packages': [
+            {
+                'name': 'pkg-1.rpm',
+                'type': 'rpm',
+                'platform_name': 'AlmaLinux-9',
+            },
+            {
+                'name': 'pkg-2.rpm',
+                'type': 'rpm',
+                'platform_name': 'AlmaLinux-10',
+            },
+        ]}
+        assert signer._files_signature_required(task) is True
+
+    def test_platform_not_listed(self):
+        signer = self.make_signer(['AlmaLinux-10'])
+        task = {'packages': [
+            {
+                'name': 'pkg-1.rpm',
+                'type': 'rpm',
+                'platform_name': 'AlmaLinux-9',
+            },
+        ]}
+        assert signer._files_signature_required(task) is False
+
+    def test_missing_platform_info(self):
+        signer = self.make_signer(['AlmaLinux-10'])
+        task = {'packages': [{'name': 'pkg-1.rpm', 'type': 'rpm'}]}
+        with pytest.raises(SignError, match='no platform information'):
+            signer._files_signature_required(task)
+
+    def test_non_rpm_packages_are_ignored(self):
+        signer = self.make_signer(['AlmaLinux-10'])
+        task = {'packages': [{'name': 'pkg_1.deb', 'type': 'deb'}]}
+        assert signer._files_signature_required(task) is False
+
+
+class TestCheckSignatureFileSignatures(TestCase):
+
+    def setUp(self):
+        self.setUpPyfakefs()
+        self.config = SignNodeConfig()
+        password_db = MagicMock()
+        password_db.get_subkeys.return_value = []
+        self.signer = Signer(self.config, password_db, None)
+
+    def run_check(self, header, require_file_signature):
+        pkg_path = '/pkg/test-package.rpm'
+        self.fs.create_file(pkg_path)
+        header = dict(header)
+        header.setdefault(rpm.RPMTAG_SIGGPG, b'fake-signature')
+        ts = MagicMock()
+        ts.hdrFromFdno.return_value = header
+        pgp_msg = MagicMock()
+        signature = MagicMock()
+        signature.signer = 'aabbccdd11223344'
+        pgp_msg.signatures = [signature]
+        with (
+            patch('sign_node.signer.rpm.TransactionSet', return_value=ts),
+            patch(
+                'sign_node.signer.pgpy.PGPMessage.from_blob',
+                return_value=pgp_msg,
+            ),
+        ):
+            return self.signer._check_signature(
+                [pkg_path],
+                'AABBCCDD11223344',
+                files_require_signature=(
+                    {pkg_path} if require_file_signature else None
+                ),
+            )
+
+    def test_missing_file_signatures_reported(self):
+        header = make_header([REGULAR_FILE_MODE], [])
+        errors = self.run_check(header, require_file_signature=True)
+        assert len(errors) == 1
+        assert 'does not contain file (IMA) signatures' in errors[0]
+
+    def test_present_file_signatures_pass(self):
+        header = make_header([REGULAR_FILE_MODE], ['aabb'])
+        errors = self.run_check(header, require_file_signature=True)
+        assert errors == []
+
+    def test_file_signatures_not_required(self):
+        header = make_header([REGULAR_FILE_MODE], [])
+        errors = self.run_check(header, require_file_signature=False)
+        assert errors == []