Introduce FetchKernelCareGPGKey action to obtain the GPG key for the …

…leapp utility

This action retrieves the KernelCare GPG key from the configured repository specifically for leapp.
Normally, leapp includes all necessary GPG keys within a designated configuration directory.
However, since KernelCare is not supported by AlmaLinux, we must manually fetch the GPG key
to ensure leapp can proceed with the conversion when KernelCare repository packages are installed.

centos2almaconverter/actions/extensions.py

@@ -1,5 +1,9 @@
 # Copyright 1999 - 2024. Plesk International GmbH. All rights reserved.
-from pleskdistup.common import action, util, leapp_configs, files
+from pleskdistup.common import action, util, leapp_configs, log, plesk, rpm, files
+
+import os
+import typing
+import urllib.request
 
 
 class FixupImunify(action.ActiveAction):
@@ -64,3 +68,72 @@ def estimate_prepare_time(self) -> int:
 
     def estimate_post_time(self) -> int:
         return 2 * 60
+
+
+class FetchKernelCareGPGKey(action.ActiveAction):
+    """This action fetches the KernelCare GPG key from the configured repository for leapp.
+    Usually leapp brings all required GPG keys with it inside specific configuration directory.
+    But KernelCare is not supported by AlmaLinux, so we need to fetch the GPG key manually to make
+    sure leapp will be able to proceed with the conversion when packages from KernelCare repository installed.
+    """
+
+    kernelcare_repofile: str = "/etc/yum.repos.d/kernelcare.repo"
+    leapp_gpg_keys_store: str = "/etc/leapp/files/vendors.d/rpm-gpg"
+
+    def __init__(self):
+        self.name = "fetching KernelCare GPG key"
+        self.kernelcare_gpg_keys_urls = self._get_kernelcare_gpg_keys_urls()
+
+    def _is_kernelcare_extension_installed(self) -> bool:
+        return "kernelcare-plesk" in dict(plesk.list_installed_extensions())
+
+    def _is_kernelcare_gpg_key_missing(self) -> bool:
+        return any(
+            not os.path.exists(self._get_kernelcare_gpg_target_path(key_url))
+            for key_url in self.kernelcare_gpg_keys_urls
+        )
+
+    def _is_required(self) -> bool:
+        return self._is_kernelcare_extension_installed() and self._is_kernelcare_gpg_key_missing()
+
+    def _get_kernelcare_gpg_keys_urls(self) -> typing.List[str]:
+        result = []
+        for repo_id, _, _, _, _, additional in rpm.extract_repodata(self.kernelcare_repofile):
+            if repo_id != "kernelcare":
+                continue
+
+            for line in additional:
+                if line.startswith("gpgkey="):
+                    result.append(line[len("gpgkey="):].rstrip())
+
+        return result
+
+    def _get_kernelcare_gpg_target_path(self, key_url: str) -> str:
+        return f"{self.leapp_gpg_keys_store}/{key_url.split('/')[-1]}"
+
+    def _prepare_action(self) -> action.ActionResult:
+        for key_url in self.kernelcare_gpg_keys_urls:
+            gpg_key_target_path = self._get_kernelcare_gpg_target_path(key_url)
+            log.debug(f"Going to save KernelCare GPG key from {key_url!r} to {gpg_key_target_path!r}")
+            if os.path.exists(gpg_key_target_path):
+                continue
+
+            try:
+                with urllib.request.urlopen(key_url) as response:
+                    with open(gpg_key_target_path, 'wb') as out_file:
+                        out_file.write(response.read())
+            except Exception as e:
+                raise RuntimeError(
+                    f"Unable to fetch KernelCare GPG key from '{key_url}': {e}. "
+                    f"To continue with the conversion, please manually install the key into "
+                    f"'{self.leapp_gpg_keys_store}' or remove the KernelCare extension."
+                ) from e
+
+        return action.ActionResult()
+
+    def _post_action(self) -> action.ActionResult:
+        return action.ActionResult()
+
+    def _revert_action(self) -> action.ActionResult:
+        # Since it's part of leapp configuration, it is fine to keep the key in the store.
+        return action.ActionResult()

centos2almaconverter/upgrader.py

@@ -127,6 +127,7 @@ def construct_actions(
                 common_actions.RevertChangesInGrub(),
                 centos2alma_actions.PrepareLeappConfigurationBackup(),
                 centos2alma_actions.RemoveOldMigratorThirparty(),
+                centos2alma_actions.FetchKernelCareGPGKey(),
                 centos2alma_actions.LeapReposConfiguration(),
                 centos2alma_actions.LeapChoicesConfiguration(),
                 centos2alma_actions.AdoptKolabRepositories(),