91 lines
3.1 KiB
Python
Executable file
91 lines
3.1 KiB
Python
Executable file
from oauth2_provider.oauth2_validators import OAuth2Validator
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from oauth2_provider.settings import oauth2_settings
|
|
|
|
from apps.gooyal_oauth2.models import Resource
|
|
|
|
USER_MODEL = get_user_model()
|
|
|
|
|
|
class MultiGatewayOAuth2Validator(OAuth2Validator): # pylint: disable=w0223
|
|
""" Primarily extend the functionality of token generation """
|
|
|
|
def validate_user(self, username, password, client, request, *args, **kwargs):
|
|
""" Here, you would be able to access the MOBILE/ OTP fields
|
|
which you will be sending in the request.post body. """
|
|
# otp = request.otp
|
|
# mobile = request.mobile
|
|
# user = AppropriateModel.objects.get(otp=otp, mobile=mobile)
|
|
auth_fields = getattr(request, 'auth_fields', 'username:password').split(':')
|
|
user_field = auth_fields[0]
|
|
pass_field = auth_fields[1]
|
|
|
|
user = None
|
|
if user_field == 'phone_number':
|
|
user = USER_MODEL.objects.get(
|
|
phone_number=username
|
|
)
|
|
|
|
elif user_field == 'username':
|
|
user = USER_MODEL.objects.get(
|
|
username=username
|
|
)
|
|
|
|
elif user_field == 'email':
|
|
user = USER_MODEL.objects.get(
|
|
email=username
|
|
)
|
|
|
|
if user is None:
|
|
return False
|
|
|
|
if not user.check_auth(pass_field, password):
|
|
return False
|
|
|
|
if user.is_active:
|
|
request.user = user
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
class IntrospectOAuth2Validator(OAuth2Validator):
|
|
def validate_bearer_token(self, token, scopes, request):
|
|
"""
|
|
When users try to access resources, check that provided token is valid
|
|
"""
|
|
if not token:
|
|
return False
|
|
|
|
introspection_url = oauth2_settings.RESOURCE_SERVER_INTROSPECTION_URL
|
|
introspection_token = oauth2_settings.RESOURCE_SERVER_AUTH_TOKEN
|
|
introspection_credentials = oauth2_settings.RESOURCE_SERVER_INTROSPECTION_CREDENTIALS
|
|
|
|
try:
|
|
access_token = Resource.objects.select_related("application", "user").get(token=token)
|
|
except Resource.DoesNotExist:
|
|
access_token = None
|
|
|
|
# if there is no token or it's invalid then introspect the token if there's an external OAuth server
|
|
if not access_token or not access_token.is_valid(scopes):
|
|
if introspection_url and (introspection_token or introspection_credentials):
|
|
access_token = self._get_token_from_authentication_server(
|
|
token,
|
|
introspection_url,
|
|
introspection_token,
|
|
introspection_credentials
|
|
)
|
|
|
|
if access_token and access_token.is_valid(scopes):
|
|
request.client = access_token.application
|
|
request.user = access_token.user
|
|
request.scopes = scopes
|
|
|
|
# this is needed by django rest framework
|
|
request.access_token = access_token
|
|
return True
|
|
else:
|
|
self._set_oauth2_error_on_request(request, access_token, scopes)
|
|
return False
|
|
|