ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-24 13:01:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

docs(kubernetes): improve docstrings for methods (#5717)

Browse Source
This commit is contained in:
Pedro Martín
2024-11-18 15:18:57 +01:00
committed by GitHub
parent f53a887291
commit 0a63e707c2
1 changed files with 152 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/providers/kubernetes/kubernetes_provider.py
+152
View File
@@ -33,6 +33,28 @@ from prowler.providers.kubernetes.models import (
class KubernetesProvider(Provider):
"""
Represents the Kubernetes provider.
Attributes:
_type (str): The provider type, wich is 'kubernetes'.
_session (KubernetesSession): The Kubernetes session.
_namespaces (list): The list of namespaces to audit.
_audit_config (dict): The audit configuration.
_identity (KubernetesIdentityInfo): The Kubernetes identity information.
_mutelist (dict): The mutelist.
audit_metadata (Audit_Metadata): The audit metadata.
Methods:
setup_session: Sets up the Kubernetes session.
test_connection: Tests the connection to the Kubernetes cluster.
search_and_save_roles: Searches for and saves roles.
get_context_user_roles: Retrieves the context user roles.
get_all_namespaces: Retrieves all namespaces.
get_pod_current_namespace: Retrieves the current namespace from the pod's mounted service account info.
print_credentials: Prints the Kubernetes credentials.
"""
_type: str = "kubernetes"
_session: KubernetesSession
_namespaces: list
@@ -56,6 +78,7 @@ class KubernetesProvider(Provider):
):
"""
Initializes the KubernetesProvider instance.
Args:
kubeconfig_file (str): Path to the kubeconfig file.
kubeconfig_content (dict): Content of the kubeconfig file.
@@ -66,6 +89,60 @@ class KubernetesProvider(Provider):
fixer_config (dict): Fixer configuration.
mutelist_path (str): Path to the mutelist file.
mutelist_content (dict): Mutelist content.
Raises:
KubernetesCloudResourceManagerAPINotUsedError: If the Kubernetes Cloud Resource Manager API is not used.
KubernetesError: If an error occurs.
Returns:
KubernetesProvider: The KubernetesProvider instance.
Usage:
- Authentication: The provider can be instantiated in two ways:
1. Using the kubeconfig file.
>>> provider = KubernetesProvider(
... kubeconfig_file="~/.kube/config",
... context="my-context",
... namespace=["default"],
... )
2. Using the kubeconfig content.
>>> provider = KubernetesProvider(
... kubeconfig_content={"kubecofig": "content"},
... context="my-context",
... namespace=["default"],
... )
- Namespace and context: The provider can be instantiated with or without specifying the namespace and context.
- Without specifying the namespace:
>>> provider = KubernetesProvider(
... kubeconfig_file="~/.kube/config",
... context="my-context",
... )
- With specifying the namespace:
>>> provider = KubernetesProvider(
... kubeconfig_file="~/.kube/config",
... context="my-context",
... namespace=["default"],
... )
- With specifying the context:
>>> provider = KubernetesProvider(
... kubeconfig_file="~/.kube/config",
... context="my-context",
... namespace=["default"],
... )
- Configuration: The provider can be instantiated with or without specifying the configuration.
- Without specifying the configuration:
>>> provider = KubernetesProvider(
... kubeconfig_file="~/.kube/config",
... context="my-context",
... namespace=["default"],
... )
- With specifying the configuration:
>>> provider = KubernetesProvider(
... kubeconfig_file="~/.kube/config",
... context="my-context",
... namespace=["default"],
... config_path="path/to/config.yaml",
... )
"""
logger.info("Instantiating Kubernetes Provider ...")
@@ -160,6 +237,11 @@ class KubernetesProvider(Provider):
Returns:
Tuple: A tuple containing the API client and the context.
Raises:
KubernetesInvalidKubeConfigFileError: If the kubeconfig file is invalid.
KubernetesInvalidProviderIdError: If the provider ID is invalid.
KubernetesSetUpSessionError: If an error occurs while setting up the session.
"""
logger.info(f"Using kubeconfig file: {kubeconfig_file}")
try:
@@ -250,8 +332,45 @@ class KubernetesProvider(Provider):
namespace (str): Namespace name.
provider_id (str): Provider ID to use, in this case, the Kubernetes context.
raise_on_exception (bool): Whether to raise an exception on error.
Returns:
Connection: A Connection object.
Raises:
KubernetesInvalidKubeConfigFileError: If the kubeconfig file is invalid.
KubernetesInvalidProviderIdError: If the provider ID is invalid.
KubernetesSetUpSessionError: If an error occurs while setting up the session.
KubernetesAPIError: If an error occurs while testing the connection.
Usage:
- Using the kubeconfig file:
>>> connection = KubernetesProvider.test_connection(
... kubeconfig_file="~/.kube/config",
... namespace="default",
... provider_id="my-context",
... raise_on_exception=True,
... )
- Using the kubeconfig content:
>>> connection = KubernetesProvider.test_connection(
... kubeconfig_content={"kubecofig": "content"},
... namespace="default",
... provider_id="my-context",
... raise_on_exception=True,
... )
- Using the namespace:
>>> connection = KubernetesProvider.test_connection(
... kubeconfig_file="~/.kube/config",
... namespace="default",
... provider_id="my-context",
... raise_on_exception=True,
... )
- Without raising an exception:
>>> connection = KubernetesProvider.test_connection(
... kubeconfig_file="~/.kube/config",
... namespace="default",
... provider_id="my-context",
... raise_on_exception=False,
... )
"""
try:
KubernetesProvider.setup_session(
@@ -316,6 +435,17 @@ class KubernetesProvider(Provider):
Returns:
list: A list containing the roles.
Raises:
KubernetesError: If an error occurs.
Usage:
>>> roles = self.search_and_save_roles(
... roles=[],
... role_bindings=rbac_api.list_cluster_role_binding().items,
... context_user="my-user",
... role_binding_type="ClusterRole",
... )
"""
try:
for rb in role_bindings:
@@ -343,6 +473,12 @@ class KubernetesProvider(Provider):
Returns:
list: A list containing the context user roles.
Raises:
KubernetesError: If an error occurs.
Usage:
>>> roles = self.get_context_user_roles()
"""
try:
rbac_api = client.RbacAuthorizationV1Api()
@@ -379,8 +515,17 @@ class KubernetesProvider(Provider):
def get_all_namespaces(self) -> list[str]:
"""
Retrieves all namespaces.
Returns:
list: A list containing all namespace names.
Raises:
KubernetesAPIError: If an error occurs while retrieving the namespaces.
KubernetesTimeoutError: If a timeout occurs while retrieving the namespaces.
KubernetesError: If an error occurs.
Usage:
>>> namespaces = self.get_all_namespaces()
"""
try:
v1 = client.CoreV1Api()
@@ -413,8 +558,12 @@ class KubernetesProvider(Provider):
def get_pod_current_namespace(self):
"""
Retrieves the current namespace from the pod's mounted service account info.
Returns:
str: The current namespace.
Usage:
>>> namespace = self.get_pod_current_namespace()
"""
try:
with open(
@@ -430,6 +579,9 @@ class KubernetesProvider(Provider):
def print_credentials(self):
"""
Prints the Kubernetes credentials.
Usage:
>>> self.print_credentials()
"""
if self._identity.context == "In-Cluster":
report_lines = [