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

Question

Я пытаюсь использовать AssumeRole таким образом, что я пересекаю несколько учетных записей и извлекаю активы для этих учетных записей. Я сделал это к этому моменту:

import boto3
stsclient = boto3.client('sts')

assumedRoleObject = sts_client.assume_role(
RoleArn="arn:aws:iam::account-of-role-to-assume:role/name-of-role",
RoleSessionName="AssumeRoleSession1")

Отлично, у меня есть предполагаемый RoleObject. Но теперь я хочу использовать это для перечисления таких вещей, как ELB или что-то, что не является встроенным ресурсом низкого уровня.

Как можно это сделать? Если я могу спросить - пожалуйста, напишите полный пример, чтобы каждый мог получить пользу.

Answer 1

Существует более простой и почему-то недокументированный способ создания такого сеанса:

session = boto3.Session(
    role_arn = 'arn:aws:iam::${ACCOUNT}:role/${ROLE NAME}'
)

ec2 = session.client('ec2') # ... etc.

РЕДАКТИРОВАТЬ: это не похоже на работу, я был уверен, что это сработало, но сейчас я пытаюсь снова, и это не так.

Answer 2

Вот фрагмент кода из официальной документации 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)

Answer 3

Вы можете принять роль, используя токен 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

Answer 4

со ссылкой на решение @jarrad, которое не работает по состоянию на февраль 2021 года, см. следующее

      
import boto3
import botocore.session
from botocore.credentials import AssumeRoleCredentialFetcher, DeferredRefreshableCredentials


def get_boto3_session(assume_role_arn=None):
    session = boto3.Session(aws_access_key_id="abc", aws_secret_access_key="def")
    if not assume_role_arn:
        return session

    fetcher = AssumeRoleCredentialFetcher(
        client_creator=_get_client_creator(session),
        source_credentials=session.get_credentials(),
        role_arn=assume_role_arn,
    )
    botocore_session = botocore.session.Session()
    botocore_session._credentials = DeferredRefreshableCredentials(
        method='assume-role',
        refresh_using=fetcher.fetch_credentials
    )

    return boto3.Session(botocore_session=botocore_session)


def _get_client_creator(session):
    def client_creator(service_name, **kwargs):
        return session.client(service_name, **kwargs)

    return client_creator

Answer 5

Вот фрагмент кода, который я использовал

      sts_client = boto3.client('sts')
assumed_role_object = sts_client.assume_role(
    RoleArn=<arn of the role to assume>,
    RoleSessionName="<role session name>"
)
print(assumed_role_object)
credentials = assumed_role_object['Credentials']
session = Session(
    aws_access_key_id=credentials['AccessKeyId'],
    aws_secret_access_key=credentials['SecretAccessKey'],
    aws_session_token=credentials['SessionToken']
)
self.s3 = session.client('s3')

Answer 6

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

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())

Answer 7

      #!/usr/bin/env python3

import boto3

sts_client = boto3.client('sts')
assumed_role = sts_client.assume_role(RoleArn =  "arn:aws:iam::123456789012:role/example_role",
                                      RoleSessionName = "AssumeRoleSession1",
                                      DurationSeconds = 1800)
session = boto3.Session(
    aws_access_key_id     = assumed_role['Credentials']['AccessKeyId'],
    aws_secret_access_key = assumed_role['Credentials']['SecretAccessKey'],
    aws_session_token     = assumed_role['Credentials']['SessionToken'],
    region_name           = 'us-west-1'
)

# now we make use of the role to retrieve a parameter from SSM
client = session.client('ssm')
response = client.get_parameter(
    Name = '/this/is/a/path/parameter',
    WithDecryption = True
)
print(response)

Answer 8

import json
import boto3


roleARN = 'arn:aws:iam::account-of-role-to-assume:role/name-of-role'
client = boto3.client('sts')
response = client.assume_role(RoleArn=roleARN, 
                              RoleSessionName='RoleSessionName', 
                              DurationSeconds=900)

dynamodb_client = boto3.client('dynamodb', region_name='us-east-1',
                    aws_access_key_id=response['Credentials']['AccessKeyId'],
                    aws_secret_access_key=response['Credentials']['SecretAccessKey'],
                    aws_session_token = response['Credentials']['SessionToken'])

response = dynamodb_client.get_item(
Key={
    'key1': {
        'S': '1',
    },
    'key2': {
        'S': '2',
    },
},
TableName='TestTable')
print(response)

Answer 9

Предполагая, что 1) ~/.aws/config или ~/.aws/credentials файл заполняется каждой из ролей, которые вы хотите принять, и что 2) роль по умолчанию имеет AssumeRole, определенную в ее политике IAM для каждой из этих ролей, тогда вы можете просто (в псевдокоде) сделать следующее и не должны возня с СТС:

import boto3

# get all of the roles from the AWS config/credentials file using a config file parser
profiles = get_profiles()

for profile in profiles:

    # this is only used to fetch the available regions
    initial_session = boto3.Session(profile_name=profile)

    # get the regions
    regions = boto3.Session.get_available_regions('ec2')

    # cycle through the regions, setting up session, resource and client objects
    for region in regions:
        boto3_session = boto3.Session(profile_name=profile, region_name=region)
        boto3_resource = boto3_session.resource(service_name='s3', region_name=region)
        boto3_client = boto3_sessoin.client(service_name='s3', region_name=region)

        [ do something interesting with your session/resource/client here ]

• Настройка учетных данных (boto3 - файл общих учетных данных)
• Предположительная установка ролей (AWS)

Answer 10

• Я только что создал решение в Airflow для тех, кто не использует CLI, где учетная запись службы использует предполагаемую роль. Я использую Google Secret Manager для хранения учетных данных в форме JSON. Вы можете передавать учетные данные как хотите, наша компания просто использует эту услугу.
• Я получаю секретное местоположение из переменной Airflow, поэтому вы видите, что эти начальные блоки кода происходят именно таким образом.
• Мой JSON для s3_bucket используетs3://bucket/prefix/нотации, поэтому я разделяю строку знаком '/'.
• The RoleSessionNameпараметр вassume_roleМетод может быть любой строкой, которую вы захотите. Однако,DurationSecondsимеет минимально допустимое значение 900.
• Это решение в целом будет работать при создании любых сервисов, которые обрабатывает клиент boto3. Вы просто подставите имя сервиса AWS в определение клиента вместо использования S3.

      import boto3
import logging
import json

from airflow.models import Variable
from google.cloud import secretmanager

sm_client = secretmanager.SecretManagerServiceClient()
config = Variable.get("project_variable", deserialize_json=True)
secret_location = config["s3_secret"]
secret_request = sm_client.access_secret_version(name=secret_location)
s3_creds = json.loads(secret_request.payload.data.decode("UTF-8"))

logging.info('Creating boto3 session.')
session = boto3.Session(
    aws_access_key_id=s3_creds["access_key"],
    aws_secret_access_key=s3_creds["secret_key"],
)

logging.info(f'Assuming S3 Role: {s3_creds["role_arn"]}')
sts_connection = session.client('sts')
assume_role_object = sts_connection.assume_role(RoleArn=s3_creds["role_arn"],
                                                RoleSessionName='DataEngineering',
                                                DurationSeconds=900)
credentials = assume_role_object['Credentials']

logging.info('Creating S3 resource.')
s3 = boto3.client('s3',
                  aws_access_key_id=credentials['AccessKeyId'],
                  aws_secret_access_key=credentials['SecretAccessKey'],
                  aws_session_token=credentials['SessionToken'])

bucket = s3_creds["s3_bucket"].split("/")[2]
prefix = f'{s3_creds["s3_bucket"].split("/")[3]}/'
objects = s3.list_objects_v2(Bucket=bucket, Prefix=prefix, Delimiter='/')

Answer 11

После нескольких дней поиска это самое простое решение, которое я нашел. объяснено здесь, но не имеет примера использования.

import boto3


for profile in boto3.Session().available_profiles:

    boto3.DEFAULT_SESSION = boto3.session.Session(profile_name=profile)

    s3 = boto3.resource('s3')

    for bucket in s3.buckets.all():
        print(bucket)

Это переключит роль по умолчанию, которую вы будете использовать. Чтобы профиль не использовался по умолчанию, просто не назначайте его для boto3.DEFAULT_SESSION. но вместо этого сделайте следующее.

testing_profile = boto3.session.Session(profile_name='mainTesting')
s3 = testing_profile.resource('s3')

for bucket in s3.buckets.all():
    print(bucket)

Важно отметить, что учетные данные.aws нужно устанавливать определенным образом.

[default]
aws_access_key_id = default_access_id
aws_secret_access_key = default_access_key

[main]
aws_access_key_id = main_profile_access_id
aws_secret_access_key = main_profile_access_key

[mainTesting]
source_profile = main
role_arn = Testing role arn
mfa_serial = mfa_arn_for_main_role

[mainProduction]
source_profile = main
role_arn = Production role arn
mfa_serial = mfa_arn_for_main_role

Я не знаю почему, но ключ mfa_serial должен быть в ролях, чтобы это работало, а не в исходной учетной записи, что было бы лучше.