Skip to content

Instantly share code, notes, and snippets.

@srossross
Last active December 18, 2015 04:29
Show Gist options
  • Save srossross/5725907 to your computer and use it in GitHub Desktop.
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.
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