boto3 是否有与 awscli 相当的凭证缓存



使用 awscli,在 ~/.aws/cli/cache 中有一个凭据缓存,它允许我在一段时间内缓存凭据。这在使用 MFA 时非常有用。boto3 是否具有类似的功能,或者我是否必须显式缓存从 session = boto3.session.Session(profile_name='CTO:Admin') 返回的凭据?

它已经在那里了。

http://boto3.readthedocs.org/en/latest/guide/configuration.html#assume-role-provider

当您指定具有 IAM 角色配置的配置文件时,boto3 将进行 AssumeRole 调用以检索临时凭证。后续的 boto3 API 调用将使用缓存的临时凭证,直到它们过期,在这种情况下,boto3 将自动刷新凭证。BOTO3 不会将这些临时凭证写入磁盘。这意味着来自 AssumeRole 调用的临时凭证仅缓存在单个会话的内存中。从该会话创建的所有客户端将共享相同的临时凭证。

我创建了一个 Python 库来为你提供这个 - 请参阅 https://github.com/mixja/boto3-session-cache

例:

import boto3_session_cache
# This returns a regular boto3 client object with the underlying session configured with local credential cache 
client = boto3_session_cache.client('ecs')
ecs_clusters = client.list_clusters()

总结以上几点,举一个工作的例子:

from os import path
import os
import sys
import json
import datetime
from distutils.spawn import find_executable
from botocore.exceptions import ProfileNotFound
import boto3
import botocore

def json_encoder(obj):
    """JSON encoder that formats datetimes as ISO8601 format."""
    if isinstance(obj, datetime.datetime):
        return obj.isoformat()
    else:
        return obj

class JSONFileCache(object):
    """JSON file cache.
    This provides a dict like interface that stores JSON serializable
    objects.
    The objects are serialized to JSON and stored in a file.  These
    values can be retrieved at a later time.
    """
    CACHE_DIR = path.expanduser(path.join('~', '.aws', 'ansible-ec2', 'cache'))
    def __init__(self, working_dir=CACHE_DIR):
        self._working_dir = working_dir
    def __contains__(self, cache_key):
        actual_key = self._convert_cache_key(cache_key)
        return path.isfile(actual_key)
    def __getitem__(self, cache_key):
        """Retrieve value from a cache key."""
        actual_key = self._convert_cache_key(cache_key)
        try:
            with open(actual_key) as f:
                return json.load(f)
        except (OSError, ValueError, IOError):
            raise KeyError(cache_key)
    def __setitem__(self, cache_key, value):
        full_key = self._convert_cache_key(cache_key)
        try:
            file_content = json.dumps(value, default=json_encoder)
        except (TypeError, ValueError):
            raise ValueError("Value cannot be cached, must be "
                             "JSON serializable: %s" % value)
        if not path.isdir(self._working_dir):
            os.makedirs(self._working_dir)
        with os.fdopen(os.open(full_key,
                               os.O_WRONLY | os.O_CREAT, 0o600), 'w') as f:
            f.truncate()
            f.write(file_content)
    def _convert_cache_key(self, cache_key):
        full_path = path.join(self._working_dir, cache_key + '.json')
        return full_path

session = boto3.session.Session()
try:
    cred_chain = session._session.get_component('credential_provider')
except ProfileNotFound:
    print "Invalid Profile"
    sys.exit(1)
provider = cred_chain.get_provider('assume-role')
provider.cache = JSONFileCache()
# Do something with the session...
ec2 = session.resource('ec2')

最初,凭证缓存和临时凭证的自动续订是 AWSCLI 的一部分,但此提交(以及一些后续提交)将该功能移至 botocore,这意味着它现在也可以在 boto3 中使用。