Day 9. Python Use Cases For DevOps

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)