diff --git a/prowler/changelog.d/cloudwatch-metric-filter-missing-log-group.fixed.md b/prowler/changelog.d/cloudwatch-metric-filter-missing-log-group.fixed.md new file mode 100644 index 0000000000..b17ec3af4b --- /dev/null +++ b/prowler/changelog.d/cloudwatch-metric-filter-missing-log-group.fixed.md @@ -0,0 +1 @@ +CloudWatch log metric filter checks no longer crash with `AttributeError` when the account has a metric filter whose log group was not retrieved diff --git a/prowler/providers/aws/services/cloudwatch/lib/metric_filters.py b/prowler/providers/aws/services/cloudwatch/lib/metric_filters.py index e5d104b840..069f542b2d 100644 --- a/prowler/providers/aws/services/cloudwatch/lib/metric_filters.py +++ b/prowler/providers/aws/services/cloudwatch/lib/metric_filters.py @@ -48,7 +48,27 @@ def check_cloudwatch_log_metric_filter( metric_filters: list, metric_alarms: list, metadata: dict, -): +) -> Check_Report_AWS | None: + """Report whether a trail's log group has a matching metric filter and an alarm. + + Only metric filters attached to a log group that a CloudTrail trail delivers to + are considered, and only those whose own pattern matches + ``metric_filter_pattern``. A filter whose log group was not retrieved is skipped + rather than dereferenced. One compliant filter anywhere short-circuits to PASS; + otherwise the last matching filter found without an alarm is returned as FAIL. + + Args: + metric_filter_pattern: regex from ``build_metric_filter_pattern``, matched + against each filter's pattern with ``re.DOTALL``. + trails: CloudTrail trails keyed by ARN; only those with a log group count. + metric_filters: CloudWatch Logs metric filters to evaluate. + metric_alarms: CloudWatch alarms, matched to a filter by metric name. + metadata: check metadata for the emitted report. + + Returns: + A ``Check_Report_AWS`` for the deciding log group, or ``None`` when no + filter matched or an inventory was not collected. + """ report = None # 1. Iterate for CloudWatch Log Group in CloudTrail trails log_groups = [] @@ -58,6 +78,10 @@ def check_cloudwatch_log_metric_filter( log_groups.append(trail.log_group_arn.split(":")[6]) # 2. Describe metric filters for previous log groups for metric_filter in metric_filters: + # A filter whose log group was not retrieved cannot be matched against + # the trail log groups, so it cannot satisfy the requirement. + if metric_filter.log_group is None: + continue if metric_filter.log_group.name in log_groups and re.search( metric_filter_pattern, metric_filter.pattern, flags=re.DOTALL ): diff --git a/tests/providers/aws/services/cloudwatch/lib/metric_filters_test.py b/tests/providers/aws/services/cloudwatch/lib/metric_filters_test.py new file mode 100644 index 0000000000..e29b2f0a5f --- /dev/null +++ b/tests/providers/aws/services/cloudwatch/lib/metric_filters_test.py @@ -0,0 +1,149 @@ +from pathlib import Path + +import prowler +from prowler.lib.check.models import CheckMetadata +from prowler.providers.aws.services.cloudtrail.cloudtrail_service import Trail +from prowler.providers.aws.services.cloudwatch.cloudwatch_service import ( + LogGroup, + MetricAlarm, + MetricFilter, +) +from prowler.providers.aws.services.cloudwatch.lib.metric_filters import ( + check_cloudwatch_log_metric_filter, +) + +AWS_REGION = "eu-west-1" +AWS_ACCOUNT_NUMBER = "123456789012" + +TRAIL_ARN = f"arn:aws:cloudtrail:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:trail/trail-test" +TRAIL_LOG_GROUP_NAME = "trail-log-group" +TRAIL_LOG_GROUP_ARN = ( + f"arn:aws:logs:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:log-group:{TRAIL_LOG_GROUP_NAME}:*" +) + +# Matches the filter patterns built below, so a filter is only skipped because of +# its missing log group and never because the pattern failed to match. +PATTERN = r"\$\.eventName\s*=\s*.?PutBucketPolicy" +FILTER_PATTERN = "{ ($.eventName = PutBucketPolicy) }" +METRIC_NAME = "PutBucketPolicyCount" + +METADATA = CheckMetadata.parse_file( + Path(prowler.__file__).parent + / "providers/aws/services/cloudwatch/cloudwatch_log_metric_filter_unauthorized_api_calls/cloudwatch_log_metric_filter_unauthorized_api_calls.metadata.json" +).json() + + +def _trails(): + """One trail delivering to TRAIL_LOG_GROUP_NAME, putting that log group in scope. + + Without a trail carrying a log group ARN the check has nothing to match filters + against and returns None for any input, which would make every assertion below + pass for the wrong reason. + """ + return {TRAIL_ARN: Trail(region=AWS_REGION, log_group_arn=TRAIL_LOG_GROUP_ARN)} + + +def _trail_log_group(): + """The trail's log group as the CloudWatch service would have collected it.""" + return LogGroup( + arn=TRAIL_LOG_GROUP_ARN, + name=TRAIL_LOG_GROUP_NAME, + retention_days=7, + never_expire=False, + region=AWS_REGION, + ) + + +def _metric_filter(name, log_group, metric=METRIC_NAME): + """Build a metric filter whose pattern always matches PATTERN. + + Args: + name: filter name, which the report echoes in status_extended. + log_group: the collected LogGroup, or None to model a filter whose log + group was never retrieved -- the input that used to raise. + metric: metric name an alarm has to carry for the filter to be compliant. + """ + return MetricFilter( + arn=f"arn:aws:logs:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:metric-filter/{name}", + name=name, + metric=metric, + pattern=FILTER_PATTERN, + log_group=log_group, + region=AWS_REGION, + ) + + +def _alarm(metric=METRIC_NAME): + """Build an alarm on metric; a non-default name models an unrelated alarm. + + The check pairs alarms to filters by metric name alone, so passing a metric no + filter uses is how a filter with no alarm of its own is expressed. + """ + return MetricAlarm( + arn=f"arn:aws:cloudwatch:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:alarm:{metric}-alarm", + name=f"{metric}-alarm", + metric=metric, + name_space="CloudTrailMetrics", + region=AWS_REGION, + alarm_actions=[f"arn:aws:sns:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:topic-test"], + actions_enabled=True, + ) + + +class Test_check_cloudwatch_log_metric_filter: + def test_metric_filter_without_collected_log_group(self): + """A metric filter whose log group was not retrieved must be skipped + instead of raising AttributeError on the None log group.""" + report = check_cloudwatch_log_metric_filter( + PATTERN, + _trails(), + [_metric_filter("orphan-filter", None)], + [_alarm()], + METADATA, + ) + + assert report is None + + def test_uncollected_log_group_does_not_mask_a_compliant_filter(self): + """The skip must not abandon the remaining filters: a compliant filter + listed after an uncollected one still has to be evaluated.""" + report = check_cloudwatch_log_metric_filter( + PATTERN, + _trails(), + [ + _metric_filter("orphan-filter", None), + _metric_filter("trail-filter", _trail_log_group()), + ], + [_alarm()], + METADATA, + ) + + assert report is not None + assert report.status == "PASS" + assert ( + report.status_extended + == f"CloudWatch log group {TRAIL_LOG_GROUP_NAME} found with metric filter trail-filter and alarms set." + ) + assert report.resource_id == TRAIL_LOG_GROUP_NAME + assert report.resource_arn == TRAIL_LOG_GROUP_ARN + assert report.region == AWS_REGION + + def test_uncollected_log_group_does_not_mask_a_missing_alarm(self): + """The skip must not turn a filter with no alarm into a pass.""" + report = check_cloudwatch_log_metric_filter( + PATTERN, + _trails(), + [ + _metric_filter("orphan-filter", None), + _metric_filter("trail-filter", _trail_log_group()), + ], + [_alarm(metric="UnrelatedMetric")], + METADATA, + ) + + assert report is not None + assert report.status == "FAIL" + assert ( + report.status_extended + == f"CloudWatch log group {TRAIL_LOG_GROUP_NAME} found with metric filter trail-filter but no alarms associated." + )