""" * Version: 0.1.0.0.0 * Purposes: * Diagraming all AWS VPCs in the Region. * System parameters: * None """ import sys, os from pingcap.pe.flash.util import ( get_name_by_datetime, get_boto_session, speak_3rd_title, speak_title, speak_result, speak_2nd_title, speak_2nd_result, speak_3rd_result, speak_4th_result, get_md5, ) from pingcap.pe.flash.util_aws import ( describe_az_ebs, describe_region_asg_instances, describe_region_vpcs, describe_subnet_running_ec2, describe_subnet_eni, describe_subnet_ngw, describe_vpc_by_id, describe_vpc_endpoint, describe_vpc_igw, describe_vpc_nacl, describe_vpc_rt, describe_vpc_sg, describe_vpc_subnet, describe_vpc_vpgw, describe_vpc_eoigw, extract_eni_name_from_eni_list, extract_sg_name_from_sgs_list, generate_sg_inference_diagram, get_12_digits_account_number, get_all_iam_groups, get_all_iam_roles, get_all_iam_users, get_attached_group_policies, get_attached_role_policies, get_attached_user_policies, get_aws_iam_policies, get_groups_by_user, get_inline_group_policies, get_inline_role_policies, get_inline_user_policies, get_local_iam_policies, map_az_name_to_subnet_list, get_nametag_value, get_non_az_level_resource_usages, map_rt_id_to_route_list, get_short_name, get_vpc_default_rt, map_ebs_id_to_asg_name, map_sg_id_to_permissions, merge_sec_line, ) from diagrams import Diagram, Cluster, Edge from diagrams.aws.security import IAMRole, IAMPermissions from diagrams.aws.integration import ( SQS, SimpleQueueServiceSqs, SimpleQueueServiceSqsQueue, ) from diagrams.aws.compute import AMI, Lambda, EC2, EC2AutoScaling from diagrams.aws.storage import ( EBS, ElasticBlockStoreEBSVolume, ElasticBlockStoreEBSSnapshot, ElasticFileSystemEFS, SimpleStorageServiceS3, ElasticFileSystemEFSFileSystem, ) from diagrams.aws.analytics import EMRCluster from diagrams.aws.network import ( VPC, PublicSubnet, PrivateSubnet, RouteTable, InternetGateway, VpnGateway, ElbApplicationLoadBalancer, ElbClassicLoadBalancer, Nacl, Endpoint, VPCElasticNetworkInterface, NATGateway, ) from diagrams.onprem.client import User, Users def aws_show_vpcs(vpc_name: str, profile=None): """ 1. Print VPC information to desired format. 2. Draw a simple png to describe above structure. 3. Print regional security analytic report to CSV. resource_type(A), resource_name(B), related_resource(C), best_practice(D), current_practice(E), comment(F) """ try: os.mkdir("./stage/") except FileExistsError: None profile_prefix = "default-profile" if profile != None: profile_prefix = profile # Security content # 5 fields in sec_line: # key: "resource_type,resource_tag,resource_name" # value: {related_resource, best_practice, current_practice, advice} sec_line_header = "resource_type,resource_tag,resource_name,related_resource,best_practice,current_practice,comment" sec_line_building_dict = {} session = get_boto_session(profile) vpc_count = 0 print(speak_result("Gathering information...")) vpcs = describe_region_vpcs(profile) services_pack = get_non_az_level_resource_usages(profile) # Prefetch regional resource information # EFS efss = services_pack["efs"] efss_regional = [efs for efs in efss if not "AvailabilityZoneId" in efs] efss_onezone = [efs for efs in efss if "AvailabilityZoneId" in efs] # ASG asgs_regional = services_pack["autoscaling"] asg_instances_regional = describe_region_asg_instances(profile) # Snapshots snapshots_regional = services_pack["snapshot"] # AMIs amis_regional = services_pack["ami"] # NICs enis_regional = [] # Will be incrementally built as the progress goging # Setup diagram elements for drawing and lining # Bag: { ObjectId: diagrams_object } service_bag = {} efs_bag = {} # Refactored subnet_bag = {} # Refactored gateway_bag = {} ebs_bag = {} asg_bag = {} snapshot_bag = {} eni_bag = {} ec2_bag = {} ami_bag = {} # Compact the subnet_to_gateway lines subnet_to_gateway_hash_list = [] subnet_to_endpoint_hash_list = [] endpoint_to_service_hash_list = [] for vpc_query in vpcs: vpc_count += 1 vpc_id = vpc_query["VpcId"] vpc = describe_vpc_by_id(vpc_id, profile) name = get_nametag_value(vpc, None, vpc_id) if vpc_name not in [vpc_id, name]: continue if name: title = name + " | " + vpc["VpcId"] else: title = vpc["VpcId"] vpc_title = ( "VPC: " + title + ", " + vpc["InstanceTenancy"] + " tenancy, " + vpc["CidrBlock"] + ", " + "IsDefault: " + str(vpc["IsDefault"]) ) print(speak_title(vpc_title)) diagram_name = "./stage/" + get_name_by_datetime( profile_prefix + "-" + vpc["VpcId"] ) with Diagram( "Resource Map: " + vpc_title + ", @" + session.region_name, filename=diagram_name, show=True, direction="TB", graph_attr={"fontsize": "40"}, edge_attr={"fontsize": "3"}, node_attr={"fontsize": "11"}, curvestyle="curved", ): # Regional and external resources # Place services in use print(speak_2nd_title("General Resources")) print( speak_2nd_result( ",".join([k for k in services_pack if len(services_pack[k]) > 0]) ) ) if "s3" in [*services_pack] and len(services_pack["s3"]) > 0: service_bag["s3"] = SimpleStorageServiceS3("S3") if "sqs" in [*services_pack] and len(services_pack["sqs"]) > 0: service_bag["sqs"] = SimpleQueueServiceSqs("SQS") # Place used regional services # EFS with Cluster("EFS - Regional"): for efs in efss_regional: name = get_nametag_value(efs, "Name", efs["FileSystemId"]) efs_bag[efs["FileSystemId"]] = ElasticFileSystemEFSFileSystem(name) # EBS snapshots for snapshot in snapshots_regional: id = snapshot["SnapshotId"] snapshot_bag[id] = ElasticBlockStoreEBSSnapshot( get_nametag_value(snapshot, None, id) ) # AMIs for ami in amis_regional: id = ami["ImageId"] ami_bag[id] = AMI(get_short_name(get_nametag_value(ami, "Name", id))) for s in ami["BlockDeviceMappings"]: if "Ebs" in s: ( ami_bag[id] << Edge(color="darkblue") << snapshot_bag[s["Ebs"]["SnapshotId"]] ) # VPC resources print(speak_2nd_title("VPC Scoped Toolkits")) with Cluster(vpc_title): ### Prefetch all subnets subnets = describe_vpc_subnet(vpc_id, profile) az_to_subnets = map_az_name_to_subnet_list(subnets) ### Prefetch all routetables rt_default = get_vpc_default_rt(vpc_id, profile) rts = describe_vpc_rt(vpc_id, profile) rt_to_routes = map_rt_id_to_route_list(rts) # print(rt_to_routes) with Cluster("VPC Scoped Toolkits"): ############ VPC scoped # IGW print(speak_3rd_title("Gateways")) igws = describe_vpc_igw(vpc_id, profile) igw_count = 0 for igw in igws: igw_count += 1 name = get_nametag_value(igw) if name: igw_title = name + " | " + igw["InternetGatewayId"] else: igw_title = igw["InternetGatewayId"] igw_result = "IGW" + str(igw_count) + ": " + igw_title print(speak_2nd_result(igw_result)) igw_display_name = get_nametag_value( igw, None, igw["InternetGatewayId"] ) gateway_bag[igw["InternetGatewayId"]] = InternetGateway( igw_display_name ) # Egress only IGW eoigws = describe_vpc_eoigw(vpc_id, profile) eoigw_count = 0 for eoigw in eoigws: eoigw_count += 1 name = get_nametag_value(eoigw) if name: eoigw_title = ( name + " | " + eoigw["EgressOnlyInternetGatewayId"] ) else: eoigw_title = eoigw["EgressOnlyInternetGatewayId"] eoigw_result = "EOIGW" + str(eoigw_count) + ": " + eoigw_title print(speak_2nd_result(eoigw_result)) # Virtual private GW vpgws = describe_vpc_vpgw(vpc_id, profile) vpgw_count = 0 for vpgw in vpgws: vpgw_count += 1 name = get_nametag_value(vpgw) if name: vpgw_title = name + " | " + vpgw["VpnGatewayId"] else: vpgw_title = vpgw["VpnGatewayId"] vpgw_result = ( "VPGW" + str(vpgw_count) + ": " + vpgw_title + ", " + vpgw["State"] + ", ASN: " + str(vpgw["AmazonSideAsn"]) ) if "AvailabilityZone" in vpgw: vpgw_result += ", @" + vpgw["AvailabilityZone"] print(speak_2nd_result(vpgw_result)) gateway_bag[vpgw["VpnGatewayId"]] = VpnGateway( get_nametag_value(vpgw, None, vpgw["VpnGatewayId"]) ) # Route table print(speak_3rd_title("Route Tables")) rt_count = 0 for rt in rts: rt_count += 1 name = get_nametag_value(rt) if name: rt_title = name + " | " + rt["RouteTableId"] else: rt_title = rt["RouteTableId"] rt_result = ( "RT" + str(rt_count) + ": " + rt_title + ", Subnets-In-Use: " + str(len(rt["Associations"])) + ", PropagtingVPGWs: " + str(len(rt["PropagatingVgws"])) + ", Routes: " + str(len(rt["Routes"])) ) if rt["RouteTableId"] == rt_default["RouteTableId"]: rt_result += ", main" print(speak_2nd_result(rt_result)) # Route table rules for route in rt_to_routes[rt["RouteTableId"]]: print(speak_3rd_result(route)) # Network ACL print(speak_3rd_title("Networking ACLs")) nacls = describe_vpc_nacl(vpc_id, profile) nacl_count = 0 for nacl in nacls: nacl_count += 1 name = get_nametag_value(nacl) if name: nacl_title = name + " | " + nacl["NetworkAclId"] else: nacl_title = nacl["NetworkAclId"] nacl_result = ( "NACL" + str(nacl_count) + ": " + nacl_title + ", Subnets-In-Use: " + str(len(nacl["Associations"])) + ", Entries: " + str(len(nacl["Entries"])) + ", IsDefault: " + str(nacl["IsDefault"]) ) print(speak_2nd_result(nacl_result)) # Rules egress_entries = [ entry for entry in nacl["Entries"] if entry["Egress"] ] ingress_entries = [ entry for entry in nacl["Entries"] if not entry["Egress"] ] egress_sorted_index = sorted( [entry["RuleNumber"] for entry in egress_entries] ) ingress_sorted_index = sorted( [entry["RuleNumber"] for entry in ingress_entries] ) entry_count = 0 for idx in egress_sorted_index: for entry in egress_entries: if entry["RuleNumber"] == idx: entry_count += 1 target = "" keys = [*entry] for k in keys: if k.endswith("CidrBlock"): target += "To: " + entry[k] + ", " action = entry["RuleAction"].title() protocol = "Protocol: " + entry["Protocol"] if "PortRange" in entry: detail = ( "PortRange: " + str(entry["PortRange"]["From"]) + "-" + str(entry["PortRange"]["To"]) ) elif "IcmpTypeCode" in entry: detail = ( "IcmpTypeCode: " + str(entry["IcmpTypeCode"]["Type"]) + "-" + str(entry["IcmpTypeCode"]["Code"]) ) elif protocol == "Protocol: -1": detail = "All" entry_result = ( "#" + str(idx) + ": " + action + ", " + target + protocol + ", " + detail ) print(speak_3rd_result(entry_result)) for idx in ingress_sorted_index: for entry in ingress_entries: if entry["RuleNumber"] == idx: entry_count += 1 target = "" keys = [*entry] for k in keys: if k.endswith("CidrBlock"): target += "From: " + entry[k] + ", " action = entry["RuleAction"].title() protocol = "Protocol: " + entry["Protocol"] if "PortRange" in entry: detail = ( "PortRange: " + str(entry["PortRange"]["From"]) + "-" + str(entry["PortRange"]["To"]) ) elif "IcmpTypeCode" in entry: detail = ( "IcmpTypeCode: " + str(entry["IcmpTypeCode"]["Type"]) + "-" + str(entry["IcmpTypeCode"]["Code"]) ) elif protocol == "Protocol: -1": detail = "All" entry_result = ( "#" + str(idx) + ": " + action + ", " + target + protocol + ", " + detail ) print(speak_3rd_result(entry_result)) # Security group print(speak_3rd_title("Security Groups")) sgs = describe_vpc_sg(vpc_id, profile) sg_to_permissions = map_sg_id_to_permissions(sgs) sg_count = 0 for sg in sgs: sg_count += 1 name = get_nametag_value(sg) if name: sg_title = name + " | " + sg["GroupId"] else: sg_title = sg["GroupName"] + " | " + sg["GroupId"] sg_result = ( "SG" + str(sg_count) + ": " + sg_title + ', [" ' + sg["Description"] + ' "]' ) print(speak_2nd_result(sg_result)) for p in sg_to_permissions[sg["GroupId"]]["in"]: print(speak_3rd_result(p)) for p in sg_to_permissions[sg["GroupId"]]["out"]: print(speak_3rd_result(p)) # SG Permissions # No icon for AWS securigy group for now. # if sg_count > 0: # sg_d = ApplicationSecurityGroups('Security Group x '+str(sg_count)) # VPC endpoint: GW, GW loadbalancer and interface print(speak_3rd_title("VPC Endpoints")) vpces = describe_vpc_endpoint(vpc_id, profile) vpce_count = 0 for vpce in vpces: vpce_count += 1 name = get_nametag_value(vpce) if name: vpce_title = name + ", " + vpce["VpcEndpointId"] else: vpce_title = vpce["VpcEndpointId"] vpce_result = ( "VPCE" + str(vpce_count) + "-" + vpce["VpcEndpointType"] + ": " + vpce_title + ", " + vpce["ServiceName"] + ", " + vpce["State"] ) print(speak_2nd_result(vpce_result)) if vpce["VpcEndpointType"] == "Gateway": gateway_bag[vpce["VpcEndpointId"]] = Endpoint( get_nametag_value(vpce, None, vpce["VpcEndpointId"]) ) for k in [*service_bag]: if k == vpce["ServiceName"].split(".")[-1]: hash = get_md5(vpce["VpcEndpointId"] + k) if not hash in endpoint_to_service_hash_list: ( gateway_bag[vpce["VpcEndpointId"]] >> Edge(color="darkgreen") << service_bag[k] ) endpoint_to_service_hash_list += [hash] associated_rts = "Used in RT: " for rt_id in vpce["RouteTableIds"]: for rt in rts: if rt_id == rt["RouteTableId"]: associated_rts += ( get_nametag_value(rt, None, rt_id) + ", " ) print(speak_3rd_result(associated_rts)) elif vpce["VpcEndpointType"] == "Interface": print(vpce) gateway_bag[vpce["VpcEndpointId"]] = Endpoint( get_nametag_value(vpce, None, vpce["VpcEndpointId"]) ) for k in [*service_bag]: if k == vpce["ServiceName"].split(".")[-1]: hash = get_md5(vpce["VpcEndpointId"] + k) if not hash in endpoint_to_service_hash_list: ( gateway_bag[vpce["VpcEndpointId"]] >> Edge(color="darkgreen") << service_bag[k] ) endpoint_to_service_hash_list += [hash] associated_interface = ( "Via interface: " + ",".join(vpce["NetworkInterfaceIds"]) + "@" + ",".join(vpce["SubnetIds"]) ) print(speak_3rd_result(associated_interface)) for dns in vpce["DnsEntries"]: address = dns["DnsName"] print(speak_3rd_result("Address: " + address)) associated_sgs = "protected by: " + ",".join( [g["GroupName"] for g in vpce["Groups"]] ) print(speak_3rd_result(associated_sgs)) # VPC endpoint service # VPC peering connection: requster and accepter # ELB V2: application and network # ELB classic # ECS # ElasticSearch # Lambda # ElastiCache # RDS # Redshift ## Subnet or AZ resources eni_bag = {} ec2_bag = {} ############ Subnet scoped # with Cluster('Availability Zones In-Use', direction='LR'): subnet_to_routetable = {} subnet_to_instances = {} instance_to_nics = {} nic_to_sgs = {} subnet_count = 0 for az in az_to_subnets: with Cluster("AZ " + az): # EBS all_ebs_in_az = describe_az_ebs(az, profile) # print("all_ebs_in_az", all_ebs_in_az) ebsid_to_asg = map_ebs_id_to_asg_name( all_ebs_in_az, asg_instances_regional ) for ebs in all_ebs_in_az: # if not ebs["VolumeId"] in [*ebsid_to_asg]: if ebs["Encrypted"]: enc_status = "Encrypted" else: enc_status = "Not Encrypted" merge_sec_line( { "EBS," + get_nametag_value(ebs, None, "N/A") + "," + ebs["VolumeId"]: "N/A," + "Encrypt" + "," + str(ebs["Encrypted"]) + "," + "N/A," }, sec_line_building_dict, ) ebs_bag[ebs["VolumeId"]] = EBS( ebs["VolumeId"] + "\n" + ebs["VolumeType"] + "-" + str(ebs["Size"]) + "\n" + enc_status ) for s in snapshots_regional: if ( s["State"] == "completed" and s["VolumeId"] == ebs["VolumeId"] ): ( ebs_bag[ebs["VolumeId"]] >> Edge(color="darkblue") >> snapshot_bag[s["SnapshotId"]] ) # EFS one-zone if len(efss_onezone) > 0: for efs in efss_onezone: if efs["AvailabilityZoneName"] == az: name = get_nametag_value( efs, "Name", efs["FileSystemId"] ) efs_bag[ efs["FileSystemId"] ] = ElasticFileSystemEFSFileSystem(name) # Subnet for subnet in az_to_subnets[az]: ec2_count = 0 subnet_count += 1 name = get_nametag_value(subnet) if name: subnet_title = name else: subnet_title = subnet["SubnetId"] rt_id = None subnet_rt = None for rt in rts: for association in rt["Associations"]: if "SubnetId" in association: if ( association["SubnetId"] == subnet["SubnetId"] ): rt_id = rt["RouteTableId"] rt_name = get_nametag_value(rt) subnet_rt = rt break if not rt_id: rt_title = "main" subnet_rt = rt_default elif not rt_name: rt_title = rt_id else: rt_title = rt_name + " | " + rt_id subnet_to_routetable[subnet["SubnetId"]] = rt_id subnet_result = ( "Subnet" + str(subnet_count) + ": " + subnet_title + ", " + subnet["CidrBlock"] + ", @" + subnet["AvailabilityZone"] + ", AvailableIp: " + str(subnet["AvailableIpAddressCount"]) + ", Ipv6:" + str(subnet["AssignIpv6AddressOnCreation"]) + ", RouteTable: " + rt_title ) print(speak_2nd_title(subnet_result)) with Cluster("Subnet " + subnet_title): public_subnet = False for route in subnet_rt["Routes"]: if "GatewayId" in route: if route["GatewayId"] in [*gateway_bag]: if route["GatewayId"].startswith("igw"): public_subnet = True break if public_subnet: subnet_bag[subnet["SubnetId"]] = PublicSubnet( subnet["CidrBlock"] ) else: subnet_bag[subnet["SubnetId"]] = PrivateSubnet( subnet["CidrBlock"] ) for route in subnet_rt["Routes"]: if "GatewayId" in route: if route["GatewayId"] in [*gateway_bag] and ( not route["GatewayId"].startswith("igw") ): hash = get_md5( subnet["SubnetId"] + route["GatewayId"] ) if hash not in subnet_to_gateway_hash_list: ( subnet_bag[subnet["SubnetId"]] >> Edge(color="orange") << gateway_bag[route["GatewayId"]] ) subnet_to_gateway_hash_list += [hash] elif "NatGatewayId" in route and route[ "NatGatewayId" ] in [*gateway_bag]: hash = get_md5( subnet["SubnetId"] + route["NatGatewayId"] ) if hash not in subnet_to_gateway_hash_list: ( subnet_bag[subnet["SubnetId"]] >> Edge(color="orange") >> gateway_bag[route["NatGatewayId"]] ) subnet_to_gateway_hash_list += [hash] for endpoint in vpces: if endpoint["VpcEndpointType"] == "Interface": for endpoint_subnet_id in endpoint["SubnetIds"]: if subnet["SubnetId"] == endpoint_subnet_id: hash = get_md5( endpoint_subnet_id + endpoint["VpcEndpointId"] ) if ( not hash in subnet_to_endpoint_hash_list ): ( subnet_bag[endpoint_subnet_id] >> Edge(color="darkgreen") << gateway_bag[ endpoint["VpcEndpointId"] ] ) subnet_to_endpoint_hash_list += [ hash ] # NAT GW ngws = describe_subnet_ngw(subnet["SubnetId"], profile) ngw_count = 0 for ngw in ngws: ngw_count += 1 name = get_nametag_value(ngw) if name: ngw_title = name + " | " + ngw["NatGatewayId"] else: ngw_title = ngw["NatGatewayId"] gateway_bag[ngw["NatGatewayId"]] = NATGateway( get_nametag_value( ngw, None, ngw["NatGatewayId"] ) ) ngw_result = ( "NATGW" + str(ngw_count) + ": " + ngw_title + ", " + ngw["State"] ) print(speak_2nd_result(ngw_result)) for address in ngw["NatGatewayAddresses"]: address_result = ( "Nic:" + address["NetworkInterfaceId"] + ", " + address["PrivateIp"] + ", " + address["PublicIp"] ) print(speak_3rd_result(address_result)) # ENI in-use enis = describe_subnet_eni(subnet["SubnetId"], profile) enis_regional += enis eni_managed_count = 0 for eni in enis: name = get_nametag_value(eni) if not name: name = eni["NetworkInterfaceId"] # eni_bag[name] = VPCElasticNetworkInterface(name) if ( eni["Attachment"]["InstanceOwnerId"] == "amazon-aws" ): eni_managed_count += 1 eni_result = ( "ENI" + str(eni_managed_count) + ": " + name + ', ["' + eni["Description"] + '"]' ) print(speak_2nd_result(eni_result)) desc = eni["Description"] if "EFS mount target for" in desc: fsid = desc.split(" ")[4] # efs_bag[fsid] >> Edge(color='darkgreen') >> eni_bag[name] ( efs_bag[fsid] >> Edge(color="darkgreen") >> subnet_bag[subnet["SubnetId"]] ) elif "Interface for NAT Gateway" in desc: natid = desc.split(" ")[4] # gateway_bag[natid] >> Edge(color='darkgreen') >> eni_bag[name] hash = get_md5(subnet["SubnetId"] + natid) if hash not in subnet_to_gateway_hash_list: ( gateway_bag[natid] >> Edge(color="darkgreen") >> subnet_bag[subnet["SubnetId"]] ) subnet_to_gateway_hash_list += [hash] # ASG asg_bag = {} for asg in asgs_regional: asg_subnets_string = asg["VPCZoneIdentifier"] if subnet["SubnetId"] in asg_subnets_string: instance_types = set() for asg_instance in asg["Instances"]: instance_types.add( "(" + asg_instance["InstanceType"] + ")" ) asg_display_name = ( get_short_name(asg["AutoScalingGroupName"]) + "\n" + "Min: " + str(asg["MinSize"]) + ", Max: " + str(asg["MaxSize"]) + ", Desired: " + str(asg["DesiredCapacity"]) + "\n" + ", ".join(instance_types) ) asg_bag[ asg["AutoScalingGroupName"] ] = EC2AutoScaling(asg_display_name) if asg["ImageId"] not in ami_bag: # Add to ami_bag ami_bag[asg["ImageId"]] = AMI( get_short_name(asg["ImageId"]) ) # with Cluster("ALERT: Non Self AMI"): # ( # AMI(asg["ImageId"]) # >> Edge(color="darkred") # >> asg_bag[ # asg["AutoScalingGroupName"] # ] # ) # else: ( ami_bag[asg["ImageId"]] >> Edge(color="darkblue") >> asg_bag[asg["AutoScalingGroupName"]] ) # EBS to ASG for k in [*ebsid_to_asg]: if ebsid_to_asg[k] in [*asg_bag]: ( ebs_bag[k] >> Edge(color="grey") >> asg_bag[ebsid_to_asg[k]] ) # EC2 instances = describe_subnet_running_ec2( subnet["SubnetId"], profile ) asg_subnet_instance_counter = { asg["AutoScalingGroupName"]: 0 for asg in asgs_regional } if not instances: None else: subnet_to_instances[ subnet["SubnetId"] ] = instances # cache subnet to instances information for i in instances: # asg_instance_ids = [] # for asg in asgs: # asg_instance_ids += [a['InstanceId'] for a in asg['Instances']] # ASG instances for asg_i in asg_instances_regional: if i["InstanceId"] == asg_i["InstanceId"]: asg_subnet_instance_counter[ asg_i["AutoScalingGroupName"] ] += 1 # None ASG instances if i["InstanceId"] not in [ asg_i["InstanceId"] for asg_i in asg_instances_regional ]: ec2_count += 1 name = get_nametag_value(i) if name: ec2_title = ( name + " | " + i["InstanceId"] ) k = name else: ec2_title = i["InstanceId"] k = ec2_title ec2_bag[k] = EC2( k + "\n(" + i["InstanceType"] + ")" ) if i["ImageId"] not in ami_bag: with Cluster( "ALERT: Amazon managed AMI" ): ( AMI(i["ImageId"]) >> Edge(color="darkred") >> ec2_bag[k] ) else: ( ami_bag[i["ImageId"]] >> Edge(color="darkblue") >> ec2_bag[k] ) detail_monitoring = i["Monitoring"]["State"] ec2_result = ( "EC2-" + str(ec2_count) + ": " + ec2_title + ", " + i["InstanceType"] + ", " + i["State"]["Name"] + ", EbsO: " + str(i["EbsOptimized"]) + ", Ena: " + str(i["EnaSupport"]) + ", Key: " + i["KeyName"] + ", DetailMonitor: " + detail_monitoring ) print(speak_2nd_result(ec2_result)) ebss = [] for dm in i["BlockDeviceMappings"]: if "Ebs" in dm: ebss += [dm["Ebs"]["VolumeId"]] for vol in all_ebs_in_az: if ( vol["VolumeId"] == dm["Ebs"]["VolumeId"] ): ebs_result = ( "EBS: " + dm["DeviceName"] + ", " + vol["VolumeType"] + ", IOPS: " + str(vol["Iops"]) + ", Enc: " + str(vol["Encrypted"]) + ", MultiAttach: " + str( vol[ "MultiAttachEnabled" ] ) + ", From: " + vol["VolumeId"] ) merge_sec_line( { "EBS" + "," + get_nametag_value( vol, None, "N/A" ) + "," + vol[ "VolumeId" ]: i["InstanceId"] + "(" + get_nametag_value( i, None, "N/A" ) + ")" + "," + "Encrypt" + "," + str( vol["Encrypted"] ) + ",N/A," }, sec_line_building_dict, ) # sec_line += 'EBS'+','+get_nametag_value(vol, None, 'N/A')+','+vol['VolumeId']+','+i['InstanceId']+'('+get_nametag_value(i,None,'N/A')+')'+','+'Encrypt'+','+str(vol['Encrypted'])+','+',\n' print( speak_3rd_result( ebs_result ) ) ( ebs_bag[vol["VolumeId"]] >> Edge( color="darkblue" ) >> ec2_bag[k] ) break nics = [] for nic in i["NetworkInterfaces"]: nic_result = ( "Nic: " + nic["NetworkInterfaceId"] + ", " + nic["Attachment"]["Status"] ) nics += [nic["NetworkInterfaceId"]] instance_to_nics[ i["InstanceId"] ] = nics # cache instance_to_nics print(speak_3rd_result(nic_result)) k = extract_eni_name_from_eni_list( nic["NetworkInterfaceId"], enis_regional, ) # if name: # eni_bag[k] >> Edge(color='purple') >> ec2_bag[name] # else: # eni_bag[k] >> Edge(color='purple') >> ec2_bag[ec2_title] nic_addresses = [] for private_address in nic[ "PrivateIpAddresses" ]: nic_addresses += [ private_address[ "PrivateIpAddress" ] ] if "Association" in private_address: nic_addresses += [ private_address[ "Association" ]["PublicIp"] ] address_result = "Addresses: " + str( ", ".join(nic_addresses) ) print(speak_4th_result(address_result)) nsgs = [] for nsg in nic["Groups"]: nsgs += [nsg["GroupId"]] nsg_result = "Protected by SG: " + str( ", ".join( [ extract_sg_name_from_sgs_list( gid, sgs ) for gid in nsgs ] ) ) nic_to_sgs[ nic["NetworkInterfaceId"] ] = nsgs print(speak_3rd_result(nsg_result)) # ASG Summary asg_in_subnet_count = 0 for k in [*asg_subnet_instance_counter]: if asg_subnet_instance_counter[k] > 0: asg_in_subnet_count += 1 asg_instance_result = ( "ASG" + str(asg_in_subnet_count) + ": " + k + ": InstanceCount: " + str(asg_subnet_instance_counter[k]) ) print(speak_2nd_result(asg_instance_result)) ( EC2( "( X " + str( asg_subnet_instance_counter[k] ) + " )" ) << Edge(style="dotted") << asg_bag[k] ) # Points to it's ASG # EMR # Security groups inference # Removed trailing parameters: instance_to_nics, nic_to_sgs # generate_sg_inference_diagram( # vpc_id, # profile, # ) print(speak_title("End")) sec_report_file_name = get_name_by_datetime( "data-sec-" + profile_prefix + "@" + session.region_name ) print("Dumping region security report...") with open("./stage/" + sec_report_file_name + ".csv", "w") as f: f.write(sec_line_header + "\n") for k in [*sec_line_building_dict]: f.write(",".join([k, sec_line_building_dict[k]]) + "\n") print("Done") if __name__ == "__main__": aws_profile_name = None if len(sys.argv) > 1: aws_profile_name = sys.argv[1] aws_show_vpcs(aws_profile_name)