I am writing a python function that requires temporary credentials to be rotated every N minutes. Here is how the implementation looks like:

if valid_bucket(bucket_name):
    try:
        aws_token = get_token(account_id=account_id, region="us-east-1")
        s3 = get_boto_client_with_creds('s3', aws_token=aws_token)
        print("this is s3 object", s3)
    except Exception as e:
        print("error ", e)
    bucket_parse_function(s3)

My use-case is to call the aws_token variable if bucket_parse_function() has already run for N minutes. Can anyone help me change the implementation in such a way that I am able to achieve the goal of refreshing token every N minutes of execution of bucket_parse_function().

CodePudding user response：

You would need to save the time at which you acquired the token last and then each time you run this check if the token has expired.

So something like

if valid_bucket(bucket_name):
    if token_acquired_time is None or time.time() - TOKEN_EXPIRATION_SECONDS > token_acquired_time:
        try:
            aws_token = get_token(account_id=account_id, region="us-east-1")
            s3 = get_boto_client_with_creds('s3', aws_token=aws_token)
            token_acquired_time = time.time()
            print("this is s3 object", s3)
       except Exception as e:
            print("error ", e)
     bucket_parse_function(s3)

And before all this you need to set

token_acquired_time = None 
TOKEN_EXPIRATION_SECONDS = 60 * N  # convert N minutes to seconds

somewhere where it will only run once.

Note this won't refresh the token every N minutes, but it will refresh the token the next time you attempt to use it if it is after the N minutes.

CodePudding user response：

You can use AWS STS AssumeRole API here.

From AWS docs

To assume a role, an application calls the AWS STS AssumeRole API operation and passes the ARN of the role to use. The operation creates a new session with temporary credentials. This session has the same permissions as the identity-based policies for that role.

May be you can do something like this:

import boto3
import time

# Assuming you do this with your IAM user credentials initially
bucket_name = "your-bucket-name"
def valid_bucket(bucket_name):
  s3 = boto3.client('s3')
  try:
    s3.client.head_bucket(Bucket=bucket_name)
    print("Bucket Exists!")
    return True
  except botocore.exceptions.ClientError as e:
    return False

def temp_credentials(bucket_name):
  if valid_bucket(bucket_name):
    try:
       sts_client = boto3.client('sts')
       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']
       # now acess your bucket with temporary credentials.
       # Use the temporary credentials that AssumeRole returns to make a 
       # connection to Amazon S3  
    except Exception as e:
    print("error ", e)
  return credentials
  
# get credentials here and call parse_bucket_function() for the first time
check=True
while check:
   credentials = temp_credentials()
   # create client object here
   s3_obj=boto3.client(
                       's3',
                       aws_access_key_id=credentials['AccessKeyId'],
                       aws_secret_access_key=credentials['SecretAccessKey'],
                       aws_session_token=credentials['SessionToken'],
                      )
   start = time.time()
   bucket_parse_function(s3_obj)
   elapsed = time.time()-start
   # convert N into a readable format in minutes.
   if elapsed != N:
     check=False
     break
   else:
     continue:
  

By default, your role session lasts for one hour. You can specify the value for the DurationSeconds parameter.

Reference: AWS docs