mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 13:01:56 +00:00
feat(k8s): Add kubeconfig content static authentication (#5370)
Co-authored-by: Pepe Fagoaga <pepe@prowler.com>
This commit is contained in:
Sergio Garcia
@@ -5,26 +5,30 @@ class KubernetesBaseException(ProwlerException):
|
||||
"""Base class for Kubernetes errors."""
|
||||
|
||||
KUBERNETES_ERROR_CODES = {
|
||||
(1925, "KubernetesCloudResourceManagerAPINotUsedError"): {
|
||||
(1931, "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"): {
|
||||
(1932, "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"): {
|
||||
(1933, "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"): {
|
||||
(1934, "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"): {
|
||||
(1935, "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__(
|
||||
@@ -48,25 +52,30 @@ class KubernetesBaseException(ProwlerException):
|
||||
|
||||
|
||||
class KubernetesError(KubernetesBaseException):
|
||||
def __init__(self, code, file=None, original_exception=None, message=None):
|
||||
super().__init__(code, file, original_exception, message)
|
||||
def __init__(self, file=None, original_exception=None, message=None):
|
||||
super().__init__(1935, 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)
|
||||
def __init__(self, file=None, original_exception=None, message=None):
|
||||
super().__init__(1931, 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)
|
||||
def __init__(self, file=None, original_exception=None, message=None):
|
||||
super().__init__(1932, 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)
|
||||
def __init__(self, file=None, original_exception=None, message=None):
|
||||
super().__init__(1933, 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)
|
||||
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)
|
||||
|
||||
@@ -4,6 +4,7 @@ from colorama import Fore, Style
|
||||
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 @@ from prowler.providers.kubernetes.exceptions.exceptions import (
|
||||
KubernetesAPIError,
|
||||
KubernetesCloudResourceManagerAPINotUsedError,
|
||||
KubernetesError,
|
||||
KubernetesInvalidKubeConfigFileError,
|
||||
KubernetesSetUpSessionError,
|
||||
KubernetesTimeoutError,
|
||||
)
|
||||
@@ -42,11 +44,13 @@ class KubernetesProvider(Provider):
|
||||
namespace: list = None,
|
||||
audit_config: dict = {},
|
||||
fixer_config: dict = {},
|
||||
kubeconfig_content: dict = None,
|
||||
):
|
||||
"""
|
||||
Initializes the KubernetesProvider instance.
|
||||
Args:
|
||||
kubeconfig_file (str): Path to the kubeconfig file.
|
||||
kubeconfig_content (dict): Content of the kubeconfig file.
|
||||
context (str): Context name.
|
||||
namespace (list): List of namespaces.
|
||||
audit_config (dict): Audit configuration.
|
||||
@@ -54,7 +58,7 @@ class KubernetesProvider(Provider):
|
||||
"""
|
||||
|
||||
logger.info("Instantiating Kubernetes Provider ...")
|
||||
self._session = self.setup_session(kubeconfig_file, context)
|
||||
self._session = self.setup_session(kubeconfig_file, kubeconfig_content, context)
|
||||
if not namespace:
|
||||
logger.info("Retrieving all namespaces ...")
|
||||
self._namespaces = self.get_all_namespaces()
|
||||
@@ -139,12 +143,17 @@ class KubernetesProvider(Provider):
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def setup_session(kubeconfig_file, input_context) -> KubernetesSession:
|
||||
def setup_session(
|
||||
kubeconfig_file: str = None,
|
||||
kubeconfig_content: dict = None,
|
||||
input_context: str = None,
|
||||
) -> KubernetesSession:
|
||||
"""
|
||||
Sets up the Kubernetes session.
|
||||
|
||||
Args:
|
||||
kubeconfig_file (str): Path to the kubeconfig file.
|
||||
kubeconfig_content (dict): Content of the kubeconfig file.
|
||||
input_context (str): Context name.
|
||||
|
||||
Returns:
|
||||
@@ -152,26 +161,34 @@ class KubernetesProvider(Provider):
|
||||
"""
|
||||
logger.info(f"Using kubeconfig file: {kubeconfig_file}")
|
||||
try:
|
||||
config.load_kube_config(
|
||||
config_file=(
|
||||
os.path.abspath(kubeconfig_file)
|
||||
if kubeconfig_file != "~/.kube/config"
|
||||
else os.path.expanduser(kubeconfig_file)
|
||||
),
|
||||
context=input_context,
|
||||
)
|
||||
except ConfigException:
|
||||
# If the kubeconfig file is not found, try to use the in-cluster config
|
||||
logger.info("Using in-cluster config")
|
||||
config.load_incluster_config()
|
||||
context = {
|
||||
"name": "In-Cluster",
|
||||
"context": {
|
||||
"cluster": "in-cluster", # Placeholder, as the real cluster name is not available
|
||||
"user": "service-account-name", # Also a placeholder
|
||||
},
|
||||
}
|
||||
else:
|
||||
if kubeconfig_content:
|
||||
config.load_kube_config_from_dict(
|
||||
kubeconfig_content, context=input_context
|
||||
)
|
||||
else:
|
||||
try:
|
||||
config.load_kube_config(
|
||||
config_file=(
|
||||
os.path.abspath(kubeconfig_file)
|
||||
if kubeconfig_file != "~/.kube/config"
|
||||
else os.path.expanduser(kubeconfig_file)
|
||||
),
|
||||
context=input_context,
|
||||
)
|
||||
except ConfigException:
|
||||
# If the kubeconfig file is not found, try to use the in-cluster config
|
||||
logger.info("Using in-cluster config")
|
||||
config.load_incluster_config()
|
||||
context = {
|
||||
"name": "In-Cluster",
|
||||
"context": {
|
||||
"cluster": "in-cluster", # Placeholder, as the real cluster name is not available
|
||||
"user": "service-account-name", # Also a placeholder
|
||||
},
|
||||
}
|
||||
return KubernetesSession(
|
||||
api_client=client.ApiClient(), context=context
|
||||
)
|
||||
if input_context:
|
||||
contexts = config.list_kube_config_contexts()[0]
|
||||
for context_item in contexts:
|
||||
@@ -179,12 +196,29 @@ class KubernetesProvider(Provider):
|
||||
context = context_item
|
||||
else:
|
||||
context = config.list_kube_config_contexts()[1]
|
||||
return KubernetesSession(api_client=client.ApiClient(), context=context)
|
||||
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}"
|
||||
)
|
||||
raise KubernetesSetUpSessionError(
|
||||
original_exception=error, file=os.path.abspath(__file__)
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def test_connection(
|
||||
kubeconfig_file: str = "~/.kube/config",
|
||||
input_context: str = "",
|
||||
kubeconfig_content: dict = None,
|
||||
namespace: str = None,
|
||||
context: str = None,
|
||||
raise_on_exception: bool = True,
|
||||
) -> Connection:
|
||||
"""
|
||||
@@ -192,15 +226,32 @@ class KubernetesProvider(Provider):
|
||||
|
||||
Args:
|
||||
kubeconfig_file (str): Path to the kubeconfig file.
|
||||
input_context (str): Context name.
|
||||
|
||||
kubeconfig_content (dict): Content of the kubeconfig file.
|
||||
namespace (str): Namespace name.
|
||||
context (str): Context name.
|
||||
raise_on_exception (bool): Whether to raise an exception on error.
|
||||
Returns:
|
||||
Connection: A Connection object.
|
||||
"""
|
||||
try:
|
||||
KubernetesProvider.setup_session(kubeconfig_file, input_context)
|
||||
client.CoreV1Api().list_namespace(timeout_seconds=2, _request_timeout=2)
|
||||
kubeconfig_content = safe_load(kubeconfig_content)
|
||||
KubernetesProvider.setup_session(
|
||||
kubeconfig_file, kubeconfig_content, context
|
||||
)
|
||||
if namespace:
|
||||
client.CoreV1Api().list_namespaced_pod(
|
||||
namespace, timeout_seconds=2, _request_timeout=2
|
||||
)
|
||||
else:
|
||||
client.CoreV1Api().list_namespace(timeout_seconds=2, _request_timeout=2)
|
||||
return Connection(is_connected=True)
|
||||
except KubernetesSetUpSessionError as setup_session_error:
|
||||
logger.critical(
|
||||
f"KubernetesSetUpSessionError[{setup_session_error.__traceback__.tb_lineno}]: {setup_session_error}"
|
||||
)
|
||||
if raise_on_exception:
|
||||
raise setup_session_error
|
||||
return Connection(error=setup_session_error)
|
||||
except ApiException as api_error:
|
||||
logger.critical(
|
||||
f"ApiException[{api_error.__traceback__.tb_lineno}]: {api_error}"
|
||||
|
||||
@@ -78,118 +78,116 @@ class TestKubernetesProvider:
|
||||
|
||||
assert kubernetes_provider.audit_config == KUBERNETES_CONFIG
|
||||
|
||||
# @patch("kubernetes.client.RbacAuthorizationV1Api")
|
||||
# @patch("kubernetes.config.list_kube_config_contexts")
|
||||
# @patch("kubernetes.config.load_incluster_config")
|
||||
# @patch("kubernetes.config.load_kube_config")
|
||||
# def test_get_context_user_roles(
|
||||
# self,
|
||||
# mock_list_kube_config_contexts,
|
||||
# mock_rbac_api,
|
||||
# ):
|
||||
# mock_list_kube_config_contexts.return_value = (
|
||||
# [
|
||||
# {
|
||||
# "name": "context_name",
|
||||
# "context": {"cluster": "test-cluster", "user": "test-user"},
|
||||
# }
|
||||
# ],
|
||||
# 0,
|
||||
# )
|
||||
@patch(
|
||||
"prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespace"
|
||||
)
|
||||
@patch("kubernetes.config.list_kube_config_contexts")
|
||||
@patch("kubernetes.config.load_kube_config_from_dict")
|
||||
def test_kubernetes_test_connection_with_kubeconfig_content(
|
||||
self,
|
||||
mock_load_kube_config_from_dict,
|
||||
mock_list_kube_config_contexts,
|
||||
mock_list_namespace,
|
||||
):
|
||||
mock_load_kube_config_from_dict.return_value = None
|
||||
mock_list_kube_config_contexts.return_value = (
|
||||
[
|
||||
{
|
||||
"name": "example-context",
|
||||
"context": {
|
||||
"cluster": "example-cluster",
|
||||
"user": "example-user",
|
||||
},
|
||||
}
|
||||
],
|
||||
None,
|
||||
)
|
||||
mock_list_namespace.return_value.items = [
|
||||
client.V1Namespace(metadata=client.V1ObjectMeta(name="namespace-1")),
|
||||
]
|
||||
|
||||
# # Mock the RbacAuthorizationV1Api methods
|
||||
# cluster_role_binding = MagicMock()
|
||||
# role_binding = MagicMock()
|
||||
# cluster_role_binding.list_cluster_role_binding.return_value = MagicMock(
|
||||
# items=[]
|
||||
# )
|
||||
# role_binding.list_role_binding_for_all_namespaces.return_value = MagicMock(
|
||||
# items=[]
|
||||
# )
|
||||
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"}}]}'
|
||||
|
||||
# mock_rbac_api.return_value = MagicMock(
|
||||
# list_cluster_role_binding=cluster_role_binding.list_cluster_role_binding,
|
||||
# list_role_binding_for_all_namespaces=role_binding.list_role_binding_for_all_namespaces,
|
||||
# )
|
||||
connection = KubernetesProvider.test_connection(
|
||||
kubeconfig_file=None,
|
||||
kubeconfig_content=kubeconfig_content,
|
||||
context="example-context",
|
||||
raise_on_exception=False,
|
||||
)
|
||||
|
||||
# args = Namespace(kubeconfig_file=None, context=None, only_logs=False)
|
||||
# provider = KubernetesProvider(args)
|
||||
assert connection.is_connected
|
||||
assert connection.error is None
|
||||
|
||||
# roles = provider.get_context_user_roles()
|
||||
@patch(
|
||||
"prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespace"
|
||||
)
|
||||
@patch("kubernetes.config.list_kube_config_contexts")
|
||||
@patch("kubernetes.config.load_kube_config")
|
||||
def test_kubernetes_test_connection_with_kubeconfig_file(
|
||||
self, mock_load_kube_config, mock_list_kube_config_contexts, mock_list_namespace
|
||||
):
|
||||
mock_load_kube_config.return_value = None
|
||||
mock_list_kube_config_contexts.return_value = (
|
||||
[
|
||||
{
|
||||
"name": "test-context",
|
||||
"context": {
|
||||
"cluster": "test-cluster",
|
||||
"user": "test-user",
|
||||
},
|
||||
}
|
||||
],
|
||||
None,
|
||||
)
|
||||
mock_list_namespace.return_value.items = [
|
||||
client.V1Namespace(metadata=client.V1ObjectMeta(name="namespace-1")),
|
||||
]
|
||||
|
||||
# assert isinstance(roles, list)
|
||||
connection = KubernetesProvider.test_connection(
|
||||
kubeconfig_file="dummy_kubeconfig_path",
|
||||
kubeconfig_content="",
|
||||
context="test-context",
|
||||
raise_on_exception=False,
|
||||
)
|
||||
|
||||
# @patch("kubernetes.client.RbacAuthorizationV1Api")
|
||||
# @patch("kubernetes.client.ApiClient")
|
||||
# @patch("kubernetes.config.load_kube_config")
|
||||
# @patch("kubernetes.config.load_incluster_config")
|
||||
# @patch("kubernetes.config.list_kube_config_contexts")
|
||||
# @patch("sys.stdout", new_callable=MagicMock)
|
||||
# def test_print_credentials(
|
||||
# self,
|
||||
# mock_list_kube_config_contexts,
|
||||
# ):
|
||||
# mock_list_kube_config_contexts.return_value = (
|
||||
# [
|
||||
# {
|
||||
# "name": "context_name",
|
||||
# "context": {"cluster": "test-cluster", "user": "test-user"},
|
||||
# }
|
||||
# ],
|
||||
# 0,
|
||||
# )
|
||||
assert connection.is_connected
|
||||
assert connection.error is None
|
||||
|
||||
# args = Namespace(kubeconfig_file=None, context=None, only_logs=False)
|
||||
# provider = KubernetesProvider(args)
|
||||
# provider.context = {
|
||||
# "context": {"cluster": "test-cluster", "user": "test-user"},
|
||||
# "namespace": "default",
|
||||
# }
|
||||
# provider.get_context_user_roles = MagicMock(return_value=["ClusterRole: admin"])
|
||||
@patch(
|
||||
"prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespaced_pod"
|
||||
)
|
||||
@patch("kubernetes.config.list_kube_config_contexts")
|
||||
@patch("kubernetes.config.load_kube_config")
|
||||
def test_kubernetes_test_connection_with_namespace_input(
|
||||
self,
|
||||
mock_load_kube_config,
|
||||
mock_list_kube_config_contexts,
|
||||
mock_list_namespaced_pod,
|
||||
):
|
||||
mock_load_kube_config.return_value = None
|
||||
mock_list_kube_config_contexts.return_value = (
|
||||
[
|
||||
{
|
||||
"name": "test-context",
|
||||
"context": {
|
||||
"cluster": "test-cluster",
|
||||
"user": "test-user",
|
||||
},
|
||||
}
|
||||
],
|
||||
None,
|
||||
)
|
||||
mock_list_namespaced_pod.return_value.items = [
|
||||
client.V1Pod(metadata=client.V1ObjectMeta(name="pod-1")),
|
||||
]
|
||||
|
||||
# # Capture print output
|
||||
# captured_output = io.StringIO()
|
||||
# sys.stdout = captured_output
|
||||
connection = KubernetesProvider.test_connection(
|
||||
kubeconfig_file="dummy_kubeconfig_path",
|
||||
kubeconfig_content="",
|
||||
namespace="test-namespace",
|
||||
context="test-context",
|
||||
raise_on_exception=False,
|
||||
)
|
||||
|
||||
# provider.print_credentials()
|
||||
|
||||
# # Reset standard output
|
||||
# sys.stdout = sys.__stdout__
|
||||
|
||||
# output = captured_output.getvalue()
|
||||
# assert "[test-cluster]" in output
|
||||
# assert "[test-user]" in output
|
||||
# assert "[default]" in output
|
||||
# assert "[ClusterRole: admin]" in output
|
||||
|
||||
# @patch("kubernetes.client.RbacAuthorizationV1Api")
|
||||
# @patch("kubernetes.config.list_kube_config_contexts")
|
||||
# @patch("kubernetes.config.load_incluster_config")
|
||||
# @patch("kubernetes.config.load_kube_config")
|
||||
# def test_search_and_save_roles(
|
||||
# self,
|
||||
# mock_list_kube_config_contexts,
|
||||
# mock_rbac_api,
|
||||
# ):
|
||||
# mock_list_kube_config_contexts.return_value = (
|
||||
# [
|
||||
# {
|
||||
# "name": "context_name",
|
||||
# "context": {"cluster": "test-cluster", "user": "test-user"},
|
||||
# }
|
||||
# ],
|
||||
# 0,
|
||||
# )
|
||||
# mock_rbac_api.return_value.list_cluster_role_binding.return_value = MagicMock(
|
||||
# items=[]
|
||||
# )
|
||||
# mock_rbac_api.return_value.list_role_binding_for_all_namespaces.return_value = (
|
||||
# MagicMock(items=[])
|
||||
# )
|
||||
|
||||
# args = Namespace(kubeconfig_file=None, context=None, only_logs=False)
|
||||
# provider = KubernetesProvider(args)
|
||||
# provider.context = {"context": {"user": "test-user"}}
|
||||
|
||||
# roles = provider.search_and_save_roles([], [], "test-user", "ClusterRole")
|
||||
# assert isinstance(roles, list)
|
||||
assert connection.is_connected
|
||||
assert connection.error is None
|
||||
|
||||
Reference in New Issue
Block a user