ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-23 20:42:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(kubernetes): add custom exception class (#4912)

Browse Source
This commit is contained in:
Pedro Martín
2024-09-05 16:52:34 +02:00
committed by GitHub
parent bd336250ee
commit d8d348f609
2 changed files with 120 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/providers/kubernetes/exceptions/exceptions.py
+72
View File
@@ -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)
prowler/providers/kubernetes/kubernetes_provider.py
+48 -10
View File
@@ -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):
"""