OAuth Request Signing

I've started writing a little app that needs to talk to Twitter - to do this it needs to speak OAuth. It runs on Google App Engine and is written in Python and there are many good OAuth (and even dedicated Twitter) modules written in Python so I could have grabbed one of these and got on with it but that wouldn't have been much fun and wouldn't get me any closer to an understanding of OAuth - so I went ahead and implemented it myself.

Since at this stage I don't need my app to allow users to log in and interact as themselves (actually I'm actively avoiding this sort of requirement) I didn't need to worry about the 'dance' part of OAuth (the token request/authentication/exchange bit) - I only need the request signing part and the keys provided by my Twitter app's dashboard page.

I tweeted from an interactive Python session using my OAuth script for authentication, and have pulled down mentions etc. The implementation is deliberately unoptimised and fragmented - this lets it serve better as an executable reference to the OAuth standard (the various functions link to the specific sections of the standard that they implement) - and also it is good enough for what I want to do right now.

Here it is

'''implementing OAuth ( request signing only ) as set out in http://tools.ietf.org/html/rfc5849

License: This code is free for use by any person, for any purpose but is provided as-is, with no guarantees - use at your own risk. One proviso: drop me a mail and let me know how it worked for you.'''

from urllib import quote
import urllib2
import time
import os
import base64
import hashlib
import hmac

def encode( value ):  
  '''#section-3.6'''
  return quote( value or '', '-._~' )

def enc_params( params ):
  '''#section-3.4.1.3.2'''
  ep = [ '%s=%s' % ( encode( k ), encode( v ) ) for k, v in params.items( ) ]
  return '&'.join( sorted( ep ) )

def b64e( value ):
  '''#section-6.8'''
  return base64.b64encode( value )

def make_base_string( method, url, params ):
  '''#section-3.4.1'''
  return '&'.join( [ encode( method.upper( ) ), encode( url ), encode( enc_params( params ) ) ] )

def get_timestamp( ):
  '''#section-3.3'''
  return str( int( time.time( ) ) )

def get_nonce( ):
  '''#section-3.3'''
  return b64e( os.urandom( 32 ) ).strip( '=' )

def hmac_sha1_sig( base_string, consumer_secret, token_secret='' ):
  '''#section-3.4.2'''
  key = '%s&%s' % ( encode( consumer_secret ), encode( token_secret ) )
  h=hmac.new( key, base_string, hashlib.sha1 )
  return b64e( h.digest( ) )

class OAuthClient( object ):
  def __init__( self, consumer_pair, token_pair ):
    '''takes (consumer_key, consumer_secret), (oauth_token, token_secret)'''
    self.consumer_key, self.consumer_secret = consumer_pair
    self.oauth_token, self.token_secret = token_pair

  def create_oauth_params( self, url, data=None ):
    method = 'POST' if data else 'GET'
    params = {
      'oauth_consumer_key'     : self.consumer_key
    , 'oauth_token'            : self.oauth_token
    , 'oauth_signature_method' : 'HMAC-SHA1'
    , 'oauth_timestamp'        : get_timestamp( )
    , 'oauth_nonce'            : get_nonce( )
    , 'oauth_version'          : '1.0'
    }

    if data: params.update( data )

    base_string = make_base_string( method, url, params )
    params[ 'oauth_signature' ] = hmac_sha1_sig( base_string, self.consumer_secret, self.token_secret )

    return params

  def create_oauth_header( self, url, data=None ):
    params = self.create_oauth_params( url, data )
    header = 'OAuth ' + ', '.join( [ '%s="%s"' % ( k, encode( v ) ) for k, v in params.items( ) ] )

    return header

  def open_url( self, url, data=None ):
    '''returns the resource represented by the url, or a tuple of error code and response text'''
    if data: post_data = enc_params( data )
    else: post_data = None
    r=urllib2.Request( url, post_data )
    r.headers[ 'Authorization' ] = self.create_oauth_header( url, data )

    try:
      return urllib2.urlopen( r ).read( )
    except urllib2.URLError, e:
      return e.code, e.read( )

If you save this as something like oauth.py and insert your own consumer/oauth keys over the placeholders you can try it out at a prompt like so

>>> CONSUMER_KEY = 'CONSUMER-KEY'
>>> CONSUMER_SECRET = 'CONSUMER-SECRET'
>>> OAUTH_TOKEN = 'OAUTH-TOKEN'
>>> TOKEN_SECRET = 'TOKEN-SECRET'
>>> import oauth
>>> c = oauth.OAuthClient( ( CONSUMER_KEY, CONSUMER_SECRET ), ( OAUTH_TOKEN, TOKEN_SECRET ) )
>>> mentions = c.open_url( 'http://api.twitter.com/1/statuses/mentions.json' )
>>> c.open_url( 'http://api.twitter.com/1/statuses/update.json', { 'status' : 'here is my status update, blah, blah. Read this blog: http://blog.tosh.me/. } )

That's it, more to follow on the project it was written for.