fix(errors): solve CloudWatch, KMS, EMR and OpenSearch service errors (#1843)

Co-authored-by: sergargar <sergargar@users.noreply.github.com>

prowler/providers/aws/services/cloudwatch/cloudwatch_service.py

@@ -1,5 +1,6 @@
 import threading
 from dataclasses import dataclass
+from typing import Optional
 
 from prowler.lib.logger import logger
 from prowler.lib.scan_filters.scan_filters import is_resource_filtered
@@ -46,12 +47,15 @@ class CloudWatch:
                         metric_name = None
                         if "MetricName" in alarm:
                             metric_name = alarm["MetricName"]
+                        namespace = None
+                        if "Namespace" in alarm:
+                            namespace = alarm["Namespace"]
                         self.metric_alarms.append(
                             MetricAlarm(
                                 alarm["AlarmArn"],
                                 alarm["AlarmName"],
                                 metric_name,
-                                alarm["Namespace"],
+                                namespace,
                                 regional_client.region,
                             )
                         )
@@ -147,8 +151,8 @@ class Logs:
 class MetricAlarm:
     arn: str
     name: str
-    metric: str
-    name_space: str
+    metric: Optional[str]
+    name_space: Optional[str]
     region: str
 
     def __init__(

prowler/providers/aws/services/emr/emr_service.py

@@ -1,5 +1,6 @@
 import threading
 from enum import Enum
+from typing import Optional
 
 from pydantic import BaseModel
 
@@ -163,7 +164,7 @@ class ClusterStatus(Enum):
 
 class Node(BaseModel):
     security_group_id: str = ""
-    additional_security_groups_id: list[str] = []
+    additional_security_groups_id: Optional[list[str]] = []
 
 
 class Cluster(BaseModel):

prowler/providers/aws/services/kms/kms_service.py

@@ -71,22 +71,22 @@ class KMS:
 
     def __get_key_rotation_status__(self):
         logger.info("KMS - Get Key Rotation Status...")
-        for key in self.keys:
-            try:
+        try:
+            for key in self.keys:
                 if "EXTERNAL" not in key.origin:
                     regional_client = self.regional_clients[key.region]
                     key.rotation_enabled = regional_client.get_key_rotation_status(
                         KeyId=key.id
                     )["KeyRotationEnabled"]
-            except Exception as error:
-                logger.error(
-                    f"{regional_client.region} -- {error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
-                )
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
+            )
 
     def __get_key_policy__(self):
         logger.info("KMS - Get Key Policy...")
-        for key in self.keys:
-            try:
+        try:
+            for key in self.keys:
                 if key.manager == "CUSTOMER":  # only customer KMS have policies
                     regional_client = self.regional_clients[key.region]
                     key.policy = json.loads(
@@ -94,10 +94,10 @@ class KMS:
                             KeyId=key.id, PolicyName="default"
                         )["Policy"]
                     )
-            except Exception as error:
-                logger.error(
-                    f"{regional_client.region} -- {error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
-                )
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
+            )
 
 
 @dataclass

prowler/providers/aws/services/opensearch/opensearch_service.py

@@ -100,7 +100,11 @@ class OpenSearchService:
                         domain.endpoint_vpc = describe_domain["DomainStatus"][
                             "Endpoints"
                         ]["vpc"]
-                domain.vpc_id = describe_domain["DomainStatus"]["VPCOptions"]["VPCId"]
+                domain.vpc_id = None
+                if "VPCOptions" in describe_domain["DomainStatus"]:
+                    domain.vpc_id = describe_domain["DomainStatus"]["VPCOptions"][
+                        "VPCId"
+                    ]
                 domain.cognito_options = describe_domain["DomainStatus"][
                     "CognitoOptions"
                 ]["Enabled"]