supporting code for IAM blog post

ast-modify.py

def modify_policy(filename, queue_name):
    with open(filename, "r") as fh:
        policy = json.load(policy)
    allowed_queue = [f"arn:aws:sqs:*:*:{queue_name}"]

    for idx, statement in enumerate(policy["Statement"])
        services = list(set([x.split(":")[0] for x in statement["Action"]]))
        if services == ['sqs']:
            policy[idx]['Action'] = allowed_queue
    return policy

ast-policy.py

import ast

def find_queue_name(tree):
    assigns = [
        node.value.s
        for node
        in ast.walk(tree)
        if isinstance(node, ast.Assign)
        and node.targets[0].id == "QUEUE_NAME"
    ]
    return assigns[0]

def parse_source():
     source = open("app.py, "r").read()
     tree = ast.parse(source, "app.py")
     queue_name = find_queue_name(tree)

boto-function.py

import json
from chalice import Chalice 
import boto3

app = Chalice(app_name="my-app")

@app.route('/add_to_queue')
def add_to_queue(event, context, *args, **kwargs):
    sqs = boto3.client("sqs")
    my_queue = sqs.create_queue(QueueName="my-queue")
    queue_url = my_queue['QueueUrl']
    
    msg = {"hello": "world"}
    sqs.send_message(QueueUrl=queue_url, MessageBody=json.encode(msg))
    return {"enqueued": msg}

chalice-generated-iam-policy.json

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "sqs:CreateQueue",
        "sqs:SendMessage"
      ],
      "Resource": [
        "*"
      ],
      "Sid": "792755c9ec914628bafd2158ecd9d5b1"
    }
  ]
}

modified-chalice-app.py

import json
from chalice import Chalice 
import boto3

app = Chalice(app_name="my-app")

QUEUE_NAME = "my-queue"

@app.route('/add_to_queue')
def add_to_queue():
    sqs = boto3.client("sqs")
    my_queue = sqs.create_queue(QueueName=QUEUE_NAME)
    queue_url = my_queue['QueueUrl']
    
    msg = {"hello": "world"}
    sqs.send_message(QueueUrl=queue_url, MessageBody=json.encode(msg))
    return {"enqueued": msg}