Added NAT GATEWAY and INTERNET GATEWAY checks #7

Author: leandrodamascena

shared/awscommands.py

@@ -1,6 +1,6 @@
 from shared.common import *
 from shared.internal.security import IAM, IAMPOLICY
-from shared.internal.network import VPC
+from shared.internal.network import VPC, IGW, NATGATEWAY
 from shared.internal.compute import LAMBDA, EC2
 from shared.internal.database import RDS, ELASTICACHE, DOCUMENTDB
 from shared.internal.storage import EFS, S3POLICY
@@ -27,3 +27,5 @@ def run(self):
         DOCUMENTDB(self.vpc_options).run()
         SQSPOLICY(self.vpc_options).run()
         MSK(self.vpc_options).run()
+        IGW(self.vpc_options).run()
+        NATGATEWAY(self.vpc_options).run()

shared/internal/network.py

@@ -21,3 +21,84 @@ def run(self):
         except Exception as e:
             message = "There is no VpcID \"{0}\" in region {1}.\nError {2}".format(self.vpc_options.vpc_id, self.vpc_options.region_name, str(e))
             exit_critical(message)
+
+class INTERNETGATEWAY(object):
+    
+    def __init__(self, vpc_options: VpcOptions):
+        self.vpc_options = vpc_options
+
+    def run(self):
+
+        try:
+            client = self.vpc_options.client('ec2')
+
+            filters = [{'Name': 'attachment.vpc-id',
+                        'Values': [self.vpc_options.vpc_id]}]
+
+            response = client.describe_internet_gateways(Filters=filters)
+
+            message_handler("\nChecking INTERNET GATEWAYS...", "HEADER")
+
+            """ One VPC has only 1 IGW then it's a direct check """
+            if len(response["InternetGateways"]) == 0:
+                    message_handler("Found 0 Internet Gateway in region {0}".format(self.vpc_options.region_name), "OKBLUE")
+            else:
+            
+                found = 1
+
+                message = "\nInternetGatewayId: {} -> VPC id {}".format(
+                            response['InternetGateways'][0]['InternetGatewayId'],
+                            self.vpc_options.vpc_id
+                            )
+                
+                message_handler("Found {0} Internet Gateway using VPC {1} {2}".format(str(found), \
+                                                                                    self.vpc_options.vpc_id, message), \
+                                                                                    'OKBLUE')
+
+        except Exception as e:
+            message = "Can't list Internet Gateway\nError {0}".format(str(e))
+            exit_critical(message)
+
+
+class NATGATEWAY(object):
+    
+    def __init__(self, vpc_options: VpcOptions):
+        self.vpc_options = vpc_options
+
+    def run(self):
+
+        try:
+            client = self.vpc_options.client('ec2')
+
+            filters = [{'Name': 'vpc-id',
+                        'Values': [self.vpc_options.vpc_id]}]
+
+            response = client.describe_nat_gateways(Filters=filters)
+
+            message_handler("\nChecking NAT GATEWAYS...", "HEADER")
+
+            if len(response["NatGateways"]) == 0:
+                    message_handler("Found 0 NAT Gateways in region {0}".format(self.vpc_options.region_name), "OKBLUE")
+            else:
+            
+                found = 0
+                message = ""
+
+                for data in response["NatGateways"]:
+
+                    if data['VpcId'] == self.vpc_options.vpc_id:
+
+                        found += 1
+                        message = message + "\nNatGatewayId: {} -> VPC id {}".format(
+                            data['NatGatewayId'],
+                            self.vpc_options.vpc_id
+                            )
+
+                message_handler("Found {0} NAT Gateways using VPC {1} {2}".format(str(found), self.vpc_options.vpc_id, message),'OKBLUE')
+
+        except Exception as e:
+            message = "Can't list NAT Gateways\nError {0}".format(str(e))
+            exit_critical(message)
+
+""" alias """
+IGW = INTERNETGATEWAY