Last active
December 18, 2015 04:29
-
-
Save srossross/5725907 to your computer and use it in GitHub Desktop.
This is a small gist to help with signed http post uploads to amazon s3 using boto.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import boto | |
import requests | |
from datetime import datetime, timedelta | |
import base64 | |
import json | |
import hmac | |
import hashlib | |
REQUIRED_FIELDS = 'key', 'AWSAccessKeyId', 'acl', 'success_action_status' | |
def make_policy(bucket, data, timeout): | |
policy = {"expiration": (datetime.utcnow() + timedelta(seconds=timeout)).strftime("%Y-%m-%dT%H:%M:%SZ"), | |
"conditions": [ | |
{"bucket": bucket.name}, | |
{"acl": data['acl']}, | |
{"success_action_status": str(data['success_action_status'])}, | |
["eq", "$key", data['key']], | |
#TODO: should this be here? | |
["content-length-range", 0, 1048576], | |
] | |
} | |
for key, value in data.items(): | |
if key in REQUIRED_FIELDS: continue | |
policy['conditions'].append(['eq', '$%s' % key, value]) | |
return policy | |
def update_with_policy(bucket, data, timeout): | |
policy_doc = make_policy(bucket, data, timeout) | |
policy = base64.b64encode(json.dumps(policy_doc)) | |
provider = bucket.connection.provider | |
signature = base64.b64encode(hmac.new(provider.secret_key, policy, hashlib.sha1).digest()) | |
data['policy'] = policy | |
data['signature'] = signature | |
def generate_post_url(key, expires_in, success_action_status=201, acl='private', extra_fields=None): | |
url = "https://%s.s3.amazonaws.com/" % (key.bucket.name,) | |
provider = key.bucket.connection.provider | |
data = {'key':key.name, | |
'AWSAccessKeyId': provider.access_key, | |
'acl':acl, | |
'success_action_status':success_action_status, | |
} | |
if extra_fields: | |
data.update(extra_fields) | |
if provider.security_token: | |
data['x-amz-security-token'] = provider.security_token | |
update_with_policy(key.bucket, data, expires_in) | |
return url, data | |
def main(): | |
s3 = boto.connect_s3() | |
bucket = s3.get_bucket('mybucket') | |
key = bucket.new_key('test_key') | |
content = 'asdfadsfarerawe rdsfasdf ' | |
url, form_data = generate_post_url(key, 60, extra_fields={'Content-Length': len(content)}) | |
print url | |
print form_data | |
res = requests.post(url, form_data, files={'file':('filename', content)}) | |
print res.status_code | |
print res.text | |
if __name__ == '__main__': | |
main() | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment