如何使用Python正确刷新AWS凭证

huangapple go评论68阅读模式
英文:

How to correctly refresh aws credentials with Python

问题

I'm trying to use the RefreshableCredentials module from botocore in order to manage automatically the credentials update.

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

def get_aws_credentials(aws_role_arn, session_name):
    sts_client = boto3.client('sts')
    assumed_role_object = sts_client.assume_role(
        RoleArn = aws_role_arn,
        RoleSessionName = session_name,
        DurationSeconds = 900
    )
    return {
        'access_key': assumed_role_object['Credentials']['AccessKeyId'],
        'secret_key': assumed_role_object['Credentials']['SecretAccessKey'],
        'token': assumed_role_object['Credentials']['SessionToken'],
        'expiry_time': assumed_role_object['Credentials']['Expiration'].isoformat()
    }

def get_aws_autorefresh_session(aws_role_arn, session_name):
    session_credentials = RefreshableCredentials.create_from_metadata(
        metadata = get_aws_credentials(aws_role_arn, session_name),
        refresh_using = get_aws_credentials,
        method = 'sts-assume-role'
    )

    session = get_session()
    session._credentials = session_credentials
    autorefresh_session = boto3.Session(botocore_session=session)

    return autorefresh_session, session_credentials

Generating the credentials like this:

arn = "1234"
session = "Test"
session, credentials = get_aws_autorefresh_session(arn, session)

And then I'm passing the session_credentials from get_aws_autorefresh_session to whatever function may need them.

With this approach, I've noticed that everything works, but after 300 seconds this exception is raised:

get_aws_credentials() missing 2 required positional arguments: 'aws_role_arn' and 'session_name'

On the contrary, if I modify the function get_aws_credentials eliminating the variables, and passing static values for them:

def get_aws_credentials():
    sts_client = boto3.client('sts')
    assumed_role_object = sts_client.assume_role(
        RoleArn = "1234",
        RoleSessionName = "Test",
        DurationSeconds = 900
    )
    return {
        'access_key': assumed_role_object['Credentials']['AccessKeyId'],
        'secret_key': assumed_role_object['Credentials']['SecretAccessKey'],
        'token': assumed_role_object['Credentials']['SessionToken'],
        'expiry_time': assumed_role_object['Credentials']['Expiration'].isoformat()
    }

def get_aws_autorefresh_session():
    session_credentials = RefreshableCredentials.create_from_metadata(
        metadata = get_aws_credentials(),
        refresh_using = get_aws_credentials,
        method = 'sts-assume-role'
    )

    session = get_session()
    session._credentials = session_credentials
    autorefresh_session = boto3.Session(botocore_session=session)

    return autorefresh_session, session_credentials

Everything works smoothly.

My question is how to retrieve the credentials using variables for the role_arn.

英文:

I'm trying to use the RefreshableCredentials module from botocore in order to manage automatically the credentials update.

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

def get_aws_credentials(aws_role_arn, session_name):
    sts_client = boto3.client('sts')
    assumed_role_object = sts_client.assume_role(
        RoleArn = aws_role_arn,
        RoleSessionName = session_name,
        DurationSeconds = 900
    )
    return {
        'access_key': assumed_role_object['Credentials']['AccessKeyId'],
        'secret_key': assumed_role_object['Credentials']['SecretAccessKey'],
        'token': assumed_role_object['Credentials']['SessionToken'],
        'expiry_time': assumed_role_object['Credentials']['Expiration'].isoformat()
    }

def get_aws_autorefresh_session(aws_role_arn, session_name):
    session_credentials = RefreshableCredentials.create_from_metadata(
        metadata = get_aws_credentials(aws_role_arn, session_name),
        refresh_using = get_aws_credentials,
        method = 'sts-assume-role'
    )

    session = get_session()
    session._credentials = session_credentials
    autorefresh_session = boto3.Session(botocore_session=session)

    return autorefresh_session, session_credentials

Generating the credentials like this:

arn = "1234"
session = "Test"
session, credentials = get_aws_autorefresh_session(arn, session)

And then I'm passing the session_credentials from get_aws_autorefresh_session to wathever function may need them.

With this approach, I've noticed that everything works, but after 300 seconds this exception is raised:

get_aws_credentials() missing 2 required positional arguments: 'aws_role_arn' and 'session_name'

On the contrary, if I modify the function get_aws_credentials eliminating the variables, and passing static values for them:

def get_aws_credentials():
    sts_client = boto3.client('sts')
    assumed_role_object = sts_client.assume_role(
        RoleArn = "1234",
        RoleSessionName = "Test",
        DurationSeconds = 900
    )
    return {
        'access_key': assumed_role_object['Credentials']['AccessKeyId'],
        'secret_key': assumed_role_object['Credentials']['SecretAccessKey'],
        'token': assumed_role_object['Credentials']['SessionToken'],
        'expiry_time': assumed_role_object['Credentials']['Expiration'].isoformat()
    }

def get_aws_autorefresh_session():
    session_credentials = RefreshableCredentials.create_from_metadata(
        metadata = get_aws_credentials(),
        refresh_using = get_aws_credentials,
        method = 'sts-assume-role'
    )

    session = get_session()
    session._credentials = session_credentials
    autorefresh_session = boto3.Session(botocore_session=session)

    return autorefresh_session, session_credentials

Everything works smoothly.

My question is how to retrieve the credentials using variables for the role_arn.

答案1

得分: 1

你可以使用 functools.partial 来创建一个偏函数,像这样:

from functools import partial

...

session_credentials = RefreshableCredentials.create_from_metadata(
        metadata = get_aws_credentials(aws_role_arn, session_name),
        refresh_using = partial(get_aws_credentials, aws_role_arn, session_name),
        method = 'sts-assume-role'
)
英文:

You can create a partial function using functools.partial like this:

from functools import partial

...

session_credentials = RefreshableCredentials.create_from_metadata(
        metadata = get_aws_credentials(aws_role_arn, session_name),
        refresh_using = partial(get_aws_credentials, aws_role_arn, session_name),
        method = 'sts-assume-role'
)

答案2

得分: 0

这是一个适合Lambda用例的好选择。

session_credentials = RefreshableCredentials.create_from_metadata(
    metadata = get_aws_credentials(aws_role_arn, session_name),
    refresh_using = lambda: get_aws_credentials(aws_role_arn, session_name),
    method = 'sts-assume-role',
)
英文:

That's a good fit for a lambda usecase

session_credentials = RefreshableCredentials.create_from_metadata(
    metadata = get_aws_credentials(aws_role_arn, session_name),
    refresh_using = lambda: get_aws_credentials(aws_role_arn, session_name),
    method = 'sts-assume-role',
)

huangapple
  • 本文由 发表于 2023年2月24日 17:18:30
  • 转载请务必保留本文链接:https://go.coder-hub.com/75554694.html
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定