/
views.py
166 lines (135 loc) · 6.58 KB
/
views.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
from urllib.parse import urlencode
import oauthlib.oauth2 as oauth
from django.conf import settings
from django.contrib.auth.decorators import login_required
from django.http import HttpResponse, HttpResponseBadRequest, HttpResponseRedirect
from django.views.decorators.csrf import csrf_exempt
from django.utils.translation import ugettext as _
from django.core.urlresolvers import get_callable
from .decorators import oauth_required
from .forms import AuthorizeRequestTokenForm
from .store import store, InvalidConsumerError, InvalidTokenError
from .utils import verify_oauth_request, get_oauth_request, require_params, send_oauth_error
from .consts import OUT_OF_BAND
OAUTH_AUTHORIZE_VIEW = 'OAUTH_AUTHORIZE_VIEW'
OAUTH_CALLBACK_VIEW = 'OAUTH_CALLBACK_VIEW'
INVALID_PARAMS_RESPONSE = send_oauth_error( oauth.OAuth2Error( _('Invalid request parameters.') ) )
@csrf_exempt
def request_token(request):
oauth_request = get_oauth_request(request)
if oauth_request is None:
return INVALID_PARAMS_RESPONSE
missing_params = require_params(oauth_request, ('oauth_callback',))
if missing_params is not None:
return missing_params
try:
consumer = store.get_consumer(request, oauth_request, oauth_request['oauth_consumer_key'])
except InvalidConsumerError:
return HttpResponseBadRequest('Invalid Consumer.')
if not verify_oauth_request(request, oauth_request, consumer):
return HttpResponseBadRequest('Could not verify OAuth request.')
try:
request_token = store.create_request_token(request, oauth_request, consumer, oauth_request['oauth_callback'])
except oauth.OAuth2Error as err:
return send_oauth_error(err(_('Error while creating request token')))
ret = urlencode({
'oauth_token': request_token.key,
'oauth_token_secret': request_token.secret,
'oauth_callback_confirmed': 'true'
})
return HttpResponse(ret, content_type='application/x-www-form-urlencoded')
@login_required
def user_authorization(request, form_class=AuthorizeRequestTokenForm):
if 'oauth_token' not in request.REQUEST:
return HttpResponseBadRequest('No request token specified.')
oauth_request = get_oauth_request(request)
try:
request_token = store.get_request_token(request, oauth_request, request.REQUEST['oauth_token'])
except InvalidTokenError:
return HttpResponseBadRequest('Invalid request token.')
consumer = store.get_consumer_for_request_token(request, oauth_request, request_token)
if request.method == 'POST':
form = form_class(request.POST)
if request.session.get('oauth', '') == request_token.key and form.is_valid():
request.session['oauth'] = ''
if form.cleaned_data['authorize_access']:
request_token = store.authorize_request_token(request, oauth_request, request_token)
args = { 'oauth_token': request_token }
else:
args = { 'error': _('Access not granted by user.') }
if request_token.callback is not None and request_token.callback != OUT_OF_BAND:
response = HttpResponseRedirect(request_token.get_callback_url(args))
else:
# try to get custom callback view
callback_view_str = getattr(settings, OAUTH_CALLBACK_VIEW,
'oauth_provider.views.fake_callback_view')
try:
callback_view = get_callable(callback_view_str)
except AttributeError:
raise Exception("%s view doesn't exist." % callback_view_str)
response = callback_view(request, **args)
else:
response = send_oauth_error(oauth.OAuth2Error(_('Action not allowed.')))
else:
# try to get custom authorize view
authorize_view_str = getattr(settings, OAUTH_AUTHORIZE_VIEW,
'oauth_provider.views.fake_authorize_view')
try:
authorize_view = get_callable(authorize_view_str)
except AttributeError:
raise Exception("%s view doesn't exist." % authorize_view_str)
params = oauth_request.get_normalized_parameters()
# set the oauth flag
request.session['oauth'] = request_token.key
response = authorize_view(request, request_token, request_token.get_callback_url(), params)
return response
@csrf_exempt
def access_token(request):
oauth_request = get_oauth_request(request)
if oauth_request is None:
return INVALID_PARAMS_RESPONSE
missing_params = require_params(oauth_request, ('oauth_token', 'oauth_verifier'))
if missing_params is not None:
return missing_params
try:
request_token = store.get_request_token(request, oauth_request, oauth_request['oauth_token'])
except InvalidTokenError:
return HttpResponseBadRequest('Invalid request token.')
try:
consumer = store.get_consumer(request, oauth_request, oauth_request['oauth_consumer_key'])
except InvalidConsumerError:
return HttpResponseBadRequest('Invalid consumer.')
if not verify_oauth_request(request, oauth_request, consumer, request_token):
return HttpResponseBadRequest('Could not verify OAuth request.')
if oauth_request.get('oauth_verifier', None) != request_token.verifier:
return HttpResponseBadRequest('Invalid OAuth verifier.')
if not request_token.is_approved:
return HttpResponseBadRequest('Request Token not approved by the user.')
access_token = store.create_access_token(request, oauth_request, consumer, request_token)
ret = urlencode({
'oauth_token': access_token.key,
'oauth_token_secret': access_token.secret
})
return HttpResponse(ret, content_type='application/x-www-form-urlencoded')
@oauth_required
def protected_resource_example(request):
"""
Test view for accessing a Protected Resource.
"""
return HttpResponse('Protected Resource access!')
@login_required
def fake_authorize_view(request, token, callback, params):
"""
Fake view for tests. It must return an ``HttpResponse``.
You need to define your own in ``settings.OAUTH_AUTHORIZE_VIEW``.
"""
return HttpResponse('Fake authorize view for %s with params: %s.' % (token.consumer.name, params))
def fake_callback_view(request, **args):
"""
Fake view for tests. It must return an ``HttpResponse``.
You can define your own in ``settings.OAUTH_CALLBACK_VIEW``.
"""
if "oauth_token" in args:
return HttpResponse('Verifier: ' + str(args["oauth_token"].verifier))
else:
return HttpResponse('Fake callback view.')