211 lines
8.8 KiB
Python
Executable file
211 lines
8.8 KiB
Python
Executable file
import base64
|
|
import logging
|
|
from datetime import datetime, timedelta
|
|
|
|
import requests
|
|
import service_clients
|
|
from django.conf import settings
|
|
from django.contrib.auth import get_user_model
|
|
from django.utils.timezone import make_aware
|
|
from oauth2_provider.models import get_access_token_model
|
|
from oauth2_provider.oauth2_validators import OAuth2Validator as BaseOAuth2Validator
|
|
from .settings import oauth2_settings
|
|
|
|
log = logging.getLogger("oauth2_provider")
|
|
|
|
AccessTokenModel = get_access_token_model()
|
|
UserModel = get_user_model()
|
|
|
|
|
|
|
|
class OAuth2Validator(BaseOAuth2Validator): # pylint: disable=w0223
|
|
def validate_user(self, username, password, client, request, *args, **kwargs):
|
|
auth_fields = getattr(request, 'auth_fields', 'username:password').split(':')
|
|
|
|
if len(auth_fields) != 2:
|
|
return False
|
|
|
|
user_field, pass_field = auth_fields
|
|
|
|
if user_field not in ['phone_number', 'username', 'email']:
|
|
return False
|
|
|
|
if pass_field not in ['password', 'otp']:
|
|
return False
|
|
|
|
if not username or not password:
|
|
return False
|
|
|
|
user = UserModel.objects.filter(**{user_field: username}).first()
|
|
|
|
if not user:
|
|
return False
|
|
|
|
if not user.check_auth(pass_field, password):
|
|
return False
|
|
|
|
if user.is_active:
|
|
request.user = user
|
|
return True
|
|
|
|
return False
|
|
|
|
def _create_user(self, content):
|
|
content = {'active': True, 'scope': 'ipg.payment:submit accounts.account:retrieve', 'exp': 1602619137,
|
|
'client_id': 'bd2hEGXytrqMjbFnplRyHJTeoW1vwKzCZJtH6ro0', 'username': '',
|
|
'uuid': '94255117-117c-4a9f-af50-35a6c47503ac', 'email': '', 'phone_number': '+989106853582',
|
|
'first_name': '', 'last_name': '', 'name': '', 'balance': 0,
|
|
'avatar': 'http://accounts.gooyal.com/media/avatars/1691899.jpg', 'iban': '', 'iban_verified': None}
|
|
|
|
# TODO: create user?
|
|
# user, _created = UserModel.objects.get_or_create(
|
|
# **{UserModel.USERNAME_FIELD: content["username"]}
|
|
# )
|
|
return None
|
|
|
|
def _get_token_from_gooyal_authentication_server(
|
|
self, token, introspection_url, introspection_token, introspection_credentials, introspection_client_id,
|
|
introspection_client_secret
|
|
):
|
|
"""Use external introspection endpoint to "crack open" the token.
|
|
:param introspection_url: introspection endpoint URL
|
|
:param introspection_token: Bearer token
|
|
:param introspection_credentials: Basic Auth credentials (id,secret)
|
|
:return: :class:`models.AccessToken`
|
|
|
|
Some RFC 7662 implementations (including this one) use a Bearer token while others use Basic
|
|
Auth. Depending on the external AS's implementation, provide either the introspection_token
|
|
or the introspection_credentials.
|
|
|
|
If the resulting access_token identifies a username (e.g. Authorization Code grant), add
|
|
that user to the UserModel. Also cache the access_token up until its expiry time or a
|
|
configured maximum time.
|
|
|
|
"""
|
|
|
|
headers = None
|
|
if introspection_token:
|
|
headers = {"Authorization": "Bearer {}".format(introspection_token)}
|
|
try:
|
|
response = requests.post(
|
|
introspection_url,
|
|
data={"token": token}, headers=headers
|
|
)
|
|
except requests.exceptions.RequestException:
|
|
log.exception("Introspection: Failed POST to %r in token lookup", introspection_url)
|
|
return None
|
|
|
|
elif introspection_credentials:
|
|
client_id = introspection_credentials[0].encode("utf-8")
|
|
client_secret = introspection_credentials[1].encode("utf-8")
|
|
basic_auth = base64.b64encode(client_id + b":" + client_secret)
|
|
headers = {"Authorization": "Basic {}".format(basic_auth.decode("utf-8"))}
|
|
try:
|
|
response = requests.post(
|
|
introspection_url,
|
|
data={"token": token}, headers=headers
|
|
)
|
|
except requests.exceptions.RequestException:
|
|
log.exception("Introspection: Failed POST to %r in token lookup", introspection_url)
|
|
return None
|
|
|
|
elif introspection_client_id and introspection_client_secret:
|
|
introspection_client = service_clients.Client(client_id=introspection_client_id,
|
|
client_secret=introspection_client_secret,
|
|
grant_type=service_clients.AccountsClient.GRANT_CLIENT_CREDENTIALS,
|
|
scopes=['introspection'])
|
|
data = {"token": token}
|
|
response = introspection_client.request(url=introspection_url, method='post', data=data,
|
|
required_scopes=['introspection'], login_required=True)
|
|
|
|
try:
|
|
content: dict = response.json()
|
|
except ValueError:
|
|
log.exception("Introspection: Failed to parse response as json")
|
|
return None
|
|
|
|
user = None
|
|
if "active" in content and content["active"] is True:
|
|
if "username" in content:
|
|
user_client = service_clients.Client(access_token=token,
|
|
scopes=content.get('scope','').split(),
|
|
expires_in=100)
|
|
user_account = user_client.accounts.account()
|
|
content.update(user_account)
|
|
user = self._create_user(content)
|
|
|
|
max_caching_time = datetime.now() + timedelta(
|
|
seconds=oauth2_settings.RESOURCE_SERVER_TOKEN_CACHING_SECONDS
|
|
)
|
|
|
|
if "exp" in content:
|
|
expires = datetime.utcfromtimestamp(content["exp"])
|
|
if expires > max_caching_time:
|
|
expires = max_caching_time
|
|
else:
|
|
expires = max_caching_time
|
|
|
|
scope = content.get("scope", "")
|
|
expires = make_aware(expires)
|
|
|
|
try:
|
|
access_token = AccessTokenModel.objects.select_related("application", "user").get(token=token)
|
|
except AccessTokenModel.DoesNotExist:
|
|
access_token = AccessTokenModel.objects.create(
|
|
user=user,
|
|
token=token,
|
|
application=None,
|
|
scope=scope,
|
|
expires=expires,
|
|
detail=content
|
|
)
|
|
else:
|
|
access_token.expires = expires
|
|
access_token.scope = scope
|
|
access_token.detail = content
|
|
access_token.save()
|
|
|
|
return access_token
|
|
|
|
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
|
|
introspection_client_id = oauth2_settings.RESOURCE_SERVER_CLIENT_ID
|
|
introspection_client_secret = oauth2_settings.RESOURCE_SERVER_CLIENT_SECRET
|
|
|
|
try:
|
|
access_token = AccessTokenModel.objects.select_related("application", "user").get(token=token)
|
|
except AccessTokenModel.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 or (introspection_client_id and
|
|
introspection_client_secret)):
|
|
access_token = self._get_token_from_gooyal_authentication_server(
|
|
token,
|
|
introspection_url,
|
|
introspection_token,
|
|
introspection_credentials,
|
|
introspection_client_id,
|
|
introspection_client_secret
|
|
)
|
|
|
|
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
|