rTtzYT43

1import requests, boto3, base64, hashlib
2
3access_key = "..."
4secret_key = "..."
5bucket = "..."
6s3 = boto3.client('s3',
7                  aws_access_key_id=access_key,
8                  aws_secret_access_key=secret_key)
9
10test_encrypt_key = os.urandom(32)
11test_uuid_file = str(uuid.uuid4())
12
13def get_url(filename):
14    return s3.generate_presigned_url(
15        ClientMethod='get_object',
16        Params={
17            'Bucket': bucket,
18            'Key': filename,
19            'SSECustomerAlgorithm': 'AES256',
20        }
21    )
22
23
24def post_url(filename):
25    return s3.generate_presigned_url(
26        ClientMethod='put_object',
27        Params={
28            'Bucket': bucket,
29            'Key': filename,
30            'SSECustomerAlgorithm': 'AES256',
31        }
32    )
33
34def test_uploading_with_signed_urls():
35    encoded_key = base64.b64encode(test_encrypt_key)
36    md5_key = base64.b64encode(hashlib.md5(test_encrypt_key).digest())
37
38    post_address = post_url(test_uuid_file)
39    print("POST: %s" % (post_address))
40
41    resp = requests.put(post_address, data="hello world", headers={
42        "x-amz-server-side-encryption-customer-algorithm": "AES256",
43        "x-amz-server-side-encryption-customer-key": encoded_key,
44        "x-amz-server-side-encryption-customer-key-MD5": md5_key
45    })
46
47    resp.raise_for_status()
48
49
50
51    get_address = get_url(test_uuid_file)
52    print("GET: %s" % (get_address))
53
54    resp = requests.get(get_address, headers={
55        "x-amz-server-side-encryption-customer-algorithm": "AES256",
56        "x-amz-server-side-encryption-customer-key": encoded_key,
57        "x-amz-server-side-encryption-customer-key-MD5": md5_key
58    })
59
60    resp.raise_for_status()
61    print("Got Payload: %s" % resp.content)
62
63if __name__ == "__main__":
64    test_uploading_with_signed_urls()