Lambda Catalog

list-functions.py

# export AWS_PROFILE=profile_name
# python3 list-functions.py <region> 

import sys
import boto3
import csv
import json
from datetime import datetime
from datetime import timedelta

try:
    region = sys.argv[1]
except:
    region = "sa-east-1"

lambda_client     = boto3.client('lambda', region_name=region)
iam_client        = boto3.client('iam', region_name=region)
cloudwatch_client = boto3.client('cloudwatch', region_name=region)

paginator     = lambda_client.get_paginator('list_functions')
functions     = paginator.paginate().build_full_result()

csv_fields = ['Name', 'Runtime', 'Last Modified', 'State', 'Memory Size', 'Timeout', 'Services', 'Triggers', 'Last Run (18 months)', 'Last Run Invocations']
csv_file   = "lambda_catalog-"+region+"-"+datetime.now().strftime('%Y-%m-%d_%H-%M-%S')+".csv"

with open(csv_file, 'w') as csvfile:
    
    csvwriter = csv.writer(csvfile)
    csvwriter.writerow(csv_fields)

    for function in functions['Functions']:

        print(function['FunctionName'])

        name          = function['FunctionName']
        runtime       = function['Runtime']
        last_modified = function['LastModified'].split("T")[0]
        state         = lambda_client.get_function_configuration(FunctionName=name)['State']
        memory_size   = function['MemorySize']
        timeout       = function['Timeout']

        services      = [ ]
        triggers      = [ ]

        role          = function['Role'].split('/')[-1]

        metrics_paginator = cloudwatch_client.get_paginator('get_metric_data')
        metric_data       = metrics_paginator.paginate(
            MetricDataQueries=[
                {
                    'Id': 'lambdametric',
                    'MetricStat': {
                        'Metric': {
                            'Namespace': 'AWS/Lambda',
                            'MetricName': 'Invocations',
                            'Dimensions': [
                                {
                                    'Name': 'FunctionName',
                                    'Value': name
                                },
                            ]
                        },
                        'Period': 86400,
                        'Stat': 'Sum',
                    }
                },
            ],
            StartTime=datetime.utcnow() - timedelta(days=548),
            EndTime=datetime.utcnow(),
        ).build_full_result()

        try:
            last_run             = metric_data['MetricDataResults'][0]['Timestamps'][0].strftime('%Y-%m-%d')
            last_run_invocations = metric_data['MetricDataResults'][0]['Values'][0]
        except:
            last_run             = "none"
            last_run_invocations = "none"

        try:
            policies = iam_client.list_attached_role_policies(RoleName=role)

            for policy in policies['AttachedPolicies']:

                policy_arn        = policy['PolicyArn']
                policy_version    = iam_client.get_policy(PolicyArn=policy_arn)['Policy']['DefaultVersionId']
                policy_statements = iam_client.get_policy_version(PolicyArn=policy_arn,VersionId=policy_version)['PolicyVersion']['Document']['Statement']

                def list_services(actions):

                    if type(actions) is list:

                        for action in actions:

                            service = action.split(':')[0]

                            if service not in services:
                                services.append(service)

                    else:

                        service = actions.split(':')[0]

                    if service not in services:
                        services.append(service)                

                if type(policy_statements) is list:

                    for statement in policy_statements:

                        actions = statement['Action']

                        list_services(actions)

                else:

                    actions = policy_statements['Action']

                    list_services(actions)

            services_str = '/'.join(services)
        except:
            services_str = "none"

        try:
            execution_policy = json.loads(lambda_client.get_policy(FunctionName=name)['Policy'])

            for statement in execution_policy['Statement']:

                trigger = statement['Principal']['Service']

                if trigger not in triggers:
                    triggers.append(trigger)

            triggers_str = '/'.join(triggers)
        except: 
            triggers_str = "none"
        
        csv_row = [name, runtime, last_modified, state, memory_size, timeout, services_str, triggers_str, last_run, last_run_invocations]
        csvwriter.writerow(csv_row)