Automating SQS Encryption for Enhanced Security with Python and boto3

Python for devops v1.1 — Writing Python script to find unencrypted SQS queues and encrypt them with a customer-managed KMS key.

Akhilesh Mishra
5 min readJun 17, 2024
Photo by Fernando Hernandez on Unsplash

This is the first blog post on my blog series Python for DevOps. This is the sub-series of my ongoing blog series DevOps Zero to Hero.

The purpose of this series is to teach building automation with Python to anyone who is starting their career as a DevOps engineers

Python for Devops part 2 — Automate Outdated AWS Lambda Runtime Updates.

What you will get out of it?

Modern-day DevOps roles require writing some code to automate boring tasks. Don’t even think about having a sustainable job let alone excelling at work if you are still shying away from writing some code. There are many options out there but IMO Python is the one you should start with.

I will help you learn Python for Devops by doing useful real-world Devops scenarios/projects.

Let’s start learning

Scenario:

A recent security scan identified hundreds of unencrypted SQS queues across various AWS accounts. To address this issue, I promptly developed a Python script. This script efficiently scans for unencrypted queues and encrypts them using the appropriate KMS key for each account.

What is the SQS?

Amazon Simple Queue Service (SQS) provides a reliable, highly scalable hosted queue for storing messages in transit between computers

In this blog post, I will teach you how to build simple automation that automatically encrypts all the SQS queues in your AWS account using Python and boto3. You can apply the same logic for other use cases as well.

Pre-requisite:

  • Create a CLI user with admin permissions, and generate access and secret keys.

Open the AWS console and head to IAM.

Create the user and attach the administrator permissions to it. Then go to user scroll down to the security credentials tab and create.

Note: I have deleted the key after taking a screenshot of it. You must protect the secret key with your life unless you want someone to misuse it and get you a huge bill from AWS.

  • Create some SQS queues without encryption.
  • Create one KMS key that we will use to encrypt the queues.

Here are the logical steps I will follow for this script.

  1. List SQS queues
  2. Check if it is using server-side encryption
  3. Get the list of SQS queues without encryption
  4. Apply encryption with the KMS key you created.

Instead of writing the code like a newbie, we will build separate functions for each logical task.

Note: I have placed all the code in my public GitHub repo, here is the link.

First thing first — Install boto3 package

pip install boto3
  • Import the boto3 module in your script and create an SQS client.

Creating a client in Boto3 establishes a connection to a specific AWS service, allowing you to interact with it using the provided methods. This client uses your AWS credentials to authenticate and authorize these operations, enabling tasks like sending messages in SQS or managing objects in S3.

Building functions

We will separate the logic into small functions.

  • list_queue_urls: This function lists all SQS queues in your AWS account.
  • get_kms_key: This function will check if the Queue is encrypted.
    It checks for the KMS key ID in queue attributes, if the KMS key exists it should be present in attributes.
  • queue_without_encryption: This function will loop through all the queues and return the list of unencrypted SQS queues.
  • encrypt_queue: This function will encrypt the SQS queue with the KMS key.
  • run: This function will execute the complete flow.

I have placed all the code in my public GitHub repo, here is the link.

Running the script

if __name__ == "__main__":
kms_key = "your-key-id" # or alias/alias_name
run(kms_key)

Note: The `if __name__ == “__main__”:` statement ensures that the enclosed code block runs only when the script is executed directly, not when it is imported as a module in another script.

Things we can do to improve the code.

  • Assume role as per AWS account and role
  • Use paginators while listing the queues.
  • Allow the script to take the KMS key as a command line argument.
  • Use logging and type hinting
  • Instead of using response[‘QueueUrls’], use response.get(‘QueueUrls’, [])
  • Build functions separately and call them in the main script.
  • Could use proper exception handling

You can find a better version of the script in my public GitHub repo — Python for devops.

If you found this blog post useful, clap, follow, and subscribe so you don’t miss my future articles.

Connect with me on Linkedin: https://www.linkedin.com/in/akhilesh-mishra-0ab886124/

--

--

Akhilesh Mishra

Self taught DevOps engineer with expertise in multi-cloud, and various DevOps tools. Open for mentorship - https://topmate.io/akhilesh_mishra