|
|
|
import boto3
|
|
|
|
import logging
|
|
logging.basicConfig()
|
|
logging.getLogger().setLevel("DEBUG")
|
|
|
|
iam_client = boto3.client('iam',
|
|
aws_access_key_id=<access key of bucket owner (TESTER)>,
|
|
aws_secret_access_key=<secret key of bucket owner (TESTER)>
|
|
endpoint_url="https://object.example.com",
|
|
region_name='RegionOne'
|
|
)
|
|
|
|
policy_document = '''{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"AWS":["arn:aws:iam::a4c52e22f45b4a0fa3ea3b9a42b35808:root"]},"Action":["sts:AssumeRole"]}]}'''
|
|
|
|
role_response = iam_client.create_role(
|
|
AssumeRolePolicyDocument=policy_document,
|
|
Path='/',
|
|
RoleName='S3Access',
|
|
)
|
|
|
|
role_policy = '''{"Version":"2012-10-17","Statement":{"Effect":"Allow","Action":"s3:*","Resource":"*"}}'''
|
|
|
|
response = iam_client.put_role_policy(
|
|
RoleName='S3Access',
|
|
PolicyName='Policy1',
|
|
PolicyDocument=role_policy
|
|
)
|
|
|
|
sts_client = boto3.client('sts',
|
|
aws_access_key_id=<access key of bucket accessor (TESTER1)>,
|
|
aws_secret_access_key=<secret key of bucket accessor (TESTER1)>
|
|
endpoint_url="https://object.example.com",
|
|
region_name='RegionOne',
|
|
)
|
|
|
|
response = sts_client.assume_role(
|
|
RoleArn=role_response['Role']['Arn'],
|
|
RoleSessionName='Bob',
|
|
DurationSeconds=3600
|
|
)
|
|
|
|
s3client = boto3.client('s3',
|
|
aws_access_key_id = response['Credentials']['AccessKeyId'],
|
|
aws_secret_access_key = response['Credentials']['SecretAccessKey'],
|
|
aws_session_token = response['Credentials']['SessionToken'],
|
|
endpoint_url="https://object.example.com",
|
|
region_name='RegionOne',)
|
|
|
|
# Check we can list objects in a bucket
|
|
bucket_name = "cross-account-poc"
|
|
resp = s3client.list_objects_v2(Bucket=bucket_name)
|
|
print("Bucket objects:")
|
|
for object in resp["Contents"]:
|
|
print(object["Key"])
|
|
|
|
# Test listing all buckets
|
|
resp = s3client.list_buckets()
|
|
bucket_list = resp["Buckets"]
|
|
|
|
s3_client_2 = boto3.client('s3',
|
|
aws_access_key_id=<access key of bucket owner (TESTER)>,
|
|
aws_secret_access_key=<secret key of bucket owner (TESTER)>
|
|
endpoint_url="https://object.example.com",
|
|
region_name='RegionOne')
|
|
|
|
resp = s3_client_2.list_buckets()
|
|
bucket_list_2 = resp["Buckets"]
|
|
|
|
print("Checking bucket list comparison")
|
|
assert(bucket_list == bucket_list_2)
|