from datetime import timedelta from django.conf import settings from django.contrib import messages from django.contrib.auth import login as auth_login from django.http import HttpResponseRedirect from django.utils.translation import gettext_lazy as _ from django.contrib.auth import get_user_model from django.contrib.auth.decorators import login_required from django.contrib.auth.views import LoginView from django.shortcuts import render from django.urls import reverse from django.utils import timezone from django.utils.decorators import method_decorator from django.views.generic import DetailView, UpdateView, CreateView from oauth2_provider.contrib.rest_framework import IsAuthenticatedOrTokenHasScope from rest_framework import generics, status, permissions from rest_framework.response import Response from rest_framework.views import APIView from apps.gooyal_oauth2.rest_framework import IsAuthenticatedOrTokenMatchesOASRequirements from apps.gooyal_oauth2.validators import OAuth2Validator from apps.users.forms import OAUTHLoginRequestForm from apps.users.models import User, OAuthCode from apps.users.serializers import AccountSerializer from django.contrib.auth import login UserModel = get_user_model() @method_decorator(login_required, name='dispatch') class AccountView(generics.RetrieveUpdateAPIView): serializer_class = AccountSerializer permission_classes = [IsAuthenticatedOrTokenMatchesOASRequirements] required_alternate_scopes = { "GET": [["accounts.account:retrieve"]], "POST": [["accounts.account:update"]], "PUT": [["accounts.account:update"]], "PATCH": [["accounts.account:update"]], } def get_object(self): return self.request.user def perform_update(self, serializer): serializer.save(last_update=timezone.now()) @method_decorator(login_required, name='dispatch') class ProfileDetailView(DetailView): model = User template_name = 'users/profile_detail.html' pk_url_kwarg = 'uuid' def get_object(self, queryset=None): return self.request.user class OAUTHLoginRequestView(CreateView): model = OAuthCode form_class = OAUTHLoginRequestForm template_name = 'users/login_request.html' def form_valid(self, form): self.object: OAuthCode = form.save(commit=False) self.object.generate_code_verifier() self.object.save() return HttpResponseRedirect(self.object.generate_login_url()) def form_invalid(self, form): for error_key in form.errors: for error_list in form.errors[error_key].data: for error in error_list: key = _(error_key) messages.error(self.request, f"{key}: {error}") return super().form_invalid(form) class OAUTHLoginCallbackView(DetailView): model = OAuthCode template_name = 'users/login_callback.html' def get_object(self, queryset=None): state = self.request.GET.get('state') code = self.request.GET.get('code') obj = OAuthCode.objects.get(pk=state) obj.code = code obj.save() return obj def introspect_token(self, token): url = settings.OAUTH2_PROVIDER['RESOURCE_SERVER_INTROSPECTION_URL'] credentials = settings.OAUTH2_PROVIDER['RESOURCE_SERVER_INTROSPECTION_CREDENTIALS'] validator = OAuth2Validator() access_token = validator._get_token_from_authentication_server(token, url, None, credentials) return access_token def get_context_data(self, **kwargs): data = self.object.validate_code() # {'access_token': 'WhnIAfci6yIyTsh63PPqM1HXfXqKvr', 'expires_in': 36000, 'token_type': 'Bearer', # 'scope': 'wallet.wallet:get_balance wallet.transaction:list', # 'refresh_token': 'qRHFXXc3R3EiQKUq5BcSY2b9xuTm0d'} # # from oauth2_provider.models import get_access_token_model, get_application_model access_token = self.introspect_token(data['access_token']) login(self.request, access_token.user, backend='django.contrib.auth.backends.ModelBackend') return super().get_context_data(**kwargs) # AccessToken = get_access_token_model() # access_token = AccessToken.objects.create( # token=data['access_token'], # scope=data['scope'], # expires=timezone.now() + timedelta(data['scope']), # ) print(data) return super().get_context_data(**kwargs)