AWS: Boto3: пример AssumeRole, который включает использование роли

Ответ 1

Чтобы получить сеанс с предполагаемой ролью:

import botocore
import boto3
import datetime
from dateutil.tz import tzlocal

assume_role_cache: dict = {}
def assumed_role_session(role_arn: str, base_session: botocore.session.Session = None):
    base_session = base_session or boto3.session.Session()._session
    fetcher = botocore.credentials.AssumeRoleCredentialFetcher(
        client_creator = base_session.create_client,
        source_credentials = base_session.get_credentials(),
        role_arn = role_arn,
        extra_args = {
        #    'RoleSessionName': None # set this if you want something non-default
        }
    )
    creds = botocore.credentials.DeferredRefreshableCredentials(
        method = 'assume-role',
        refresh_using = fetcher.fetch_credentials,
        time_fetcher = lambda: datetime.datetime.now(tzlocal())
    )
    botocore_session = botocore.session.Session()
    botocore_session._credentials = creds
    return boto3.Session(botocore_session = botocore_session)

# usage:
session = assumed_role_session('arn:aws:iam::ACCOUNTID:role/ROLE_NAME')
ec2 = session.client('ec2') # ... etc.

Полученные учетные данные сеанса будут автоматически обновляться при необходимости, что довольно неплохо.

Примечание: мой предыдущий ответ был совершенно неверным, но я не могу удалить его, поэтому я заменил его более эффективным и рабочим ответом.

Ответ 2

Вы можете принять роль, используя токен STS, например:

class Boto3STSService(object):
    def __init__(self, arn):
        sess = Session(aws_access_key_id=ARN_ACCESS_KEY,
                       aws_secret_access_key=ARN_SECRET_KEY)
        sts_connection = sess.client('sts')
        assume_role_object = sts_connection.assume_role(RoleArn=arn, RoleSessionName=ARN_ROLE_SESSION_NAME,DurationSeconds=3600)
        self.credentials = assume_role_object['Credentials']

Это даст вам временный ключ доступа и секретные ключи с токеном сеанса. С помощью этих временных учетных данных вы можете получить доступ к любой службе. Например, если вы хотите получить доступ к ELB, вы можете использовать следующий код:

self.tmp_credentials = Boto3STSService(arn).credentials

def get_boto3_session(self):
    tmp_access_key = self.tmp_credentials['AccessKeyId']
    tmp_secret_key = self.tmp_credentials['SecretAccessKey']
    security_token = self.tmp_credentials['SessionToken']

    boto3_session = Session(
        aws_access_key_id=tmp_access_key,
        aws_secret_access_key=tmp_secret_key, aws_session_token=security_token
    )
    return boto3_session

def get_elb_boto3_connection(self, region):
    sess = self.get_boto3_session()
    elb_conn = sess.client(service_name='elb', region_name=region)
    return elb_conn

Ответ 3

Вот фрагмент кода из официальной документации AWS, где создан ресурс s3 для перечисления всех сегментов s3. boto3 или клиенты для других сервисов могут создаваться аналогичным образом.

# create an STS client object that represents a live connection to the 
# STS service
sts_client = boto3.client('sts')

# Call the assume_role method of the STSConnection object and pass the role
# ARN and a role session name.
assumed_role_object=sts_client.assume_role(
    RoleArn="arn:aws:iam::account-of-role-to-assume:role/name-of-role",
    RoleSessionName="AssumeRoleSession1"
)

# From the response that contains the assumed role, get the temporary 
# credentials that can be used to make subsequent API calls
credentials=assumed_role_object['Credentials']

# Use the temporary credentials that AssumeRole returns to make a 
# connection to Amazon S3  
s3_resource=boto3.resource(
    's3',
    aws_access_key_id=credentials['AccessKeyId'],
    aws_secret_access_key=credentials['SecretAccessKey'],
    aws_session_token=credentials['SessionToken'],
)

# Use the Amazon S3 resource object that is now configured with the 
# credentials to access your S3 buckets. 
for bucket in s3_resource.buckets.all():
    print(bucket.name)

Ответ 4

Если вы хотите функциональную реализацию, я остановился на этом:

def filter_none_values(kwargs: dict) -> dict:
    """Returns a new dictionary excluding items where value was None"""
    return {k: v for k, v in kwargs.items() if v is not None}


def assume_session(
    role_session_name: str,
    role_arn: str,
    duration_seconds: Union[int, None] = None,
    region_name: Union[str, None] = None,
) -> boto3.Session:
    """
    Returns a session with the given name and role.
    If not specified, duration will be set by AWS, probably at 1 hour.
    If not specified, region will be left unset.
    Region can be overridden by each client or resource spawned from this session.
    """
    assume_role_kwargs = filter_none_values(
        {
            "RoleSessionName": role_session_name,
            "RoleArn": role_arn,
            "DurationSeconds": duration_seconds,
        }
    )
    credentials = boto3.client("sts").assume_role(**assume_role_kwargs)["Credentials"]
    create_session_kwargs = filter_none_values(
        {
            "aws_access_key_id": credentials["AccessKeyId"],
            "aws_secret_access_key": credentials["SecretAccessKey"],
            "aws_session_token": credentials["SessionToken"],
            "region_name": region_name,
        }
    )
    return boto3.Session(**create_session_kwargs)


def main() -> None:
    session = assume_session(
        "MyCustomSessionName",
        "arn:aws:iam::XXXXXXXXXXXX:role/TheRoleIWantToAssume",
        region_name="us-east-1",
    )
    client = session.client(service_name="ec2")
    print(client.describe_key_pairs())

Ответ 5

Пример из официальной документации AWS

https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_use_switch-role-api.html

Ответ 6

Обратите внимание, что учетные данные, которые вы получаете от вызова sts.AssumeRole, действительно истекают через некоторое время (по умолчанию через 15 минут, но вы можете установить его дольше при выполнении вызова AssumeRole, вплоть до максимального времени продолжительности сеанса API текущего сеанса по умолчанию - 1 час), если вы не обновите их.

Если вам нужно автоматически обновить учетные данные, я делюсь своим кодом, который я написал после нескольких часов изучения кода boto и пытаюсь обойти зависимость boto от файловой системы и профилей.

Здесь я просто использовал встроенный механизм кэширования boto и регулярно обновлял предполагаемые учетные данные, не затрагивая никаких файлов:

from datetime import datetime

import boto3
from botocore.credentials import (
    AssumeRoleProvider,
    AssumeRoleCredentialFetcher,
    DeferredRefreshableCredentials,
    CredentialResolver
)
from dateutil.tz import tzlocal


class CustomAssumeRoleProvider(AssumeRoleProvider):
    """
    Overrides default AssumeRoleProvider to not use profiles from filesystem.
    """

    def __init__(self,
                 source_session: boto3.Session,
                 assume_role_arn: str,
                 expiry_window_seconds: int):
        super().__init__(
            load_config=lambda: source_session._session.full_config,
            client_creator=source_session._session.create_client,
            cache={},
            profile_name='not-used'
        )
        self.expiry_window_seconds = expiry_window_seconds
        self.source_session = source_session
        self.assume_role_arn = assume_role_arn
        assert assume_role_arn, "assume_role_arn is required"

    def load(self):
        fetcher = AssumeRoleCredentialFetcher(
            client_creator=self.source_session._session.create_client,
            source_credentials=self.source_session.get_credentials(),
            role_arn=self.assume_role_arn,
            expiry_window_seconds=self.expiry_window_seconds,
            cache=self.cache,
        )

        return DeferredRefreshableCredentials(
            method=self.METHOD,
            refresh_using=fetcher.fetch_credentials,
            time_fetcher=lambda: datetime.now(tzlocal())
        )


def get_assume_role_session(
    source_session: boto3.Session,
    assume_role_arn: str,
    expiry_window_seconds=15 * 60
) -> boto3.Session:
    """
    Creates a new boto3 session that will operate as of another user.

    Source session must have permission to call sts:AssumeRole on the provided ARN,
    and that ARN role must have been trusted to be assumed from this account (where source_session is from).

    See https://docs.aws.amazon.com/IAM/latest/UserGuide/tutorial_cross-account-with-roles.html

    Uses internal session._session to hack it together, as I haven't found another way.
    """
    # must have .load() method to be used in CredentialsResolver.
    provider = CustomAssumeRoleProvider(
        source_session=source_session,
        assume_role_arn=assume_role_arn,
        expiry_window_seconds=expiry_window_seconds
    )

    # must have .load_credentials() method to be used in register_component()
    resolver = CredentialResolver([provider])
    new_session = boto3.Session()
    new_session._session.register_component('credential_provider', resolver)
    return new_session