eksupdate/src/eksctlfinal.py [35:92]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
def get_bearer_token(cluster_id, region):
    '''' AUthenticating the session with sts token'''
    STS_TOKEN_EXPIRES_IN = 60
    session = boto3.session.Session()

    client = session.client('sts', region_name=region)
    service_id = client.meta.service_model.service_id

    signer = RequestSigner(
        service_id,
        region,
        'sts',
        'v4',
        session.get_credentials(),
        session.events
    )

    params = {
        'method': 'GET',
        'url': 'https://sts.{}.amazonaws.com/?Action=GetCallerIdentity&Version=2011-06-15'.format(region),
        'body': {},
        'headers': {
            'x-k8s-aws-id': cluster_id
        },
        'context': {}
    }
    '''Getting a presigned Url'''

    signed_url = signer.generate_presigned_url(
        params,
        region_name=region,
        expires_in=STS_TOKEN_EXPIRES_IN,
        operation_name=''
    )

    base64_url = base64.urlsafe_b64encode(
        signed_url.encode('utf-8')).decode('utf-8')

    # remove any base64 encoding padding and returing the kubernets token
    return 'k8s-aws-v1.' + re.sub(r'=*', '', base64_url)


def loading_config(cluster_name,regionName):
    ''' loading kubeconfig with sts'''
    eks = boto3.client('eks',region_name=regionName)
    resp = eks.describe_cluster(

        name=cluster_name
    )
    endPoint = resp["cluster"]["endpoint"]
    configs = client.Configuration()
    configs.host = endPoint
    configs.verify_ssl = False
    configs.debug = False 
    configs.api_key = {"authorization": "Bearer " +
                       get_bearer_token(cluster_name, regionName)}
    client.Configuration.set_default(configs)
    return "Initialiazed"
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



eksupdate/src/k8s_client.py [32:89]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
def get_bearer_token(cluster_id, region):
    '''' AUthenticating the session with sts token'''
    STS_TOKEN_EXPIRES_IN = 60
    session = boto3.session.Session()

    client = session.client('sts', region_name=region)
    service_id = client.meta.service_model.service_id

    signer = RequestSigner(
        service_id,
        region,
        'sts',
        'v4',
        session.get_credentials(),
        session.events
    )

    params = {
        'method': 'GET',
        'url': 'https://sts.{}.amazonaws.com/?Action=GetCallerIdentity&Version=2011-06-15'.format(region),
        'body': {},
        'headers': {
            'x-k8s-aws-id': cluster_id
        },
        'context': {}
    }
    '''Getting a presigned Url'''

    signed_url = signer.generate_presigned_url(
        params,
        region_name=region,
        expires_in=STS_TOKEN_EXPIRES_IN,
        operation_name=''
    )

    base64_url = base64.urlsafe_b64encode(
        signed_url.encode('utf-8')).decode('utf-8')

    # remove any base64 encoding padding and returing the kubernets token
    return 'k8s-aws-v1.' + re.sub(r'=*', '', base64_url)


def loading_config(cluster_name,regionName):
    ''' loading kubeconfig with sts'''
    eks = boto3.client('eks',region_name=regionName)
    resp = eks.describe_cluster(

        name=cluster_name
    )
    endPoint = resp["cluster"]["endpoint"]
    configs = client.Configuration()
    configs.host = endPoint
    configs.verify_ssl = False
    configs.debug = False 
    configs.api_key = {"authorization": "Bearer " +
                       get_bearer_token(cluster_name, regionName)}
    client.Configuration.set_default(configs)
    return "Initialiazed"
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -