python 脚本无限期运行时如何刷新 boto3 凭据

Question

我正在尝试编写一个 python 脚本，该脚本使用 watchdog 来查找文件创建并使用 boto3 将其上传到 s3。但是，我的 boto3 凭据每 12 小时就会过期，因此我需要更新它们。我将我的 boto3 凭据存储在~/.aws/credentials. 所以现在我正在尝试捕获S3UploadFailedError、更新凭据并将它们写入~/.aws/credentials. 但是，尽管凭证正在更新，但我boto3.client('s3')再次调用它的抛出异常。

我究竟做错了什么？或者我该如何解决？

下面是代码片段

try:
     s3 = boto3.client('s3')
     s3.upload_file(event.src_path,'bucket-name',event.src_path)

except boto3.exceptions.S3UploadFailedError as e:
     print(e)
     get_aws_credentials()
     s3 = boto3.client('s3')

Answer 1

我找到了一个很好的例子来刷新此链接中的凭据： https: //pritul95.github.io/blogs/boto3/2020/08/01/refreshable-boto3-session/

但里面有一个小错误。对此要小心。这是更正后的代码：

from uuid import uuid4
from datetime import datetime
from time import time

import pytz
from boto3 import Session
from botocore.credentials import RefreshableCredentials
from botocore.session import get_session


class RefreshableBotoSession:
    """
    Boto Helper class which lets us create a refreshable session so that we can cache the client or resource.

    Usage
    -----
    session = RefreshableBotoSession().refreshable_session()

    client = session.client("s3") # we now can cache this client object without worrying about expiring credentials
    """

    def __init__(
        self,
        region_name: str = None,
        profile_name: str = None,
        sts_arn: str = None,
        session_name: str = None,
        session_ttl: int = 3000
    ):
        """
        Initialize `RefreshableBotoSession`

        Parameters
        ----------
        region_name : str (optional)
            Default region when creating a new connection.

        profile_name : str (optional)
            The name of a profile to use.

        sts_arn : str (optional)
            The role arn to sts before creating a session.

        session_name : str (optional)
            An identifier for the assumed role session. (required when `sts_arn` is given)

        session_ttl : int (optional)
            An integer number to set the TTL for each session. Beyond this session, it will renew the token.
            50 minutes by default which is before the default role expiration of 1 hour
        """

        self.region_name = region_name
        self.profile_name = profile_name
        self.sts_arn = sts_arn
        self.session_name = session_name or uuid4().hex
        self.session_ttl = session_ttl

    def __get_session_credentials(self):
        """
        Get session credentials
        """
        session = Session(region_name=self.region_name, profile_name=self.profile_name)

        # if sts_arn is given, get credential by assuming the given role
        if self.sts_arn:
            sts_client = session.client(service_name="sts", region_name=self.region_name)
            response = sts_client.assume_role(
                RoleArn=self.sts_arn,
                RoleSessionName=self.session_name,
                DurationSeconds=self.session_ttl,
            ).get("Credentials")

            credentials = {
                "access_key": response.get("AccessKeyId"),
                "secret_key": response.get("SecretAccessKey"),
                "token": response.get("SessionToken"),
                "expiry_time": response.get("Expiration").isoformat(),
            }
        else:
            session_credentials = session.get_credentials().get_frozen_credentials()
            credentials = {
                "access_key": session_credentials.access_key,
                "secret_key": session_credentials.secret_key,
                "token": session_credentials.token,
                "expiry_time": datetime.fromtimestamp(time() + self.session_ttl).replace(tzinfo=pytz.utc).isoformat(),
            }

        return credentials

    def refreshable_session(self) -> Session:
        """
        Get refreshable boto3 session.
        """
        # Get refreshable credentials
        refreshable_credentials = RefreshableCredentials.create_from_metadata(
            metadata=self.__get_session_credentials(),
            refresh_using=self.__get_session_credentials,
            method="sts-assume-role",
        )

        # attach refreshable credentials current session
        session = get_session()
        session._credentials = refreshable_credentials
        session.set_config_variable("region", self.region_name)
        autorefresh_session = Session(botocore_session=session)

        return autorefresh_session

请注意，您需要安装pytz 。

Answer 2

根据文档，客户端在多个位置查找凭据，并且您可能需要考虑其他更编程友好的选项，而不是.aws/credentials。

引用文档：

Boto3 搜索凭据的顺序是：

• 在 boto.client() 方法中将凭据作为参数传递
• 创建 Session 对象时将凭据作为参数传递
• 环境变量
• 共享凭证文件 (~/.aws/credentials)
• AWS 配置文件 (~/.aws/config)
• 承担角色提供者

在您的情况下，由于您已经捕获了异常并更新了凭据，我只需将新的凭据传递给客户端的新实例，如下所示：

client = boto3.client(
    's3',
    aws_access_key_id=NEW_ACCESS_KEY,
    aws_secret_access_key=NEW_SECRET_KEY,
    aws_session_token=NEW_SESSION_TOKEN
)

相反，如果您在代码中的其他位置使用这些相同的凭据来创建其他客户端，我会考虑将它们设置为环境变量：

import os

os.environ['AWS_ACCESS_KEY_ID'] = NEW_ACCESS_KEY
os.environ['AWS_SECRET_ACCESS_KEY'] = NEW_SECRET_KEY
os.environ['AWS_SESSION_TOKEN'] = NEW_SESSION_TOKEN

再次引用文档：

仅当您使用临时凭证时才需要您的 AWS 账户的会话密钥[...]。