There are 14 posts tagged Python (this is page 1 of 3).
In this video I show how setup Flask and Apache on an Ubuntu VM in Digital Ocean with a custom domain. This was made after someone in the comments on my other DigitalOcean video requested it. If there is something else anyone would like to see, please just let me know I am happy to provide these walk through’s.
Note: I hit a number of challenges with DNS in this one, I think it’s fun to watch me struggle. Enjoy!
AWS Lambda functions are a great way to run some code on a trigger/schedule without needing a whole server dedicated to it. They can be cost effective, but be careful depending on how long they run, and the number of executions per hour, they can be quite costly as well.
For my use case, I wanted to create snapshot backups of EBS volumes for a Mongo Database every day. I originally implemented this using only CloudWatch, which is a monitoring service, but because it’s focused on scheduling, AWS also uses it for other things that require scheduling/cron like features. Unfortunately, the CloudWatch implementation of snapshot backups was very limited. I could not ‘tag’ the backups, which was certainly something I needed for easy finding and cleanups later (past a retention period).
Anyway, there were a couple pitfalls I ran into when creating this function.
AWSLambdaVPCAccessExecutionRole, with permissions for the necessary EC2 actions (ec2:CreateNetworkInterface, ec2:DescribeNetworkInterfaces, and ec2:DeleteNetworkInterface) that you can use when creating a role”
I think that was it, after that everything worked fine. To finish this short article off, screenshots and the code!
And finally the code…
# Backup cis volumes
import boto3
def lambda_handler(event, context):
ec2 = boto3.client('ec2')
reg = 'us-east-1'
# Connect to region
ec2 = boto3.client('ec2', region_name=reg)
response = ec2.describe_instances(Filters=[{'Name': 'instance-state-name', 'Values': ['running']},
{'Name': 'tag-key', 'Values': ['Name']},
{'Name': 'tag-value', 'Values': ['cis-mongo*']},
])
for r in response['Reservations']:
for i in r['Instances']:
for mapping in i['BlockDeviceMappings']:
volId = mapping['Ebs']['VolumeId']
# Create snapshot
result = ec2.create_snapshot(VolumeId=volId,
Description='Created by Lambda backup function ebs-snapshots')
# Get snapshot resource
ec2resource = boto3.resource('ec2', region_name=reg)
snapshot = ec2resource.Snapshot(result['SnapshotId'])
# Add volume name to snapshot for easier identification
snapshot.create_tags(Tags=[{'Key': 'Name', 'Value': 'cis-mongo-snapshot-backup'}])
And here is an additional function to add for cleanup
import boto3
from datetime import timedelta, datetime
def lambda_handler(event, context):
# if older than days delete
days = 14
filters = [{'Name': 'tag:Name', 'Values': ['cis-mongo-snapshot-backup']}]
ec2 = boto3.setup_default_session(region_name='us-east-1')
client = boto3.client('ec2')
snapshots = client.describe_snapshots(Filters=filters)
for snapshot in snapshots["Snapshots"]:
start_time = snapshot["StartTime"]
delete_time = datetime.now(start_time.tzinfo) - timedelta(days=days)
if start_time < delete_time:
print 'Deleting {id}'.format(id=snapshot["SnapshotId"])
client.delete_snapshot(SnapshotId=snapshot["SnapshotId"], DryRun=False)
The end, happy server-lessing (ha !)
Recently, while having to work a lot more in Jira than normal I got annoyed with the Jira Web GUI. So I wrote a script to do simple management of our jira issues.
(env) ➜ jira git:(master) ✗ ./jira-ctl.py
usage: jira-ctl.py [-h] [-lp] [-li LIST_PROJECT_ISSUES] [-ua UPDATE_ASSIGNEE]
[-ud UPDATE_DATE] [-usts UPDATE_STATUS]
[-usum UPDATE_SUMMARY]
optional arguments:
-h, --help show this help message and exit
-lp, --list-projects List all projects
-li LIST_PROJECT_ISSUES, --list-issues LIST_PROJECT_ISSUES
List Issues for a specific project
-ua UPDATE_ASSIGNEE, --update-assignee UPDATE_ASSIGNEE
Update an issues assignee format: <issue
number>,<first_last>
-ud UPDATE_DATE, --update-date UPDATE_DATE
Update an issues due date format: <issue number
>,<yyyy-mm-dd>
-usts UPDATE_STATUS, --update-status UPDATE_STATUS
Update an issues status format: <issue
number>,'<Open|In Progress|Resolved>'
-usum UPDATE_SUMMARY, --update-summary UPDATE_SUMMARY
Update an issues summary format: <issue
number>,'<summary>'
(env) ➜ jira git:(master) ✗
from jira.client import JIRA
import os, sys
import prettytable
import argparse
parser = argparse.ArgumentParser()
parser.add_argument('-lp', '--list-projects', action="store_true", dest="list_projects", required=False, help="List all projects")
parser.add_argument('-li', '--list-issues', action="store", dest="list_project_issues", required=False, help="List Issues for a specific project")
parser.add_argument('-ua', '--update-assignee', action="store", dest="update_assignee", required=False, help="Update an issues assignee format: <issue number>,<first_last>")
parser.add_argument('-ud', '--update-date', action="store", dest="update_date", required=False, help="Update an issues due date format: <issue number>,<yyyy-mm-dd>")
parser.add_argument('-usts', '--update-status', action="store", dest="update_status", required=False, help="Update an issues status format: <issue number>,'<Open|In Progress|Resolved>'")
parser.add_argument('-usum', '--update-summary', action="store", dest="update_summary", required=False, help="Update an issues summary format: <issue number>,'<summary>'")
args = parser.parse_args()
def get_pass():
if os.environ.get('JIRA_PASS') == None:
print "you must first export your JIRA_PASS in the shell by running: source getpass.sh"
print 'or "export JIRA_PASS=<jira_password>'
sys.exit()
jira_password = str(os.environ['JIRA_PASS'])
return jira_password
def connect_jira(jira_server, jira_user, jira_password):
'''
Connect to JIRA. Return None on error
'''
try:
#print "Connecting to JIRA: %s" % jira_server
jira_options = {'server': jira_server}
jira = JIRA(options=jira_options,
# Note the tuple
basic_auth=(jira_user,
jira_password))
return jira
except Exception,e:
print "Failed to connect to JIRA: %s" % e
return None
def list_projects():
projects = jira.projects()
header = ["Projects"]
table = prettytable.PrettyTable(header)
for project in projects:
row = [str(project)]
table.add_row(row)
print table
def list_project_issues(project):
query = 'project = %s and status != Resolved order by due asc' % (project)
#query = 'project = %s and status != Resolved order by due desc' % (project)
#query = 'project = %s order by due desc' % (project)
issues = jira.search_issues(query, maxResults=100)
if issues:
header = ["Ticket", "Summary", "Assignee", "Status", "Due Date"]
table = prettytable.PrettyTable(header)
for issue in issues:
summary = issue.fields.summary
st = str(issue.fields.status)
dd = issue.fields.duedate
assignee = issue.fields.assignee
row = [issue, summary, assignee, st, dd]
table.add_row(row)
print table
def update_issue_assignee(issue_number, assignee):
issue = jira.issue(issue_number)
issue.update(assignee=assignee)
print "updated %s with new Assignee: %s" % (issue_number, assignee)
def update_issue_duedate(issue_number, dd):
issue = jira.issue(issue_number)
issue.update(duedate=dd)
print "updated %s with new Due Date: %s" % (issue_number, dd)
def update_issue_summary(issue_number, summary):
issue = jira.issue(issue_number)
issue.update(summary=summary)
print "updated %s with new Summary: %s" % (issue_number, summary)
def update_issue_status(issue_number, status):
# To figure out transitions view the following URL
# https://jira.yourcompany.com/rest/api/2/issue/XXXXXX-22/transitions?expand=transitions.fields
transitions = {
'Resolved': 11,
'Open': 41,
'In Progress': 21,
'Testing': 71,
'Transition': 81,
'Closed': 91,
}
issue = jira.issue(issue_number)
jira.transition_issue(issue, transitions[status])
print "updated %s with new Status: %s" % (issue_number, status)
if __name__ == "__main__":
if not args:
parser.print_help()
else:
jira_user = '<your_login>'
jira_password = get_pass()
jira_server = 'https://jira.yourcompany.com'
jira = connect_jira(jira_server, jira_user, jira_password)
if args.list_projects:
list_projects()
elif args.list_project_issues:
project = args.list_project_issues
list_project_issues(project)
elif args.update_assignee:
(issue_number, assignee) = args.update_assignee.split(',')
update_issue_assignee(issue_number, assignee)
elif args.update_date:
(issue_number, dd) = args.update_date.split(',')
update_issue_duedate(issue_number, dd)
elif args.update_status:
(issue_number, status) = args.update_status.split(',')
update_issue_status(issue_number, status)
elif args.update_summary:
(issue_number, summary) = args.update_summary.split(',')
update_issue_summary(issue_number, summary)
else:
parser.print_help()
Later, I found out that there is a Jira command line utility, I didn’t check and see what functionality it provided, but I enjoyed writing this anyway.
Happy coding !
Boto is a Software Development Kit for accessing the AWS API’s using Python.
Recently, I needed to determine how many of my EC2 instances were spawned in a public subnet, that also had security groups with wide open access on any port via any protocol to the instances. Because I have an IGW (Internet Gateway) in my VPC’s and properly setup routing tables, instances launched in the public subnets with wide open security groups (allowing ingress traffic from any source) is a really bad thing 🙂
Here is the code I wrote to identify these naughty instances. It will require slight modifications in your own environment, to match your public subnet IP Ranges, EC2 Tags, and Account names.
#!/usr/bin/env python
__author__ = 'Jason Riedel'
__description__ = 'Find EC2 instances provisioned in a public subnet that have security groups allowing ingress traffic from any source.'
__date__ = 'June 5th 2016'
__version__ = '1.0'
import boto3
def find_public_addresses(ec2):
public_instances = {}
instance_public_ips = {}
instance_private_ips = {}
instance_ident = {}
instances = ec2.instances.filter(Filters=[{'Name': 'instance-state-name', 'Values': ['running'] }])
# Ranges that you define as public subnets in AWS go here.
public_subnet_ranges = ['10.128.0', '192.168.0', '172.16.0']
for instance in instances:
# I only care if the private address falls into a public subnet range
# because if it doesnt Internet ingress cant reach it directly anyway even with a public IP
if any(cidr in instance.private_ip_address for cidr in public_subnet_ranges):
owner_tag = "None"
instance_name = "None"
if instance.tags:
for i in range(len(instance.tags)):
#comment OwnerEmail tag out if you do not tag your instances with it.
if instance.tags[i]['Key'] == "OwnerEmail":
owner_tag = instance.tags[i]['Value']
if instance.tags[i]['Key'] == "Name":
instance_name = instance.tags[i]['Value']
instance_ident[instance.id] = "Name: %s\n\tKeypair: %s\n\tOwner: %s" % (instance_name, instance.key_name, owner_tag)
if instance.public_ip_address is not None:
values=[]
for i in range(len(instance.security_groups)):
values.append(instance.security_groups[i]['GroupId'])
public_instances[instance.id] = values
instance_public_ips[instance.id] = instance.public_ip_address
instance_private_ips[instance.id] = instance.private_ip_address
return (public_instances, instance_public_ips,instance_private_ips, instance_ident)
def inspect_security_group(ec2, sg_id):
sg = ec2.SecurityGroup(sg_id)
open_cidrs = []
for i in range(len(sg.ip_permissions)):
to_port = ''
ip_proto = ''
if 'ToPort' in sg.ip_permissions[i]:
to_port = sg.ip_permissions[i]['ToPort']
if 'IpProtocol' in sg.ip_permissions[i]:
ip_proto = sg.ip_permissions[i]['IpProtocol']
if '-1' in ip_proto:
ip_proto = 'All'
for j in range(len(sg.ip_permissions[i]['IpRanges'])):
cidr_string = "%s %s %s" % (sg.ip_permissions[i]['IpRanges'][j]['CidrIp'], ip_proto, to_port)
if sg.ip_permissions[i]['IpRanges'][j]['CidrIp'] == '0.0.0.0/0':
#preventing an instance being flagged for only ICMP being open
if ip_proto != 'icmp':
open_cidrs.append(cidr_string)
return open_cidrs
if __name__ == "__main__":
#loading profiles from ~/.aws/config & credentials
profile_names = ['de', 'pe', 'pde', 'ppe']
#Translates profile name to a more friendly name i.e. Account Name
translator = {'de': 'Platform Dev', 'pe': 'Platform Prod', 'pde': 'Products Dev', 'ppe': 'Products Prod'}
for profile_name in profile_names:
session = boto3.Session(profile_name=profile_name)
ec2 = session.resource('ec2')
(public_instances, instance_public_ips, instance_private_ips, instance_ident) = find_public_addresses(ec2)
for instance in public_instances:
for sg_id in public_instances[instance]:
open_cidrs = inspect_security_group(ec2, sg_id)
if open_cidrs: #only print if there are open cidrs
print "=================================="
print " %s, %s" % (instance, translator[profile_name])
print "=================================="
print "\tprivate ip: %s\n\tpublic ip: %s\n\t%s" % (instance_private_ips[instance], instance_public_ips[instance], instance_ident[instance])
print "\t=========================="
print "\t open ingress rules"
print "\t=========================="
for cidr in open_cidrs:
print "\t\t" + cidr
To run this you also need to setup your .aws/config and .aws/credentials file.
http://docs.aws.amazon.com/cli/latest/userguide/cli-chap-getting-started.html#cli-config-files
Email me tuxninja [at] tuxlabs.com if you have any issues.
Boto is awesome 🙂 Note so is the AWS CLI, but requires some shell scripting as opposed to Python to do cool stuff.
The github for this code here https://github.com/jasonriedel/AWS/blob/master/sg-audit.py
Enjoy !