"""Avoid using subnets that are short on available IPs. Works for the next resources: - aws_lb - aws_ecs_service - aws_instance - aws_network_interface """ from checkov.common.models.enums import CheckCategories, CheckResult from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck import logging import math import botocore import ipaddress import tools class SubnetsWithIPShortage(BaseResourceCheck): def __init__(self): name = 'Avoid using subnets that are short on available IPs' id = 'ORCD_AWS_1' categories = [CheckCategories.NETWORKING] supported_resources = [ 'aws_lb', 'aws_ecs_service', 'aws_instance', 'aws_network_interface'] guideline = 'https://www.notion.so/Checkov-guide-e5c30d67d35248ebbb0806df59775e0e#46166c83eab648edb5ea9212d8197626' self.min_available_ips = 25 self.min_available_percent = 10 self.aws_ec2 = None self.subnets_cache = {} super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources, guideline=guideline) def scan_resource_conf(self, conf): subnets = [] # aws_lb data = tools.flatten(conf.get('subnets')) if data and isinstance(data, list): subnets = data # aws_ecs_service data = tools.flatten(conf.get('network_configuration')) if data and isinstance(data, dict): data = tools.flatten(data.get('subnets')) if data and isinstance(data, list): subnets = data # aws_instance, aws_network_interface data = tools.flatten(conf.get('subnet_id')) if data: subnets.append(data) return self._analyze_subnets(subnets) def _analyze_subnets(self, subnets): try: session = tools.get_aws_session() self.aws_ec2 = session.resource('ec2') except Exception as exc: logging.error(f'Failed to create AWS session: {exc}') return CheckResult.SKIPPED for subnet_id in subnets: try: subnet_info = self._get_subnet_info(subnet_id) if subnet_info['available_ips'] < self.min_available_ips: return CheckResult.FAILED if subnet_info['available_percent'] < self.min_available_percent: return CheckResult.FAILED except botocore.exceptions.ClientError as exc: logging.error(f'Failed to describe subnet {subnet_id}: {exc}') return CheckResult.SKIPPED return CheckResult.PASSED def _get_subnet_info(self, subnet_id): if subnet_id in self.subnets_cache: return self.subnets_cache[subnet_id] subnet = self.aws_ec2.Subnet(subnet_id) network = ipaddress.ip_network(subnet.cidr_block) total_ips = network.num_addresses - 5 available_ips = subnet.available_ip_address_count available_percent = math.floor(100 * available_ips / total_ips) subnet_info = { 'total_ips': total_ips, 'available_ips': available_ips, 'available_percent': available_percent } self.subnets_cache[subnet_id] = subnet_info return subnet_info check = SubnetsWithIPShortage()