Use only one key/user for backup codes, better UX, handle recovery mode deactivation
This commit is contained in:
@@ -4,35 +4,42 @@ from django.http import HttpResponse
|
||||
from .Common import get_redirect_url
|
||||
from .models import *
|
||||
import simplejson
|
||||
from django.conf import settings
|
||||
import random
|
||||
import string
|
||||
import logging
|
||||
|
||||
def invalidate_token(key):
|
||||
key.enabled = 0
|
||||
key.save()
|
||||
|
||||
def token_left(request, username=None):
|
||||
if not username:
|
||||
username = request.user.username
|
||||
return len(User_Keys.objects.filter(username=username, key_type="RECOVERY", enabled=True))
|
||||
#TODO :
|
||||
# - Show authtificator panel on login everytime if RECOVERY is not deactivated
|
||||
# - Generation abuse checks
|
||||
|
||||
def token_left(request):
|
||||
uk = User_Keys.objects.filter(username=request.user.username, key_type="RECOVERY", enabled=True)
|
||||
keyLeft=0
|
||||
for key in uk:
|
||||
keyEnabled = key.properties["enabled"]
|
||||
for i in range(len(keyEnabled)):
|
||||
if keyEnabled[i]:
|
||||
keyLeft += 1
|
||||
return keyLeft
|
||||
|
||||
def delTokens(request):
|
||||
#Only when all MFA have been deactivated, or to generate new !
|
||||
#We iterate only to clean if any error happend and multiple entry of RECOVERY created for one user
|
||||
for key in User_Keys.objects.filter(username=request.user.username, key_type = "RECOVERY"):
|
||||
if key.username == request.user.username:
|
||||
key.delete()
|
||||
|
||||
def newTokens(username):
|
||||
for newkey in range(5):
|
||||
token = ''.join(random.choice(string.ascii_uppercase + string.ascii_lowercase + string.digits) for _ in range(6))
|
||||
uk=User_Keys()
|
||||
uk.username=username
|
||||
uk.properties={"secret_key":token}
|
||||
uk.key_type="RECOVERY"
|
||||
uk.enabled=True
|
||||
uk.save()
|
||||
# Separated from genTokens to be able to regenerate codes after login if last code has been used
|
||||
newKeys = []
|
||||
for i in range(5):
|
||||
token = ''.join(random.choice(string.ascii_uppercase + string.ascii_lowercase + string.digits) for _ in range(10))
|
||||
newKeys.append(token)
|
||||
uk=User_Keys()
|
||||
uk.username=username
|
||||
uk.properties={"secret_keys":newKeys, "enabled":[True for j in range(5)]}
|
||||
uk.key_type="RECOVERY"
|
||||
uk.save()
|
||||
|
||||
def genTokens(request, softGen=False):
|
||||
if not softGen or (softGen and token_left(request) == 0):
|
||||
@@ -44,24 +51,26 @@ def genTokens(request, softGen=False):
|
||||
return HttpResponse("Success")
|
||||
|
||||
|
||||
def verify_login(username,token):
|
||||
def verify_login(request, username,token):
|
||||
for key in User_Keys.objects.filter(username=username, key_type = "RECOVERY"):
|
||||
logging.critical(key.properties["secret_key"])
|
||||
if key.properties["secret_key"] == token and key.enabled:
|
||||
invalidate_token(key)
|
||||
newRecoveryGen = False
|
||||
if token_left(None, username) == 0:
|
||||
newRecoveryGen = True
|
||||
newTokens(username)
|
||||
return [True, key.id, newRecoveryGen]
|
||||
secret_keys = key.properties["secret_keys"]
|
||||
for i in range(len(secret_keys)):
|
||||
if token == secret_keys[i] and key.properties["enabled"][i]:
|
||||
key.properties["enabled"][i] = False
|
||||
key.save()
|
||||
if token_left(request) == 0:
|
||||
newTokens(username)
|
||||
return [True, key.id]
|
||||
return [False]
|
||||
|
||||
def getTokens(request):
|
||||
tokens = []
|
||||
enable = []
|
||||
for key in User_Keys.objects.filter(username=request.user.username, key_type = "RECOVERY"):
|
||||
tokens.append(key.properties["secret_key"])
|
||||
enable.append(1 if key.enabled else 0)
|
||||
secret_keys = key.properties["secret_keys"]
|
||||
for i in range(len(secret_keys)):
|
||||
tokens.append(secret_keys[i])
|
||||
enable.append(key.properties["enabled"][i])
|
||||
return HttpResponse(simplejson.dumps({"keys":tokens, "enable":enable}))
|
||||
|
||||
@never_cache
|
||||
|
||||
@@ -51,7 +51,7 @@
|
||||
<span class="input-group-addon input-group-text">
|
||||
<i class="glyphicon glyphicon-lock bi bi-lock"></i>
|
||||
</span>
|
||||
<input class="form-control" size="6" MaxLength="6" value="" placeholder="e.g 55552" name="otp" type="text" id="otp" autofocus>
|
||||
<input class="form-control" size="6" MaxLength="10" value="" placeholder="e.g 55552" name="otp" type="text" id="otp" autofocus>
|
||||
|
||||
</div>
|
||||
</div>
|
||||
|
||||
@@ -15,7 +15,7 @@
|
||||
{% for method in request.session.mfa_methods %}
|
||||
|
||||
<li><a href="{% url "mfa_goto" method %}">
|
||||
{% if method == "TOTP" %}Authenticator App
|
||||
{% if method == "TOTP" %}Authenticator App / Backup codes
|
||||
{% elif method == "Email" %}Send OTP by Email
|
||||
{% elif method == "U2F" %}Secure Key
|
||||
{% elif method == "FIDO2" %}FIDO2 Secure Key
|
||||
|
||||
41
mfa/totp.py
41
mfa/totp.py
@@ -41,24 +41,29 @@ def recheck(request):
|
||||
def auth(request):
|
||||
context=csrf(request)
|
||||
if request.method=="POST":
|
||||
res=verify_login(request,request.session["base_username"],token = request.POST["otp"])
|
||||
resBackup=recovery.verify_login(request.session["base_username"], token=request.POST["otp"])
|
||||
if res[0]:
|
||||
mfa = {"verified": True, "method": "TOTP","id":res[1]}
|
||||
if getattr(settings, "MFA_RECHECK", False):
|
||||
mfa["next_check"] = datetime.datetime.timestamp((datetime.datetime.now()
|
||||
+ datetime.timedelta(
|
||||
seconds=random.randint(settings.MFA_RECHECK_MIN, settings.MFA_RECHECK_MAX))))
|
||||
request.session["mfa"] = mfa
|
||||
return login(request)
|
||||
elif resBackup[0]:
|
||||
mfa = {"verified": True, "method": "TOTP","id":resBackup[1], "newRecoveryGen":resBackup[2]}
|
||||
if getattr(settings, "MFA_RECHECK", False):
|
||||
mfa["next_check"] = datetime.datetime.timestamp((datetime.datetime.now()
|
||||
+ datetime.timedelta(
|
||||
seconds=random.randint(settings.MFA_RECHECK_MIN, settings.MFA_RECHECK_MAX))))
|
||||
request.session["mfa"] = mfa
|
||||
return login(request)
|
||||
tokenLength = len(request.POST["otp"])
|
||||
if tokenLength == 6:
|
||||
#TOTO code check
|
||||
res=verify_login(request,request.session["base_username"],token = request.POST["otp"])
|
||||
if res[0]:
|
||||
mfa = {"verified": True, "method": "TOTP","id":res[1]}
|
||||
if getattr(settings, "MFA_RECHECK", False):
|
||||
mfa["next_check"] = datetime.datetime.timestamp((datetime.datetime.now()
|
||||
+ datetime.timedelta(
|
||||
seconds=random.randint(settings.MFA_RECHECK_MIN, settings.MFA_RECHECK_MAX))))
|
||||
request.session["mfa"] = mfa
|
||||
return login(request)
|
||||
elif tokenLength == 10 and "RECOVERY" not in settings.MFA_UNALLOWED_METHODS:
|
||||
#Backup code check
|
||||
resBackup=recovery.verify_login(request.session["base_username"], token=request.POST["otp"])
|
||||
if resBackup[0]:
|
||||
mfa = {"verified": True, "method": "RECOVERY","id":resBackup[1]}
|
||||
if getattr(settings, "MFA_RECHECK", False):
|
||||
mfa["next_check"] = datetime.datetime.timestamp((datetime.datetime.now()
|
||||
+ datetime.timedelta(
|
||||
seconds=random.randint(settings.MFA_RECHECK_MIN, settings.MFA_RECHECK_MAX))))
|
||||
request.session["mfa"] = mfa
|
||||
return login(request)
|
||||
context["invalid"]=True
|
||||
return render(request,"TOTP/Auth.html", context)
|
||||
|
||||
|
||||
@@ -39,6 +39,11 @@ def verify(request,username):
|
||||
return login(request)
|
||||
methods.remove("Trusted Device")
|
||||
request.session["mfa_methods"] = methods
|
||||
|
||||
if "TOTP" not in methods and "RECOVERY" not in settings.MFA_UNALLOWED_METHODS:
|
||||
#Add the "totp" option if user doesn't have totp auth (case with fido auth and backup code for instace)
|
||||
methods.append("TOTP")
|
||||
|
||||
if len(methods)==1:
|
||||
return HttpResponseRedirect(reverse(methods[0].lower()+"_auth"))
|
||||
return show_methods(request)
|
||||
|
||||
Reference in New Issue
Block a user