Skip to content

Instantly share code, notes, and snippets.

@reizist
Last active October 5, 2023 06:33
Show Gist options
  • Star 1 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save reizist/e7dfc77dc5b0267b4083044356fb77cc to your computer and use it in GitHub Desktop.
Save reizist/e7dfc77dc5b0267b4083044356fb77cc to your computer and use it in GitHub Desktop.
run aws command on gcp with keyless"
import boto3
import json
import os
from botocore.credentials import Credentials
from google.oauth2 import id_token
from google.oauth2 import service_account
import google.auth
import google.auth.transport.requests
ASSUME_ROLE_ARN = os.environ.get("ASSUME_ROLE_ARN")
TARGET_AUDIENCE = "sts.amazonaws.com"
S3_BUCKET_NAME = os.environ.get("S3_BUCKET")
def get_metadata(path: str, parameter: str):
metadata_url = 'http://metadata.google.internal/computeMetadata/v1/{}/{}'.format(path, parameter)
headers = {'Metadata-Flavor': 'Google'}
try:
meta_request = requests.get(metadata_url, headers=headers)
except requests.exceptions.RequestException as e:
raise SystemExit(e)
if meta_request.ok:
return meta_request.text
else:
raise SystemExit('Compute Engine meta data error')
def get_id_token_via_metadata():
return get_metadata('instance', 'service-accounts/default/identity?format=standard&audience={}'.format(TARGET_AUDIENCE))
def get_id_token():
creds = os.environ.get("SA")
info = json.loads(creds)
creds = service_account.IDTokenCredentials.from_service_account_info(
info,
target_audience=TARGET_AUDIENCE)
request = google.auth.transport.requests.Request()
creds.refresh(request)
return creds.token
def verify_token(token: str, audience: str) -> dict:
request = google.auth.transport.requests.Request()
payload = id_token.verify_token(token, request=request, audience=audience)
return payload['email_verified']
def command(token):
assumed_role_object = sts.assume_role_with_web_identity(
RoleArn=ASSUME_ROLE_ARN,
RoleSessionName="AssumeRoleSession1",
WebIdentityToken=token,
DurationSeconds=900
)
credentials = assumed_role_object['Credentials']
s3_resource = boto3.resource(
's3',
aws_access_key_id=credentials['AccessKeyId'],
aws_secret_access_key=credentials['SecretAccessKey'],
aws_session_token=credentials['SessionToken'],
)
bkt = s3_resource.Bucket(S3_BUCKET_NAME)
for my_bucket_object in bkt.objects.all():
print(my_bucket_object)
sts = boto3.client('sts', aws_access_key_id='', aws_secret_access_key='')
token = get_id_token()
if verify_token(token, TARGET_AUDIENCE):
command(token)
else:
print('Verify failed.')
boto3
google-auth
@reizist
Copy link
Author

reizist commented Jun 29, 2022

❯ ASSUME_ROLE_ARN=arn:aws:iam::xxxx:role/your_role S3_BUCKET=your_s3_bucket python s3_ls.py
s3.ObjectSummary(bucket_name='your_s3_bucket', key='xxx')

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment