I'm using Python Social Auth and Django OAuth Toolkit to manage my user accounts and restrict access to my REST API.
I can create a token for users that sign up manually with my app by using the regular
curl -X POST -d "grant_type=password&username=<user_name>&password=<password>" -u"<client_id>:<client_secret>" http://localhost:8000/o/token/
But when I register my users with PSA by their access token, I'd like to create a OAuth2 Toolkit token for my own app and return it as JSON to the client so it can use it for making requests with my API.
Presently, I generate token simply using generate_token
from oauthlib, is that good practice? Should I take into consideration other factors?
from oauthlib.common import generate_token
...
@psa('social:complete')
def register_by_access_token(request, backend):
# This view expects an access_token GET parameter, if it's needed,
# request.backend and request.strategy will be loaded with the current
# backend and strategy.
token = request.GET.get('access_token')
user = request.backend.do_auth(token)
if user:
login(request, user)
app = Application.objects.get(name="myapp")
# We delete the old one
try:
old = AccessToken.objects.get(user=user, application=app)
except:
pass
else:
old.delete()
# We create a new one
tok = generate_token()
AccessToken.objects.get_or_create(user=user,
application=app,
expires=now() + timedelta(days=365),
token=tok)
return "OK" # I will eventually return JSON with the token
else:
return "ERROR"
I've recently used https://github.com/PhilipGarnero/django-rest-framework-social-oauth2 for this purpose like user Felix D. suggested. Below is my implementation:
class TokenHandler:
application = Application.objects.filter(name=APPLICATION_NAME).values('client_id', 'client_secret')
def handle_token(self, request):
"""
Gets the latest token (to access my API) and if it's expired, check to see if the social token has expired.
If the social token has expired, then the user must log back in to access the API. If it hasn't expired,
(my) token is refreshed.
"""
try:
token_list = AccessToken.objects.filter(user=request.user)\
.order_by('-id').values('token', 'expires')
if token_list[0]['expires'] < datetime.now(timezone.utc):
if not self.social_token_is_expired(request):
token = self.refresh_token(request)
else:
token = 'no_valid_token'
else:
token = token_list[0]['token']
except IndexError: # happens where there are no old tokens to check
token = self.convert_social_token(request)
except TypeError: # happens when an anonymous user attempts to get a token for the API
token = 'no_valid_token'
return token
def convert_social_token(self, request):
grant_type = 'convert_token'
client_id = self.application[0]['client_id']
client_secret = self.application[0]['client_secret']
try:
user_social_auth = request.user.social_auth.filter(user=request.user).values('provider', 'extra_data')[0]
backend = user_social_auth['provider']
token = user_social_auth['extra_data']['access_token']
url = get_base_url(request) + reverse('convert_token')
fields = {'client_id': client_id, 'client_secret': client_secret, 'grant_type': grant_type,
'backend': backend,
'token': token}
if backend == 'azuread-oauth2':
fields['id_token'] = user_social_auth['extra_data']['id_token']
response = requests.post(url, data=fields)
response_dict = json.loads(response.text)
except IndexError:
return {'error': 'You must use an OAuth account to access the API.'}
except UserSocialAuth.DoesNotExist:
return {'error': 'You must use an OAuth account to access the API.'}
return response_dict['access_token']
def refresh_token(self, request):
grant_type = 'refresh_token'
client_id = self.application[0]['client_id']
client_secret = self.application[0]['client_secret']
try:
refresh_token_object = RefreshToken.objects.filter(user=request.user).order_by('-id').values('token')[0]
token = refresh_token_object['token']
url = get_base_url(request) + reverse('token')
fields = {'client_id': client_id, 'client_secret': client_secret, 'grant_type': grant_type,
'refresh_token': token}
response = requests.post(url, data=fields)
response_dict = json.loads(response.text)
except RefreshToken.DoesNotExist:
return {'error': 'You must use an OAuth account to access the API.'}
return response_dict['access_token']
@staticmethod
def social_token_is_expired(request):
user_social_auth = UserSocialAuth.objects.filter(user=request.user).values('provider', 'extra_data')[0]
try:
return float(user_social_auth['extra_data']['expires_on']) <= datetime.now().timestamp()
except KeyError: # social API did not provide an expiration
return True # if our token is expired and social API did not provide a time, we do consider them expired
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With