access_key = 'AK!!!!!!!!!!!!!!!!7A'
secret_key = 'A!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!Qg'
from datetime import datetime, timedelta
from base64 import b64encode
from json import dumps
import hmac
import hashlib
import requests
class S3Policy:
def __init__(self, bucket, key_prefix, access_key, content_max_size=4048576, acl='public-read', time_frame=120):
self.bucket_url = 'https://{}.s3.amazonaws.com'.format(bucket)
self.bucket = bucket
self.acl = acl
self.access_key = access_key
self.key_prefix = key_prefix
self.time_frame = time_frame
self.content_max_size = content_max_size
def generate_policy_str(self):
expiration = (datetime.utcnow() + timedelta(seconds=self.time_frame)).strftime('%Y-%m-%dT%H:%M:%S.000Z')
policy_document = {
"expiration": expiration,
"conditions": [
{"bucket": self.bucket},
["starts-with", "$key", self.key_prefix],
{"acl": self.acl},
{"success_action_status": "201"},
["starts-with", "$Content-Type", ""],
["content-length-range", 0, self.content_max_size]
]
}
self.expiration = expiration
self.policy = b64encode((dumps(policy_document).replace('\n', '').replace('\r', '')).encode('ascii'))
def signed(self, secret_key):
self.generate_policy_str()
signed = b64encode(hmac.new(str.encode(secret_key), self.policy, hashlib.sha1).digest()).decode('unicode_escape')
to_unicode = lambda s: s if type(s)==str else s.decode('unicode_escape')
return {
"policy": to_unicode(self.policy),
"signature": signed,
"bucket_url": to_unicode(self.bucket_url),
"AWSAccessKeyId": to_unicode(self.access_key),
"acl": to_unicode(self.acl),
"key_prefix": to_unicode(self.key_prefix),
}
policy = S3Policy(bucket='my-bucket-name', key_prefix='media/tmp/test.png', access_key=access_key).signed(secret_key=secret_key)
policy
{'AWSAccessKeyId': 'AK!!!!!!!!!!!!!!!7A',
'acl': 'public-read',
'bucket_url': 'https://my-bucket-name.s3.amazonaws.com',
'key_prefix': 'media/tmp/test.png',
'policy': 'eyJjb25kaXRpb25zIjogW3siYnVja2V0IjogImxvd2VzLW1lZGlhLWRldiJ9LCBbInN0YXJ0cy13aXRoIiwgIiRrZXkiLCAibWVkaWEvdG1wL3Rlc3QucG5nIl0sIHsiYWNsIjogInB1YmxpYy1yZWFkIn0sIHsic3VjY2Vzc19hY3Rpb25fc3RhdHVzIjogIjIwMSJ9LCBbInN0YXJ0cy13aXRoIiwgIiRDb250ZW50LVR5cGUiLCAiIl0sIFsiY29udGVudC1sZW5ndGgtcmFuZ2UiLCAwLCA0MDQ4NTc2XV0sICJleHBpcmF0aW9uIjogIjIwMTYtMDItMjNUMTc6NDY6MTQuMDAwWiJ9',
'signature': 'grWm5mcVUGt7D7ucWpEHsqngPZk='}
files = {'file': open('/home/pssu/Pictures/No_image_available.png', 'rb')}
# files = {'file': ('report.csv', 'some,data,to,send\nanother,row,to,send\n')}
policy.update({'success_action_status': '201', 'Content-Type': 'image/png', 'key': policy.pop('key_prefix')})
policy
{'AWSAccessKeyId': 'AK!!!!!!!!!!!!!!!!!7A',
'Content-Type': 'image/png',
'acl': 'public-read',
'bucket_url': 'https://my-bucket-name.s3.amazonaws.com',
'key': 'media/tmp/test.png',
'policy': 'eyJjb25kaXRpb25zIjogW3siYnVja2V0IjogImxvd2VzLW1lZGlhLWRldiJ9LCBbInN0YXJ0cy13aXRoIiwgIiRrZXkiLCAibWVkaWEvdG1wL3Rlc3QucG5nIl0sIHsiYWNsIjogInB1YmxpYy1yZWFkIn0sIHsic3VjY2Vzc19hY3Rpb25fc3RhdHVzIjogIjIwMSJ9LCBbInN0YXJ0cy13aXRoIiwgIiRDb250ZW50LVR5cGUiLCAiIl0sIFsiY29udGVudC1sZW5ndGgtcmFuZ2UiLCAwLCA0MDQ4NTc2XV0sICJleHBpcmF0aW9uIjogIjIwMTYtMDItMjNUMTc6NDY6MTQuMDAwWiJ9',
'signature': 'grWm5mcVUGt7D7ucWpEHsqngPZk=',
'success_action_status': '201'}
r = requests.post(policy.pop('bucket_url'), data = policy, files=files)
print(r.status_code, r.reason, r.text, r.headers)
201 Created <?xml version="1.0" encoding="UTF-8"?>
<PostResponse><Location>https://my-bucket-name.s3.amazonaws.com/media%2Ftmp%2Ftest.png</Location><Bucket>my-bucket-name</Bucket><Key>media/tmp/test.png</Key><ETag>"80b0040331073af367b02fcc2fc9027f"</ETag></PostResponse> {'x-amz-request-id': 'DEBBF522CCADDCF6', 'x-amz-id-2': 'CiVcJghWFcIm4+l7Xjyza67sOxDkzj7TJra0/XKq17PfwwdvYkHf/QSe3GX9kKf3GSdRU6EGkOQ=', 'Location': 'https://my-bucket-name.s3.amazonaws.com/media%2Ftmp%2Ftest.png', 'Content-Length': '260', 'Content-Type': 'application/xml', 'ETag': '"80b0040331073af367b02fcc2fc9027f"', 'Server': 'AmazonS3', 'Date': 'Tue, 23 Feb 2016 17:44:18 GMT'}