chore(cloudwatch): Improve checks related with function check_cloudwatch_log_metric_filter (#5286)

This commit is contained in:
Rubén De la Torre Vico
2024-10-04 17:18:46 +02:00
committed by GitHub
parent d5acdc766a
commit df10253056
17 changed files with 275 additions and 274 deletions
@@ -15,24 +15,24 @@ class cloudwatch_changes_to_network_acls_alarm_configured(Check):
def execute(self):
pattern = r"\$\.eventName\s*=\s*.?CreateNetworkAcl.+\$\.eventName\s*=\s*.?CreateNetworkAclEntry.+\$\.eventName\s*=\s*.?DeleteNetworkAcl.+\$\.eventName\s*=\s*.?DeleteNetworkAclEntry.+\$\.eventName\s*=\s*.?ReplaceNetworkAclEntry.+\$\.eventName\s*=\s*.?ReplaceNetworkAclAssociation.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
report,
)
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
self.metadata(),
)
if report is not None:
if report == Check_Report_AWS(self.metadata()):
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report.resource_tags = []
findings.append(report)
return findings
@@ -15,24 +15,24 @@ class cloudwatch_changes_to_network_gateways_alarm_configured(Check):
def execute(self):
pattern = r"\$\.eventName\s*=\s*.?CreateCustomerGateway.+\$\.eventName\s*=\s*.?DeleteCustomerGateway.+\$\.eventName\s*=\s*.?AttachInternetGateway.+\$\.eventName\s*=\s*.?CreateInternetGateway.+\$\.eventName\s*=\s*.?DeleteInternetGateway.+\$\.eventName\s*=\s*.?DetachInternetGateway.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
report,
)
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
self.metadata(),
)
if report is not None:
if report == Check_Report_AWS(self.metadata()):
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report.resource_tags = []
findings.append(report)
return findings
@@ -15,24 +15,24 @@ class cloudwatch_changes_to_network_route_tables_alarm_configured(Check):
def execute(self):
pattern = r"\$\.eventSource\s*=\s*.?ec2.amazonaws.com.+\$\.eventName\s*=\s*.?CreateRoute.+\$\.eventName\s*=\s*.?CreateRouteTable.+\$\.eventName\s*=\s*.?ReplaceRoute.+\$\.eventName\s*=\s*.?ReplaceRouteTableAssociation.+\$\.eventName\s*=\s*.?DeleteRouteTable.+\$\.eventName\s*=\s*.?DeleteRoute.+\$\.eventName\s*=\s*.?DisassociateRouteTable.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
report,
)
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
self.metadata(),
)
if report is not None:
if report == Check_Report_AWS(self.metadata()):
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report.resource_tags = []
findings.append(report)
return findings
@@ -15,24 +15,24 @@ class cloudwatch_changes_to_vpcs_alarm_configured(Check):
def execute(self):
pattern = r"\$\.eventName\s*=\s*.?CreateVpc.+\$\.eventName\s*=\s*.?DeleteVpc.+\$\.eventName\s*=\s*.?ModifyVpcAttribute.+\$\.eventName\s*=\s*.?AcceptVpcPeeringConnection.+\$\.eventName\s*=\s*.?CreateVpcPeeringConnection.+\$\.eventName\s*=\s*.?DeleteVpcPeeringConnection.+\$\.eventName\s*=\s*.?RejectVpcPeeringConnection.+\$\.eventName\s*=\s*.?AttachClassicLinkVpc.+\$\.eventName\s*=\s*.?DetachClassicLinkVpc.+\$\.eventName\s*=\s*.?DisableVpcClassicLink.+\$\.eventName\s*=\s*.?EnableVpcClassicLink.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
report,
)
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
self.metadata(),
)
if report is not None:
if report == Check_Report_AWS(self.metadata()):
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report.resource_tags = []
findings.append(report)
return findings
@@ -17,24 +17,24 @@ class cloudwatch_log_metric_filter_and_alarm_for_aws_config_configuration_change
def execute(self):
pattern = r"\$\.eventSource\s*=\s*.?config.amazonaws.com.+\$\.eventName\s*=\s*.?StopConfigurationRecorder.+\$\.eventName\s*=\s*.?DeleteDeliveryChannel.+\$\.eventName\s*=\s*.?PutDeliveryChannel.+\$\.eventName\s*=\s*.?PutConfigurationRecorder.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
report,
)
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
self.metadata(),
)
if report is not None:
if report == Check_Report_AWS(self.metadata()):
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report.resource_tags = []
findings.append(report)
return findings
@@ -17,24 +17,24 @@ class cloudwatch_log_metric_filter_and_alarm_for_cloudtrail_configuration_change
def execute(self):
pattern = r"\$\.eventName\s*=\s*.?CreateTrail.+\$\.eventName\s*=\s*.?UpdateTrail.+\$\.eventName\s*=\s*.?DeleteTrail.+\$\.eventName\s*=\s*.?StartLogging.+\$\.eventName\s*=\s*.?StopLogging.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
report,
)
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
self.metadata(),
)
if report is not None:
if report == Check_Report_AWS(self.metadata()):
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report.resource_tags = []
findings.append(report)
return findings
@@ -15,24 +15,24 @@ class cloudwatch_log_metric_filter_authentication_failures(Check):
def execute(self):
pattern = r"\$\.eventName\s*=\s*.?ConsoleLogin.+\$\.errorMessage\s*=\s*.?Failed authentication.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
report,
)
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
self.metadata(),
)
if report is not None:
if report == Check_Report_AWS(self.metadata()):
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report.resource_tags = []
findings.append(report)
return findings
@@ -15,24 +15,24 @@ class cloudwatch_log_metric_filter_aws_organizations_changes(Check):
def execute(self):
pattern = r"\$\.eventSource\s*=\s*.?organizations\.amazonaws\.com.+\$\.eventName\s*=\s*.?AcceptHandshake.+\$\.eventName\s*=\s*.?AttachPolicy.+\$\.eventName\s*=\s*.?CancelHandshake.+\$\.eventName\s*=\s*.?CreateAccount.+\$\.eventName\s*=\s*.?CreateOrganization.+\$\.eventName\s*=\s*.?CreateOrganizationalUnit.+\$\.eventName\s*=\s*.?CreatePolicy.+\$\.eventName\s*=\s*.?DeclineHandshake.+\$\.eventName\s*=\s*.?DeleteOrganization.+\$\.eventName\s*=\s*.?DeleteOrganizationalUnit.+\$\.eventName\s*=\s*.?DeletePolicy.+\$\.eventName\s*=\s*.?EnableAllFeatures.+\$\.eventName\s*=\s*.?EnablePolicyType.+\$\.eventName\s*=\s*.?InviteAccountToOrganization.+\$\.eventName\s*=\s*.?LeaveOrganization.+\$\.eventName\s*=\s*.?DetachPolicy.+\$\.eventName\s*=\s*.?DisablePolicyType.+\$\.eventName\s*=\s*.?MoveAccount.+\$\.eventName\s*=\s*.?RemoveAccountFromOrganization.+\$\.eventName\s*=\s*.?UpdateOrganizationalUnit.+\$\.eventName\s*=\s*.?UpdatePolicy.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
report,
)
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
self.metadata(),
)
if report is not None:
if report == Check_Report_AWS(self.metadata()):
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report.resource_tags = []
findings.append(report)
return findings
@@ -15,24 +15,24 @@ class cloudwatch_log_metric_filter_disable_or_scheduled_deletion_of_kms_cmk(Chec
def execute(self):
pattern = r"\$\.eventSource\s*=\s*.?kms.amazonaws.com.+\$\.eventName\s*=\s*.?DisableKey.+\$\.eventName\s*=\s*.?ScheduleKeyDeletion.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
report,
)
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
self.metadata(),
)
if report is not None:
if report == Check_Report_AWS(self.metadata()):
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report.resource_tags = []
findings.append(report)
return findings
@@ -15,25 +15,24 @@ class cloudwatch_log_metric_filter_for_s3_bucket_policy_changes(Check):
def execute(self):
pattern = r"\$\.eventSource\s*=\s*.?s3.amazonaws.com.+\$\.eventName\s*=\s*.?PutBucketAcl.+\$\.eventName\s*=\s*.?PutBucketPolicy.+\$\.eventName\s*=\s*.?PutBucketCors.+\$\.eventName\s*=\s*.?PutBucketLifecycle.+\$\.eventName\s*=\s*.?PutBucketReplication.+\$\.eventName\s*=\s*.?DeleteBucketPolicy.+\$\.eventName\s*=\s*.?DeleteBucketCors.+\$\.eventName\s*=\s*.?DeleteBucketLifecycle.+\$\.eventName\s*=\s*.?DeleteBucketReplication.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
report,
)
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
self.metadata(),
)
if report is not None:
if report == Check_Report_AWS(self.metadata()):
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report.resource_tags = []
findings.append(report)
return findings
@@ -15,24 +15,24 @@ class cloudwatch_log_metric_filter_policy_changes(Check):
def execute(self):
pattern = r"\$\.eventName\s*=\s*.?DeleteGroupPolicy.+\$\.eventName\s*=\s*.?DeleteRolePolicy.+\$\.eventName\s*=\s*.?DeleteUserPolicy.+\$\.eventName\s*=\s*.?PutGroupPolicy.+\$\.eventName\s*=\s*.?PutRolePolicy.+\$\.eventName\s*=\s*.?PutUserPolicy.+\$\.eventName\s*=\s*.?CreatePolicy.+\$\.eventName\s*=\s*.?DeletePolicy.+\$\.eventName\s*=\s*.?CreatePolicyVersion.+\$\.eventName\s*=\s*.?DeletePolicyVersion.+\$\.eventName\s*=\s*.?AttachRolePolicy.+\$\.eventName\s*=\s*.?DetachRolePolicy.+\$\.eventName\s*=\s*.?AttachUserPolicy.+\$\.eventName\s*=\s*.?DetachUserPolicy.+\$\.eventName\s*=\s*.?AttachGroupPolicy.+\$\.eventName\s*=\s*.?DetachGroupPolicy.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
report,
)
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
self.metadata(),
)
if report is not None:
if report == Check_Report_AWS(self.metadata()):
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report.resource_tags = []
findings.append(report)
return findings
@@ -15,24 +15,24 @@ class cloudwatch_log_metric_filter_root_usage(Check):
def execute(self):
pattern = r"\$\.userIdentity\.type\s*=\s*.?Root.+\$\.userIdentity\.invokedBy NOT EXISTS.+\$\.eventType\s*!=\s*.?AwsServiceEvent.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
report,
)
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
self.metadata(),
)
if report is not None:
if report == Check_Report_AWS(self.metadata()):
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report.resource_tags = []
findings.append(report)
return findings
@@ -15,24 +15,24 @@ class cloudwatch_log_metric_filter_security_group_changes(Check):
def execute(self):
pattern = r"\$\.eventName\s*=\s*.?AuthorizeSecurityGroupIngress.+\$\.eventName\s*=\s*.?AuthorizeSecurityGroupEgress.+\$\.eventName\s*=\s*.?RevokeSecurityGroupIngress.+\$\.eventName\s*=\s*.?RevokeSecurityGroupEgress.+\$\.eventName\s*=\s*.?CreateSecurityGroup.+\$\.eventName\s*=\s*.?DeleteSecurityGroup.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
report,
)
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
self.metadata(),
)
if report is not None:
if report == Check_Report_AWS(self.metadata()):
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report.resource_tags = []
findings.append(report)
return findings
@@ -15,24 +15,24 @@ class cloudwatch_log_metric_filter_sign_in_without_mfa(Check):
def execute(self):
pattern = r"\$\.eventName\s*=\s*.?ConsoleLogin.+\$\.additionalEventData\.MFAUsed\s*!=\s*.?Yes.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
report,
)
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
self.metadata(),
)
if report is not None:
if report == Check_Report_AWS(self.metadata()):
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report.resource_tags = []
findings.append(report)
return findings
@@ -15,24 +15,24 @@ class cloudwatch_log_metric_filter_unauthorized_api_calls(Check):
def execute(self):
pattern = r"\$\.errorCode\s*=\s*.?\*UnauthorizedOperation.+\$\.errorCode\s*=\s*.?AccessDenied\*.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
report,
)
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
self.metadata(),
)
if report is not None:
if report == Check_Report_AWS(self.metadata()):
report.status = "FAIL"
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
report.resource_tags = []
findings.append(report)
return findings
@@ -134,7 +134,7 @@ class Logs(AWSService):
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
if not self.metric_filters:
self.metric_filters = []
self.metric_filters = None
else:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -8,11 +8,13 @@ def check_cloudwatch_log_metric_filter(
trails: list,
metric_filters: list,
metric_alarms: list,
report: Check_Report_AWS,
metadata: dict,
):
report = None
# 1. Iterate for CloudWatch Log Group in CloudTrail trails
log_groups = []
if trails is not None:
if trails is not None and metric_filters is not None and metric_alarms is not None:
report = Check_Report_AWS(metadata)
for trail in trails.values():
if trail.log_group_arn:
log_groups.append(trail.log_group_arn.split(":")[6])
@@ -24,7 +26,7 @@ def check_cloudwatch_log_metric_filter(
report.resource_id = metric_filter.log_group.name
report.resource_arn = metric_filter.log_group.arn
report.region = metric_filter.log_group.region
report.resource_tags = metric_filter.log_group.tags
report.resource_tags = getattr(metric_filter.log_group, "tags", [])
report.status = "FAIL"
report.status_extended = f"CloudWatch log group {metric_filter.log_group.name} found with metric filter {metric_filter.name} but no alarms associated."
# 3. Check if there is an alarm for the metric