diff --git a/aws_quota/check/quota_check.py b/aws_quota/check/quota_check.py index 919ac09..a493c1c 100644 --- a/aws_quota/check/quota_check.py +++ b/aws_quota/check/quota_check.py @@ -17,6 +17,8 @@ class QuotaCheck: scope: QuotaScope = None service_code: str = None quota_code: str = None + warning_threshold: float = None + error_threshold: float = None def __init__(self, boto_session: boto3.Session) -> None: super().__init__() diff --git a/aws_quota/check/vpc.py b/aws_quota/check/vpc.py index f07391e..e1bb029 100644 --- a/aws_quota/check/vpc.py +++ b/aws_quota/check/vpc.py @@ -27,6 +27,13 @@ def get_vpc_by_id(session: boto3.Session, vpc_id: str) -> dict: except StopIteration: raise KeyError +@cachetools.cached(cache=cachetools.TTLCache(1, 60)) +def get_vpc_peering_connections(session: boto3.Session) -> typing.List[dict]: + return session.client('ec2').describe_vpc_peering_connections( + Filters=[ + {'Name': 'status-code', 'Values': ['active']}, + ] + )['VpcPeeringConnections'] @cachetools.cached(cache=cachetools.TTLCache(1, 60)) def get_all_sgs(session: boto3.Session) -> typing.List[dict]: @@ -80,6 +87,16 @@ class InternetGatewayCountCheck(QuotaCheck): def current(self): return len(self.boto_session.client('ec2').describe_internet_gateways()['InternetGateways']) +class VpcEndpointCountCheck(QuotaCheck): + key = "vpc_endpoint" + description = "Gateway VPC endpoints per Region" + scope = QuotaScope.REGION + service_code = 'vpc' + quota_code = 'L-1B52E74A' + + @property + def current(self): + return len(self.boto_session.client('ec2').describe_vpc_endpoints()['VpcEndpoints']) class NetworkInterfaceCountCheck(QuotaCheck): key = "ni_count" @@ -104,6 +121,16 @@ class SecurityGroupCountCheck(QuotaCheck): def current(self): return len(self.boto_session.client('ec2').describe_security_groups()['SecurityGroups']) +class NatGatewayCountCheck(QuotaCheck): + key = "nat_count" + description = "NAT gateways per Region" + scope = QuotaScope.REGION + service_code = 'vpc' + quota_code = 'L-FE5A380F' + + @property + def current(self): + return len(self.boto_session.client('ec2').describe_nat_gateways()['NatGateways']) class RulesPerSecurityGroupCheck(InstanceQuotaCheck): key = "vpc_rules_per_sg" @@ -275,3 +302,29 @@ def current(self) -> int: return len(list(filter(lambda cbas: cbas['Ipv6CidrBlockState']['State'] == 'associated', vpc['Ipv6CidrBlockAssociationSet']))) except KeyError: raise InstanceWithIdentifierNotFound(self) + +class ActiveVpcPeeringConnectionsPerVpcCheck(InstanceQuotaCheck): + key = "vpc_peering_connections_per_vpc" + description = "Active VPC peering connections per VPC" + service_code = 'vpc' + quota_code = 'L-7E9ECCDB' + instance_id = 'VPC ID' + + @staticmethod + def get_all_identifiers(session: boto3.Session) -> typing.List[str]: + return [vpc['VpcId'] for vpc in get_all_vpcs(session)] + + @property + def current(self) -> int: + peering_connections_per_vpc = 0 + try: + vpc = get_vpc_by_id(self.boto_session, self.instance_id) + vpc_peering_connections = get_vpc_peering_connections(self.boto_session) + for peering_connection in vpc_peering_connections: + for vpc_info in [peering_connection['AccepterVpcInfo'], peering_connection['RequesterVpcInfo']]: + if vpc_info['VpcId'] == vpc['VpcId'] and self.boto_session.region_name == vpc_info['Region']: + peering_connections_per_vpc += 1 + + return peering_connections_per_vpc + except KeyError: + raise InstanceWithIdentifierNotFound(self)