fix(checks): add getattr to fix NoneType errors

This commit is contained in:
pedrooot
2024-12-18 16:00:17 +01:00
parent fd5e7b809f
commit 9575d22c72
24 changed files with 128 additions and 86 deletions
@@ -7,7 +7,7 @@ from prowler.providers.aws.services.cloudwatch.cloudwatch_client import (
class cloudwatch_alarm_actions_alarm_state_configured(Check):
def execute(self):
findings = []
for metric_alarm in cloudwatch_client.metric_alarms:
for metric_alarm in getattr(cloudwatch_client, "metric_alarms", []):
report = Check_Report_AWS(self.metadata())
report.region = metric_alarm.region
report.resource_id = metric_alarm.name
@@ -7,7 +7,7 @@ from prowler.providers.aws.services.cloudwatch.cloudwatch_client import (
class cloudwatch_alarm_actions_enabled(Check):
def execute(self):
findings = []
for metric_alarm in cloudwatch_client.metric_alarms:
for metric_alarm in getattr(cloudwatch_client, "metric_alarms", []):
report = Check_Report_AWS(self.metadata())
report.region = metric_alarm.region
report.resource_id = metric_alarm.name
@@ -18,9 +18,9 @@ class cloudwatch_changes_to_network_acls_alarm_configured(Check):
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
getattr(cloudtrail_client, "trails", []),
getattr(logs_client, "metric_filters", []),
getattr(cloudwatch_client, "metric_alarms", []),
self.metadata(),
)
@@ -18,9 +18,9 @@ class cloudwatch_changes_to_network_gateways_alarm_configured(Check):
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
getattr(cloudtrail_client, "trails", []),
getattr(logs_client, "metric_filters", []),
getattr(cloudwatch_client, "metric_alarms", []),
self.metadata(),
)
@@ -18,9 +18,9 @@ class cloudwatch_changes_to_network_route_tables_alarm_configured(Check):
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
getattr(cloudtrail_client, "trails", []),
getattr(logs_client, "metric_filters", []),
getattr(cloudwatch_client, "metric_alarms", []),
self.metadata(),
)
@@ -18,9 +18,9 @@ class cloudwatch_changes_to_vpcs_alarm_configured(Check):
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
getattr(cloudtrail_client, "trails", []),
getattr(logs_client, "metric_filters", []),
getattr(cloudwatch_client, "metric_alarms", []),
self.metadata(),
)
@@ -7,14 +7,14 @@ class cloudwatch_log_group_not_publicly_accessible(Check):
def execute(self):
findings = []
public_log_groups = []
if (
logs_client.resource_policies is not None
and logs_client.log_groups is not None
if getattr(logs_client, "resource_policies", None) and getattr(
logs_client, "log_groups", None
):
for resource_policies in logs_client.resource_policies.values():
for resource_policy in resource_policies:
if is_policy_public(
resource_policy.policy, logs_client.audited_account
getattr(resource_policy, "policy", None),
getattr(logs_client, "audited_account", None),
):
for statement in resource_policy.policy.get("Statement", []):
public_resources = statement.get("Resource", [])
@@ -34,7 +34,7 @@ class cloudwatch_log_group_not_publicly_accessible(Check):
report.status_extended = (
f"Log Group {log_group.name} is not publicly accessible."
)
if log_group.arn in public_log_groups:
if getattr(log_group, "arn", None) in public_log_groups:
report.status = "FAIL"
report.status_extended = (
f"Log Group {log_group.name} is publicly accessible."
@@ -20,9 +20,9 @@ class cloudwatch_log_metric_filter_and_alarm_for_aws_config_configuration_change
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
getattr(cloudtrail_client, "trails", []),
getattr(logs_client, "metric_filters", []),
getattr(cloudwatch_client, "metric_alarms", []),
self.metadata(),
)
@@ -20,9 +20,9 @@ class cloudwatch_log_metric_filter_and_alarm_for_cloudtrail_configuration_change
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
getattr(cloudtrail_client, "trails", []),
getattr(logs_client, "metric_filters", []),
getattr(cloudwatch_client, "metric_alarms", []),
self.metadata(),
)
@@ -18,9 +18,9 @@ class cloudwatch_log_metric_filter_authentication_failures(Check):
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
getattr(cloudtrail_client, "trails", []),
getattr(logs_client, "metric_filters", []),
getattr(cloudwatch_client, "metric_alarms", []),
self.metadata(),
)
@@ -18,9 +18,9 @@ class cloudwatch_log_metric_filter_aws_organizations_changes(Check):
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
getattr(cloudtrail_client, "trails", []),
getattr(logs_client, "metric_filters", []),
getattr(cloudwatch_client, "metric_alarms", []),
self.metadata(),
)
@@ -18,9 +18,9 @@ class cloudwatch_log_metric_filter_disable_or_scheduled_deletion_of_kms_cmk(Chec
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
getattr(cloudtrail_client, "trails", []),
getattr(logs_client, "metric_filters", []),
getattr(cloudwatch_client, "metric_alarms", []),
self.metadata(),
)
@@ -18,9 +18,9 @@ class cloudwatch_log_metric_filter_for_s3_bucket_policy_changes(Check):
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
getattr(cloudtrail_client, "trails", []),
getattr(logs_client, "metric_filters", []),
getattr(cloudwatch_client, "metric_alarms", []),
self.metadata(),
)
@@ -18,9 +18,9 @@ class cloudwatch_log_metric_filter_policy_changes(Check):
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
getattr(cloudtrail_client, "trails", []),
getattr(logs_client, "metric_filters", []),
getattr(cloudwatch_client, "metric_alarms", []),
self.metadata(),
)
@@ -18,9 +18,9 @@ class cloudwatch_log_metric_filter_root_usage(Check):
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
getattr(cloudtrail_client, "trails", []),
getattr(logs_client, "metric_filters", []),
getattr(cloudwatch_client, "metric_alarms", []),
self.metadata(),
)
@@ -18,9 +18,9 @@ class cloudwatch_log_metric_filter_security_group_changes(Check):
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
getattr(cloudtrail_client, "trails", []),
getattr(logs_client, "metric_filters", []),
getattr(cloudwatch_client, "metric_alarms", []),
self.metadata(),
)
@@ -18,9 +18,9 @@ class cloudwatch_log_metric_filter_sign_in_without_mfa(Check):
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
getattr(cloudtrail_client, "trails", []),
getattr(logs_client, "metric_filters", []),
getattr(cloudwatch_client, "metric_alarms", []),
self.metadata(),
)
@@ -18,9 +18,9 @@ class cloudwatch_log_metric_filter_unauthorized_api_calls(Check):
report = check_cloudwatch_log_metric_filter(
pattern,
cloudtrail_client.trails,
logs_client.metric_filters,
cloudwatch_client.metric_alarms,
getattr(cloudtrail_client, "trails", []),
getattr(logs_client, "metric_filters", []),
getattr(cloudwatch_client, "metric_alarms", []),
self.metadata(),
)
@@ -13,11 +13,13 @@ class codebuild_project_logging_enabled(Check):
report.resource_tags = project.tags
report.status = "PASS"
if project.cloudwatch_logs.enabled and project.s3_logs.enabled:
if getattr(
getattr(project, "cloudwatch_logs", None), "enabled", False
) and getattr(getattr(project, "s3_logs", None), "enabled", False):
report.status_extended = f"CodeBuild project {project.name} has enabled CloudWartch logs in log group {project.cloudwatch_logs.group_name} and S3 logs in bucket {project.s3_logs.bucket_location}."
elif project.cloudwatch_logs.enabled:
elif getattr(getattr(project, "cloudwatch_logs", None), "enabled", False):
report.status_extended = f"CodeBuild project {project.name} has CloudWatch logging enabled in log group {project.cloudwatch_logs.group_name}."
elif project.s3_logs.enabled:
elif getattr(getattr(project, "s3_logs", None), "enabled", False):
report.status_extended = f"CodeBuild project {project.name} has S3 logging enabled in bucket {project.s3_logs.bucket_location}."
else:
report.status = "FAIL"
@@ -6,7 +6,7 @@ class codebuild_project_s3_logs_encrypted(Check):
def execute(self):
findings = []
for project in codebuild_client.projects.values():
if project.s3_logs.enabled:
if getattr(getattr(project, "s3_logs", None), "enabled", False):
report = Check_Report_AWS(self.metadata())
report.resource_id = project.name
report.resource_arn = project.arn
@@ -16,18 +16,29 @@ class network_http_internet_access_restricted(Check):
report.status_extended = f"Security Group {security_group.name} from subscription {subscription} has HTTP internet access restricted."
rule_fail_condition = any(
(
rule.destination_port_range == "80"
getattr(rule, "destination_port_range", "") == "80"
or (
"-" in rule.destination_port_range
and int(rule.destination_port_range.split("-")[0]) <= 80
and int(rule.destination_port_range.split("-")[1]) >= 80
"-" in getattr(rule, "destination_port_range", "")
and int(
getattr(rule, "destination_port_range", "0-0").split(
"-"
)[0]
)
<= 80
and int(
getattr(rule, "destination_port_range", "0-0").split(
"-"
)[1]
)
>= 80
)
)
and rule.protocol in ["TCP", "Tcp", "*"]
and rule.source_address_prefix in ["Internet", "*", "0.0.0.0/0"]
and rule.access == "Allow"
and rule.direction == "Inbound"
for rule in security_group.security_rules
and getattr(rule, "protocol", "").lower() in ["tcp", "*"]
and getattr(rule, "source_address_prefix", "")
in ["Internet", "*", "0.0.0.0/0"]
and getattr(rule, "access", "") == "Allow"
and getattr(rule, "direction", "") == "Inbound"
for rule in getattr(security_group, "security_rules", []) or []
)
if rule_fail_condition:
report.status = "FAIL"
@@ -16,18 +16,29 @@ class network_rdp_internet_access_restricted(Check):
report.status_extended = f"Security Group {security_group.name} from subscription {subscription} has RDP internet access restricted."
rule_fail_condition = any(
(
rule.destination_port_range == "3389"
getattr(rule, "destination_port_range", "") == "3389"
or (
"-" in rule.destination_port_range
and int(rule.destination_port_range.split("-")[0]) <= 3389
and int(rule.destination_port_range.split("-")[1]) >= 3389
"-" in getattr(rule, "destination_port_range", "")
and int(
getattr(rule, "destination_port_range", "0-0").split(
"-"
)[0]
)
<= 3389
and int(
getattr(rule, "destination_port_range", "0-0").split(
"-"
)[1]
)
>= 3389
)
)
and rule.protocol in ["TCP", "Tcp", "*"]
and rule.source_address_prefix in ["Internet", "*", "0.0.0.0/0"]
and rule.access == "Allow"
and rule.direction == "Inbound"
for rule in security_group.security_rules
and getattr(rule, "protocol", "").lower() in ["tcp", "*"]
and getattr(rule, "source_address_prefix", "")
in ["Internet", "*", "0.0.0.0/0"]
and getattr(rule, "access", "") == "Allow"
and getattr(rule, "direction", "") == "Inbound"
for rule in getattr(security_group, "security_rules", []) or []
)
if rule_fail_condition:
report.status = "FAIL"
@@ -16,18 +16,29 @@ class network_ssh_internet_access_restricted(Check):
report.status_extended = f"Security Group {security_group.name} from subscription {subscription} has SSH internet access restricted."
rule_fail_condition = any(
(
rule.destination_port_range == "22"
getattr(rule, "destination_port_range", "") == "22"
or (
"-" in rule.destination_port_range
and int(rule.destination_port_range.split("-")[0]) <= 22
and int(rule.destination_port_range.split("-")[1]) >= 22
"-" in getattr(rule, "destination_port_range", "")
and int(
getattr(rule, "destination_port_range", "0-0").split(
"-"
)[0]
)
<= 22
and int(
getattr(rule, "destination_port_range", "0-0").split(
"-"
)[1]
)
>= 22
)
)
and rule.protocol in ["TCP", "Tcp", "*"]
and rule.source_address_prefix in ["Internet", "*", "0.0.0.0/0"]
and rule.access == "Allow"
and rule.direction == "Inbound"
for rule in security_group.security_rules
and getattr(rule, "protocol", "").lower() in ["tcp", "*"]
and getattr(rule, "source_address_prefix", "")
in ["Internet", "*", "0.0.0.0/0"]
and getattr(rule, "access", "") == "Allow"
and getattr(rule, "direction", "") == "Inbound"
for rule in getattr(security_group, "security_rules", []) or []
)
if rule_fail_condition:
report.status = "FAIL"
@@ -16,9 +16,16 @@ class vm_trusted_launch_enabled(Check):
report.status_extended = f"VM {vm.resource_name} has trusted launch disabled in subscription {subscription_name}"
if (
vm.security_profile.security_type == "TrustedLaunch"
and vm.security_profile.uefi_settings.secure_boot_enabled
and vm.security_profile.uefi_settings.v_tpm_enabled
getattr(vm, "security_profile", None)
and getattr(vm.security_profile, "security_type", None)
== "TrustedLaunch"
and getattr(vm.security_profile, "uefi_settings", None)
and getattr(
vm.security_profile.uefi_settings, "secure_boot_enabled", False
)
and getattr(
vm.security_profile.uefi_settings, "v_tpm_enabled", False
)
):
report.status = "PASS"
report.status_extended = f"VM {vm.resource_name} has trusted launch enabled in subscription {subscription_name}"