Ceph » rgw

Pritha Srivastava wrote:

The AssumeRole functionality has been tested for local RGW users and not for external Openstack Keystone users. Also the format of Principal for a user : arn:aws:iam::${PROJECT_ID}:root) is not supported in RGW.

Does that mean this still works for local RGW users? As in running the script below doesn't raise an AssertionError? I've attached the log output when I run it for comparison, and also the version of the same script I'm using.

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=<IAM_URL>,
region_name='RegionOne'
)

policy_document = '''{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"AWS":["arn:aws:iam:::user/TESTER1"]},"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=<STS_URL>,
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=<S3_URL>,
region_name='RegionOne',)

# Check we can list objects in a bucket
bucket_name = "my-bucket" 
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=<S3_URL>,
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)