chore(providers): Remove get_output_mapping (#5484)

docs/developer-guide/provider.md

@@ -44,7 +44,6 @@ class Provider(ABC):
     Methods:
         print_credentials(): Displays the provider's credentials used for auditing in the command-line interface.
         setup_session(): Sets up the session for the provider.
-        get_output_mapping(): Returns the output mapping between the provider and the generic model.
         validate_arguments(): Validates the arguments for the provider.
         get_checks_to_execute_by_audit_resources(): Returns a set of checks based on the input resources to scan.
 
@@ -131,15 +130,6 @@ class Provider(ABC):
         """
         raise NotImplementedError()
 
-    @abstractmethod
-    def get_output_mapping(self) -> dict:
-        """
-        get_output_mapping returns the output mapping between the provider and the generic model.
-
-        This method needs to be created in each provider.
-        """
-        raise NotImplementedError()
-
     def validate_arguments(self) -> None:
         """
         validate_arguments validates the arguments for the provider.

prowler/lib/outputs/common.py

@@ -1,26 +1,8 @@
-from operator import attrgetter
-
 from prowler.config.config import timestamp
-from prowler.lib.logger import logger
 from prowler.lib.outputs.utils import unroll_tags
 from prowler.lib.utils.utils import outputs_unix_timestamp
 
 
-def get_provider_data_mapping(provider) -> dict:
-    data = {}
-    for generic_field, provider_field in provider.get_output_mapping.items():
-        try:
-            provider_value = attrgetter(provider_field)(provider)
-            data[generic_field] = provider_value
-        except AttributeError:
-            data[generic_field] = ""
-        except Exception as error:
-            logger.error(
-                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
-            )
-    return data
-
-
 # TODO: add test for outputs_unix_timestamp
 def fill_common_finding_data(finding: dict, unix_timestamp: bool) -> dict:
     finding_data = {

prowler/lib/outputs/finding.py

@@ -7,12 +7,9 @@ from pydantic import BaseModel, Field
 from prowler.config.config import prowler_version
 from prowler.lib.check.models import Check_Report, CheckMetadata
 from prowler.lib.logger import logger
-from prowler.lib.outputs.common import (
-    fill_common_finding_data,
-    get_provider_data_mapping,
-)
+from prowler.lib.outputs.common import fill_common_finding_data
 from prowler.lib.outputs.compliance.compliance import get_check_compliance
-from prowler.lib.utils.utils import dict_to_lowercase
+from prowler.lib.utils.utils import dict_to_lowercase, get_nested_attribute
 from prowler.providers.common.provider import Provider
 
 
@@ -118,9 +115,6 @@ class Finding(BaseModel):
             finding_output (Finding): the finding output object
 
         """
-        # TODO: think about get_provider_data_mapping
-        provider_data_mapping = get_provider_data_mapping(provider)
-
         # TODO: move fill_common_finding_data
         unix_timestamp = False
         if hasattr(output_options, "unix_timestamp"):
@@ -128,7 +122,6 @@ class Finding(BaseModel):
 
         common_finding_data = fill_common_finding_data(check_output, unix_timestamp)
         output_data = {}
-        output_data.update(provider_data_mapping)
         output_data.update(common_finding_data)
 
         bulk_checks_metadata = {}
@@ -139,9 +132,35 @@ class Finding(BaseModel):
             check_output, provider.type, bulk_checks_metadata
         )
         try:
+            output_data["provider"] = provider.type
+
             if provider.type == "aws":
+                output_data["account_uid"] = get_nested_attribute(
+                    provider, "identity.account"
+                )
+                output_data["account_name"] = get_nested_attribute(
+                    provider, "organizations_metadata.account_name"
+                )
+                output_data["account_email"] = get_nested_attribute(
+                    provider, "organizations_metadata.account_email"
+                )
+                output_data["account_organization_uid"] = get_nested_attribute(
+                    provider, "organizations_metadata.organization_arn"
+                )
+                output_data["account_organization_name"] = get_nested_attribute(
+                    provider, "organizations_metadata.organization_id"
+                )
+                output_data["account_tags"] = get_nested_attribute(
+                    provider, "organizations_metadata.account_tags"
+                )
+                output_data["partition"] = get_nested_attribute(
+                    provider, "identity.partition"
+                )
+
                 # TODO: probably Organization UID is without the account id
-                output_data["auth_method"] = f"profile: {output_data['auth_method']}"
+                output_data["auth_method"] = (
+                    f"profile: {get_nested_attribute(provider, 'identity.profile')}"
+                )
                 output_data["resource_name"] = check_output.resource_id
                 output_data["resource_uid"] = check_output.resource_arn
                 output_data["region"] = check_output.region
@@ -152,9 +171,9 @@ class Finding(BaseModel):
                     f"{provider.identity.identity_type}: {provider.identity.identity_id}"
                 )
                 # Get the first tenant domain ID, just in case
-                output_data["account_organization_uid"] = output_data[
-                    "account_organization_uid"
-                ][0]
+                output_data["account_organization_uid"] = get_nested_attribute(
+                    provider, "identity.tenant_ids"
+                )[0]
                 output_data["account_uid"] = (
                     output_data["account_organization_uid"]
                     if "Tenant:" in check_output.subscription
@@ -164,15 +183,33 @@ class Finding(BaseModel):
                 output_data["resource_name"] = check_output.resource_name
                 output_data["resource_uid"] = check_output.resource_id
                 output_data["region"] = check_output.location
+                # TODO: check the tenant_ids
+                # TODO: we have to get the account organization, the tenant is not that
+                output_data["account_organization_name"] = get_nested_attribute(
+                    provider, "identity.tenant_domain"
+                )
+
+                output_data["partition"] = get_nested_attribute(
+                    provider, "region_config.name"
+                )
+                # TODO: pending to get the subscription tags
+                # "account_tags": "organizations_metadata.account_details_tags",
+                # TODO: store subscription_name + id pairs
+                # "account_name": "organizations_metadata.account_details_name",
+                # "account_email": "organizations_metadata.account_details_email",
 
             elif provider.type == "gcp":
-                output_data["auth_method"] = f"Principal: {output_data['auth_method']}"
+                output_data["auth_method"] = (
+                    f"Principal: {get_nested_attribute(provider, 'identity.profile')}"
+                )
                 output_data["account_uid"] = provider.projects[
                     check_output.project_id
                 ].id
                 output_data["account_name"] = provider.projects[
                     check_output.project_id
                 ].name
+                # There is no concept as project email in GCP
+                # "account_email": "organizations_metadata.account_details_email",
                 output_data["account_tags"] = provider.projects[
                     check_output.project_id
                 ].labels
@@ -203,6 +240,9 @@ class Finding(BaseModel):
                 output_data["resource_name"] = check_output.resource_name
                 output_data["resource_uid"] = check_output.resource_id
                 output_data["account_name"] = f"context: {provider.identity.context}"
+                output_data["account_uid"] = get_nested_attribute(
+                    provider, "identity.cluster"
+                )
                 output_data["region"] = f"namespace: {check_output.namespace}"
 
             # check_output Unique ID

prowler/lib/utils/utils.py

@@ -1,5 +1,6 @@
 import json
 import os
+from operator import attrgetter
 
 try:
     import grp
@@ -16,7 +17,7 @@ from io import TextIOWrapper
 from ipaddress import ip_address
 from os.path import exists
 from time import mktime
-from typing import Optional
+from typing import Any, Optional
 
 from colorama import Style
 from detect_secrets import SecretsCollection
@@ -293,3 +294,23 @@ def dict_to_lowercase(d):
             v = dict_to_lowercase(v)
         new_dict[k.lower()] = v
     return new_dict
+
+
+def get_nested_attribute(obj: Any, attr: str) -> Any:
+    """
+    Get a nested attribute from an object.
+    Args:
+        obj (Any): The object to get the attribute from.
+        attr (str): The attribute to get.
+    Returns:
+        Any: The attribute value if present, otherwise "".
+    """
+    try:
+        return attrgetter(attr)(obj)
+    except AttributeError:
+        return ""
+    except Exception as error:
+        logger.error(
+            f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+        )
+        return ""

prowler/providers/aws/aws_provider.py

@@ -360,20 +360,6 @@ class AwsProvider(Provider):
         """
         return self._mutelist
 
-    @property
-    def get_output_mapping(self):
-        return {
-            "auth_method": "identity.profile",
-            "provider": "type",
-            "account_uid": "identity.account",
-            "account_name": "organizations_metadata.account_name",
-            "account_email": "organizations_metadata.account_email",
-            "account_organization_uid": "organizations_metadata.organization_arn",
-            "account_organization_name": "organizations_metadata.organization_id",
-            "account_tags": "organizations_metadata.account_tags",
-            "partition": "identity.partition",
-        }
-
     # TODO: This can be moved to another class since it doesn't need self
     def get_organizations_info(
         self, organizations_session: Session, aws_account_id: str

prowler/providers/azure/azure_provider.py

@@ -87,7 +87,6 @@ class AzureProvider(Provider):
         fixer_config(self): Returns the fixer configuration.
         output_options(self, options: tuple): Sets the output options for the Azure provider.
         mutelist(self) -> AzureMutelist: Returns the mutelist object associated with the Azure provider.
-        get_output_mapping(self): Returns a dictionary that maps output keys to their corresponding values.
         validate_arguments(cls, az_cli_auth, sp_env_auth, browser_auth, managed_identity_auth, tenant_id): Validates the authentication arguments for the Azure provider.
         setup_region_config(cls, region): Sets up the region configuration for the Azure provider.
         print_credentials(self): Prints the Azure credentials information.
@@ -266,26 +265,6 @@ class AzureProvider(Provider):
         """Mutelist object associated with this Azure provider."""
         return self._mutelist
 
-    @property
-    def get_output_mapping(self):
-        """Dictionary that maps output keys to their corresponding values."""
-        return {
-            # identity_type: identity_id
-            # "auth_method": "identity.profile",
-            "provider": "type",
-            # "account_uid": "identity.account",
-            # TODO: store subscription_name + id pairs
-            # "account_name": "organizations_metadata.account_details_name",
-            # "account_email": "organizations_metadata.account_details_email",
-            # TODO: check the tenant_ids
-            # TODO: we have to get the account organization, the tenant is not that
-            "account_organization_uid": "identity.tenant_ids",
-            "account_organization_name": "identity.tenant_domain",
-            # TODO: pending to get the subscription tags
-            # "account_tags": "organizations_metadata.account_details_tags",
-            "partition": "region_config.name",
-        }
-
     # TODO: this should be moved to the argparse, if not we need to enforce it from the Provider
     # previously was using the AzureException
     @staticmethod

prowler/providers/common/provider.py

@@ -41,7 +41,6 @@ class Provider(ABC):
     Methods:
         print_credentials(): Displays the provider's credentials used for auditing in the command-line interface.
         setup_session(): Sets up the session for the provider.
-        get_output_mapping(): Returns the output mapping between the provider and the generic model.
         validate_arguments(): Validates the arguments for the provider.
         get_checks_to_execute_by_audit_resources(): Returns a set of checks based on the input resources to scan.
 
@@ -108,15 +107,6 @@ class Provider(ABC):
         """
         raise NotImplementedError()
 
-    @abstractmethod
-    def get_output_mapping(self) -> dict:
-        """
-        get_output_mapping returns the output mapping between the provider and the generic model.
-
-        This method needs to be created in each provider.
-        """
-        raise NotImplementedError()
-
     # TODO: uncomment this once all the providers have implemented the test_connection method
     # @abstractmethod
     def test_connection(self) -> Any:

prowler/providers/gcp/gcp_provider.py

@@ -224,27 +224,6 @@ class GcpProvider(Provider):
         """
         return self._mutelist
 
-    @property
-    def get_output_mapping(self):
-        return {
-            # Account: identity.profile
-            "auth_method": "identity.profile",
-            "provider": "type",
-            # TODO: comes from finding, finding.project_id
-            # "account_uid": "",
-            # TODO: get project name from GCP
-            # "account_name": "organizations_metadata.account_details_name",
-            # There is no concept as project email in GCP
-            # "account_email": "organizations_metadata.account_details_email",
-            # TODO: get project organization ID from GCP
-            # "account_organization_uid": "organizations_metadata.account_details_arn",
-            # TODO: get project organization from GCP
-            # "account_organization": "",
-            # TODO: get project tags organization from GCP
-            # "account_tags": "organizations_metadata.account_details_tags",
-            # "partition": "identity.partition",
-        }
-
     @staticmethod
     def setup_session(
         credentials_file: str, service_account: str, gcp_credentials: dict = None

prowler/providers/kubernetes/kubernetes_provider.py

@@ -144,22 +144,6 @@ class KubernetesProvider(Provider):
         """
         return self._mutelist
 
-    @property
-    def get_output_mapping(self):
-        return {
-            # "in-cluster/kubeconfig"
-            # "auth_method": "identity.profile",
-            "provider": "type",
-            # cluster: <context>
-            "account_uid": "identity.cluster",
-            # "account_name": "organizations_metadata.account_details_name",
-            # "account_email": "organizations_metadata.account_details_email",
-            # "account_organization_uid": "organizations_metadata.account_details_arn",
-            # "account_organization": "organizations_metadata.account_details_org",
-            # "account_tags": "organizations_metadata.account_details_tags",
-            # "partition": "identity.partition",
-        }
-
     @staticmethod
     def setup_session(
         kubeconfig_file: str = None,

tests/lib/outputs/finding_test.py

@@ -47,52 +47,11 @@ def mock_check_metadata(provider):
     )
 
 
-def mock_get_provider_data_mapping_aws(_):
-    return {
-        "auth_method": "mock_auth",
-        "provider": "aws",
-        "account_uid": "mock_account_uid",
-        "account_name": "mock_account_name",
-        "account_email": "mock_account_email",
-        "account_organization_uid": "mock_account_org_uid",
-        "account_organization_name": "mock_account_org_name",
-        "account_tags": {"tag1": "value1"},
-        "partition": "aws",
-    }
-
-
-def mock_get_provider_data_mapping_azure(_):
-    return {
-        "provider": "azure",
-        "account_organization_uid": "mock_account_org_uid",
-        "account_organization_name": "mock_account_org_name",
-        "partition": "AzureCloud",
-    }
-
-
-def mock_get_provider_data_mapping_gcp(_):
-    return {
-        "auth_method": "mock_auth",
-        "provider": "gcp",
-    }
-
-
-def mock_get_provider_data_mapping_kubernetes(_):
-    return {
-        "provider": "kubernetes",
-        "account_uid": "test_cluster",
-    }
-
-
 def mock_get_check_compliance(*_):
     return {"mock_compliance_key": "mock_compliance_value"}
 
 
 class TestFinding:
-    @patch(
-        "prowler.lib.outputs.finding.get_provider_data_mapping",
-        new=mock_get_provider_data_mapping_aws,
-    )
     @patch(
         "prowler.lib.outputs.finding.get_check_compliance",
         new=mock_get_check_compliance,
@@ -101,6 +60,14 @@ class TestFinding:
         # Mock provider
         provider = MagicMock()
         provider.type = "aws"
+        provider.identity.profile = "mock_auth"
+        provider.identity.account = "mock_account_uid"
+        provider.identity.partition = "aws"
+        provider.organizations_metadata.account_name = "mock_account_name"
+        provider.organizations_metadata.account_email = "mock_account_email"
+        provider.organizations_metadata.organization_arn = "mock_account_org_uid"
+        provider.organizations_metadata.organization_id = "mock_account_org_name"
+        provider.organizations_metadata.account_tags = {"tag1": "value1"}
 
         # Mock check result
         check_output = MagicMock()
@@ -123,7 +90,7 @@ class TestFinding:
         finding_output = Finding.generate_output(provider, check_output, output_options)
 
         # Finding
-        assert finding_output is not None
+        assert isinstance(finding_output, Finding)
         assert finding_output.auth_method == "profile: mock_auth"
         assert finding_output.resource_name == "test_resource_id"
         assert finding_output.resource_uid == "test_resource_arn"
@@ -179,10 +146,6 @@ class TestFinding:
         assert finding_output.service_name == "mock_service_name"
         assert finding_output.raw == {}
 
-    @patch(
-        "prowler.lib.outputs.finding.get_provider_data_mapping",
-        new=mock_get_provider_data_mapping_azure,
-    )
     @patch(
         "prowler.lib.outputs.finding.get_check_compliance",
         new=mock_get_check_compliance,
@@ -196,6 +159,9 @@ class TestFinding:
         provider.identity.subscriptions = {
             "mock_subscription_id": "mock_subscription_name"
         }
+        provider.identity.tenant_ids = ["mock_tenant_id_1", "mock_tenant_id_2"]
+        provider.identity.tenant_domain = "mock_tenant_domain"
+        provider.region_config.name = "AzureCloud"
 
         # Mock check result
         check_output = MagicMock()
@@ -220,8 +186,12 @@ class TestFinding:
         finding_output = Finding.generate_output(provider, check_output, output_options)
 
         # Finding
-        assert finding_output is not None
+        assert isinstance(finding_output, Finding)
         assert finding_output.auth_method == "mock_identity_type: mock_identity_id"
+        assert finding_output.account_organization_uid == "mock_tenant_id_1"
+        assert finding_output.account_organization_name == "mock_tenant_domain"
+        assert finding_output.account_uid == "mock_subscription_name"
+        assert finding_output.account_name == "mock_subscription_id"
         assert finding_output.resource_name == "test_resource_name"
         assert finding_output.resource_uid == "test_resource_id"
         assert finding_output.region == "us-west-1"
@@ -233,6 +203,7 @@ class TestFinding:
         assert finding_output.muted is False
         assert finding_output.resource_tags == {}
         assert finding_output.partition == "AzureCloud"
+
         assert isinstance(finding_output.timestamp, int)
 
         # Metadata
@@ -261,10 +232,6 @@ class TestFinding:
         assert finding_output.metadata.Notes == "mock_notes"
         assert finding_output.metadata.Compliance == []
 
-    @patch(
-        "prowler.lib.outputs.finding.get_provider_data_mapping",
-        new=mock_get_provider_data_mapping_gcp,
-    )
     @patch(
         "prowler.lib.outputs.finding.get_check_compliance",
         new=mock_get_check_compliance,
@@ -273,9 +240,12 @@ class TestFinding:
         # Mock provider
         provider = MagicMock()
         provider.type = "gcp"
+        provider.identity.profile = "mock_auth"
+        # Organization
         organization = MagicMock()
         organization.id = "mock_organization_id"
         organization.display_name = "mock_organization_name"
+        # Project
         project = MagicMock()
         project.id = "mock_project_id"
         project.name = "mock_project_name"
@@ -304,7 +274,7 @@ class TestFinding:
         finding_output = Finding.generate_output(provider, check_output, output_options)
 
         # Finding
-        assert finding_output is not None
+        assert isinstance(finding_output, Finding)
         assert finding_output.auth_method == "Principal: mock_auth"
         assert finding_output.resource_name == "test_resource_name"
         assert finding_output.resource_uid == "test_resource_id"
@@ -351,10 +321,6 @@ class TestFinding:
         assert finding_output.metadata.Notes == "mock_notes"
         assert finding_output.metadata.Compliance == []
 
-    @patch(
-        "prowler.lib.outputs.finding.get_provider_data_mapping",
-        new=mock_get_provider_data_mapping_kubernetes,
-    )
     @patch(
         "prowler.lib.outputs.finding.get_check_compliance",
         new=mock_get_check_compliance,
@@ -363,9 +329,8 @@ class TestFinding:
         # Mock provider
         provider = MagicMock()
         provider.type = "kubernetes"
-        identity = MagicMock()
-        identity.context = "In-Cluster"
-        provider.identity = identity
+        provider.identity.context = "In-Cluster"
+        provider.identity.cluster = "test_cluster"
 
         # Mock check result
         check_output = MagicMock()
@@ -387,11 +352,12 @@ class TestFinding:
         finding_output = Finding.generate_output(provider, check_output, output_options)
 
         # Finding
-        assert finding_output is not None
+        assert isinstance(finding_output, Finding)
         assert finding_output.auth_method == "in-cluster"
         assert finding_output.resource_name == "test_resource_name"
         assert finding_output.resource_uid == "test_resource_id"
         assert finding_output.region == "namespace: test_namespace"
+        assert finding_output.account_name == "context: In-Cluster"
         assert finding_output.compliance == {
             "mock_compliance_key": "mock_compliance_value"
         }

tests/providers/aws/aws_provider_test.py

@@ -459,22 +459,6 @@ class TestAWSProvider:
                 aws_provider.session.current_session.region_name == AWS_REGION_US_EAST_1
             )
 
-    @mock_aws
-    def test_aws_provider_get_output_mapping(self):
-        aws_provider = AwsProvider()
-
-        assert aws_provider.get_output_mapping == {
-            "auth_method": "identity.profile",
-            "provider": "type",
-            "account_uid": "identity.account",
-            "account_name": "organizations_metadata.account_name",
-            "account_email": "organizations_metadata.account_email",
-            "account_organization_uid": "organizations_metadata.organization_arn",
-            "account_organization_name": "organizations_metadata.organization_id",
-            "account_tags": "organizations_metadata.account_tags",
-            "partition": "identity.partition",
-        }
-
     @mock_aws
     def test_aws_provider_assume_role_with_mfa(self):
         # Variables