How to refresh the boto3 credentials when python script is running indefinitely

Question

I am trying to write a python script that uses watchdog to look for file creation and upload that to s3 using boto3. However, my boto3 credentials expire after every 12hrs, So I need to renew them. I am storing my boto3 credentials in ~/.aws/credentials. So right now I am trying to catch the S3UploadFailedError, renew the credentials, and write them to ~/.aws/credentials. But though the credentials are getting renewed and I am calling boto3.client('s3') again its throwing exception.

What am I doing wrong? Or how can I resolve it?

Below is the code snippet

try:
     s3 = boto3.client('s3')
     s3.upload_file(event.src_path,'bucket-name',event.src_path)

except boto3.exceptions.S3UploadFailedError as e:
     print(e)
     get_aws_credentials()
     s3 = boto3.client('s3')

Answer 1

According to the documentation, the client looks in several locations for credentials and there are other options that are also more programmatic-friendly that you might want to consider instead of the .aws/credentials file.

Quoting the docs:

The order in which Boto3 searches for credentials is:

• Passing credentials as parameters in the boto.client() method
• Passing credentials as parameters when creating a Session object
• Environment variables
• Shared credential file (~/.aws/credentials)
• AWS config file (~/.aws/config)
• Assume Role provider

In your case, since you are already catching the exception and renewing the credentials, I would simply pass the new ones to a new instance of the client like so:

client = boto3.client(
    's3',
    aws_access_key_id=NEW_ACCESS_KEY,
    aws_secret_access_key=NEW_SECRET_KEY,
    aws_session_token=NEW_SESSION_TOKEN
)

If instead you are using these same credentials elsewhere in the code to create other clients, I'd consider setting them as environment variables:

import os

os.environ['AWS_ACCESS_KEY_ID'] = NEW_ACCESS_KEY
os.environ['AWS_SECRET_ACCESS_KEY'] = NEW_SECRET_KEY
os.environ['AWS_SESSION_TOKEN'] = NEW_SESSION_TOKEN

Again, quoting the docs:

The session key for your AWS account [...] is only needed when you are using temporary credentials.

Answer 2

I have found a good example to refresh the credentials within this link: https://pritul95.github.io/blogs/boto3/2020/08/01/refreshable-boto3-session/

but there's a little bug inside. Be careful about that. Here is the corrected code:

from uuid import uuid4
from datetime import datetime
from time import time

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


class RefreshableBotoSession:
    """
    Boto Helper class which lets us create a refreshable session so that we can cache the client or resource.

    Usage
    -----
    session = RefreshableBotoSession().refreshable_session()

    client = session.client("s3") # we now can cache this client object without worrying about expiring credentials
    """

    def __init__(
        self,
        region_name: str = None,
        profile_name: str = None,
        sts_arn: str = None,
        session_name: str = None,
        session_ttl: int = 3000
    ):
        """
        Initialize `RefreshableBotoSession`

        Parameters
        ----------
        region_name : str (optional)
            Default region when creating a new connection.

        profile_name : str (optional)
            The name of a profile to use.

        sts_arn : str (optional)
            The role arn to sts before creating a session.

        session_name : str (optional)
            An identifier for the assumed role session. (required when `sts_arn` is given)

        session_ttl : int (optional)
            An integer number to set the TTL for each session. Beyond this session, it will renew the token.
            50 minutes by default which is before the default role expiration of 1 hour
        """

        self.region_name = region_name
        self.profile_name = profile_name
        self.sts_arn = sts_arn
        self.session_name = session_name or uuid4().hex
        self.session_ttl = session_ttl

    def __get_session_credentials(self):
        """
        Get session credentials
        """
        session = Session(region_name=self.region_name, profile_name=self.profile_name)

        # if sts_arn is given, get credential by assuming the given role
        if self.sts_arn:
            sts_client = session.client(service_name="sts", region_name=self.region_name)
            response = sts_client.assume_role(
                RoleArn=self.sts_arn,
                RoleSessionName=self.session_name,
                DurationSeconds=self.session_ttl,
            ).get("Credentials")

            credentials = {
                "access_key": response.get("AccessKeyId"),
                "secret_key": response.get("SecretAccessKey"),
                "token": response.get("SessionToken"),
                "expiry_time": response.get("Expiration").isoformat(),
            }
        else:
            session_credentials = session.get_credentials().get_frozen_credentials()
            credentials = {
                "access_key": session_credentials.access_key,
                "secret_key": session_credentials.secret_key,
                "token": session_credentials.token,
                "expiry_time": datetime.fromtimestamp(time() + self.session_ttl).replace(tzinfo=pytz.utc).isoformat(),
            }

        return credentials

    def refreshable_session(self) -> Session:
        """
        Get refreshable boto3 session.
        """
        # Get refreshable credentials
        refreshable_credentials = RefreshableCredentials.create_from_metadata(
            metadata=self.__get_session_credentials(),
            refresh_using=self.__get_session_credentials,
            method="sts-assume-role",
        )

        # attach refreshable credentials current session
        session = get_session()
        session._credentials = refreshable_credentials
        session.set_config_variable("region", self.region_name)
        autorefresh_session = Session(botocore_session=session)

        return autorefresh_session

Note that you need pytz to be installed.

Answer 3

1. create boto3 session
2. replace it's botocore credential with DeferredRefreshableCredentials

from botocore.credentials import create_assume_role_refresher as carr
from botocore.credentials import DeferredRefreshableCredentials as DRC
from boto3 import Session

session = Session(region_name='us-east-1')
session._session._credentials=DRC(
            refresh_using=carr(session.client("sts"),
                               {'RoleArn':'your arn',
                               'RoleSessionName':'your name'}),
            method='sts-assume-role')

Answer 4

Here is my implementation which only generates new credentials if existing credentials expire using a singleton design pattern

import boto3
from datetime import datetime
from dateutil.tz import tzutc
import os
import binascii


class AssumeRoleProd:
    __credentials = None

    def __init__(self):
        assert True==False

    @staticmethod
    def __setCredentials():
        print("\n\n ======= GENERATING NEW SESSION TOKEN ======= \n\n")
        # 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=your_role_here,
            RoleSessionName=f"AssumeRoleSession{binascii.b2a_hex(os.urandom(15)).decode('UTF-8')}"
        )

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

    @staticmethod
    def getTempCredentials():
        credsExpired = False

        # Return object for the first time 
        if AssumeRoleProd.__credentials is None:
            AssumeRoleProd.__setCredentials()
            credsExpired = True

        # Generate if only 5 minutes are left for expiry. You may setup for entire 60 minutes by catching botocore ClientException
        elif (AssumeRoleProd.__credentials['Expiration']-datetime.now(tzutc())).seconds//60<=5: 
            AssumeRoleProd.__setCredentials()
            credsExpired = True

        return AssumeRoleProd.__credentials

And then I am using singleton design pattern for client as well which would generate a new client only if new session is generated. You can add region as well if required.

class lambdaClient:
    __prodClient = None

    def __init__(self):
        assert True==False

    @staticmethod
    def __initProdClient():
        credsExpired, credentials =  AssumeRoleProd.getTempCredentials()
        if lambdaClient.__prodClient is None or credsExpired:
            lambdaClient.__prodClient = boto3.client('lambda',
                                                       aws_access_key_id=credentials['AccessKeyId'],
                                                       aws_secret_access_key=credentials['SecretAccessKey'],
                                                       aws_session_token=credentials['SessionToken'])
        return lambdaClient.__prodClient


    @staticmethod
    def getProdClient():
        return lambdaClient.__initProdClient()

Answer 5

Why does this happen?

Upon looking at boto code we can see the problem. the boto3.client(..) function calls _get_default_session(..) (line no 92) where we can see that DEFAULT_SESSION is instantiated just once (line no 80) and afterwards same session is always returned (line no 79 and line no 83)

Solution

1. I manually read ~/.aws/credentials file and pass aws_access_key_id, aws_secret_access_key & aws_session_token while instantiating boto3 client
2. instantiate boto3 client on every call
   • in my case no of calls to boto3 were relatively few
   • otherwise you can instantiate client only when you get an exception
3. I created and ran a small bash script in background that keeps refreshing AWS credentials and updating the ~/.aws/credentials file
4. My python script was running with 30 parallel processes and since I was reading ~/.aws/credentials file which would itself be updated by bash script every 10 minutes, there was a risk that read-during-write could cause disruptions. So i just added some very basic retry logic in my python script so that it tries every failed operation at least 3 times. That ways it became resilient to read-during-write problem.

Code-snippets

Here's an excerpt from my Python script where I was reading AWS credentials file while instantiating client

import boto3
from typing import Dict, List, Optional, Callable, Tuple, Any, Union
import configparser

...

# helper method to read AWS Credentials file (which is in a standard INI file format)
def _read_aws_credentials():
    config: configparser.ConfigParser = configparser.ConfigParser()
    config.read('/home/alias/.aws/credentials')
    return config

...

    # read AWS credentials from credential file
    aws_credentials = _read_aws_credentials()
    # use credentials values read above to instantiate client
    client = boto3.client(
        "logs",
        region_name='us-east-1',
        aws_access_key_id=aws_credentials.get("default", "aws_access_key_id"),
        aws_secret_access_key=aws_credentials.get("default", "aws_secret_access_key"),
        aws_session_token=aws_credentials.get("default", "aws_session_token"))

(i can't share credentials refresh Bash script since it uses proprietary tools)

References

I used linux screens to run both my long-running Python script (ran for ~ 2 days) and AWS credentials refresh bash script

Here are some StackOverflow threads that i referenced

1. Best way to read aws credentials file

2. How to read and write INI file with Python3?