如何让 boto3 脚本作为 Lambda 函数运行？

Question

这适用于 CLI，但不适用于 AWS Lambda：

#This will list out all volumes that aren't in use or attached to instances
import boto3
import sys

ec2 = boto3.resource('ec2')
client = boto3.client('sns')
vol_array = ec2.volumes.all()
vol_avail = []

#def chk_vols(event, context):
for v in vol_array:
    if v.state == 'available':
        vol_avail.append(v.id)

response = client.publish(
    TopicArn='arn:aws:sns:us-east-1:444444444444:adm-group',
    Message=str(vol_avail),
    Subject='AWS Volumes Available'
)

我想将此代码转换为 AWS Lambda 函数。我取消了该行的注释： #def chk_vols（事件，上下文）： 并正确地将 For 循环移至“def”语句下。

此代码将检查是否有任何 AWS 卷处于可用状态并向 adm-group 主题发送 SNS 消息。预先感谢您！

Answer 1

对于通过任何受支持的方法访问 AWS 的任何资源，都需要对请求进行身份验证。如果您在本地配置了 AWS CLI，

boto3

知道如何使用这些凭证进行身份验证，因此您可以访问资源并修改它们。

不幸的是，Lambda 不支持凭证，因为您需要考虑其他方式。最安全的方法是使用

STS

承担角色并获取临时凭证。

    try:
        sts_client = boto3.client('sts')

        assumedRoleObject = sts_client.assume_role(
            RoleArn="arn:aws:iam::" + "AwsAccountNumber" + ":role/" + "RoleWithSTSPermissions",
            RoleSessionName="NameOfTheSession"
        )

        credentials = assumedRoleObject['Credentials']

        _client = boto3.client(
            'ec2',
            aws_access_key_id=credentials['AccessKeyId'],
            aws_secret_access_key=credentials['SecretAccessKey'],
            aws_session_token=credentials['SessionToken'],
            region_name=_region
        )
    except Exception as e:
        print(e.message)
        raise

如果这些东西是变量，那就太理想了，这样你就可以根据需要动态更改它们。

示例

import os
import boto3
import logging

AwsAccount = dict(
    number=None,
    role=None,
    region=None
)
def authenticate(_region, _resource):
    try:
        sts_client = boto3.client('sts')

        assumedRoleObject = sts_client.assume_role(
            RoleArn="arn:aws:iam::" + AwsAccount['number'] + ":role/" + AwsAccount['role'],
            RoleSessionName="LambdaAssumeRoleSession"
        )

        credentials = assumedRoleObject['Credentials']

        _client = boto3.client(
            _resource,
            aws_access_key_id=credentials['AccessKeyId'],
            aws_secret_access_key=credentials['SecretAccessKey'],
            aws_session_token=credentials['SessionToken'],
            region_name=_region
        )
    except Exception as e:
        log.warn(e.message)
        log.warn("Switching to local credentials")

        try:
            # TODO: Remove failover profile in production
            _session = boto3.session.Session(region_name=_region)
            _client = _session.client(_resource, region_name=_region)
            log.info("Successfully authenticated using local credentials")
        except Exception as e:
            log.error(e.message)
            raise

    return _client

def main():
    _client = authenticate(AwsAccount['region'], 'ec2')

    return

def lambda_handler(event, context):
    try:
        AwsAccount['number'] = os.environ['AWS_ACCOUNT'].strip()
        AwsAccount['region'] = os.environ['AWS_REGION'].strip()
        AwsAccount['role'] = os.environ['AWS_LAMBDA_ROLE'].strip()
    except Exception as e:
        log.error(e)
        raise

    main()

    return

if __name__ == '__main__':
    lambda_handler(None, None)

如果您已在本地和 Lambda 上配置了环境变量，则可以在本地和 Lambda 上测试此代码

创建 Lambda 函数时，请确保将

lambda_handler

定义为处理程序。

Answer 2

您需要将 IAM 角色与具有所需权限的 Lambda 相关联。参考文档：https://docs.aws.amazon.com/lambda/latest/dg/intro-permission-model.html#lambda-intro-execution-role

Answer 3

我能够让我的原始代码正常工作。

import boto3

ec2 = boto3.resource('ec2')
sns = boto3.client('sns')

def chk_vols(event, context):
    vol_array = ec2.volumes.all()
    vol_avail = []

    for v in vol_array:
        if v.state == 'available':
            vol_avail.append(v.id)

    if vol_avail:
        sns.publish(
            TopicArn='arn:aws:sns:us-east-1:444444444444:adm-group',
            Message=str(vol_avail),
            Subject='AWS Volumes Available'
        )