import base64 import binascii import logging from datetime import datetime, timedelta from urllib.parse import unquote_plus import requests # import service_clients 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 from django.conf import settings log = logging.getLogger("oauth2_provider") AccessTokenModel = get_access_token_model() UserModel = get_user_model() class OAuth2Validator(BaseOAuth2Validator): # pylint: disable=w0223 introspection_client = None def get_introspection_client(self, introspection_client_id, introspection_client_secret): if not OAuth2Validator.introspection_client: OAuth2Validator.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']) return OAuth2Validator.introspection_client 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 _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 response = 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: data = {"token": token} introspection_client = self.get_introspection_client(introspection_client_id, introspection_client_secret) 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, content = oauth2_settings.INTROSPECTION_USER_CREATE_METHOD(token, 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) access_token, _created = AccessTokenModel.objects.update_or_create( token=token, defaults={ "user": user, "application": None, "scope": scope, "expires": expires, "detail": content, }) # 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 def _authenticate_basic_auth(self, request): """ Authenticates with HTTP Basic Auth. Note: as stated in rfc:`2.3.1`, client_id and client_secret must be encoded with "application/x-www-form-urlencoded" encoding algorithm. """ auth_string = self._extract_basic_auth(request) if not auth_string: return False try: encoding = request.encoding or settings.DEFAULT_CHARSET or "utf-8" except AttributeError: encoding = "utf-8" try: b64_decoded = base64.b64decode(auth_string) except (TypeError, binascii.Error): log.debug("Failed basic auth: %r can't be decoded as base64", auth_string) return False try: auth_string_decoded = b64_decoded.decode(encoding) except UnicodeDecodeError: log.debug("Failed basic auth: %r can't be decoded as unicode by %r", auth_string, encoding) return False try: client_id, client_secret = map(unquote_plus, auth_string_decoded.split(":", 1)) except ValueError: log.debug("Failed basic auth, Invalid base64 encoding.") return False if self._load_application(client_id, request) is None: log.debug("Failed basic auth: Application %s does not exist" % client_id) return False elif request.client.client_id != client_id: log.debug("Failed basic auth: wrong client id %s" % client_id) return False # TODO: check why not work elif not client_secret == request.client.client_secret: log.debug("Failed basic auth: wrong client secret %s" % client_secret) return False else: return True