Day 9. Python Use Cases For DevOps

Day 9. Python Use Cases For DevOps

Table of contents

Python Use Cases

  1. Python Boto3 program to retrieve secrets from parameter store and secrets manager.

    AWS system parameter store helps store values, secure information such as passwords, database strings, Amazon Machine Image (AMI) code IDs while running.

     #Retrieve secrets from parameter store
     import json
     import boto3
     ssm = boto3.client('ssm', 'us-east-1')
     def get_parameters():
         response = ssm.get_parameters(
             Names=['My-Password'],WithDecryption=True
         )
         for parameter in response['Parameters']:
             return parameter['Value']
     value=get_parameters()
     print(value)
    

    AWS Secrets Manager helps you manage, retrieve, and rotate database credentials, API keys, and other secrets throughout their lifecycles.

     #Retrieve secrets from secret manager
     import json
     import boto3
     def get_secret_value(name):
        session = boto3.session.Session()
        s3_client = session.client('secretsmanager')
        secret_id=name
        response =   s3_client.get_secret_value(SecretId=secret_id)
        return response  
    
     get_secret_value("")
    
  2. Python script to fetch IPs of live servers in an autoscaling group.

     import json
     from logging import Logger
     from botocore.exceptions import ClientError
     import boto3
    
     class AutoScalingServers:
         def __init__(self):
             client = boto3.client('autoscaling')
             self.autoscaling_client=client
    
         def describe_group(self,group_name):
             try:
                 response = self.autoscaling_client.describe_auto_scaling_groups(
                             AutoScalingGroupNames=[group_name])
             except ClientError as e:
                 Logger.error("Couls not describe group")
             else:
                 groups=response.get("AutoScalingGroups", [])
                 return groups
    
     ec2= boto3.resource("ec2")
     ASG= AutoScalingServers()
     groups =ASG.describe_group("My_ASG")
     for asg in groups:
         print(asg["AutoScalingGroupName"])
         instances =[i for i in asg['Instances']]
         ec2_instances= ec2.instances.filter(Filters=[{}])
         for instance in ec2_instances:
             print(instance.id,instance.public_dns_name,instance.private_ip_address)
    
  3. Python AWS Lambda function to stop running instances on weekends.

     import boto3
     region="us-east-1"
     instances=["first instance id","second instance id"]
     ec2= boto3.client('ec2',region_name=region)
    
     ec2.stop_instances(Instances=instances)
     print('stopped instances',str(instances))
    
  4. Python script to take backups.

     import os
     import shutil
     import schedule
     import time
     import zipfile
    
     def doprocess(source_folder, target_zip):
         zipf = zipfile.ZipFile(target_zip, "w")
         for subdir, dirs, files in os.walk(source_folder):
             for file in files:
                 print(os.path.join(subdir, file))
                 zipf.write(os.path.join(subdir, file))
         print("Created ", target_zip)
    
     def backup(src, dst):    
         shutil.copy(src,destination)
    
     def schedule_backup():
         time = "07:26"  # schedule time in 24h format
         schedule.every().day.at(time).do(backup, destzip, destination)
         print(f"Backup scheduled for {time} every day")
         while True:
             schedule.run_pending()
             time.sleep(1)
    
     if __name__ == "__main__":
         source = "path/to/source/directory"
         destination = "path/to/destination/directory"
         schedule_backup()
    
  5. Find SSL expiry date using python.

     from cryptography import x509
     import socket
     import ssl
     import sys
     import json
    
     hostname = "google.com"
     port="443"
     context = ssl.create_default_context()
    
     with socket.create_connection((hostname, port)) as sock:
         with context.wrap_socket(sock, server_hostname=hostname) as ssock:
             print(ssock.version())
             data =ssock.getpeercert(True)      
             # print(ssock.getpeercert())
    
     pem_data = ssl.DER_cert_to_PEM_cert(data)
     cert_data = x509.load_pem_x509_certificate(str.encode(pem_data))
     print (cert_data.not_valid_after_utc)