fix(kustomize): support kubectl 1.28+ (#5480)

support kubectl 1.28+

checkov/kustomize/runner.py

@@ -30,7 +30,7 @@
 from checkov.kubernetes.runner import Runner as K8sRunner
 from checkov.kubernetes.runner import _get_entity_abs_path
 from checkov.kustomize.image_referencer.manager import KustomizeImageReferencerManager
-from checkov.kustomize.utils import get_kustomize_version
+from checkov.kustomize.utils import get_kustomize_version, get_kubectl_version
 from checkov.runner_filter import RunnerFilter
 from checkov.common.graph.checks_infra.registry import BaseRegistry
 from checkov.common.typing import LibraryGraphConnector
@@ -452,22 +452,13 @@ def check_system_deps(self) -> str | None:
         logging.info(f"Checking necessary system dependancies for {self.check_type} checks.")
 
         if shutil.which(self.kubectl_command) is not None:
-            try:
-                proc = subprocess.run([self.kubectl_command, 'version', '--client=true'], capture_output=True)  # nosec
-                version_output = proc.stdout.decode("utf-8")
-
-                if "Client Version:" in version_output:
-                    kubectl_version_major = version_output.split('\n')[0].split('Major:\"')[1].split('"')[0]
-                    kubectl_version_minor = version_output.split('\n')[0].split('Minor:\"')[1].split('"')[0]
-                    kubectl_version = float(f"{kubectl_version_major}.{kubectl_version_minor}")
-                    if kubectl_version >= 1.14:
-                        logging.info(f"Found working version of {self.check_type} dependancy {self.kubectl_command}: {kubectl_version}")
-                        self.templateRendererCommand = self.kubectl_command
-                        return None
-
-            except Exception:
-                logging.debug(f"An error occured testing the {self.kubectl_command} command:", exc_info=True)
-
+            kubectl_version = get_kubectl_version(kubectl_command=self.kubectl_command)
+            if kubectl_version and kubectl_version >= 1.14:
+                logging.info(f"Found working version of {self.check_type} dependancy {self.kubectl_command}: {kubectl_version}")
+                self.templateRendererCommand = self.kubectl_command
+                return None
+            else:
+                return self.check_type
         elif shutil.which(self.kustomize_command) is not None:
             kustomize_version = get_kustomize_version(kustomize_command=self.kustomize_command)
             if kustomize_version:
@@ -482,8 +473,6 @@ def check_system_deps(self) -> str | None:
             logging.info(f"Could not find usable tools locally to process {self.check_type} checks. Framework will be disabled for this run.")
             return self.check_type
 
-        return None
-
     def _handle_overlay_case(self, file_path: str) -> None:
         for parent in pathlib.Path(file_path).parents:
             for potentialBase in self.potentialBases:
@@ -553,12 +542,15 @@ def _get_parsed_output(
             line_num += 1
         return cur_writer
 
-    def _get_kubectl_output(self, filePath: str, template_renderer_command: str, source_type: str | None) -> bytes:
+    def _get_kubectl_output(self, filePath: str, template_renderer_command: str, source_type: str | None) -> bytes | None:
         # Template out the Kustomizations to Kubernetes YAML
         if template_renderer_command == "kubectl":
             template_render_command_options = "kustomize"
-        if template_renderer_command == "kustomize":
+        elif template_renderer_command == "kustomize":
             template_render_command_options = "build"
+        else:
+            logging.error(f"Template renderer command has an invalid value: {template_renderer_command}")
+            return None
 
         add_origin_annotations_return_code = None
 
@@ -612,6 +604,9 @@ def get_binary_output(
         logging.debug(f"Kustomization at {file_path} likley a {source_type}")
         try:
             output = self._get_kubectl_output(file_path, template_renderer_command, source_type)
+            if output is None:
+                return None, None
+
             return output, file_path
         except Exception:
             logging.warning(f"Error building Kustomize output at dir: {file_path}.", exc_info=True)

checkov/kustomize/utils.py

@@ -20,6 +20,29 @@ def get_kustomize_version(kustomize_command: str) -> str | None:
 
         return kustomize_version
     except Exception:
-        logging.debug(f"An error occured testing the {kustomize_command} command:", exc_info=True)
+        logging.debug(f"An error occurred testing the {kustomize_command} command:", exc_info=True)
+
+    return None
+
+
+def get_kubectl_version(kubectl_command: str) -> float | None:
+    try:
+        proc = subprocess.run([kubectl_command, "version", "--client=true"], capture_output=True)  # nosec
+        version_output = proc.stdout.decode("utf-8")
+
+        if "Client Version:" in version_output:
+            if "Major:" in version_output and "Minor:" in version_output:
+                # version <= 1.27 output looks like 'Client Version: version.Info{Major:"1", Minor:"27", GitVersion:...}\n...'
+                kubectl_version_major = version_output.split("\n")[0].split('Major:"')[1].split('"')[0]
+                kubectl_version_minor = version_output.split("\n")[0].split('Minor:"')[1].split('"')[0]
+            else:
+                # version >= 1.28 output looks like 'Client Version: v1.28.0\n...'
+                kubectl_version_str = version_output.split("\n")[0].replace("Client Version: v", "")
+                kubectl_version_major, kubectl_version_minor, *_ = kubectl_version_str.split(".")
+            kubectl_version = float(f"{kubectl_version_major}.{kubectl_version_minor}")
+
+            return kubectl_version
+    except Exception:
+        logging.debug(f"An error occurred testing the {kubectl_command} command:", exc_info=True)
 
     return None

tests/kustomize/test_utils.py

@@ -2,7 +2,49 @@
 
 from pytest_mock import MockerFixture
 
-from checkov.kustomize.utils import get_kustomize_version
+from checkov.kustomize.utils import get_kustomize_version, get_kubectl_version
+
+
+def test_get_kubectl_version_v1_27(mocker: MockerFixture):
+    # given
+    subprocess_mock = MagicMock()
+    subprocess_mock.stdout = b'Client Version: version.Info{Major:"1", Minor:"27", GitVersion:"v1.27.2", GitCommit:"7f6f68fdabc4df88cfea2dcf9a19b2b830f1e647", GitTreeState:"clean", BuildDate:"2023-05-17T14:20:07Z", GoVersion:"go1.20.4", Compiler:"gc", Platform:"darwin/amd64"}\nKustomize Version: v5.0.1\n'
+
+    mocker.patch("checkov.kustomize.utils.subprocess.run", return_value=subprocess_mock)
+
+    # when
+    version = get_kubectl_version(kubectl_command="kubectl")
+
+    # then
+    assert version == 1.27
+
+
+def test_get_kubectl_version_v1_28(mocker: MockerFixture):
+    # given
+    subprocess_mock = MagicMock()
+    subprocess_mock.stdout = b"Client Version: v1.28.0\nKustomize Version: v5.0.4-0.20230601165947-6ce0bf390ce3\n"
+
+    mocker.patch("checkov.kustomize.utils.subprocess.run", return_value=subprocess_mock)
+
+    # when
+    version = get_kubectl_version(kubectl_command="kubectl")
+
+    # then
+    assert version == 1.28
+
+
+def test_get_kubectl_version_none(mocker: MockerFixture):
+    # given
+    subprocess_mock = MagicMock()
+    subprocess_mock.stdout = b"command not found: kubectl\n"
+
+    mocker.patch("checkov.kustomize.utils.subprocess.run", return_value=subprocess_mock)
+
+    # when
+    version = get_kubectl_version(kubectl_command="kubectl")
+
+    # then
+    assert version is None
 
 
 def test_get_kustomize_version_v4(mocker: MockerFixture):