Python requests lib based communication to AWS EKS K8S cluster

main.py

from botocore.signers import RequestSigner
import boto3
import re
import base64
import requests
import json
import sys
import sigv4
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def get_bearer_token(cluster_name: str, region: str="eu-central-1") -> str:
    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_name
        },
        'context': {}
    }

    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:
    return 'k8s-aws-v1.' + re.sub(r'=*', '', base64_url)


def main():
    cluster_name = "my-cluster-name"
    cluster_endpoint = "https://<MY_CLUSTER_ENDPOINT>.eks.amazonaws.com"

    stream = False  # For watch requests

    # Takes 1 arg from command line
    # Queries the namespaces by default if nothing supplied
    try:
        api_call = sys.argv[1]
    except IndexError:
        api_call = 'namespaces'

    api_path = f'api/v1/{api_call}'
    target = f'{cluster_endpoint}/{api_path}'
    headers = {'Authorization': 'Bearer ' + get_bearer_token(cluster_name)}

    if stream:
        tc = 0
        with requests.get(target, verify=False, headers=headers, stream=True) as s:
            s.raise_for_status()
            for chunk in s.iter_content(chunk_size=8193):
                print(str(chunk, 'utf-8'))
                print(f"{len(chunk)} | t {tc}")
                tc += len(chunk)
    else:
        resp = requests.get(target, verify=False, headers=headers)
        try:
            print(json.dumps(resp.json(), indent=4))
            print(len(resp.content))
        except Exception as e:
            print(e)
            print(resp.text)
    return

if __name__ == "__main__":
    main()