added lots of checks

rhythmictech · Aug 9, 2024 · 3e76bfa · 3e76bfa
1 parent 1bd801b
commit 3e76bfa
Show file tree

Hide file tree

Showing 13 changed files with 976 additions and 38 deletions.
diff --git a/README.md b/README.md
@@ -84,6 +84,7 @@ We open source the vast majority of the resources we use to deliver our managed
 | [aws_iam_role.monitor_service_quotas_execution](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/iam_role) | resource |
 | [aws_iam_role_policy_attachment.monitor_ami_usage_execution](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/iam_role_policy_attachment) | resource |
 | [aws_iam_role_policy_attachment.monitor_service_quotas_execution](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/iam_role_policy_attachment) | resource |
+| [aws_iam_role_policy_attachment.monitor_service_quotas_security_analyst](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/iam_role_policy_attachment) | resource |
 | [aws_lambda_function.monitor_ami_usage](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/lambda_function) | resource |
 | [aws_lambda_function.monitor_service_quotas](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/lambda_function) | resource |
 | [aws_lambda_permission.monitor_ami_usage](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/lambda_permission) | resource |

diff --git a/lambda/monitor_service_quotas/cloudformation_checks.py b/lambda/monitor_service_quotas/cloudformation_checks.py
@@ -0,0 +1,44 @@
+from abc import ABC, abstractmethod
+import boto3
+from datetime import datetime
+
+class CloudFormationClientSingleton:
+    _instances = {}
+
+    @classmethod
+    def get_client(cls, region):
+        if region not in cls._instances:
+            cls._instances[region] = boto3.client('cloudformation', region_name=region)
+        return cls._instances[region]
+
+class CloudFormationUsageChecker(ABC):
+
+    @abstractmethod
+    def get_usage(self, region, quota_name):
+        pass
+
+class StackInstancesPerStackSetChecker(CloudFormationUsageChecker):
+    def get_usage(self, region, quota_name):
+        cfn_client = CloudFormationClientSingleton.get_client(region)
+        paginator = cfn_client.get_paginator('list_stack_sets')
+        max_instances = 0
+
+        for page in paginator.paginate():
+            for stack_set in page['Summaries']:
+                stack_set_name = stack_set['StackSetName']
+                instances = cfn_client.list_stack_instances(StackSetName=stack_set_name)['Summaries']
+                instance_count = len(instances)
+                max_instances = max(max_instances, instance_count)
+        return max_instances
+
+
+class StackCountChecker(CloudFormationUsageChecker):
+    def get_usage(self, region, quota_name):
+        cfn_client = CloudFormationClientSingleton.get_client(region)
+        paginator = cfn_client.get_paginator('list_stacks')
+        stack_count = 0
+
+        for page in paginator.paginate(StackStatusFilter=['CREATE_COMPLETE', 'UPDATE_COMPLETE', 'UPDATE_ROLLBACK_COMPLETE']):
+            stack_count += len(page['StackSummaries'])
+
+        return stack_count
diff --git a/lambda/monitor_service_quotas/ebs_checks.py b/lambda/monitor_service_quotas/ebs_checks.py
@@ -0,0 +1,140 @@
+from abc import ABC, abstractmethod
+import boto3
+from datetime import datetime
+
+class EBSClientSingleton:
+    _instances = {}
+
+    @classmethod
+    def get_client(cls, region):
+        if region not in cls._instances:
+            cls._instances[region] = boto3.client('ec2', region_name=region)
+        return cls._instances[region]
+
+class EBSUsageChecker(ABC):
+
+    @abstractmethod
+    def get_usage(self, region, quota_name):
+        pass
+
+class VolumeTypeStorageChecker(EBSUsageChecker):
+    VOLUME_TYPE_MAP = {
+        'Cold HDD (sc1)': 'sc1',
+        'Throughput Optimized HDD (st1)': 'st1',
+        'General Purpose SSD (gp2)': 'gp2',
+        'General Purpose SSD (gp3)': 'gp3',
+        'Provisioned IOPS SSD (io1)': 'io1',
+        'Provisioned IOPS SSD (io2)': 'io2',
+        'Magnetic (standard)': 'standard'
+    }
+
+    def get_usage(self, region, quota_name):
+        ebs = EBSClientSingleton.get_client(region)
+        total_storage = 0
+        next_token = None
+
+        volume_type = self.VOLUME_TYPE_MAP.get(quota_name.split(' volumes')[0].strip())
+        if not volume_type:
+            raise ValueError(f"Unknown volume type in quota: {quota_name}")
+
+        while True:
+            if next_token:
+                response = ec2.describe_volumes(
+                    Filters=[{'Name': 'volume-type', 'Values': [volume_type]}],
+                    NextToken=next_token
+                )
+            else:
+                response = ec2.describe_volumes(
+                    Filters=[{'Name': 'volume-type', 'Values': [volume_type]}]
+                )
+
+            for volume in response['Volumes']:
+                total_storage += volume['Size']
+
+            next_token = response.get('NextToken')
+            if not next_token:
+                break
+
+        return total_storage
+
+class ProvisionedIOPSChecker(EBSUsageChecker):
+    def get_usage(self, region, quota_name):
+        ec2 = EBSClientSingleton.get_client(region)
+        total_iops = 0
+        next_token = None
+
+        volume_type = quota_name.split('IOPS for Provisioned IOPS SSD (')[1].split(')')[0]
+        while True:
+            if next_token:
+                response = ec2.describe_volumes(
+                    Filters=[{'Name': 'volume-type', 'Values': ['io2']}],
+                    NextToken=next_token
+                )
+            else:
+                response = ec2.describe_volumes(
+                    Filters=[{'Name': 'volume-type', 'Values': ['io2']}]
+                )
+
+            for volume in response['Volumes']:
+                total_iops += volume['Iops']
+
+            next_token = response.get('NextToken')
+            if not next_token:
+                break
+
+        return total_iops
+
+class ArchivedSnapshotsPerVolumeChecker(EBSUsageChecker):
+    def get_usage(self, region, quota_name):
+        ec2 = EBSClientSingleton.get_client(region)
+        snapshot_counts = {}
+        next_token = None
+
+        while True:
+            if next_token:
+                response = ec2.describe_snapshots(
+                    OwnerIds=['self'],
+                    Filters=[{'Name': 'storage-tier', 'Values': ['archive']}],
+                    NextToken=next_token
+                )
+            else:
+                response = ec2.describe_snapshots(
+                    OwnerIds=['self'],
+                    Filters=[{'Name': 'storage-tier', 'Values': ['archive']}]
+                )
+
+            for snapshot in response['Snapshots']:
+                volume_id = snapshot.get('VolumeId')
+                if volume_id:
+                    snapshot_counts[volume_id] = snapshot_counts.get(volume_id, 0) + 1
+
+            next_token = response.get('NextToken')
+            if not next_token:
+                break
+
+        return max(snapshot_counts.values()) if snapshot_counts else 0
+
+class SnapshotsPerRegionChecker(EBSUsageChecker):
+    def get_usage(self, region, quota_name):
+        ec2 = EBSClientSingleton.get_client(region)
+        total_snapshots = 0
+        next_token = None
+
+        while True:
+            if next_token:
+                response = ec2.describe_snapshots(
+                    OwnerIds=['self'],
+                    NextToken=next_token
+                )
+            else:
+                response = ec2.describe_snapshots(
+                    OwnerIds=['self']
+                )
+
+            total_snapshots += len(response['Snapshots'])
+
+            next_token = response.get('NextToken')
+            if not next_token:
+                break
+
+        return total_snapshots
diff --git a/lambda/monitor_service_quotas/ec2_checks.py b/lambda/monitor_service_quotas/ec2_checks.py
@@ -1,8 +1,6 @@
 from abc import ABC, abstractmethod
 import boto3
 from datetime import datetime
-import logging
-logger = logging.getLogger(__name__)
 
 class EC2ClientSingleton:
     _instances = {}
@@ -60,12 +58,6 @@ def get_usage(self, region, quota_name):
             max_interfaces = max(max_interfaces, interfaces)
         return max_interfaces
 
-class VerifiedAccessGroupsChecker(EC2UsageChecker):
-    def get_usage(self, region, quota_name):
-        ec2 = EC2ClientSingleton.get_client(region)
-        response = ec2.describe_verified_access_groups()
-        return len(response['VerifiedAccessGroups'])
-
 class RunningDedicatedHostsChecker(EC2UsageChecker):
     def get_usage(self, region, quota_name):
         ec2 = EC2ClientSingleton.get_client(region)
@@ -205,3 +197,25 @@ def get_usage(self, region, quota_name):
             max_routes = max(max_routes, route_count)
 
         return max_routes
+
+class EC2VPCElasticIPsChecker(EC2UsageChecker):
+    def get_usage(self, region, quota_name):
+        ec2 = EC2ClientSingleton.get_client(region)
+        eip_count = 0
+        next_token = None
+
+        while True:
+            if next_token:
+                response = ec2.describe_addresses(NextToken=next_token)
+            else:
+                response = ec2.describe_addresses()
+
+            for address in response['Addresses']:
+                if 'Domain' in address and address['Domain'] == 'vpc':
+                    eip_count += 1
+
+            next_token = response.get('NextToken')
+            if not next_token:
+                break
+
+        return eip_count
diff --git a/lambda/monitor_service_quotas/efs_checks.py b/lambda/monitor_service_quotas/efs_checks.py
@@ -0,0 +1,38 @@
+from abc import ABC, abstractmethod
+import boto3
+from datetime import datetime
+
+class EFSClientSingleton:
+    _instances = {}
+
+    @classmethod
+    def get_client(cls, region):
+        if region not in cls._instances:
+            cls._instances[region] = boto3.client('elasticfilesystem', region_name=region)
+        return cls._instances[region]
+
+class EFSUsageChecker(ABC):
+
+    @abstractmethod
+    def get_usage(self, region, quota_name):
+        pass
+
+class FileSystemsPerAccountChecker(EFSUsageChecker):
+    def get_usage(self, region, quota_name):
+        efs = EFSClientSingleton.get_client(region)
+        file_system_count = 0
+        next_token = None
+
+        while True:
+            if next_token:
+                response = efs.describe_file_systems(MaxItems=100, Marker=next_token)
+            else:
+                response = efs.describe_file_systems(MaxItems=100)
+
+            file_system_count += len(response['FileSystems'])
+
+            next_token = response.get('NextMarker')
+            if not next_token:
+                break
+
+        return file_system_count