import requests from django.conf import settings from django.contrib.auth.base_user import BaseUserManager from django.contrib.auth.models import AbstractUser from django.contrib.auth.validators import UnicodeUsernameValidator from django.db import models from django.utils import timezone from django.utils.translation import gettext_lazy as _ import uuid import random import string import base64 import hashlib from apps.wallet.constans import TypeChoices class UserManager(BaseUserManager): use_in_migrations = True def _create_user(self, pk, **extra_fields): user = self.model(pk=pk **extra_fields) user.date_joined = timezone.now() user.save(using=self._db) return user def create_user(self, pk, **extra_fields): extra_fields.setdefault('is_staff', False) extra_fields.setdefault('is_superuser', False) return self._create_user(pk=pk, **extra_fields) def create_superuser(self, username, email, password, **extra_fields): if not username: raise ValueError('The given username must be set') extra_fields.setdefault('is_staff', True) extra_fields.setdefault('is_superuser', True) if extra_fields.get('is_staff') is not True: raise ValueError('Superuser must have is_staff=True.') if extra_fields.get('is_superuser') is not True: raise ValueError('Superuser must have is_superuser=True.') return self._create_user(username=username, email=email, password=password, **extra_fields) class User(AbstractUser): uuid = models.UUIDField(primary_key=True, editable=False, default=uuid.uuid4, unique=True, db_index=True) password = models.CharField(_('password'), max_length=128) username_validator = UnicodeUsernameValidator() username = models.CharField( _('username'), max_length=150, unique=True, help_text=_('Required. 150 characters or fewer. Letters, digits and @/./+/-/_ only.'), validators=[username_validator], error_messages={ 'unique': _("A user with that username already exists."), }, blank=True, null=True ) last_checkout_request = models.DateTimeField(_('otp expire'), max_length=30, blank=True, null=True) balance = models.PositiveBigIntegerField(_('balance'), default=0) iban = models.CharField(null=True, max_length=30, blank=True) iban_verified = models.BooleanField(null=True) updated_at = models.DateTimeField(_('last update'), max_length=30, blank=True, null=True, auto_now=True) last_login = models.DateTimeField(_('last login'), blank=True, null=True) date_joined = models.DateTimeField(_('date joined'), default=timezone.now) objects = UserManager() def __str__(self): return str(self.username or self.pk) @property def instance_id(self): return self.pk @property def instance_type(self): return TypeChoices.APPLICATION # TODO: user redis instead. # TODO: create a queryset class OAuthCode(models.Model): uuid = models.UUIDField(primary_key=True, editable=False, default=uuid.uuid4, unique=True, db_index=True) code_verifier = models.CharField(max_length=130, blank=True, null=True) code = models.CharField(max_length=64, blank=True, null=True) def generate_code_verifier(self): code_verifier = ''.join( random.choice(string.ascii_uppercase + string.digits) for _ in range(random.randint(43, 128))) self.code_verifier = code_verifier self.save() return code_verifier def generate_code_challenge(self): code_challenge = hashlib.sha256(self.code_verifier.encode('utf-8')).digest() code_challenge = base64.urlsafe_b64encode(code_challenge).decode('utf-8').replace('=', '') return code_challenge def generate_login_url(self): redirect_uri = 'http://127.0.0.1:8000/users/login/callback/' scope = '+'.join(settings.SCOPES.split(' ')) url = f'{settings.BASE_OAUTH2_PROVIDER_PUBLIC_URL}/authorize/?response_type=code&code_challenge={self.generate_code_challenge()}&code_challenge_method=S256&client_id={settings.CLIENT_ID}&scope={scope}&state={self.uuid}&redirect_uri={self.redirect_uri}' return url def validate_code(self, redirect_uri): headers = { "Content-Type": "application/x-www-form-urlencoded", } data = { "client_id": settings.CLIENT_ID, "client_secret": settings.CLIENT_SECRET, "code": self.code, "code_verifier": self.code_verifier, "redirect_uri": redirect_uri, "grant_type": "authorization_code" } response = requests.post(f'{settings.BASE_OAUTH2_PROVIDER_PRIVATE_URL}/token/', data=data, headers=headers) return response.json()