wallet/apps/gooyal_oauth2/validators.py

# import service_clients
import base64
import http.client
import logging
from datetime import datetime, timedelta

import requests
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,
    get_application_model,
    get_grant_model,
    get_id_token_model,
    get_refresh_token_model,
)
from oauth2_provider.oauth2_validators import OAuth2Validator as BaseOAuth2Validator
from oauth2_provider.settings import oauth2_settings
from oauth2_provider.utils import get_timezone

Application = get_application_model()
AccessToken = get_access_token_model()
IDToken = get_id_token_model()
Grant = get_grant_model()
RefreshToken = get_refresh_token_model()
UserModel = get_user_model()



log = logging.getLogger("oauth2_provider")


class OAuth2Validator(BaseOAuth2Validator):  # pylint: disable=w0223
    def get_or_create_user_from_content(self, content):
        """
        An optional layer to define where to store the profile in `UserModel` or a separate model.
        For example `UserOAuth`, where `user = models.OneToOneField(UserModel)` .

        The function is called after checking that username is in the content.

        Returns an UserModel instance;
        """
        user, _ = UserModel.objects.get_or_create(pk=content["username"])
        return user

    def _get_token_from_authentication_server(
        self, token, introspection_url, introspection_token, introspection_credentials
    ):
        # NOTICE: onlu change from orginal method is that we create application here
        """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)}
        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

        # Log an exception when response from auth server is not successful
        if response.status_code != http.client.OK:
            log.exception(
                "Introspection: Failed to get a valid response "
                "from authentication server. Status code: {}, "
                "Reason: {}.".format(response.status_code, response.reason)
            )
            return None

        try:
            content = response.json()
        except ValueError:
            log.exception("Introspection: Failed to parse response as json")
            return None

        if "active" in content and content["active"] is True:
            try:
                application_introspection_url = introspection_url.rstrip("/") + "_application/"
                response = requests.post(application_introspection_url, data={"client_id": content['client_id']}, headers=headers)
            except requests.exceptions.RequestException:
                log.exception("Introspection: Failed POST to %r in token lookup", introspection_url)
                return None

            if response.status_code != http.client.OK:
                log.exception(
                    "Application introspection: Failed to get a valid response "
                    "from authentication server. Status code: {}, "
                    "Reason: {}.".format(response.status_code, response.reason)
                )
                return None

            try:
                application_introspection_content = response.json()
            except ValueError:
                log.exception("Introspection: Failed to parse response as json")
                return None

            owner_value = None
            application_uuid = None
            if "active" in application_introspection_content and application_introspection_content["active"] is True:
                if "owner" in application_introspection_content:
                    owner_value = application_introspection_content["owner"]
                application_uuid = application_introspection_content.get("uuid")

            if owner_value:
                owner, _ = UserModel.objects.get_or_create(pk=owner_value)
            else:
                owner = None

            if "username" in content:
                user = self.get_or_create_user_from_content(content)
            else:
                user = None

            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", "")

            if settings.USE_TZ:
                expires = make_aware(
                    expires, timezone=get_timezone(oauth2_settings.AUTHENTICATION_SERVER_EXP_TIME_ZONE)
                )


            application, _created = Application.objects.get_or_create(
                client_id=content["client_id"],
                uuid=application_introspection_content["uuid"],
                defaults={
                    "user": owner  # TODO: JUST FOR WALLET
                }
            )
            access_token, _created = AccessToken.objects.update_or_create(
                token=token,
                defaults={
                    "user": user ,   # TODO: JUST FOR WALLET
                    "client_id": content["client_id"],
                    "application_owner": owner,
                    "application_id": application_uuid,
                    "scope": scope,
                    "expires": expires,
                },
            )

            return access_token