fix(aws): solve None type errors (#6268)

This commit is contained in:
Sergio Garcia
2024-12-19 11:32:33 -05:00
committed by GitHub
parent 58cd944618
commit bb5b63f62f
4 changed files with 32 additions and 25 deletions
@@ -18,7 +18,8 @@ class Backup(AWSService):
self.backup_vault_arn_template = f"arn:{self.audited_partition}:backup:{self.region}:{self.audited_account}:backup-vault"
self.backup_vaults = []
self.__threading_call__(self._list_backup_vaults)
self.__threading_call__(self._list_tags, self.backup_vaults)
if self.backup_vaults is not None:
self.__threading_call__(self._list_tags, self.backup_vaults)
self.backup_plans = []
self.__threading_call__(self._list_backup_plans)
self.__threading_call__(self._list_tags, self.backup_plans)
@@ -28,6 +29,7 @@ class Backup(AWSService):
self.__threading_call__(self._list_backup_selections)
self.recovery_points = []
self.__threading_call__(self._list_recovery_points)
self.__threading_call__(self._list_tags, self.recovery_points)
def _list_backup_vaults(self, regional_client):
logger.info("Backup - Listing Backup Vaults...")
@@ -171,10 +173,11 @@ class Backup(AWSService):
def _list_tags(self, resource):
try:
tags = self.regional_clients[resource.region].list_tags(
ResourceArn=resource.arn
)["Tags"]
resource.tags = [tags] if tags else []
if getattr(resource, "arn", None):
tags = self.regional_clients[resource.region].list_tags(
ResourceArn=resource.arn
)["Tags"]
resource.tags = [tags] if tags else []
except Exception as error:
logger.error(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -18,24 +18,24 @@ def check_cloudwatch_log_metric_filter(
for trail in trails.values():
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.name in log_groups and re.search(
metric_filter_pattern, metric_filter.pattern, flags=re.DOTALL
):
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 = 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
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.name} found with metric filter {metric_filter.name} and alarms set."
# 2. Describe metric filters for previous log groups
for metric_filter in metric_filters:
if metric_filter.log_group.name in log_groups and re.search(
metric_filter_pattern, metric_filter.pattern, flags=re.DOTALL
):
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 = 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
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.name} found with metric filter {metric_filter.name} and alarms set."
break
if report.status == "PASS":
break
if report.status == "PASS":
break
return report
@@ -19,7 +19,10 @@ class ec2_instance_uses_single_eni(Check):
)
else:
for eni_id in instance.network_interfaces:
if ec2_client.network_interfaces[eni_id].type in eni_types:
if (
eni_id in ec2_client.network_interfaces
and ec2_client.network_interfaces[eni_id].type in eni_types
):
eni_types[ec2_client.network_interfaces[eni_id].type].append(
eni_id
)
@@ -112,7 +112,8 @@ class IAM(AWSService):
[policy for policy in self.policies if policy.type == "Custom"],
)
self.__threading_call__(self._list_tags, self.server_certificates)
self.__threading_call__(self._list_tags, self.saml_providers.values())
if self.saml_providers is not None:
self.__threading_call__(self._list_tags, self.saml_providers.values())
def _get_client(self):
return self.client