refactor(cloudwatch): simplify logic (#3172)

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

@@ -1,5 +1,3 @@
-import re
-
 from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
     cloudtrail_client,
@@ -7,6 +5,9 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
 from prowler.providers.aws.services.cloudwatch.cloudwatch_client import (
     cloudwatch_client,
 )
+from prowler.providers.aws.services.cloudwatch.lib.metric_filters import (
+    check_cloudwatch_log_metric_filter,
+)
 from prowler.providers.aws.services.cloudwatch.logs_client import logs_client
 
 
@@ -22,26 +23,13 @@ class cloudwatch_changes_to_network_acls_alarm_configured(Check):
         report.region = cloudwatch_client.region
         report.resource_id = cloudtrail_client.audited_account
         report.resource_arn = cloudtrail_client.audited_account_arn
-        # 1. Iterate for CloudWatch Log Group in CloudTrail trails
+        report = check_cloudwatch_log_metric_filter(
-        log_groups = []
+            pattern,
-        for trail in cloudtrail_client.trails:
+            cloudtrail_client.trails,
-            if trail.log_group_arn:
+            logs_client.metric_filters,
-                log_groups.append(trail.log_group_arn.split(":")[6])
+            cloudwatch_client.metric_alarms,
-        # 2. Describe metric filters for previous log groups
+            report,
-        for metric_filter in logs_client.metric_filters:
+        )
-            if metric_filter.log_group in log_groups:
-                if re.search(pattern, metric_filter.pattern, flags=re.DOTALL):
-                    report.resource_id = metric_filter.log_group
-                    report.resource_arn = metric_filter.arn
-                    report.region = metric_filter.region
-                    report.status = "FAIL"
-                    report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} but no alarms associated."
-                    # 3. Check if there is an alarm for the metric
-                    for alarm in cloudwatch_client.metric_alarms:
-                        if alarm.metric == metric_filter.metric:
-                            report.status = "PASS"
-                            report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} and alarms set."
-                            break
 
         findings.append(report)
         return findings

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

@@ -1,5 +1,3 @@
-import re
-
 from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
     cloudtrail_client,
@@ -7,6 +5,9 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
 from prowler.providers.aws.services.cloudwatch.cloudwatch_client import (
     cloudwatch_client,
 )
+from prowler.providers.aws.services.cloudwatch.lib.metric_filters import (
+    check_cloudwatch_log_metric_filter,
+)
 from prowler.providers.aws.services.cloudwatch.logs_client import logs_client
 
 
@@ -22,26 +23,13 @@ class cloudwatch_changes_to_network_gateways_alarm_configured(Check):
         report.region = cloudwatch_client.region
         report.resource_id = cloudtrail_client.audited_account
         report.resource_arn = cloudtrail_client.audited_account_arn
-        # 1. Iterate for CloudWatch Log Group in CloudTrail trails
+        report = check_cloudwatch_log_metric_filter(
-        log_groups = []
+            pattern,
-        for trail in cloudtrail_client.trails:
+            cloudtrail_client.trails,
-            if trail.log_group_arn:
+            logs_client.metric_filters,
-                log_groups.append(trail.log_group_arn.split(":")[6])
+            cloudwatch_client.metric_alarms,
-        # 2. Describe metric filters for previous log groups
+            report,
-        for metric_filter in logs_client.metric_filters:
+        )
-            if metric_filter.log_group in log_groups:
-                if re.search(pattern, metric_filter.pattern, flags=re.DOTALL):
-                    report.resource_id = metric_filter.log_group
-                    report.resource_arn = metric_filter.arn
-                    report.region = metric_filter.region
-                    report.status = "FAIL"
-                    report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} but no alarms associated."
-                    # 3. Check if there is an alarm for the metric
-                    for alarm in cloudwatch_client.metric_alarms:
-                        if alarm.metric == metric_filter.metric:
-                            report.status = "PASS"
-                            report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} and alarms set."
-                            break
 
         findings.append(report)
         return findings

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

@@ -1,5 +1,3 @@
-import re
-
 from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
     cloudtrail_client,
@@ -7,6 +5,9 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
 from prowler.providers.aws.services.cloudwatch.cloudwatch_client import (
     cloudwatch_client,
 )
+from prowler.providers.aws.services.cloudwatch.lib.metric_filters import (
+    check_cloudwatch_log_metric_filter,
+)
 from prowler.providers.aws.services.cloudwatch.logs_client import logs_client
 
 
@@ -22,26 +23,13 @@ class cloudwatch_changes_to_network_route_tables_alarm_configured(Check):
         report.region = cloudwatch_client.region
         report.resource_id = cloudtrail_client.audited_account
         report.resource_arn = cloudtrail_client.audited_account_arn
-        # 1. Iterate for CloudWatch Log Group in CloudTrail trails
+        report = check_cloudwatch_log_metric_filter(
-        log_groups = []
+            pattern,
-        for trail in cloudtrail_client.trails:
+            cloudtrail_client.trails,
-            if trail.log_group_arn:
+            logs_client.metric_filters,
-                log_groups.append(trail.log_group_arn.split(":")[6])
+            cloudwatch_client.metric_alarms,
-        # 2. Describe metric filters for previous log groups
+            report,
-        for metric_filter in logs_client.metric_filters:
+        )
-            if metric_filter.log_group in log_groups:
-                if re.search(pattern, metric_filter.pattern, flags=re.DOTALL):
-                    report.resource_id = metric_filter.log_group
-                    report.resource_arn = metric_filter.arn
-                    report.region = metric_filter.region
-                    report.status = "FAIL"
-                    report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} but no alarms associated."
-                    # 3. Check if there is an alarm for the metric
-                    for alarm in cloudwatch_client.metric_alarms:
-                        if alarm.metric == metric_filter.metric:
-                            report.status = "PASS"
-                            report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} and alarms set."
-                            break
 
         findings.append(report)
         return findings

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

@@ -1,5 +1,3 @@
-import re
-
 from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
     cloudtrail_client,
@@ -7,6 +5,9 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
 from prowler.providers.aws.services.cloudwatch.cloudwatch_client import (
     cloudwatch_client,
 )
+from prowler.providers.aws.services.cloudwatch.lib.metric_filters import (
+    check_cloudwatch_log_metric_filter,
+)
 from prowler.providers.aws.services.cloudwatch.logs_client import logs_client
 
 
@@ -22,26 +23,13 @@ class cloudwatch_changes_to_vpcs_alarm_configured(Check):
         report.region = cloudwatch_client.region
         report.resource_id = cloudtrail_client.audited_account
         report.resource_arn = cloudtrail_client.audited_account_arn
-        # 1. Iterate for CloudWatch Log Group in CloudTrail trails
+        report = check_cloudwatch_log_metric_filter(
-        log_groups = []
+            pattern,
-        for trail in cloudtrail_client.trails:
+            cloudtrail_client.trails,
-            if trail.log_group_arn:
+            logs_client.metric_filters,
-                log_groups.append(trail.log_group_arn.split(":")[6])
+            cloudwatch_client.metric_alarms,
-        # 2. Describe metric filters for previous log groups
+            report,
-        for metric_filter in logs_client.metric_filters:
+        )
-            if metric_filter.log_group in log_groups:
-                if re.search(pattern, metric_filter.pattern, flags=re.DOTALL):
-                    report.resource_id = metric_filter.log_group
-                    report.resource_arn = metric_filter.arn
-                    report.region = metric_filter.region
-                    report.status = "FAIL"
-                    report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} but no alarms associated."
-                    # 3. Check if there is an alarm for the metric
-                    for alarm in cloudwatch_client.metric_alarms:
-                        if alarm.metric == metric_filter.metric:
-                            report.status = "PASS"
-                            report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} and alarms set."
-                            break
 
         findings.append(report)
         return findings

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

@@ -1,5 +1,3 @@
-import re
-
 from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
     cloudtrail_client,
@@ -7,6 +5,9 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
 from prowler.providers.aws.services.cloudwatch.cloudwatch_client import (
     cloudwatch_client,
 )
+from prowler.providers.aws.services.cloudwatch.lib.metric_filters import (
+    check_cloudwatch_log_metric_filter,
+)
 from prowler.providers.aws.services.cloudwatch.logs_client import logs_client
 
 
@@ -24,26 +25,13 @@ class cloudwatch_log_metric_filter_and_alarm_for_aws_config_configuration_change
         report.region = cloudwatch_client.region
         report.resource_id = cloudtrail_client.audited_account
         report.resource_arn = cloudtrail_client.audited_account_arn
-        # 1. Iterate for CloudWatch Log Group in CloudTrail trails
+        report = check_cloudwatch_log_metric_filter(
-        log_groups = []
+            pattern,
-        for trail in cloudtrail_client.trails:
+            cloudtrail_client.trails,
-            if trail.log_group_arn:
+            logs_client.metric_filters,
-                log_groups.append(trail.log_group_arn.split(":")[6])
+            cloudwatch_client.metric_alarms,
-        # 2. Describe metric filters for previous log groups
+            report,
-        for metric_filter in logs_client.metric_filters:
+        )
-            if metric_filter.log_group in log_groups:
-                if re.search(pattern, metric_filter.pattern, flags=re.DOTALL):
-                    report.resource_id = metric_filter.log_group
-                    report.resource_arn = metric_filter.arn
-                    report.region = metric_filter.region
-                    report.status = "FAIL"
-                    report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} but no alarms associated."
-                    # 3. Check if there is an alarm for the metric
-                    for alarm in cloudwatch_client.metric_alarms:
-                        if alarm.metric == metric_filter.metric:
-                            report.status = "PASS"
-                            report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} and alarms set."
-                            break
 
         findings.append(report)
         return findings

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

@@ -1,5 +1,3 @@
-import re
-
 from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
     cloudtrail_client,
@@ -7,6 +5,9 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
 from prowler.providers.aws.services.cloudwatch.cloudwatch_client import (
     cloudwatch_client,
 )
+from prowler.providers.aws.services.cloudwatch.lib.metric_filters import (
+    check_cloudwatch_log_metric_filter,
+)
 from prowler.providers.aws.services.cloudwatch.logs_client import logs_client
 
 
@@ -24,26 +25,13 @@ class cloudwatch_log_metric_filter_and_alarm_for_cloudtrail_configuration_change
         report.region = cloudwatch_client.region
         report.resource_id = cloudtrail_client.audited_account
         report.resource_arn = cloudtrail_client.audited_account_arn
-        # 1. Iterate for CloudWatch Log Group in CloudTrail trails
+        report = check_cloudwatch_log_metric_filter(
-        log_groups = []
+            pattern,
-        for trail in cloudtrail_client.trails:
+            cloudtrail_client.trails,
-            if trail.log_group_arn:
+            logs_client.metric_filters,
-                log_groups.append(trail.log_group_arn.split(":")[6])
+            cloudwatch_client.metric_alarms,
-        # 2. Describe metric filters for previous log groups
+            report,
-        for metric_filter in logs_client.metric_filters:
+        )
-            if metric_filter.log_group in log_groups:
-                if re.search(pattern, metric_filter.pattern, flags=re.DOTALL):
-                    report.resource_id = metric_filter.log_group
-                    report.resource_arn = metric_filter.arn
-                    report.region = metric_filter.region
-                    report.status = "FAIL"
-                    report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} but no alarms associated."
-                    # 3. Check if there is an alarm for the metric
-                    for alarm in cloudwatch_client.metric_alarms:
-                        if alarm.metric == metric_filter.metric:
-                            report.status = "PASS"
-                            report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} and alarms set."
-                            break
 
         findings.append(report)
         return findings

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

@@ -1,5 +1,3 @@
-import re
-
 from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
     cloudtrail_client,
@@ -7,6 +5,9 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
 from prowler.providers.aws.services.cloudwatch.cloudwatch_client import (
     cloudwatch_client,
 )
+from prowler.providers.aws.services.cloudwatch.lib.metric_filters import (
+    check_cloudwatch_log_metric_filter,
+)
 from prowler.providers.aws.services.cloudwatch.logs_client import logs_client
 
 
@@ -22,26 +23,13 @@ class cloudwatch_log_metric_filter_authentication_failures(Check):
         report.region = cloudwatch_client.region
         report.resource_id = cloudtrail_client.audited_account
         report.resource_arn = cloudtrail_client.audited_account_arn
-        # 1. Iterate for CloudWatch Log Group in CloudTrail trails
+        report = check_cloudwatch_log_metric_filter(
-        log_groups = []
+            pattern,
-        for trail in cloudtrail_client.trails:
+            cloudtrail_client.trails,
-            if trail.log_group_arn:
+            logs_client.metric_filters,
-                log_groups.append(trail.log_group_arn.split(":")[6])
+            cloudwatch_client.metric_alarms,
-        # 2. Describe metric filters for previous log groups
+            report,
-        for metric_filter in logs_client.metric_filters:
+        )
-            if metric_filter.log_group in log_groups:
-                if re.search(pattern, metric_filter.pattern, flags=re.DOTALL):
-                    report.resource_id = metric_filter.log_group
-                    report.resource_arn = metric_filter.arn
-                    report.region = metric_filter.region
-                    report.status = "FAIL"
-                    report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} but no alarms associated."
-                    # 3. Check if there is an alarm for the metric
-                    for alarm in cloudwatch_client.metric_alarms:
-                        if alarm.metric == metric_filter.metric:
-                            report.status = "PASS"
-                            report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} and alarms set."
-                            break
 
         findings.append(report)
         return findings

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

@@ -1,5 +1,3 @@
-import re
-
 from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
     cloudtrail_client,
@@ -7,6 +5,9 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
 from prowler.providers.aws.services.cloudwatch.cloudwatch_client import (
     cloudwatch_client,
 )
+from prowler.providers.aws.services.cloudwatch.lib.metric_filters import (
+    check_cloudwatch_log_metric_filter,
+)
 from prowler.providers.aws.services.cloudwatch.logs_client import logs_client
 
 
@@ -22,26 +23,13 @@ class cloudwatch_log_metric_filter_aws_organizations_changes(Check):
         report.region = cloudwatch_client.region
         report.resource_id = cloudtrail_client.audited_account
         report.resource_arn = cloudtrail_client.audited_account_arn
-        # 1. Iterate for CloudWatch Log Group in CloudTrail trails
+        report = check_cloudwatch_log_metric_filter(
-        log_groups = []
+            pattern,
-        for trail in cloudtrail_client.trails:
+            cloudtrail_client.trails,
-            if trail.log_group_arn:
+            logs_client.metric_filters,
-                log_groups.append(trail.log_group_arn.split(":")[6])
+            cloudwatch_client.metric_alarms,
-        # 2. Describe metric filters for previous log groups
+            report,
-        for metric_filter in logs_client.metric_filters:
+        )
-            if metric_filter.log_group in log_groups:
-                if re.search(pattern, metric_filter.pattern, flags=re.DOTALL):
-                    report.resource_id = metric_filter.log_group
-                    report.resource_arn = metric_filter.arn
-                    report.region = metric_filter.region
-                    report.status = "FAIL"
-                    report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} but no alarms associated."
-                    # 3. Check if there is an alarm for the metric
-                    for alarm in cloudwatch_client.metric_alarms:
-                        if alarm.metric == metric_filter.metric:
-                            report.status = "PASS"
-                            report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} and alarms set."
-                            break
 
         findings.append(report)
         return findings

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

@@ -1,5 +1,3 @@
-import re
-
 from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
     cloudtrail_client,
@@ -7,6 +5,9 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
 from prowler.providers.aws.services.cloudwatch.cloudwatch_client import (
     cloudwatch_client,
 )
+from prowler.providers.aws.services.cloudwatch.lib.metric_filters import (
+    check_cloudwatch_log_metric_filter,
+)
 from prowler.providers.aws.services.cloudwatch.logs_client import logs_client
 
 
@@ -22,26 +23,13 @@ class cloudwatch_log_metric_filter_disable_or_scheduled_deletion_of_kms_cmk(Chec
         report.region = cloudwatch_client.region
         report.resource_id = cloudtrail_client.audited_account
         report.resource_arn = cloudtrail_client.audited_account_arn
-        # 1. Iterate for CloudWatch Log Group in CloudTrail trails
+        report = check_cloudwatch_log_metric_filter(
-        log_groups = []
+            pattern,
-        for trail in cloudtrail_client.trails:
+            cloudtrail_client.trails,
-            if trail.log_group_arn:
+            logs_client.metric_filters,
-                log_groups.append(trail.log_group_arn.split(":")[6])
+            cloudwatch_client.metric_alarms,
-        # 2. Describe metric filters for previous log groups
+            report,
-        for metric_filter in logs_client.metric_filters:
+        )
-            if metric_filter.log_group in log_groups:
-                if re.search(pattern, metric_filter.pattern, flags=re.DOTALL):
-                    report.resource_id = metric_filter.log_group
-                    report.resource_arn = metric_filter.arn
-                    report.region = metric_filter.region
-                    report.status = "FAIL"
-                    report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} but no alarms associated."
-                    # 3. Check if there is an alarm for the metric
-                    for alarm in cloudwatch_client.metric_alarms:
-                        if alarm.metric == metric_filter.metric:
-                            report.status = "PASS"
-                            report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} and alarms set."
-                            break
 
         findings.append(report)
         return findings

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

@@ -1,5 +1,3 @@
-import re
-
 from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
     cloudtrail_client,
@@ -7,6 +5,9 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
 from prowler.providers.aws.services.cloudwatch.cloudwatch_client import (
     cloudwatch_client,
 )
+from prowler.providers.aws.services.cloudwatch.lib.metric_filters import (
+    check_cloudwatch_log_metric_filter,
+)
 from prowler.providers.aws.services.cloudwatch.logs_client import logs_client
 
 
@@ -22,26 +23,14 @@ class cloudwatch_log_metric_filter_for_s3_bucket_policy_changes(Check):
         report.region = cloudwatch_client.region
         report.resource_id = cloudtrail_client.audited_account
         report.resource_arn = cloudtrail_client.audited_account_arn
-        # 1. Iterate for CloudWatch Log Group in CloudTrail trails
+
-        log_groups = []
+        report = check_cloudwatch_log_metric_filter(
-        for trail in cloudtrail_client.trails:
+            pattern,
-            if trail.log_group_arn:
+            cloudtrail_client.trails,
-                log_groups.append(trail.log_group_arn.split(":")[6])
+            logs_client.metric_filters,
-        # 2. Describe metric filters for previous log groups
+            cloudwatch_client.metric_alarms,
-        for metric_filter in logs_client.metric_filters:
+            report,
-            if metric_filter.log_group in log_groups:
+        )
-                if re.search(pattern, metric_filter.pattern, flags=re.DOTALL):
-                    report.resource_id = metric_filter.log_group
-                    report.resource_arn = metric_filter.arn
-                    report.region = metric_filter.region
-                    report.status = "FAIL"
-                    report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} but no alarms associated."
-                    # 3. Check if there is an alarm for the metric
-                    for alarm in cloudwatch_client.metric_alarms:
-                        if alarm.metric == metric_filter.metric:
-                            report.status = "PASS"
-                            report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} and alarms set."
-                            break
 
         findings.append(report)
         return findings

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

@@ -1,5 +1,3 @@
-import re
-
 from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
     cloudtrail_client,
@@ -7,6 +5,9 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
 from prowler.providers.aws.services.cloudwatch.cloudwatch_client import (
     cloudwatch_client,
 )
+from prowler.providers.aws.services.cloudwatch.lib.metric_filters import (
+    check_cloudwatch_log_metric_filter,
+)
 from prowler.providers.aws.services.cloudwatch.logs_client import logs_client
 
 
@@ -22,26 +23,13 @@ class cloudwatch_log_metric_filter_policy_changes(Check):
         report.region = cloudwatch_client.region
         report.resource_id = cloudtrail_client.audited_account
         report.resource_arn = cloudtrail_client.audited_account_arn
-        # 1. Iterate for CloudWatch Log Group in CloudTrail trails
+        report = check_cloudwatch_log_metric_filter(
-        log_groups = []
+            pattern,
-        for trail in cloudtrail_client.trails:
+            cloudtrail_client.trails,
-            if trail.log_group_arn:
+            logs_client.metric_filters,
-                log_groups.append(trail.log_group_arn.split(":")[6])
+            cloudwatch_client.metric_alarms,
-        # 2. Describe metric filters for previous log groups
+            report,
-        for metric_filter in logs_client.metric_filters:
+        )
-            if metric_filter.log_group in log_groups:
-                if re.search(pattern, metric_filter.pattern, flags=re.DOTALL):
-                    report.resource_id = metric_filter.log_group
-                    report.resource_arn = metric_filter.arn
-                    report.region = metric_filter.region
-                    report.status = "FAIL"
-                    report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} but no alarms associated."
-                    # 3. Check if there is an alarm for the metric
-                    for alarm in cloudwatch_client.metric_alarms:
-                        if alarm.metric == metric_filter.metric:
-                            report.status = "PASS"
-                            report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} and alarms set."
-                            break
 
         findings.append(report)
         return findings

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

@@ -1,5 +1,3 @@
-import re
-
 from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
     cloudtrail_client,
@@ -7,6 +5,9 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
 from prowler.providers.aws.services.cloudwatch.cloudwatch_client import (
     cloudwatch_client,
 )
+from prowler.providers.aws.services.cloudwatch.lib.metric_filters import (
+    check_cloudwatch_log_metric_filter,
+)
 from prowler.providers.aws.services.cloudwatch.logs_client import logs_client
 
 
@@ -22,26 +23,13 @@ class cloudwatch_log_metric_filter_root_usage(Check):
         report.region = cloudwatch_client.region
         report.resource_id = cloudtrail_client.audited_account
         report.resource_arn = cloudtrail_client.audited_account_arn
-        # 1. Iterate for CloudWatch Log Group in CloudTrail trails
+        report = check_cloudwatch_log_metric_filter(
-        log_groups = []
+            pattern,
-        for trail in cloudtrail_client.trails:
+            cloudtrail_client.trails,
-            if trail.log_group_arn:
+            logs_client.metric_filters,
-                log_groups.append(trail.log_group_arn.split(":")[6])
+            cloudwatch_client.metric_alarms,
-        # 2. Describe metric filters for previous log groups
+            report,
-        for metric_filter in logs_client.metric_filters:
+        )
-            if metric_filter.log_group in log_groups:
-                if re.search(pattern, metric_filter.pattern, flags=re.DOTALL):
-                    report.resource_id = metric_filter.log_group
-                    report.resource_arn = metric_filter.arn
-                    report.region = metric_filter.region
-                    report.status = "FAIL"
-                    report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} but no alarms associated."
-                    # 3. Check if there is an alarm for the metric
-                    for alarm in cloudwatch_client.metric_alarms:
-                        if alarm.metric == metric_filter.metric:
-                            report.status = "PASS"
-                            report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} and alarms set."
-                            break
 
         findings.append(report)
         return findings

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

@@ -1,5 +1,3 @@
-import re
-
 from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
     cloudtrail_client,
@@ -7,6 +5,9 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
 from prowler.providers.aws.services.cloudwatch.cloudwatch_client import (
     cloudwatch_client,
 )
+from prowler.providers.aws.services.cloudwatch.lib.metric_filters import (
+    check_cloudwatch_log_metric_filter,
+)
 from prowler.providers.aws.services.cloudwatch.logs_client import logs_client
 
 
@@ -22,26 +23,13 @@ class cloudwatch_log_metric_filter_security_group_changes(Check):
         report.region = cloudwatch_client.region
         report.resource_id = cloudtrail_client.audited_account
         report.resource_arn = cloudtrail_client.audited_account_arn
-        # 1. Iterate for CloudWatch Log Group in CloudTrail trails
+        report = check_cloudwatch_log_metric_filter(
-        log_groups = []
+            pattern,
-        for trail in cloudtrail_client.trails:
+            cloudtrail_client.trails,
-            if trail.log_group_arn:
+            logs_client.metric_filters,
-                log_groups.append(trail.log_group_arn.split(":")[6])
+            cloudwatch_client.metric_alarms,
-        # 2. Describe metric filters for previous log groups
+            report,
-        for metric_filter in logs_client.metric_filters:
+        )
-            if metric_filter.log_group in log_groups:
-                if re.search(pattern, metric_filter.pattern, flags=re.DOTALL):
-                    report.resource_id = metric_filter.log_group
-                    report.resource_arn = metric_filter.arn
-                    report.region = metric_filter.region
-                    report.status = "FAIL"
-                    report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} but no alarms associated."
-                    # 3. Check if there is an alarm for the metric
-                    for alarm in cloudwatch_client.metric_alarms:
-                        if alarm.metric == metric_filter.metric:
-                            report.status = "PASS"
-                            report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} and alarms set."
-                            break
 
         findings.append(report)
         return findings

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

@@ -1,5 +1,3 @@
-import re
-
 from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
     cloudtrail_client,
@@ -7,6 +5,9 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
 from prowler.providers.aws.services.cloudwatch.cloudwatch_client import (
     cloudwatch_client,
 )
+from prowler.providers.aws.services.cloudwatch.lib.metric_filters import (
+    check_cloudwatch_log_metric_filter,
+)
 from prowler.providers.aws.services.cloudwatch.logs_client import logs_client
 
 
@@ -22,26 +23,13 @@ class cloudwatch_log_metric_filter_sign_in_without_mfa(Check):
         report.region = cloudwatch_client.region
         report.resource_id = cloudtrail_client.audited_account
         report.resource_arn = cloudtrail_client.audited_account_arn
-        # 1. Iterate for CloudWatch Log Group in CloudTrail trails
+        report = check_cloudwatch_log_metric_filter(
-        log_groups = []
+            pattern,
-        for trail in cloudtrail_client.trails:
+            cloudtrail_client.trails,
-            if trail.log_group_arn:
+            logs_client.metric_filters,
-                log_groups.append(trail.log_group_arn.split(":")[6])
+            cloudwatch_client.metric_alarms,
-        # 2. Describe metric filters for previous log groups
+            report,
-        for metric_filter in logs_client.metric_filters:
+        )
-            if metric_filter.log_group in log_groups:
-                if re.search(pattern, metric_filter.pattern, flags=re.DOTALL):
-                    report.resource_id = metric_filter.log_group
-                    report.resource_arn = metric_filter.arn
-                    report.region = metric_filter.region
-                    report.status = "FAIL"
-                    report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} but no alarms associated."
-                    # 3. Check if there is an alarm for the metric
-                    for alarm in cloudwatch_client.metric_alarms:
-                        if alarm.metric == metric_filter.metric:
-                            report.status = "PASS"
-                            report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} and alarms set."
-                            break
 
         findings.append(report)
         return findings

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

@@ -1,5 +1,3 @@
-import re
-
 from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
     cloudtrail_client,
@@ -7,6 +5,9 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
 from prowler.providers.aws.services.cloudwatch.cloudwatch_client import (
     cloudwatch_client,
 )
+from prowler.providers.aws.services.cloudwatch.lib.metric_filters import (
+    check_cloudwatch_log_metric_filter,
+)
 from prowler.providers.aws.services.cloudwatch.logs_client import logs_client
 
 
@@ -22,26 +23,13 @@ class cloudwatch_log_metric_filter_unauthorized_api_calls(Check):
         report.region = cloudwatch_client.region
         report.resource_id = cloudtrail_client.audited_account
         report.resource_arn = cloudtrail_client.audited_account_arn
-        # 1. Iterate for CloudWatch Log Group in CloudTrail trails
+        report = check_cloudwatch_log_metric_filter(
-        log_groups = []
+            pattern,
-        for trail in cloudtrail_client.trails:
+            cloudtrail_client.trails,
-            if trail.log_group_arn:
+            logs_client.metric_filters,
-                log_groups.append(trail.log_group_arn.split(":")[6])
+            cloudwatch_client.metric_alarms,
-        # 2. Describe metric filters for previous log groups
+            report,
-        for metric_filter in logs_client.metric_filters:
+        )
-            if metric_filter.log_group in log_groups:
-                if re.search(pattern, metric_filter.pattern, flags=re.DOTALL):
-                    report.resource_id = metric_filter.log_group
-                    report.resource_arn = metric_filter.arn
-                    report.region = metric_filter.region
-                    report.status = "FAIL"
-                    report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} but no alarms associated."
-                    # 3. Check if there is an alarm for the metric
-                    for alarm in cloudwatch_client.metric_alarms:
-                        if alarm.metric == metric_filter.metric:
-                            report.status = "PASS"
-                            report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} and alarms set."
-                            break
 
         findings.append(report)
         return findings

prowler/providers/aws/services/cloudwatch/lib/metric_filters.py

@@ -0,0 +1,34 @@
+import re
+
+from prowler.lib.check.models import Check_Report_AWS
+
+
+def check_cloudwatch_log_metric_filter(
+    metric_filter_pattern: str,
+    trails: list,
+    metric_filters: list,
+    metric_alarms: list,
+    report: Check_Report_AWS,
+):
+    # 1. Iterate for CloudWatch Log Group in CloudTrail trails
+    log_groups = []
+    for trail in trails:
+        if trail.log_group_arn:
+            log_groups.append(trail.log_group_arn.split(":")[6])
+    # 2. Describe metric filters for previous log groups
+    for metric_filter in metric_filters:
+        if metric_filter.log_group in log_groups:
+            if re.search(metric_filter_pattern, metric_filter.pattern, flags=re.DOTALL):
+                report.resource_id = metric_filter.log_group
+                report.resource_arn = metric_filter.arn
+                report.region = metric_filter.region
+                report.status = "FAIL"
+                report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} but no alarms associated."
+                # 3. Check if there is an alarm for the metric
+                for alarm in metric_alarms:
+                    if alarm.metric == metric_filter.metric:
+                        report.status = "PASS"
+                        report.status_extended = f"CloudWatch log group {metric_filter.log_group} found with metric filter {metric_filter.name} and alarms set."
+                        break
+
+    return report