Refactoring, better tests

This commit is contained in:
Spitap
2022-12-28 18:42:58 +01:00
parent 9ad9bf9ce4
commit 0146175153
9 changed files with 270 additions and 185 deletions
+1 -4
View File
@@ -8,12 +8,9 @@ export default {
return repository.post('/password_reset/', data)
},
checkSmsTotp (data) {
return repository.post('/confirm_sms_totp/', data)
return repository.post('/sms_totp/', data)
},
changePassword (data) {
return repository.post('/reset_confirm/', data)
},
resendSmsTotp (data) {
return repository.post('/reset_sms_resend/', data)
}
}
@@ -104,11 +104,13 @@ export default {
this.loading = true
auth.changePassword(payload).then(resp => {
document.getElementById('password_validation').innerHTML = ''
this.loading = false
this.returnLogin()
if (resp.status === 200) {
this.loading = false
this.returnLogin()
}
}).catch(err => {
this.loading = false
if (err.response.status === 401) {
if (err.response.status === 403) {
this.$refs.observer.setErrors({
password_confirmed: this.$gettext('Invalid reset token.')
})
@@ -116,9 +118,9 @@ export default {
this.$refs.observer.setErrors({
password_confirmed: this.$gettext('User unknown.')
})
} else if (err.response.status === 455) {
} else if (err.response.status === 400 && err.response.data.type === 'password_requirement') {
let message = ''
err.response.data.forEach(element => {
err.response.data.errors.forEach(element => {
message += this.$gettext(element) + '<br>'
})
document.getElementById('password_validation').innerHTML = message
@@ -62,30 +62,18 @@ export default {
returnLogin () {
this.$router.push({ name: 'Login' })
},
async showDialogEmail () {
const confirm = await this.$refs.dialog_response.open(
this.$gettext('Info'),
this.$gettext(
'Email sent, please check your inbox'
),
{
color: 'info',
cancelLabel: this.$gettext('Ok'),
agreeLabel: this.$gettext('Return to login')
})
if (!confirm) {
return
async showDialog (title, body, error = false) {
let color = 'info'
if (error) {
color = 'error'
}
this.returnLogin()
},
async emailFailedToSend () {
const confirm = await this.$refs.dialog_response.open(
this.$gettext('Error'),
this.$gettext(title),
this.$gettext(
'Failed to send email, please check with you administrator'
body
),
{
color: 'error',
color: color,
cancelLabel: this.$gettext('Ok'),
agreeLabel: this.$gettext('Return to login')
})
@@ -105,20 +93,23 @@ export default {
this.loading = true
auth.recoverPassword(payload).then(resp => {
this.loading = false
if (resp.status === 233) {
this.$router.push({ name: 'PasswordRecoverySmsTotpForm' })
} else if (resp.status === 210) {
this.showDialogEmail()
if (resp.status === 200) {
if (resp.data.type === 'sms') {
this.$router.push({ name: 'PasswordRecoverySmsTotpForm' })
} else if (resp.data.type === 'email') {
this.showDialog('Info', 'Email sent, please check your inbox')
}
}
}).catch(err => {
this.loading = false
if (err.response.status === 404) {
// User not found
this.$refs.observer.setErrors({
email: this.$gettext('Invalid email')
})
} else if (err.response.status === 502) {
this.emailFailedToSend()
if (err.response.data.type === 'sms' || err.response.data.type === 'email') {
this.$refs.observer.setErrors({
email: this.$gettext(err.response.data.reason)
})
}
} else if (err.response.status === 503 && err.response.data.type === 'email') {
this.showDialog('Error', err.response.data.reason, true)
}
})
}
@@ -10,7 +10,7 @@
<v-col cols="12" sm="6" class="d-flex flex-column justify-center pa-10" @keyup.enter="changePassword">
<confirm-dialog ref="confirm_wrong_user">
</confirm-dialog>
<span class="primary--text mb-6 text-h4"><translate>Password recovert</translate></span>
<span class="primary--text mb-6 text-h4"><translate>Password recovery</translate></span>
<validation-observer ref="observer">
<validation-provider
vid="sms_totp"
@@ -31,7 +31,7 @@
class="flex-grow-1"
color="primary"
large
@click="changePassword"
@click="checkSmsTotp"
:loading="loading"
>
<translate>Submit</translate>
@@ -69,9 +69,9 @@ export default {
},
async userPkError () {
const confirm = await this.$refs.confirm.open(
this.$gettext('Warning'),
this.$gettext('Error'),
this.$gettext(
'User seems wrong, return to login or restart reset process?'
'User seems wrong, return to login or restart reset the process?'
),
{
color: 'error',
@@ -86,13 +86,18 @@ export default {
}
},
async resendSms () {
auth.resendSmsTotp().then(resp => {
auth.checkSmsTotp({ type: 'resend' }).then(resp => {
this.loading = false
// Reset done !
if (resp.status === 200) {
this.$refs.observer.setErrors({
password_confirmed: this.$gettext('TOTP resent.')
})
}
}).catch(err => {
console.error(err)
this.loading = false
this.userPkError()
if (err.response.status === 400) {
this.loading = false
this.userPkError()
}
})
},
async checkSmsTotp () {
@@ -101,12 +106,13 @@ export default {
return
}
const payload = {
type: 'confirm',
sms_totp: this.sms_totp
}
this.loading = true
auth.checkSmsTotp(payload).then(resp => {
this.loading = false
if (resp.response.status === 200) {
if (resp.status === 200) {
// Code is good, redirecting to change password form.
this.$router.push({ name: 'PasswordRecoveryChangeForm', data: resp.response.data })
}
@@ -114,7 +120,7 @@ export default {
this.loading = false
if (err.response.status === 500) {
this.$refs.observer.setErrors({
password_confirmed: this.$gettext('Invalid totp, try resend')
password_confirmed: this.$gettext('Invalid totp.')
})
}
})
+72 -75
View File
@@ -8,35 +8,39 @@ from django.contrib.auth import password_validation
from django.contrib.sites.shortcuts import get_current_site
from django.contrib.auth.tokens import default_token_generator
from django.core.mail import send_mail
from django.core.validators import validate_email
from django.core.exceptions import ValidationError as djangoValidationError
from django.template import loader
from django.utils import formats
from django.utils.encoding import force_text
from django.utils.http import urlsafe_base64_encode, urlsafe_base64_decode
from django.utils.encoding import force_bytes
from django.utils.translation import ugettext_lazy, ugettext as _
from rest_framework import serializers
from rest_framework import serializers, status
from rest_framework.exceptions import APIException
from modoboa.core import constants, models, sms_backends, app_settings, context_processors
from modoboa.core.models import User
from modoboa.lib import fields as lib_fields, cryptutils
class CustomValidationError(APIException):
status_code = status.HTTP_400_BAD_REQUEST
default_detail = 'A server error occurred.'
def __init__(self, detail, status_code):
if status_code is not None:
self.status_code = status_code
if detail is not None:
self.detail = detail
else:
self.detail = {'detail': force_text(self.default_detail)}
class NoSMSAvailable(Exception):
""" Raised when no sms totp is availible for password reset (to try email)."""
pass
class UserNotFound(Exception):
""" Raised when no valid user has been found (to have a proper 404 http code instead of 400."""
pass
class EmailSendingFailure(Exception):
""" Raised when a reset email has failed to send."""
""" Raised when no sms totp is available for password reset (to try email)."""
pass
@@ -48,11 +52,6 @@ class PasswordRequirementsFailure(Exception):
self.message_list = message_list
class InvalidToken(Exception):
""" Raised during password reset if token is not valid."""
pass
class CoreGlobalParametersSerializer(serializers.Serializer):
"""A serializer for global parameters."""
@@ -281,7 +280,7 @@ class UserAPITokenSerializer(serializers.Serializer):
token = serializers.CharField()
class EmailPasswordRecoveryInitSerializer(serializers.Serializer):
class PasswordRecoveryEmailSerializer(serializers.Serializer):
"""Serializer used by password recovery route."""
email = serializers.EmailField()
@@ -293,7 +292,7 @@ class EmailPasswordRecoveryInitSerializer(serializers.Serializer):
.exclude(Q(secondary_email__isnull=True) | Q(secondary_email=""))
).first()
if self.context["user"] is None:
raise UserNotFound()
raise CustomValidationError({"type": "email", "reason": "No valid user found."}, 404)
return data
def save(self):
@@ -318,14 +317,21 @@ class EmailPasswordRecoveryInitSerializer(serializers.Serializer):
subject = ''.join(subject.splitlines())
body = loader.render_to_string(
"registration/password_reset_email_v2.html", context)
try:
if send_mail(subject, body, None, [to_email]) == 0:
raise EmailSendingFailure()
except:
raise EmailSendingFailure()
send_mail(subject, body, None, [to_email], fail_silently=False)
class SMSPasswordRecoveryInitSerializer(serializers.Serializer):
def send_sms_code(secret, backend, user):
""" Send a sms containing a TOTP code for password reset. """
code = oath.totp(secret)
text = _(
"Please use the following code to recover your Modoboa password: {}"
.format(code)
)
if not backend.send(text, [str(user.phone_number)]):
raise NoSMSAvailable()
class PasswordRecoverySmsSerializer(serializers.Serializer):
email = serializers.EmailField()
@@ -333,6 +339,10 @@ class SMSPasswordRecoveryInitSerializer(serializers.Serializer):
request = self.context["request"]
clean_email = data["email"]
user = User.objects.filter(email__iexact=clean_email, is_active=True).first()
if user is None:
raise CustomValidationError({"type": "sms", "reason": "No valid user found."}, 404)
self.context["user"] = (User.objects.filter(
email__iexact=clean_email, is_active=True)
.exclude(Q(phone_number__isnull=True) | Q(phone_number=""))
@@ -349,48 +359,63 @@ class SMSPasswordRecoveryInitSerializer(serializers.Serializer):
backend = sms_backends.get_active_backend(
request.localconfig.parameters)
secret = cryptutils.random_hex_key(20)
code = oath.totp(secret)
text = _(
"Please use the following code to recover your Modoboa password: {}"
.format(code)
)
if not backend.send(text, [str(user.phone_number)]):
raise NoSMSAvailable()
send_sms_code(secret, backend, user)
request.session["totp_secret"] = secret
request.session["user_pk"] = user.pk
return data
class PasswordRecoverySmsSerializer(serializers.Serializer):
class PasswordRecoverySmsConfirmSerializer(serializers.Serializer):
sms_totp = serializers.CharField(min_length=6, max_length=6)
def validate(self, data):
try:
totp_secret = self.context["request"].session["totp_secret"]
self.context["request"].session.pop("totp_secret")
self.context["totp_secret"] = self.context["request"].session["totp_secret"]
except KeyError:
raise serializers.ValidationError(
"totp secret not set in session", 403)
raise CustomValidationError(
{"reason": "totp secret not set in session"}, 403)
if not oath.accept_totp(totp_secret, data["sms_totp"]):
raise serializers.ValidationError("Wrong totp", 403)
if not oath.accept_totp(self.context["totp_secret"], data["sms_totp"])[0]:
raise CustomValidationError({"reason": "Wrong totp"}, 403)
# Attempt to get user, will raise an error if pk is not valid
self.user = User.objects.get(
pk=self.context["user_pk"])
self.context["user"] = User.objects.get(
pk=self.context["request"].session["user_pk"])
return data
def save(self):
self.context["request"].session.pop("totp_secret")
self.context["request"].session.pop("user")
token = default_token_generator.make_token(self.user)
uid = urlsafe_base64_encode(force_bytes(self.user.pk))
user_pk = self.context["request"].session.pop("user_pk")
token = default_token_generator.make_token(self.context["user"])
uid = urlsafe_base64_encode(force_bytes(user_pk))
self.context["response"] = (token, uid)
return True
class PasswordRecoverySmsResendSerializer(serializers.Serializer):
def validate(self, data):
try:
self.context["totp_secret"] = self.context["request"].session["totp_secret"]
except KeyError:
raise CustomValidationError(
{"reason": "totp secret not set in session"}, status.HTTP_403_FORBIDDEN)
# Attempt to get user, will raise an error if pk is not valid
self.context["user"] = User.objects.get(
pk=self.context["request"].session["user_pk"])
return data
def save(self):
backend = sms_backends.get_active_backend(
self.context["request"].localconfig.parameters)
send_sms_code(self.context["totp_secret"], backend, self.context["user"])
return True
class PasswordRecoveryConfirmSerializer(serializers.Serializer):
new_password1 = serializers.CharField()
@@ -419,10 +444,10 @@ class PasswordRecoveryConfirmSerializer(serializers.Serializer):
user = self.get_user(data["id"])
if user is None:
raise UserNotFound
raise CustomValidationError({"reason": "User not found."}, 404)
if not default_token_generator.check_token(user, data["token"]):
raise InvalidToken
raise CustomValidationError({"reason": "Wrong reset token"}, 403)
# Check that the password works with set conditions
try:
@@ -440,31 +465,3 @@ class PasswordRecoveryConfirmSerializer(serializers.Serializer):
return True
class PasswordRecoverySmsResendSerializer(serializers.Serializer):
def validate(self, data):
try:
self.context["session"].user_pk
self.context["session"].totp_token
return data
except KeyError:
raise serializers.ValidationError(
"No previous reset attempt recorded", 403)
def save(self):
user = User.objects.get(
pk=self.context["session"].user_pk)
request = self.context["request"]
backend = sms_backends.get_active_backend(
request.localconfig.parameters)
secret = cryptutils.random_hex_key(20)
code = oath.totp(secret)
text = _(
"Please use the following code to recover your Modoboa password: {}"
.format(code)
)
if not backend.send(text, [str(user.phone_number)]):
return False
request.session["totp_secret"] = secret
request.session["user_pk"] = user.pk
+104 -21
View File
@@ -1,6 +1,7 @@
"""Core API related tests."""
import copy
import oath
from unittest import mock
from django.urls import reverse
@@ -114,14 +115,6 @@ class ParametersAPITestCase(ModoAPITestCase):
class AccountViewSetTestCase(ModoAPITestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super().setUpTestData()
factories.populate_database()
cls.da_token = Token.objects.create(
user=models.User.objects.get(username="admin@test.com"))
def test_me(self):
url = reverse("v2:account-me")
resp = self.client.get(url)
@@ -209,6 +202,21 @@ class AccountViewSetTestCase(ModoAPITestCase):
resp = self.client.post(url)
self.assertEqual(resp.status_code, 201)
class PasswordResetTestCase(ModoAPITestCase):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.id = 0
self.sms_token = 0
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super().setUpTestData()
factories.populate_database()
cls.da_token = Token.objects.create(
user=models.User.objects.get(username="admin@test.com"))
@mock.patch("ovh.Client.get")
@mock.patch("ovh.Client.post")
def test_reset_password(self, client_post, client_get):
@@ -217,7 +225,29 @@ class AccountViewSetTestCase(ModoAPITestCase):
response = self.client.post(url)
self.assertEqual(response.status_code, 400)
# SMS parameter
# Invalid user
data = {"email": "doesntexist@whoami.com"}
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 404)
self.assertEqual(response.json()["type"], "sms")
account = models.User.objects.get(username="user@test.com")
# Email reset test
# No secondary email
data = {"email": account.email}
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 404)
self.assertEqual(response.json()["type"], "email")
# With secondary email
account.secondary_email = "toto@test.com"
account.is_active = True
account.save()
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()["type"], "email")
# SMS reset test
self.set_global_parameters({
"sms_password_recovery": True,
"sms_provider": "ovh",
@@ -225,24 +255,77 @@ class AccountViewSetTestCase(ModoAPITestCase):
"sms_ovh_application_secret": "secret",
"sms_ovh_consumer_key": "consumer"
}, app="core")
account = models.User.objects.get(username="user@test.com")
data = {"email": account.email}
# Provide a user w/o phone or secondary
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 404)
# Secondary email for reset
account.secondary_email = "toto@test.com"
account.is_active = True
account.save()
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 204)
# Phone number
account.phone_number = "+33612345678"
account.save()
client_get.return_value = ["service"]
client_post.return_value = {"totalCreditsRemoved": 1}
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 233)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()["type"], "sms")
@mock.patch("ovh.Client.get")
@mock.patch("ovh.Client.post")
def test_password_reset_sms_totp(self, client_post, client_get):
"""Test reset password by SMS."""
url_create_sms = reverse("v2:password_reset")
url_sms_totp = reverse("v2:sms_totp")
# Prepare account
account = models.User.objects.get(username="user@test.com")
account.phone_number = "+33612345678"
account.secondary_email = "toto@test.com"
account.is_active = True
account.save()
# Send sms
client_get.return_value = ["service"]
client_post.return_value = {"totalCreditsRemoved": 1}
self.set_global_parameters({
"sms_password_recovery": True,
"sms_provider": "ovh",
"sms_ovh_application_key": "key",
"sms_ovh_application_secret": "secret",
"sms_ovh_consumer_key": "consumer"
})
self.client.logout()
self.create_session()
data = {"email": account.email}
response = self.client.post(url_create_sms, data, format="json")
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()["type"], "sms")
self.assertIn("totp_secret", self.client.session)
# Get the secret
session = self.client.session
secret = session["totp_secret"]
# Fail to provide type
data = {"sms_totp": "123456"}
response = self.client.post(url_sms_totp, data, format="json")
self.assertEqual(response.status_code, 400)
# Resend sms
data = {"type": "resend"}
response = self.client.post(url_sms_totp, data, format="json")
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()["type"], "resend")
# False totp code
data = {"sms_totp": "123456", "type": "confirm"}
response = self.client.post(url_sms_totp, data, format="json")
self.assertEqual(response.status_code, 403)
self.assertEqual(response.json()["reason"], "Wrong totp")
# Good totp code
data = {"sms_totp": oath.totp(secret), "type": "confirm"}
response = self.client.post(url_sms_totp, data, format="json")
self.assertEqual(response.status_code, 200)
self.assertIn("token", response.json())
self.assertIn("id", response.json())
self.sms_token = response.json()["token"]
self.id = response.json()["id"]
class AuthenticationTestCase(ModoAPITestCase):
+2 -4
View File
@@ -26,8 +26,6 @@ urlpatterns += [
name="password_reset"),
path('reset_confirm/', views.PasswordResetConfirmView.as_view(),
name="password_reset_confirm_v2"),
path('confirm_sms_totp/', views.PasswordResetConfirmSmsCodeView.as_view(),
name="confirm_sms_totp"),
path('reset_sms_resend', views.PasswordResetResendSmsCodeView.as_view(),
name='reset_sms_resend'),
path('sms_totp/', views.PasswordResetSmsTOTP.as_view(),
name="sms_totp"),
]
+37 -36
View File
@@ -4,6 +4,7 @@ import logging
from django.utils.html import escape
from django.utils.translation import ugettext as _
from django.utils.datastructures import MultiValueDictKeyError
from django.contrib.auth import login
@@ -15,6 +16,8 @@ from rest_framework.views import APIView
from modoboa.core.password_hashers import get_password_hasher
from modoboa.parameters import tools as param_tools
from smtplib import SMTPException
from .serializers import *
logger = logging.getLogger("modoboa.auth")
@@ -81,51 +84,60 @@ class EmailPasswordResetView(APIView):
"""
def post(self, request, *args, **kwargs):
serializer = EmailPasswordRecoveryInitSerializer(
serializer = PasswordRecoveryEmailSerializer(
data=request.data, context={'request': request})
try:
serializer.is_valid(raise_exception=True)
except UserNotFound:
return response.Response({"no valid user found"}, 404)
serializer.is_valid(raise_exception=True)
try:
serializer.save()
except EmailSendingFailure:
return response.Response({"Email failed to send, possibly a misconfiguration?"}, 502)
except SMTPException:
return response.Response({
"type": "email",
"reason": "Error while sending the email. Please contact an administrator."
},
503)
# Email response
return response.Response(status=204)
return response.Response({"type": "email"}, 200)
class DefaultPasswordResetView(EmailPasswordResetView):
"""
"""
Works with PasswordRecoveryForm.vue.
First checks if SMS recovery is available, else switch to super (Email recovery [with seconday email]).
First checks if SMS recovery is available, else switch to super (Email recovery [with secondary email]).
"""
def post(self, request, *args, **kwargs):
"""Recover password."""
serializer = SMSPasswordRecoveryInitSerializer(
serializer = PasswordRecoverySmsSerializer(
data=request.data, context={'request': request})
try:
serializer.is_valid(raise_exception=True)
except NoSMSAvailable:
return super().post(request, *args, **kwargs)
except NoSMSAvailable as e:
if type(e) == NoSMSAvailable:
return super().post(request, *args, **kwargs)
# SMS response
return response.Response(status=233)
return response.Response({"type": "sms"}, 200)
class PasswordResetConfirmSmsCodeView(APIView):
class PasswordResetSmsTOTP(APIView):
""" Check SMS Totp code. """
def post(self, request, *args, **kwargs):
serializer = PasswordRecoverySmsSerializer(data=request.data)
try:
if request.data["type"] == "confirm":
serializer = PasswordRecoverySmsConfirmSerializer(data=request.data, context={'request': request})
elif request.data["type"] == "resend":
serializer = PasswordRecoverySmsResendSerializer(data=request.data, context={'request': request})
except (MultiValueDictKeyError, KeyError):
return response.Response({"reason": "No type provided."}, 400)
serializer.is_valid(raise_exception=True)
serializer.save()
payload = serializer.context["response"]
payload = {"token": payload[0], "id": payload[1]}
payload = {"type": "resend"}
if request.data["type"] == "confirm":
serializer_response = serializer.context["response"]
payload.update({"token": serializer_response[0], "id": serializer_response[1], "type": "confirm"})
return response.Response(payload, 200)
class PasswordResetConfirmView(APIView):
""" Get and set new user password. """
@@ -134,26 +146,15 @@ class PasswordResetConfirmView(APIView):
serializer = PasswordRecoveryConfirmSerializer(data=request.data)
try:
serializer.is_valid(raise_exception=True)
except (PasswordRequirementsFailure, UserNotFound, InvalidToken) as e:
except PasswordRequirementsFailure as e:
if type(e) == PasswordRequirementsFailure:
data = {"type": "password_requirement"}
errors = []
for element in e.message_list:
errors.append(element)
return response.Response(errors, 455)
elif type(e) == UserNotFound:
return response.Response(status=404)
elif type(e) == InvalidToken:
return response.Response(status=401)
data.update({"errors": errors})
return response.Response(data, 400)
serializer.save()
return response.Response(status=200)
class PasswordResetResendSmsCodeView(APIView):
def post(self, request, *args, **kwargs):
serializer = PasswordRecoverySmsResendSerializer(
data=request.data, context={"request": request})
serializer.is_valid(raise_exception=True)
serializer.save()
return response.Response(status=200)
+10
View File
@@ -2,9 +2,12 @@
import socket
from django.conf import settings
from django.core import management
from django.test import TestCase
from importlib import import_module
from rest_framework.authtoken.models import Token
from rest_framework.test import APITestCase
@@ -99,3 +102,10 @@ class ModoAPITestCase(ParametersMixin, APITestCase):
"""Setup."""
super(ModoAPITestCase, self).setUp()
self.client.credentials(HTTP_AUTHORIZATION="Token " + self.token.key)
def create_session(self):
"""Enable session storage across requests."""
session_engine = import_module(settings.SESSION_ENGINE)
store = session_engine.SessionStore()
store.save()
self.client.cookies[settings.SESSION_COOKIE_NAME] = store.session_key