AWS boto3 脚本访问 AWS 资源

引言
boto3
- 主要功能
- 常见用例
- 安装和基本使用
boto3.Client() 低级客户端
- 基本用法
- 关键参数
boto3.resource() 高级客户端
- 常见参数
- 用法
boto3.resource VS boto3.client
- 相似点
- 不同点
- 总结
关于身份验证凭证
- 隐式身份凭证
- 显式身份验证凭证
- - assuem role
  - 如何处理凭证过期
- 兼容隐式和显示身份凭证代码示例

引言

前面有介绍《AWS IAM 基本概念》和《AWS 资源 ARN 基本概念》，项目开发测试脚本中，怎么用脚本访问 AWS 资源呢，尤其是怎么获取 AWS 凭证 credential。

boto3

Boto3 是 Amazon Web Services (AWS) 提供的官方 Python 软件开发工具包 (Software Development Kit)。它允许开发人员使用 Python 代码与 AWS 的多种服务进行交互和管理。Boto3 提供了对 AWS 服务的全面支持，简化了开发和管理 AWS 资源的工作流程。

主要功能

资源管理：Boto3 使用户可以轻松地管理 AWS 资源，例如 EC2 实例、S3 存储桶、DynamoDB 表等。
高层次服务 API：通过资源接口，Boto3 提供了一种更高级别、更直观的方式来与 AWS 服务交互。
底层服务 API：Boto3 也提供了对 AWS API 的底层访问，使得用户可以使用更细粒度的控制来执行复杂的操作。
自动分页：对于返回大量结果的 AWS 操作，Boto3 能够自动处理分页请求，从而简化了处理大量数据的过程。

常见用例

与 S3 交互：用户可以使用 Boto3 来创建、删除、列出和管理 S3 存储桶和对象。
管理 EC2 实例：Boto3 允许用户启动、停止、终止和管理 EC2 实例。
操作 DynamoDB：用户可以使用 Boto3 来创建、查询、更新和删除 DynamoDB 表和数据。
自动化基础设施：Boto3 常与其他工具（如 AWS CloudFormation 和 Terraform）结合使用，以便自动化管理 AWS 资源。

安装和基本使用

要安装 Boto3，可以使用 pip：

pip install boto3

然后可以在 Python 脚本中导入并使用 Boto3。例如，列出所有的 S3 存储桶：

import boto3

# 创建 S3 服务的客户端
s3 = boto3.client('s3')

# 列出所有的 S3 存储桶
response = s3.list_buckets()

# 打印存储桶名称
for bucket in response['Buckets']:
    print(bucket['Name'])

在使用 boto3 之前，确保你已经配置了 AWS 凭证和配置文件。你可以通过 AWS CLI 或者直接在 ~/.aws/credentials 和 ~/.aws/config 文件中进行配置。

boto3.Client() 低级客户端

boto3.client() 函数是 Boto3 中用于创建与特定 AWS 服务交互的低级客户端。使用客户端，您可以直接调用服务 API 操作并处理更复杂的操作。

基本用法

import boto3

# 创建一个 S3 客户端
s3_client = boto3.client('s3')

# 创建一个 EC2 客户端
ec2_client = boto3.client('ec2')

在上述示例中，boto3.client(‘s3’) 和 boto3.client(‘ec2’) 分别创建了一个 S3 服务和一个 EC2 服务的客户端。这个客户端允许您直接调用 S3 或 EC2 提供的 API 操作。

关键参数

service_name:
服务名称字符串，指定要创建客户端的 AWS 服务。常见的服务名称有 ‘s3’、‘ec2’、‘dynamodb’、‘lambda’ 等。
region_name (可选):
用于指定客户端所使用的 AWS 区域。例如，如果您希望与特定区域的服务进行交互，可以设置此参数。

s3_client = boto3.client('s3', region_name='us-west-1')

aws_access_key_id , aws_secret_access_key 和 aws_session_token (可选):
这些参数用于显式设置访问 AWS 服务所需的身份验证凭证。如果您没有通过其他方式设置凭证（例如环境变量或配置文件），可以直接在这里提供。

s3_client = boto3.client(
    's3',
    aws_access_key_id='YOUR_ACCESS_KEY',
    aws_secret_access_key='YOUR_SECRET_KEY',
    aws_session_token='YOUR_SESSION_TOKEN'
)

endpoint_url (可选):
允许指定要访问的 AWS 服务的自定义端点。这通常用于连接到本地模拟器或自定义的服务端点。

s3_client = boto3.client('s3', endpoint_url='http://localhost:8000')

config (可选):
通过 botocore.client.Config 对象提供额外的配置选项，例如请求重试策略、超时时间、最大连接数等。

from botocore.config import Config

config = Config(
    retries = {
        'max_attempts': 10,
        'mode': 'standard'
    }
)

s3_client = boto3.client('s3', config=config)

boto3.resource() 高级客户端

boto3.resource 是 AWS 的 Python SDK（boto3）提供的高级接口，用于更方便地与 AWS 服务进行交互。提供了比 boto3.client 更加简洁和易用的接口。对于大多数常见的 AWS 操作，resource 提供了更高的抽象，使代码更加简洁和可读。

常见参数

要使用 boto3.resource，首先需要创建一个资源对象。常见的资源包括 S3、DynamoDB、EC2 等。创建资源对象时，可以传递一些可选参数来配置连接和请求。这些参数主要用于控制资源的配置，如区域、凭证等。

service_name
说明: 指定要创建的 AWS 服务的名称。
类型: str
示例: ‘s3’, ‘ec2’, ‘dynamodb’

s3 = boto3.resource('s3')

region_name 可选
说明: 指定要连接的 AWS 区域。如果未指定，boto3 会使用默认配置或环境变量中的区域。
类型: str
示例: ‘us-west-1’, ‘us-east-1’

s3 = boto3.resource('s3', region_name='us-west-2')

api_version 可选
说明: 指定要使用的 API 版本。如果未指定，boto3 会使用最新的可用版本。
类型: str
示例: ‘2012-11-05’

dynamodb = boto3.resource('dynamodb', api_version='2012-08-10')

use_ssl 可选
说明: 指定是否使用 SSL 与 AWS 服务通信。默认是 True。
类型: bool
示例: True, False

s3 = boto3.resource('s3', use_ssl=False)

verify 可选
说明: 指定是否验证 SSL 证书，或指定一个 CA 证书的路径。默认是 True。
类型: bool 或 str
示例: True, False, ‘/path/to/cert.pem’

s3 = boto3.resource('s3', verify='/path/to/cert.pem')

endpoint_url
说明: 指定要连接的自定义端点 URL，通常用于本地开发或测试，或与某些 AWS 兼容的服务交互。
类型: str
示例: ‘http://localhost:8000’

dynamodb = boto3.resource('dynamodb', endpoint_url='http://localhost:8000')

aws_access_key_id, aws_secret_access_key 和 aws_session_token 可选
说明: 指定 AWS 访问密钥 ID，访问密钥和 AWS 会话令牌。通常你可以通过环境变量或配置文件设置，而不需要在代码中显式指定。
类型: str
示例: ‘YOUR_ACCESS_KEY’， ‘YOUR_SECRET_KEY’ 和 ‘YOUR_SESSION_TOKEN’

s3 = boto3.resource('s3', aws_access_key_id='YOUR_ACCESS_KEY', aws_secret_access_key='YOUR_SECRET_KEY', aws_session_token='YOUR_SESSION_TOKEN')

config 可选
说明: 一个 botocore.client.Config 对象，用于自定义客户端的配置，如重试策略、超时设置等。
类型: botocore.client.Config
示例: Config(retries={‘max_attempts’: 10}, connect_timeout=5)

from botocore.client import Config

config = Config(
    retries = {
        'max_attempts': 10,
        'mode': 'standard'
    },
    connect_timeout = 5
)

s3 = boto3.resource('s3', config=config)

用法

创建资源对象后，可以直接访问或操作资源。

import boto3

# 创建一个 S3 资源对象
s3 = boto3.resource('s3')

# 列出所有 S3 桶
for bucket in s3.buckets.all():
    print(bucket.name)

# 创建一个新桶
bucket = s3.create_bucket(Bucket='my-test-bucket')

# 上传文件
bucket.upload_file('test.txt', 'test_remote.txt')

# 下载文件
bucket.download_file('test_remote.txt', 'downloaded_test.txt')

boto3.resource VS boto3.client

boto3.resource 和 boto3.client 都是用于与 AWS 服务交互的接口，但它们之间的参数使用有所不同。以下是两者在参数方面的主要区别和相似点：

相似点

许多参数在 boto3.resource 和 boto3.client 中是相同的，提供了类似的功能：

region_name: 指定连接的 AWS 区域。
api_version: 指定使用的 API 版本。
use_ssl: 指定是否使用 SSL。
verify: 指定是否验证 SSL 证书，或提供 CA 证书路径。
endpoint_url: 指定自定义端点 URL，通常用于本地开发或与 AWS 兼容的服务交互。
aws_access_key_id, aws_secret_access_key, aws_session_token: 指定 AWS 凭证信息（虽然通常通过环境变量或配置文件提供，而不是显式传递）。
config: 用于传递 botocore.client.Config 对象，以自定义连接配置（如重试策略、超时设置等）。

不同点

虽然许多参数是相似的，但 boto3.resource 和 boto3.client 之间有一些重要的区别，主要体现在使用方式和参数的意图上。

面向对象 vs 面向 API

boto3.resource: 提供了面向对象的接口，允许你直接操作 AWS 资源的对象。resource 的设计是为了提供更高的抽象层次，使开发人员可以更直观地处理 AWS 资源。
示例：

s3 = boto3.resource('s3')
bucket = s3.Bucket('my-bucket')
bucket.upload_file('file.txt', 'remote_file.txt')

boto3.client: 提供了更底层的 API 接口，直接映射到 AWS 服务的 API 调用。它通常需要你处理请求和响应的细节，适合需要更精细控制的场景。
示例：

s3_client = boto3.client('s3')
s3_client.upload_file('file.txt', 'my-bucket', 'remote_file.txt')

功能集

boto3.resource: 提供了更高层次的功能，允许你直接操作 AWS 资源，如 S3 的 Bucket 和 Object，DynamoDB 的 Table 等。它通常会自动处理一些细节，如分页、资源引用等。
boto3.client: 提供了更广泛的功能集，直接映射到 API 操作，因此可以访问 resource 不提供的某些低级功能。例如，client 可以调用所有的 API 操作，而 resource 可能只覆盖了最常用的操作。

返回值

boto3.resource: 返回的对象是资源对象，如 S3 的 Bucket、Object，它们有自己的方法和属性，可以继续进行操作。
boto3.client: 返回的通常是 Python 字典，包含 API 请求的响应数据。

参数配置

boto3.resource: 通常用于更高层次的操作，因此在配置上可能需要的参数更少，且 resource 本身封装了很多默认行为和最佳实践。
boto3.client: 由于直接与 AWS API 交互，因此提供了更多的配置选项，适合需要精细控制的场景。你可能需要更详细地设置参数，控制 API 的行为。

总结

虽然 boto3.resource 和 boto3.client 在参数上有很多相似之处，但它们的设计初衷和使用场景不同。resource 提供了更高层次、面向对象的接口，适合日常操作；而 client 提供了更底层的 API 访问，适合需要直接控制 API 请求和响应的场景。

关于身份验证凭证

隐式身份凭证

前面实例化一个 client，或 resouce 都没有显式设置访问 AWS 服务所需的身份验证凭证（aws_access_key_id, aws_secret_access_key 和 aws_session_token），会默认读取 ~/.aws/credentials 和 ~/.aws/config 文件中进行配置身份凭证。

代码实例：

使用 boto3.client()

import boto3
client = boto3.client('s3', region_name='us-east-1')

使用 session.client()

import boto3
session = boto3.Session(profile_name='default')
client = session.client('s3', region_name='us-east-1')

身份验证凭证配置示例：

通常 Devops 会创建一个基于 SSO 也就是 AD 账号信任的 Role，用于来自本地账号 assume role 访问 AWS 资源，获取临时凭证，通过脚本将凭证设置到 ~/.aws/credentials 和 ~/.aws/config 文件中。用户可以直接登入 AWS 网站上访问资源，或通过脚本通过隐式身份凭证创建 boto3.client() 访问 AWS 资源。

例如：下面这个 Role，只能是来自 AD,Azure 的用户才能 assum 该 Role，而 session duration 设置的时间比较长。
在这里插入图片描述

~/.aws/credentials 有如下配置：基于不同的环境的凭证配置
profile_name=‘default’ 指定了使用名为 default 的 AWS 配置文件，包含访问 AWS 服务所需的凭证 aws_access_key_id, aws_secret_access_key 和 aws_session_token。

在这里插入图片描述

直接使用 boto3.client() 和通过 session.client() 创建客户端有以下区别：

默认配置文件:
boto3.client() 会使用默认的配置文件和凭证。
session.client() 可以指定使用特定的配置文件和凭证。
会话管理:
boto3.client() 会自动创建一个默认会话。
session.client() 使用的是你显式创建的会话，这样可以更灵活地管理多个会话。

显式身份验证凭证

assuem role

sts_client.assume_role 是 AWS SDK for Python (boto3) 中的一个方法，用于调用 AWS Security Token Service (STS) 的 AssumeRole API 操作。这个方法允许一个已经拥有 AWS 凭证的用户获取临时安全凭证，以便以另一个角色的身份执行操作。

参数

RoleArn：要扮演的角色的 Amazon 资源名称 (ARN)。
RoleSessionName：为会话指定的名称。此名称用于唯一标识会话。
DurationSeconds（可选）：指定临时凭证的有效期，默认为 3600 秒（1 小时），最大值为 43200 秒（12 小时）。
Policy（可选）：一个 JSON 字符串，指定进一步限制角色的权限。
ExternalId（可选）：一个外部 ID，用于第三方身份验证。
SerialNumber（可选）：MFA 设备的序列号。
TokenCode（可选）：MFA 设备生成的验证码。

以下是一个使用 sts_client.assume_role 的示例：

import boto3

# 创建 STS 客户端
sts_client = boto3.client('sts')

# 假设角色
response = sts_client.assume_role(
    RoleArn='arn:aws:iam::123456789012:role/example-role',
    RoleSessionName='example-session'
)

# 获取临时凭证
credentials = response['Credentials']

# 打印临时凭证
print(f"Access Key ID: {credentials['AccessKeyId']}")
print(f"Secret Access Key: {credentials['SecretAccessKey']}")
print(f"Session Token: {credentials['SessionToken']}")

在这个示例中，我们创建了一个 STS 客户端，并使用 assume_role 方法获取了临时安全凭证。然后，我们打印了这些临时凭证。接下来可以使用 boto3.client 方法并传递 aws_access_key_id、aws_secret_access_key 和 aws_session_token 参数。

def create_client_with_credentials(service, credentials):
    client = boto3.client(service, 
                          aws_access_key_id=credentials['AccessKeyId'],
                          aws_secret_access_key=credentials['SecretAccessKey'],
                          aws_session_token=credentials['SessionToken'])
    return client

如何处理凭证过期

注意： DurationSeconds 参数不能超过角色设定的最大时间。每个角色在创建时可以指定一个最大会话持续时间（Maximum session duration），这个时间范围是 1 小时到 12 小时（3600 秒到 43200 秒）。如果在调用 assume_role 时指定的 DurationSeconds 超过了角色的最大会话持续时间，调用将会失败。

通常出于安全考虑，临时凭证只是临时用一下，所以 session duration 都不会设置过长。

例如：这面这个 role 是为了 Jenkins 执行 Automation 时需要访问 AWS 资源创建的 Role，但是最大 session duration 只设定了 1 个小时，意味 1 小时果然，这个 assume role 凭证就会失效了，将无法访问 AWS 资源。
在这里插入图片描述

所以当程序访问 AWS 资源遇到 Token 过期异常时，就需要重刷一下 token

def assume_role(role_arn, session_name):
    sts_client = boto3.client('sts')
    try:
        assumed_role_object = sts_client.assume_role(
            RoleArn=role_arn,
            RoleSessionName=session_name
        )
        return assumed_role_object
    except Exception as e:
        logging.error(f"Error in assuming role: {e}")
        return None
 

def create_client_with_credentials(service, credentials):
    client = boto3.client(service, aws_access_key_id=credentials['AccessKeyId'],
                          aws_secret_access_key=credentials['SecretAccessKey'],
                          aws_session_token=credentials['SessionToken'])
    return client


def random_string(length=9):
    characters = string.ascii_letters + string.digits  
    return ''.join(random.choice(characters) for _ in range(length))
    
def refresh_token_in_jenkins(service, account_id):
    role_arn = f'arn:aws:iam::{account_id}:role/AssetApplication_2022/JenkinsExecutionRole'
    session_name = random_string()
    assumed_role_object = assume_role(role_arn, session_name)
    if assumed_role_object is not None:
        credentials = assumed_role_object['Credentials']
        client = create_client_with_credentials(service, credentials)
        return client
    else:
        return None

def get_s3_list(s3, prefix, bucket_name):
    all_objects = []           
    # Get the list of objects in the bucket, MaxKeys is set to 1000 to get the maximum number of objects in a single call
    kwargs = {'Bucket': bucket_name, 'Prefix': prefix, 'MaxKeys': 1000}
    retry_count = 0
    max_retry = 3

    while True:
        try:
            response = s3.list_objects_v2(**kwargs)
            all_objects.append(response)
            if 'NextContinuationToken' in response:
                kwargs['ContinuationToken'] = response['NextContinuationToken']
            else:
                break
        except Exception as e:
            logging.error(f"Error for {prefix}: {e}")   
            retry_count += 1
            if retry_count >= max_retry:
                logging.error(f"Max retry to refresh s3 token for {prefix} count reached: {max_retry}")
                break
            s3 = refresh_token_in_jenkins('s3', account_id='111111111111') 
    return all_objects

兼容隐式和显示身份凭证代码示例

脚本一般都会通过本地调试，然后部署到 Jenkins 上跑。通常本地都是通过AD 身份 assume role 获取到临时凭证，Jenkins 是通过 Jenskin 账户 Assum Role 获取临时凭证，所以代码要兼容显式和隐式身份凭证访问 AWS，方便本地和 Jenkins 上执行。所以我们可以写一个兼容两种方式创建 Client 访问 AWS 资源。

import boto3
import string
import random
import logging

import sys
sys.path.append('./')
from constants import ENV

def assume_role(role_arn, session_name):
    sts_client = boto3.client('sts')
    try:
        assumed_role_object = sts_client.assume_role(
            RoleArn=role_arn,
            RoleSessionName=session_name
        )
        return assumed_role_object
    except Exception as e:
        logging.error(f"Error in assuming role: {e}")
        return None
 

def create_client_with_credentials(service, credentials):
    client = boto3.client(service, aws_access_key_id=credentials['AccessKeyId'],
                          aws_secret_access_key=credentials['SecretAccessKey'],
                          aws_session_token=credentials['SessionToken'])
    return client


def random_string(length=9):
    characters = string.ascii_letters + string.digits  
    return ''.join(random.choice(characters) for _ in range(length))


def refresh_token_in_jenkins(service, account_id):
    role_arn = f'arn:aws:iam::{account_id}:role/AssetApplication_1911/JenkinsExecutionRole'
    session_name = random_string()
    assumed_role_object = assume_role(role_arn, session_name)
    if assumed_role_object is not None:
        credentials = assumed_role_object['Credentials']
        client = create_client_with_credentials(service, credentials)
        return client
    else:
        return None


def get_client(service, account_id):
    if ENV == "jenkins":
        client = refresh_token_in_jenkins(service, account_id)
    else:
        session = boto3.Session(profile_name='default')
        client = session.client(service, region_name='us-east-1')
    return client