testing CMRESHandler with RefreshableCredentials

test.py

from cmreslogging.handlers import CMRESHandler
import logging
import os
import sys
import time
import boto3
from imdb import IMDb


from botocore.credentials import RefreshableCredentials
try:
    from datetime import timezone
except ImportError:
    pass

from datetime import datetime, timedelta
from botocore.session import get_session
from botocore.session import get_session
from boto3 import Session


################################################################################################
##                  References    
################################################################################################
# https://realpython.com/pypi-publish-python-package/
# https://dev.to/li_chastina/auto-refresh-aws-tokens-using-iam-role-and-boto3-2cjf
# https://www.owenrumney.co.uk/implementing-refreshingawscredentials-python/
################################################################################################


AWS_ACCESS_KEY_ID=os.environ['AWS_ACCESS_KEY_ID']
AWS_SECRET_ACCESS_KEY=os.environ['AWS_SECRET_ACCESS_KEY']
AWS_SESSION_TOKEN=os.environ['AWS_SESSION_TOKEN']
AWS_REGION='us-west-2'
HOSTS=[{'host': 'my-es-host.us-west-2.es.amazonaws.com', 'port': 443}]
# HOSTS=[{'host': 'localhost', 'port': 9200}]



def refresh_external_credentials():
    " Refresh tokens by calling assume_role again "
    ROLE_ARN = "arn:aws:iam::012345678901:role/testpackage"
    AWS_REGION='us-west-2'
    params = {
        "RoleArn": ROLE_ARN,
        "RoleSessionName": "testpackagesession",
        "DurationSeconds": 3600,
        "ExternalId": "992644"
    }    
    sts_client = boto3.client("sts", region_name=AWS_REGION)
    response = sts_client.assume_role(**params).get("Credentials")
    print(response)
    credentials = {
        "access_key": response.get("AccessKeyId"),
        "secret_key": response.get("SecretAccessKey"),
        "token": response.get("SessionToken"),
        "expiry_time": response.get("Expiration").isoformat(),
    }
    return credentials



session_credentials = RefreshableCredentials.create_from_metadata(
    metadata = refresh_external_credentials(),
    refresh_using = refresh_external_credentials,
    method = 'sts-assume-role'
)

initial_credentials = refresh_external_credentials()


if session_credentials:
    AWS_ACCESS_KEY_ID=initial_credentials['access_key']
    AWS_SECRET_ACCESS_KEY=initial_credentials['secret_key']
    AWS_SESSION_TOKEN=initial_credentials['token']


######################################################################
# Create and Configure Elasticsearch logging handler
######################################################################
es_handler = CMRESHandler( hosts=HOSTS,
                        # auth_type=CMRESHandler.AuthType.NO_AUTH,
                        # auth_type=CMRESHandler.AuthType.BASIC_AUTH,
                        # auth_details=('admin','admin'),
                        # auth_type=CMRESHandler.AuthType.AWS_SIGNED_AUTH,
                        # aws_access_key=AWS_ACCESS_KEY_ID,
                        # aws_secret_key=AWS_SECRET_ACCESS_KEY,
                        # aws_session_token=AWS_SESSION_TOKEN,
                        auth_type=CMRESHandler.AuthType.AWS_REFRESHABLE_CREDENTIALS,
                        aws_region=AWS_REGION,
                        aws_refreshable_credentials=session_credentials,
                        use_ssl=True,
                        verify_ssl=False,
                        es_additional_fields={'App': 'TestApp', 'Environment': 'Dev'},
                        es_index_name="mylogs")

es_handler.setLevel(logging.DEBUG)

######################################################################
# Create and Configure stdout logging handler
######################################################################
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
# stdout_handler = logging.StreamHandler(sys.stdout)
# stdout_handler.setLevel(logging.DEBUG)



######################################################################
# Configure the root logger add the handlers to the root logger
######################################################################
testlogger = logging.getLogger('testlogger')
testlogger.setLevel(logging.DEBUG)
# testlogger.addHandler(stdout_handler)
testlogger.addHandler(es_handler)



######################################################################
# Test print and Python logging 
######################################################################
# print("hello world")
testlogger.debug("hello stdout world")
testlogger.info("hello AWS world")
# logging.debug("hello DEBUG world using the root logger")


for x in range(100):
    time.sleep(0.5)
    testlogger.info("Iteration: {}".format(x))
    testlogger.info("Adrian is awesome")



exit()