[python] using AWS_WEB_IDENTITY_TOKEN_FILE with google workload identity federation

workload_identity_aws.md

using:

python custom credential provider which allows for arbitrary aws source credential usage (eg AWS_WEB_IDENTITY_TOKEN_FILE)

• https://github.com/googleapis/google-auth-library-python/blob/main/docs/user-guide.rst#accessing-resources-using-a-custom-supplier-with-aws

also see in golang here

to use this, you may have en-vars setup like so

export AWS_WEB_IDENTITY_TOKEN_FILE=/tmp/aws.txt
export AWS_ROLE_ARN="arn:aws:iam::291738886548:role/cicps3role"
export AWS_ROLE_SESSION_NAME=mysession

and a gcp workload identity provider as shown below along with an iam binding which authorizes a session (eg, for gcs)

gcloud storage buckets add-iam-policy-binding  gs://$PROJECT_ID-bucket \
   --member="principal://iam.googleapis.com/projects/995081019036/locations/global/workloadIdentityPools/aws-pool-1/subject/arn:aws:sts::291738886548:assumed-role/cicps3role/mysession" \
    --role="roles/storage.objectViewer"

from google.auth import aws
from google.auth import exceptions
import boto3
from google.cloud import storage

# client = boto3.client('sts')
# response = client.get_caller_identity()
# print(response)

class CustomAwsSecurityCredentialsSupplier(aws.AwsSecurityCredentialsSupplier):
    def get_aws_security_credentials(self, context, request):
        try:
            session = boto3.Session()
            credentials = session.get_credentials()
            return aws.AwsSecurityCredentials(credentials.access_key, credentials.secret_key, credentials.token)
        except Exception as e:
            raise exceptions.RefreshError(e, retryable=True)

    def get_aws_region(self, context, request):
        return 'us-east-1'

supplier = CustomAwsSecurityCredentialsSupplier()

audience = "//iam.googleapis.com/projects/995081019036/locations/global/workloadIdentityPools/aws-pool-1/providers/aws-provider-1"
subject_token_type = "urn:ietf:params:aws:token-type:aws4_request"
credentials = aws.Credentials(
    audience,
    subject_token_type,
    aws_security_credentials_supplier=supplier
)

# credentials, project = google.auth.default()    
client = storage.Client(project="core-eso",credentials=credentials)
blobs = client.list_blobs("core-eso-bucket")

for blob in blobs:
    print(blob.name)