Skip to content

Instantly share code, notes, and snippets.

@jweyrich
Created June 15, 2022 13:55
Show Gist options
  • Save jweyrich/08e2777b941c5046a4c452cf6d1a77c0 to your computer and use it in GitHub Desktop.
Save jweyrich/08e2777b941c5046a4c452cf6d1a77c0 to your computer and use it in GitHub Desktop.
Script that visits all deployed Lambda functions in an AWS account & region and updates all environment variables matching a specified substring.
#!/usr/bin/env python3
#
# Author: Jardel Weyrich <jweyrich at gmail dot com>
#
# How to run:
# aws-vault exec <profile> -- python3 aws-lambda-update-variables.py
#
import logging
import os
from pprint import pformat
import boto3
class VariableMutation:
def __init__(self, find_substr: str, replace_with: str):
self._find_substr = find_substr
self._replace_with = replace_with
def matches(self, key_value: tuple[str, str]) -> bool:
return self._find_substr in key_value[1]
def mutate(self, current_value: str) -> str:
return current_value.replace(self._find_substr, self._replace_with)
def lambda_apply_changes(client, func_name: str, func_vars: dict[str]) -> bool:
response = client.update_function_configuration(
FunctionName=func_name,
Environment={
'Variables': func_vars
}
)
metadata = response.get('ResponseMetadata', {})
status_code = metadata.get('HTTPStatusCode', 0)
if status_code != 200:
logging.warning('failed to update function %s', func_name)
return False
logging.info('updated %s', func_name)
return True
def filter_variables(variables: dict[str], mutation: VariableMutation) -> dict[str]:
'''
Apply the mutation's `matches` function over `variables` and return a new dictionary
with the matching key-values.
'''
return dict(list(filter(mutation.matches, variables.items())))
def merge_vars(left_vars: dict[str], right_vars: dict[str]) -> dict[str]:
'''
Copy values from right_vars over left_vars.
'''
ret_vars = left_vars.copy()
for (key, value) in right_vars.items():
ret_vars[key] = value
return ret_vars
def mutate_vars(variables: dict[str], mutation: VariableMutation) -> dict[str]:
'''
Apply the provided mutation over `variables`.
'''
ret_vars = variables.copy()
for (key, value) in variables.items():
if mutation.matches((key, value)):
ret_vars[key] = mutation.mutate(value)
return ret_vars
def patch_function(client, func: any, mutation: VariableMutation):
func_name = func.get('FunctionName', '')
func_vars = func.get('Environment', {}).get('Variables', {})
matching_vars = filter_variables(func_vars, mutation)
# If there are no matching variables to patch, skip this function.
if len(matching_vars) == 0:
return
mutated_vars = mutate_vars(matching_vars, mutation)
merged_vars = merge_vars(func_vars, mutated_vars)
logging.info('%s\nMATCHES=%s\nMUTATED=%s\nMERGED=%s',
func_name,
pformat(matching_vars, indent=4),
pformat(mutated_vars, indent=4),
pformat(merged_vars, indent=4))
lambda_apply_changes(client, func_name, merged_vars)
def run(mutation: VariableMutation):
client = boto3.client('lambda')
paginator = client.get_paginator('list_functions')
response_iterator = paginator.paginate(PaginationConfig={'PageSize': 50})
for response in response_iterator:
functions = response.get('Functions', [])
for func in functions:
patch_function(client, func, mutation)
if __name__ == "__main__":
LOG_LEVEL = os.environ.get('LOGLEVEL', 'INFO')
LOG_FORMAT = '%(message)s'
# LOG_FORMAT = '%(asctime)s %(levelname)s %(message)s'
logging.basicConfig(level=LOG_LEVEL, format=LOG_FORMAT)
run(VariableMutation('SUBSTRING_I_WANT_TO_REPLACE', 'REPLACE_WITH_THIS'))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment