|
#!/usr/bin/env python3
|
|
|
|
MY_ENDPOINT = 'https://my.rgw.endpoint'
|
|
MY_BUCKET = 'my_bucket'
|
|
|
|
MY_ORIGIN = 'https://my.origin'
|
|
REGIONS = ['us-east-1', 'us-east-2']
|
|
|
|
import logging
|
|
import requests
|
|
import boto3
|
|
from botocore.exceptions import ClientError
|
|
|
|
def create_presigned_url_expanded(client_method_name, method_parameters=None,
|
|
expiration=3600, http_method=None, region_name='us-west-2'):
|
|
"""Generate a presigned URL to invoke an S3.Client method
|
|
|
|
Not all the client methods provided in the AWS Python SDK are supported.
|
|
|
|
:param client_method_name: Name of the S3.Client method, e.g., 'list_buckets'
|
|
:param method_parameters: Dictionary of parameters to send to the method
|
|
:param expiration: Time in seconds for the presigned URL to remain valid
|
|
:param http_method: HTTP method to use (GET, etc.)
|
|
:return: Presigned URL as string. If error, returns None.
|
|
"""
|
|
|
|
# Generate a presigned URL for the S3 client method
|
|
s3_client = boto3.client('s3', endpoint_url=MY_ENDPOINT, region_name=region_name)
|
|
try:
|
|
response = s3_client.generate_presigned_url(ClientMethod=client_method_name,
|
|
Params=method_parameters,
|
|
ExpiresIn=expiration,
|
|
HttpMethod=http_method)
|
|
except ClientError as e:
|
|
logging.error(e)
|
|
return None
|
|
|
|
# The response contains the presigned URL
|
|
return response
|
|
|
|
headers = {
|
|
'Origin': MY_ORIGIN,
|
|
'Access-Control-Request-Method': 'PUT',
|
|
}
|
|
|
|
TEST = {
|
|
'Without ACL': {},
|
|
'With ACL': {
|
|
'ACL': 'private'
|
|
}
|
|
}
|
|
|
|
for region in REGIONS:
|
|
print(f"Region {region}")
|
|
for test in TEST:
|
|
print(test)
|
|
params = {
|
|
'Bucket': MY_BUCKET,
|
|
'Key': 'foo.png'
|
|
}
|
|
params.update(TEST[test])
|
|
presigned = create_presigned_url_expanded(
|
|
'put_object',
|
|
params,
|
|
600,
|
|
'PUT',
|
|
region,
|
|
)
|
|
print(presigned.replace(MY_ENDPOINT, "https://endpoint").replace(MY_BUCKET, "bucket"))
|
|
|
|
req = requests.options(
|
|
presigned,
|
|
headers=headers,
|
|
verify=True,
|
|
)
|
|
print(req.status_code)
|
|
|