fix(test-connection): Handle yaml kubeconfig and add errors

prowler/providers/kubernetes/exceptions/exceptions.py

@@ -25,6 +25,14 @@ class KubernetesBaseException(ProwlerException):
             "message": "An error occurred in the Kubernetes provider.",
             "remediation": "Check the provider code and configuration to identify the issue. For more information on troubleshooting Kubernetes providers, refer to the Kubernetes documentation: https://kubernetes.io/docs/reference/",
         },
+        (1936, "KubernetesError"): {
+            "message": "An error occurred in the Kubernetes provider.",
+            "remediation": "Check the provider code and configuration to identify the issue. For more information on troubleshooting Kubernetes providers, refer to the Kubernetes documentation: https://kubernetes.io/docs/reference/",
+        },
+        (1936, "KubernetesInvalidKubeConfigFileError"): {
+            "message": "The provided kube-config is invalid.",
+            "remediation": "Review the kube-config and the attached error to get more details. Please, refer to the Kubernetes config documentation: https://kubernetes.io/docs/reference/config-api/kubeconfig.v1/#Config",
+        },
     }
 
     def __init__(
@@ -70,3 +78,8 @@ def __init__(self, file=None, original_exception=None, message=None):
 class KubernetesTimeoutError(KubernetesBaseException):
     def __init__(self, file=None, original_exception=None, message=None):
         super().__init__(1934, file, original_exception, message)
+
+
+class KubernetesInvalidKubeConfigFileError(KubernetesBaseException):
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(1936, file, original_exception, message)

prowler/providers/kubernetes/kubernetes_provider.py

@@ -4,6 +4,7 @@
 from kubernetes.client.exceptions import ApiException
 from kubernetes.config.config_exception import ConfigException
 from requests.exceptions import Timeout
+from yaml import safe_load
 
 from kubernetes import client, config
 from prowler.config.config import get_default_mute_file_path
@@ -15,6 +16,7 @@
     KubernetesAPIError,
     KubernetesCloudResourceManagerAPINotUsedError,
     KubernetesError,
+    KubernetesInvalidKubeConfigFileError,
     KubernetesSetUpSessionError,
     KubernetesTimeoutError,
 )
@@ -195,6 +197,14 @@ def setup_session(
             else:
                 context = config.list_kube_config_contexts()[1]
             return KubernetesSession(api_client=client.ApiClient(), context=context)
+
+        except ConfigException as config_error:
+            logger.critical(
+                f"{config_error.__class__.__name__}[{config_error.__traceback__.tb_lineno}]: {config_error}"
+            )
+            raise KubernetesInvalidKubeConfigFileError(
+                original_exception=config_error, file=os.path.abspath(__file__)
+            )
         except Exception as error:
             logger.critical(
                 f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -208,7 +218,7 @@ def test_connection(
         kubeconfig_file: str = "~/.kube/config",
         kubeconfig_content: dict = None,
         namespace: str = None,
-        input_context: str = "",
+        context: str = None,
         raise_on_exception: bool = True,
     ) -> Connection:
         """
@@ -218,14 +228,15 @@ def test_connection(
             kubeconfig_file (str): Path to the kubeconfig file.
             kubeconfig_content (dict): Content of the kubeconfig file.
             namespace (str): Namespace name.
-            input_context (str): Context name.
+            context (str): Context name.
             raise_on_exception (bool): Whether to raise an exception on error.
         Returns:
             Connection: A Connection object.
         """
         try:
+            kubeconfig_content = safe_load(kubeconfig_content)
             KubernetesProvider.setup_session(
-                kubeconfig_file, kubeconfig_content, input_context
+                kubeconfig_file, kubeconfig_content, context
             )
             if namespace:
                 client.CoreV1Api().list_namespaced_pod(

tests/providers/kubernetes/kubernetes_provider_test.py

@@ -106,42 +106,12 @@ def test_kubernetes_test_connection_with_kubeconfig_content(
             client.V1Namespace(metadata=client.V1ObjectMeta(name="namespace-1")),
         ]
 
-        kubeconfig_content = {
-            "apiVersion": "v1",
-            "clusters": [
-                {
-                    "cluster": {
-                        "server": "https://kubernetes.example.com",
-                    },
-                    "name": "example-cluster",
-                }
-            ],
-            "contexts": [
-                {
-                    "context": {
-                        "cluster": "example-cluster",
-                        "user": "example-user",
-                    },
-                    "name": "example-context",
-                }
-            ],
-            "current-context": "example-context",
-            "kind": "Config",
-            "preferences": {},
-            "users": [
-                {
-                    "name": "example-user",
-                    "user": {
-                        "token": "EXAMPLE_TOKEN",
-                    },
-                }
-            ],
-        }
+        kubeconfig_content = '{"apiVersion": "v1", "clusters": [{"cluster": {"server": "https://kubernetes.example.com"}, "name": "example-cluster"}], "contexts": [{"context": {"cluster": "example-cluster", "user": "example-user"}, "name": "example-context"}], "current-context": "example-context", "kind": "Config", "preferences": {}, "users": [{"name": "example-user", "user": {"token": "EXAMPLE_TOKEN"}}]}'
 
         connection = KubernetesProvider.test_connection(
             kubeconfig_file=None,
             kubeconfig_content=kubeconfig_content,
-            input_context="example-context",
+            context="example-context",
             raise_on_exception=False,
         )
 
@@ -175,8 +145,8 @@ def test_kubernetes_test_connection_with_kubeconfig_file(
 
         connection = KubernetesProvider.test_connection(
             kubeconfig_file="dummy_kubeconfig_path",
-            kubeconfig_content={},
-            input_context="test-context",
+            kubeconfig_content="",
+            context="test-context",
             raise_on_exception=False,
         )
 
@@ -213,9 +183,9 @@ def test_kubernetes_test_connection_with_namespace_input(
 
         connection = KubernetesProvider.test_connection(
             kubeconfig_file="dummy_kubeconfig_path",
-            kubeconfig_content={},
+            kubeconfig_content="",
             namespace="test-namespace",
-            input_context="test-context",
+            context="test-context",
             raise_on_exception=False,
         )