Skip to content


Subversion checkout URL

You can clone with
Download ZIP
Tree: 2655b0e724
Fetching contributors…

Cannot retrieve contributors at this time

318 lines (248 sloc) 10.5 KB
import binascii
import oauth
from django.http import HttpResponse, HttpResponseRedirect
from django.contrib.auth.models import User, AnonymousUser
from django.contrib.auth.decorators import login_required
from django.template import loader
from django.contrib.auth import authenticate
from django.conf import settings
from django.core.urlresolvers import get_callable
from django.core.exceptions import ImproperlyConfigured
from django.shortcuts import render_to_response
from django.template import RequestContext
from piston import forms
class NoAuthentication(object):
Authentication handler that always returns
True, so no authentication is needed, nor
initiated (`challenge` is missing.)
def is_authenticated(self, request):
return True
class HttpBasicAuthentication(object):
Basic HTTP authenticater. Synopsis:
Authentication handlers must implement two methods:
- `is_authenticated`: Will be called when checking for
authentication. Receives a `request` object, please
set your `User` object on `request.user`, otherwise
return False (or something that evaluates to False.)
- `challenge`: In cases where `is_authenticated` returns
False, the result of this method will be returned.
This will usually be a `HttpResponse` object with
some kind of challenge headers and 401 code on it.
def __init__(self, auth_func=authenticate, realm='API'):
self.auth_func = auth_func
self.realm = realm
def is_authenticated(self, request):
auth_string = request.META.get('HTTP_AUTHORIZATION', None)
if not auth_string:
return False
(authmeth, auth) = auth_string.split(" ", 1)
if not authmeth.lower() == 'basic':
return False
auth = auth.strip().decode('base64')
(username, password) = auth.split(':', 1)
except (ValueError, binascii.Error):
return False
request.user = self.auth_func(username=username, password=password) \
or AnonymousUser()
return not request.user in (False, None, AnonymousUser())
def challenge(self):
resp = HttpResponse("Authorization Required")
resp['WWW-Authenticate'] = 'Basic realm="%s"' % self.realm
resp.status_code = 401
return resp
def __repr__(self):
return u'<HTTPBasic: realm=%s>' % self.realm
class HttpBasicSimple(HttpBasicAuthentication):
def __init__(self, realm, username, password):
self.user = User.objects.get(username=username)
self.password = password
super(HttpBasicSimple, self).__init__(auth_func=self.hash, realm=realm)
def hash(self, username, password):
if username == self.user.username and password == self.password:
return self.user
def load_data_store():
'''Load data store for OAuth Consumers, Tokens, Nonces and Resources
path = getattr(settings, 'OAUTH_DATA_STORE', '')
# stolen from django.contrib.auth.load_backend
i = path.rfind('.')
module, attr = path[:i], path[i+1:]
mod = __import__(module, {}, {}, attr)
except ImportError, e:
raise ImproperlyConfigured, 'Error importing OAuth data store %s: "%s"' % (module, e)
cls = getattr(mod, attr)
except AttributeError:
raise ImproperlyConfigured, 'Module %s does not define a "%s" OAuth data store' % (module, attr)
return cls
# Set the datastore here.
oauth_datastore = load_data_store()
def initialize_server_request(request):
Shortcut for initialization.
if request.method == "POST": #and \
# request.META['CONTENT_TYPE'] == "application/x-www-form-urlencoded":
params = dict(request.REQUEST.items())
params = { }
# Seems that we want to put HTTP_AUTHORIZATION into 'Authorization'
# for to understand. Lovely.
request.META['Authorization'] = request.META.get('HTTP_AUTHORIZATION', '')
oauth_request = oauth.OAuthRequest.from_request(
request.method, request.build_absolute_uri(),
headers=request.META, parameters=params,
query_string=request.environ.get('QUERY_STRING', ''))
if oauth_request:
oauth_server = oauth.OAuthServer(oauth_datastore(oauth_request))
oauth_server = None
return oauth_server, oauth_request
def send_oauth_error(err=None):
Shortcut for sending an error.
response = HttpResponse(err.message.encode('utf-8'))
response.status_code = 401
realm = 'OAuth'
header = oauth.build_authenticate_header(realm=realm)
for k, v in header.iteritems():
response[k] = v
return response
def oauth_request_token(request):
oauth_server, oauth_request = initialize_server_request(request)
if oauth_server is None:
token = oauth_server.fetch_request_token(oauth_request)
response = HttpResponse(token.to_string())
except oauth.OAuthError, err:
response = send_oauth_error(err)
return response
def oauth_auth_view(request, token, callback, params):
form = forms.OAuthAuthenticationForm(initial={
'oauth_token': token.key,
'oauth_callback': token.get_callback_url() or callback,
return render_to_response('piston/authorize_token.html',
{ 'form': form }, RequestContext(request))
def oauth_user_auth(request):
oauth_server, oauth_request = initialize_server_request(request)
if oauth_request is None:
token = oauth_server.fetch_request_token(oauth_request)
except oauth.OAuthError, err:
return send_oauth_error(err)
callback = oauth_server.get_callback(oauth_request)
callback = None
if request.method == "GET":
params = oauth_request.get_normalized_parameters()
oauth_view = getattr(settings, 'OAUTH_AUTH_VIEW', None)
if oauth_view is None:
return oauth_auth_view(request, token, callback, params)
return get_callable(oauth_view)(request, token, callback, params)
elif request.method == "POST":
form = forms.OAuthAuthenticationForm(request.POST)
if form.is_valid():
token = oauth_server.authorize_token(token, request.user)
args = '?'+token.to_string(only_key=True)
args = '?error=%s' % 'Access not granted by user.'
print "FORM ERROR", form.errors
if not callback:
callback = getattr(settings, 'OAUTH_CALLBACK_VIEW')
return get_callable(callback)(request, token)
response = HttpResponseRedirect(callback+args)
except oauth.OAuthError, err:
response = send_oauth_error(err)
response = HttpResponse('Action not allowed.')
return response
def oauth_access_token(request):
oauth_server, oauth_request = initialize_server_request(request)
if oauth_request is None:
token = oauth_server.fetch_access_token(oauth_request)
return HttpResponse(token.to_string())
except oauth.OAuthError, err:
return send_oauth_error(err)
INVALID_PARAMS_RESPONSE = send_oauth_error(oauth.OAuthError('Invalid request parameters.'))
class OAuthAuthentication(object):
OAuth authentication. Based on work by Leah Culver.
def __init__(self, realm='API'):
self.realm = realm
self.builder = oauth.build_authenticate_header
def is_authenticated(self, request):
Checks whether a means of specifying authentication
is provided, and if so, if it is a valid token.
Read the documentation on `HttpBasicAuthentication`
for more information about what goes on here.
if self.is_valid_request(request):
consumer, token, parameters = self.validate_token(request)
except oauth.OAuthError, err:
print send_oauth_error(err)
return False
if consumer and token:
request.user = token.user
request.consumer = consumer
request.throttle_extra =
return True
return False
def challenge(self):
Returns a 401 response with a small bit on
what OAuth is, and where to learn more about it.
When this was written, browsers did not understand
OAuth authentication on the browser side, and hence
the helpful template we render. Maybe some day in the
future, browsers will take care of this stuff for us
and understand the 401 with the realm we give it.
response = HttpResponse()
response.status_code = 401
realm = 'API'
for k, v in self.builder(realm=realm).iteritems():
response[k] = v
tmpl = loader.render_to_string('oauth/challenge.html',
{ 'MEDIA_URL': settings.MEDIA_URL })
response.content = tmpl
return response
def is_valid_request(request):
Checks whether the required parameters are either in
the http-authorization header sent by some clients,
which is by the way the preferred method according to
OAuth spec, but otherwise fall back to `GET` and `POST`.
must_have = [ 'oauth_'+s for s in [
'consumer_key', 'token', 'signature',
'signature_method', 'timestamp', 'nonce' ] ]
is_in = lambda l: all([ (p in l) for p in must_have ])
auth_params = request.META.get("HTTP_AUTHORIZATION", "")
req_params = request.REQUEST
return is_in(auth_params) or is_in(req_params)
def validate_token(request, check_timestamp=True, check_nonce=True):
oauth_server, oauth_request = initialize_server_request(request)
return oauth_server.verify_request(oauth_request)
Jump to Line
Something went wrong with that request. Please try again.