diff --git a/node-config/sign_node.yml b/node-config/sign_node.yml index 49eee95..838e78a 100644 --- a/node-config/sign_node.yml +++ b/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: + +# 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 diff --git a/sign_node/config.py b/sign_node/config.py index af850b4..bdf6ae1 100644 --- a/sign_node/config.py +++ b/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}, diff --git a/sign_node/signer.py b/sign_node/signer.py index 3b3fc94..774dfe6 100644 --- a/sign_node/signer.py +++ b/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) diff --git a/tests/sign_node/test_signer.py b/tests/sign_node/test_signer.py index 8f5a96b..3fa7641 100644 --- a/tests/sign_node/test_signer.py +++ b/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 == []