160 lines
5.1 KiB
Python
160 lines
5.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
#
|
|
# ITerativ GmbH
|
|
# http://www.iterativ.ch/
|
|
#
|
|
# Copyright (c) 2019 ITerativ GmbH. All rights reserved.
|
|
#
|
|
# Created on 2019-10-01
|
|
# @author: chrigu <christian.cueni@iterativ.ch>
|
|
import re
|
|
|
|
import graphene
|
|
from django.contrib.auth import authenticate, login, logout
|
|
from django.contrib.auth.forms import PasswordResetForm
|
|
from django.contrib.auth.views import PasswordResetView, PasswordResetConfirmView, INTERNAL_RESET_URL_TOKEN
|
|
from graphene import relay
|
|
|
|
from users.models import User
|
|
|
|
|
|
class FieldError(graphene.ObjectType):
|
|
code = graphene.String()
|
|
|
|
|
|
class MutationError(graphene.ObjectType):
|
|
field = graphene.String()
|
|
errors = graphene.List(FieldError)
|
|
|
|
|
|
class Login(relay.ClientIDMutation):
|
|
class Input:
|
|
username_input = graphene.String()
|
|
password_input = graphene.String()
|
|
|
|
success = graphene.Boolean()
|
|
errors = graphene.List(MutationError) # todo: change for consistency
|
|
|
|
@classmethod
|
|
def mutate_and_get_payload(cls, root, info, **kwargs):
|
|
|
|
user = authenticate(username=kwargs.get('username_input'), password=kwargs.get('password_input'))
|
|
if user is not None:
|
|
login(info.context, user)
|
|
return cls(success=True, errors=[])
|
|
else:
|
|
return cls(success=False, errors=['invalid_credentials'])
|
|
|
|
|
|
class PasswordReset(relay.ClientIDMutation):
|
|
class Input:
|
|
email_input = graphene.String()
|
|
|
|
success = graphene.Boolean()
|
|
errors = graphene.List(MutationError)
|
|
|
|
@classmethod
|
|
def mutate_and_get_payload(cls, root, info, **kwargs):
|
|
email = kwargs.get('email_input')
|
|
try:
|
|
user = User.objects.get(email=email) # todo: make lowercase
|
|
except User.DoesNotExist:
|
|
return cls(success=False, errors=['invalid_email'])
|
|
|
|
password_reset_view = PasswordResetView()
|
|
password_reset_view.request = info.context
|
|
form = password_reset_view.form_class({'email': user.email})
|
|
|
|
if not form.is_valid():
|
|
return cls(success=False, errors=form.errors)
|
|
|
|
try:
|
|
password_reset_view.form_valid(form)
|
|
except Exception:
|
|
return cls(success=False, errors=['email_error'])
|
|
|
|
return cls(success=True, errors=[])
|
|
|
|
|
|
class PasswordConfirm:
|
|
@classmethod
|
|
def verify_token_and_uidb64(cls, token, uidb64, request):
|
|
password_reset_confirm_view = PasswordResetConfirmView(request=request)
|
|
return password_reset_confirm_view.dispatch(uidb64=uidb64, token=token, request=request)
|
|
|
|
|
|
class PasswordResetVerify(relay.ClientIDMutation, PasswordConfirm):
|
|
class Input:
|
|
token_input = graphene.String()
|
|
uidb64_input = graphene.String()
|
|
|
|
success = graphene.Boolean()
|
|
errors = graphene.List(MutationError)
|
|
|
|
@classmethod
|
|
def mutate_and_get_payload(cls, root, info, **kwargs):
|
|
uidb64 = kwargs.get('uidb64_input')
|
|
token = kwargs.get('token_input')
|
|
|
|
http_response = cls.verify_token_and_uidb64(token, uidb64, info.context)
|
|
|
|
# PasswordResetConfirmView returns a webpage if either token or uidb64 are invalid (HTTP 200)
|
|
# if token and uidb64 are correct a redirect is returned (HTTP302)
|
|
|
|
if http_response.status_code == 302:
|
|
return cls(success=True, errors=[])
|
|
else:
|
|
return cls(success=False, errors=['invalid_challenge'])
|
|
|
|
|
|
class PasswordResetSetPassword(relay.ClientIDMutation, PasswordConfirm):
|
|
class Input:
|
|
new_password_input = graphene.String()
|
|
confirm_new_password_input = graphene.String()
|
|
uidb64_input = graphene.String()
|
|
|
|
success = graphene.Boolean()
|
|
errors = graphene.List(MutationError)
|
|
|
|
@classmethod
|
|
def mutate_and_get_payload(cls, root, info, **kwargs):
|
|
uidb64 = kwargs.get('uidb64_input')
|
|
new_password = kwargs.get('new_password_input')
|
|
confirm_new_password_input = kwargs.get('confirm_new_password_input')
|
|
|
|
# fake ordinary POST for view
|
|
info.context.POST = {
|
|
'new_password1': new_password,
|
|
'new_password2': confirm_new_password_input
|
|
}
|
|
|
|
if not cls.verify_strong_password(new_password):
|
|
return cls(success=False, errors=['invalid_passwords'])
|
|
|
|
http_response = cls.verify_token_and_uidb64(INTERNAL_RESET_URL_TOKEN, uidb64, info.context)
|
|
|
|
if http_response.status_code == 302:
|
|
return cls(success=True, errors=[])
|
|
else:
|
|
return cls(success=False, errors=['invalid_passwords']) # todo: check error from form
|
|
|
|
@classmethod
|
|
def verify_strong_password(cls, password):
|
|
if len(password) == 0:
|
|
return password
|
|
|
|
has_number = re.search('\d', password)
|
|
has_upper = re.search('[A-Z]', password)
|
|
has_lower = re.search('[a-z]', password)
|
|
has_special = re.search('[!@#$%^&*(),.?":{}|<>\+]', password)
|
|
|
|
return has_number and has_upper and has_lower and has_special
|
|
|
|
class UserMutations:
|
|
login = Login.Field()
|
|
password_reset = PasswordReset.Field()
|
|
password_reset_verify = PasswordResetVerify.Field()
|
|
password_reset_set_password = PasswordResetSetPassword.Field()
|
|
|
|
|