diff --git a/prowler/providers/kubernetes/exceptions/exceptions.py b/prowler/providers/kubernetes/exceptions/exceptions.py new file mode 100644 index 0000000000..b7c8d9de90 --- /dev/null +++ b/prowler/providers/kubernetes/exceptions/exceptions.py @@ -0,0 +1,72 @@ +from prowler.exceptions.exceptions import ProwlerException + + +class KubernetesBaseException(ProwlerException): + """Base class for Kubernetes errors.""" + + KUBERNETES_ERROR_CODES = { + (1925, "KubernetesCloudResourceManagerAPINotUsedError"): { + "message": "Cloud Resource Manager API is not enabled, blocking access to necessary resources.", + "remediation": "Refer to the Kubernetes documentation to enable the Cloud Resource Manager API: https://kubernetes.io/docs/reference/access-authn-authz/rbac/", + }, + (1926, "KubernetesSetUpSessionError"): { + "message": "Failed to establish a Kubernetes session, preventing further actions.", + "remediation": "Verify your session setup, including credentials and Kubernetes cluster configuration. Refer to this guide for proper setup: https://kubernetes.io/docs/tasks/access-application-cluster/configure-access-multiple-clusters/", + }, + (1930, "KubernetesAPIError"): { + "message": "An error occurred while interacting with the Kubernetes API.", + "remediation": "Check the API request and ensure it is properly formatted. Refer to the Kubernetes API documentation for guidance: https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.22/", + }, + (1931, "KubernetesTimeoutError"): { + "message": "The request to the Kubernetes API timed out.", + "remediation": "Check the network connection and the Kubernetes API server status. For information on troubleshooting timeouts, refer to the Kubernetes documentation: https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.22/#-strong-timeout-strong-", + }, + (1934, "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/", + }, + } + + def __init__( + self, + code, + file=None, + original_exception=None, + message=None, + ): + provider = "Kubernetes" + error_info = self.KUBERNETES_ERROR_CODES.get((code, self.__class__.__name__)) + if message: + error_info["message"] = message + super().__init__( + code=code, + provider=provider, + file=file, + original_exception=original_exception, + error_info=error_info, + ) + + +class KubernetesError(KubernetesBaseException): + def __init__(self, code, file=None, original_exception=None, message=None): + super().__init__(code, file, original_exception, message) + + +class KubernetesCloudResourceManagerAPINotUsedError(KubernetesBaseException): + def __init__(self, code, file=None, original_exception=None, message=None): + super().__init__(code, file, original_exception, message) + + +class KubernetesSetUpSessionError(KubernetesBaseException): + def __init__(self, code, file=None, original_exception=None, message=None): + super().__init__(code, file, original_exception, message) + + +class KubernetesAPIError(KubernetesBaseException): + def __init__(self, code, file=None, original_exception=None, message=None): + super().__init__(code, file, original_exception, message) + + +class KubernetesTimeoutError(KubernetesBaseException): + def __init__(self, code, file=None, original_exception=None, message=None): + super().__init__(code, file, original_exception, message) diff --git a/prowler/providers/kubernetes/kubernetes_provider.py b/prowler/providers/kubernetes/kubernetes_provider.py index 1f73c8de05..ff58f442bb 100644 --- a/prowler/providers/kubernetes/kubernetes_provider.py +++ b/prowler/providers/kubernetes/kubernetes_provider.py @@ -1,7 +1,9 @@ import os from colorama import Fore, Style +from kubernetes.client.exceptions import ApiException from kubernetes.config.config_exception import ConfigException +from requests.exceptions import Timeout from kubernetes import client, config from prowler.config.config import ( @@ -12,6 +14,13 @@ from prowler.lib.logger import logger from prowler.lib.utils.utils import print_boxes from prowler.providers.common.models import Audit_Metadata, Connection from prowler.providers.common.provider import Provider +from prowler.providers.kubernetes.exceptions.exceptions import ( + KubernetesAPIError, + KubernetesCloudResourceManagerAPINotUsedError, + KubernetesError, + KubernetesSetUpSessionError, + KubernetesTimeoutError, +) from prowler.providers.kubernetes.lib.mutelist.mutelist import KubernetesMutelist from prowler.providers.kubernetes.models import ( KubernetesIdentityInfo, @@ -59,8 +68,9 @@ class KubernetesProvider(Provider): if not self._session.api_client: logger.critical("Failed to set up a Kubernetes session.") - # TODO: add custom exception once we have the Kubernetes exceptions - raise SystemExit + raise KubernetesCloudResourceManagerAPINotUsedError( + message="Failed to set up a Kubernetes session." + ) self._identity = KubernetesIdentityInfo( context=self._session.context["name"], @@ -205,13 +215,19 @@ class KubernetesProvider(Provider): KubernetesProvider.setup_session(kubeconfig_file, input_context) client.CoreV1Api().list_namespace(timeout_seconds=2, _request_timeout=2) return Connection(is_connected=True) + except ApiException as api_error: + logger.critical( + f"ApiException[{api_error.__traceback__.tb_lineno}]: {api_error}" + ) + if raise_on_exception: + raise KubernetesAPIError(original_exception=api_error) + return Connection(error=api_error) except Exception as error: logger.critical( f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) if raise_on_exception: - # TODO: add custom exception once we have the Kubernetes exceptions - raise error + raise KubernetesSetUpSessionError(original_exception=error) return Connection(error=error) def search_and_save_roles( @@ -245,8 +261,9 @@ class KubernetesProvider(Provider): logger.critical( f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) - # TODO: add custom exception once we have the Kubernetes exceptions - raise error + raise KubernetesError( + original_exception=error, file=os.path.abspath(__file__) + ) def get_context_user_roles(self): """ @@ -276,12 +293,18 @@ class KubernetesProvider(Provider): ) logger.info("Context user roles retrieved successfully.") return roles + except ApiException as api_error: + logger.critical( + f"ApiException[{api_error.__traceback__.tb_lineno}]: {api_error}" + ) + raise KubernetesAPIError(original_exception=api_error) + except KubernetesError as error: + raise error except Exception as error: logger.critical( f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) - # TODO: add custom exception once we have the Kubernetes exceptions - raise error + raise KubernetesError(original_exception=error) def get_all_namespaces(self) -> list[str]: """ @@ -295,12 +318,27 @@ class KubernetesProvider(Provider): namespaces = [item.metadata.name for item in namespace_list.items] logger.info("All namespaces retrieved successfully.") return namespaces + except ApiException as api_error: + logger.critical( + f"ApiException[{api_error.__traceback__.tb_lineno}]: {api_error}" + ) + raise KubernetesAPIError( + original_exception=api_error, file=os.path.abspath(__file__) + ) + except Timeout as timeout_error: + logger.critical( + f"Timeout[{timeout_error.__traceback__.tb_lineno}]: {timeout_error}" + ) + raise KubernetesTimeoutError( + original_exception=timeout_error, file=os.path.abspath(__file__) + ) except Exception as error: logger.critical( f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) - # TODO: add custom exception once we have the Kubernetes exceptions - raise error + raise KubernetesError( + original_exception=error, file=os.path.abspath(__file__) + ) def get_pod_current_namespace(self): """