S3 request signer in python

gistfile1.py

#!/usr/bin/env python
#
# baaed on Ruby S3 client implementation in RESTful Web Services of O'Reilly.
#

import base64
import hashlib
import hmac
import os
import re
import sys
import urlparse

interesting_headers = ['content-type', 'content-md5', 'date']
amazon_header_prefix = 'x-amz-'
amazon_request_params = ['acl', 'logging', 'torrent', 'versionid', 'versioning']

def signature(uri, private_key, method='GET', headers={}, expires=None):
  parsed = urlparse.urlsplit(uri)
  path = urlparse.urlunsplit(('', '', parsed.path, parsed.query, parsed.fragment))
  return sign(private_key, canonical_string(method, path, headers, expires))

def canonical_string(method, path, headers, expires=None):
  sign_headers = {}
  for header in interesting_headers:
    sign_headers[header] = ''

  re_endpoint = re.compile(r'\.?s3[^.]*\.amazonaws\.com$')
  for header in headers:
    header_lower = header.lower()
    if header_lower in interesting_headers or header_lower.startswith(amazon_header_prefix):
      sign_headers[header_lower] = str(headers[header]).strip()

## support for VirtualHosting; http://docs.amazonwebservices.com/AmazonS3/latest/dev/index.html?VirtualHosting.html
    if header_lower == 'host':
      host = headers[header]
      if re_endpoint.search(host):
        bucket = re_endpoint.sub('', host, 1)
        if bucket is None or len(bucket) < 1:
          bucket = None # request uri is not a virtual host
      else:
        bucket = re.sub(r'[^.\-0-9A-Za-z].*$', '', host, 1) # remove non-host part (suck like port)
      if bucket is not None:
        bucket = bucket.lower() # case in dns names should be ignored
        path = '/' + bucket + path

  if sign_headers.has_key('x-amz-date'):
    sign_headers['date'] = ''

  if expires is not None:
    sign_headers['date'] = str(expires)

  canonical = str(method).upper() + "\n"
  for header in sorted(sign_headers, lambda a, b: cmp(a, b)):
    if header.startswith(amazon_header_prefix):
      canonical += header + ':'
    canonical += sign_headers[header] + "\n"

  parsed = urlparse.urlsplit(path)
  canonical += parsed.path
  qs = [ q for q in parsed.query.split('&') if q in amazon_request_params ]
  if 0 < len(qs):
    canonical += "?" + "&".join(qs)
  return canonical

def sign(private_key, s):
  digest = hmac.new(private_key, s, hashlib.sha1).digest()
  return base64.encodestring(digest).strip()

# vim:set ft=python :

asynchronous S3 REST request sample from Tornado.

import email.utils
import tornado.httpclient
import tornado.web
import urlparse

class MyHandler(tornado.web.RequestHandler):
  @tornado.web.asynchronous
  def get(self, **kwargs):
    uri = urlparse.urljoin("http://s3.amazonaws.com/mybucket/", self.request.path)
    headers = {
      'Date': email.utils.formatdate(None, False, True),
      'Content-Type': '',
    }
    signed = signature(uri, self.settings.get('s3_secret_access_key', ''), method=self.request.method, headers=headers)
    headers['Authorization'] = 'AWS {k}:{d}'.format(k=self.settings.get('s3_access_key_id', ''), d=signed)
    request = tornado.httpclient.HTTPRequest(uri, method=self.request.method, headers=headers)
    http_client = tornado.httpclient.AsyncHTTPClient()
    return http_client.fetch(request, callback=self.on_fetched)