Added support for two-step authentication. The login function will
now email a 6-digit code to users if the user is part of a group with role 'web2py TWo-Step Authentication'.
This commit is contained in:
@@ -3,7 +3,7 @@
|
|||||||
<h2>{{=T( request.args(0).replace('_',' ').capitalize() )}}</h2>
|
<h2>{{=T( request.args(0).replace('_',' ').capitalize() )}}</h2>
|
||||||
<div id="web2py_user_form">
|
<div id="web2py_user_form">
|
||||||
{{
|
{{
|
||||||
if request.args(0)=='login':
|
if request.args(0)=='login' and not session.auth_2_factor_user:
|
||||||
if not 'register' in auth.settings.actions_disabled:
|
if not 'register' in auth.settings.actions_disabled:
|
||||||
form.add_button(T('Register'),URL(args='register', vars={'_next': request.vars._next} if request.vars._next else None),_class='btn')
|
form.add_button(T('Register'),URL(args='register', vars={'_next': request.vars._next} if request.vars._next else None),_class='btn')
|
||||||
pass
|
pass
|
||||||
|
|||||||
+232
-125
@@ -2214,6 +2214,17 @@ class Auth(object):
|
|||||||
_code='INVALID TICKET'))
|
_code='INVALID TICKET'))
|
||||||
raise HTTP(200, message)
|
raise HTTP(200, message)
|
||||||
|
|
||||||
|
def _reset_2_factor_auth(self, session):
|
||||||
|
'''When two-step authentication is enabled, this function is used to
|
||||||
|
clear the session after successfully completing second challenge
|
||||||
|
or when the maximum number of tries allowed has expired.
|
||||||
|
'''
|
||||||
|
session.auth_2_factor_user = None
|
||||||
|
session.auth_2_factor = None
|
||||||
|
session.auth_2_factor_enabled = False
|
||||||
|
# Allow up to 4 attempts (the 1st one plus 3 more)
|
||||||
|
session.auth_2_factor_tries_left = 3
|
||||||
|
|
||||||
def login(
|
def login(
|
||||||
self,
|
self,
|
||||||
next=DEFAULT,
|
next=DEFAULT,
|
||||||
@@ -2293,107 +2304,108 @@ class Auth(object):
|
|||||||
old_requires = table_user[username].requires
|
old_requires = table_user[username].requires
|
||||||
table_user[username].requires = tmpvalidator
|
table_user[username].requires = tmpvalidator
|
||||||
|
|
||||||
# do we use our own login form, or from a central source?
|
# If two-factor authentication is enabled, and the maximum
|
||||||
if settings.login_form == self:
|
# number of tries allowed is used up, reset the session to
|
||||||
form = SQLFORM(
|
# pre-login state with two-factor auth
|
||||||
table_user,
|
if session.auth_2_factor_enabled and session.auth_2_factor_tries_left < 1:
|
||||||
fields=[username, passfield],
|
# Exceeded maximum allowed tries for this code. Require user to enter
|
||||||
hidden=dict(_next=next),
|
# username and password again.
|
||||||
showid=settings.showid,
|
user = None
|
||||||
submit_button=self.messages.login_button,
|
|
||||||
delete_label=self.messages.delete_label,
|
|
||||||
formstyle=settings.formstyle,
|
|
||||||
separator=settings.label_separator
|
|
||||||
)
|
|
||||||
|
|
||||||
if settings.remember_me_form:
|
|
||||||
## adds a new input checkbox "remember me for longer"
|
|
||||||
if settings.formstyle != 'bootstrap':
|
|
||||||
addrow(form, XML(" "),
|
|
||||||
DIV(XML(" "),
|
|
||||||
INPUT(_type='checkbox',
|
|
||||||
_class='checkbox',
|
|
||||||
_id="auth_user_remember",
|
|
||||||
_name="remember",
|
|
||||||
),
|
|
||||||
XML(" "),
|
|
||||||
LABEL(
|
|
||||||
self.messages.label_remember_me,
|
|
||||||
_for="auth_user_remember",
|
|
||||||
)), "",
|
|
||||||
settings.formstyle,
|
|
||||||
'auth_user_remember__row')
|
|
||||||
elif settings.formstyle == 'bootstrap':
|
|
||||||
addrow(form,
|
|
||||||
"",
|
|
||||||
LABEL(
|
|
||||||
INPUT(_type='checkbox',
|
|
||||||
_id="auth_user_remember",
|
|
||||||
_name="remember"),
|
|
||||||
self.messages.label_remember_me,
|
|
||||||
_class="checkbox"),
|
|
||||||
"",
|
|
||||||
settings.formstyle,
|
|
||||||
'auth_user_remember__row')
|
|
||||||
|
|
||||||
captcha = settings.login_captcha or \
|
|
||||||
(settings.login_captcha != False and settings.captcha)
|
|
||||||
if captcha:
|
|
||||||
addrow(form, captcha.label, captcha, captcha.comment,
|
|
||||||
settings.formstyle, 'captcha__row')
|
|
||||||
accepted_form = False
|
accepted_form = False
|
||||||
|
self._reset_2_factor_auth(session)
|
||||||
|
# Redirect to the default 'next' page without logging
|
||||||
|
# in. If that page requires login, user will be redirected
|
||||||
|
# back to the main login form
|
||||||
|
redirect(next, client_side=settings.client_side)
|
||||||
|
|
||||||
if form.accepts(request, session if self.csrf_prevention else None,
|
# Before showing the default login form, check whether
|
||||||
formname='login', dbio=False,
|
# we are already on the second step of two-step authentication.
|
||||||
onvalidation=onvalidation,
|
# If we are, then skip this login form and use the form for the
|
||||||
hideerror=settings.hideerror):
|
# second challenge instead.
|
||||||
|
# Note to devs: The code inside the if-block is unchanged from the
|
||||||
accepted_form = True
|
# previous version of this file, other than for indentation inside
|
||||||
# check for username in db
|
# to put it inside the if-block
|
||||||
entered_username = form.vars[username]
|
if session.auth_2_factor_user is None:
|
||||||
if multi_login and '@' in entered_username:
|
# do we use our own login form, or from a central source?
|
||||||
# if '@' in username check for email, not username
|
if settings.login_form == self:
|
||||||
user = table_user(email = entered_username)
|
form = SQLFORM(
|
||||||
else:
|
table_user,
|
||||||
user = table_user(**{username: entered_username})
|
fields=[username, passfield],
|
||||||
if user:
|
hidden=dict(_next=next),
|
||||||
# user in db, check if registration pending or disabled
|
showid=settings.showid,
|
||||||
temp_user = user
|
submit_button=self.messages.login_button,
|
||||||
if temp_user.registration_key == 'pending':
|
delete_label=self.messages.delete_label,
|
||||||
response.flash = self.messages.registration_pending
|
formstyle=settings.formstyle,
|
||||||
return form
|
separator=settings.label_separator
|
||||||
elif temp_user.registration_key in ('disabled', 'blocked'):
|
)
|
||||||
response.flash = self.messages.login_disabled
|
|
||||||
return form
|
if settings.remember_me_form:
|
||||||
elif (not temp_user.registration_key is None
|
## adds a new input checkbox "remember me for longer"
|
||||||
and temp_user.registration_key.strip()):
|
if settings.formstyle != 'bootstrap':
|
||||||
response.flash = \
|
addrow(form, XML(" "),
|
||||||
self.messages.registration_verifying
|
DIV(XML(" "),
|
||||||
return form
|
INPUT(_type='checkbox',
|
||||||
# try alternate logins 1st as these have the
|
_class='checkbox',
|
||||||
# current version of the password
|
_id="auth_user_remember",
|
||||||
user = None
|
_name="remember",
|
||||||
for login_method in settings.login_methods:
|
),
|
||||||
if login_method != self and \
|
XML(" "),
|
||||||
login_method(request.vars[username],
|
LABEL(
|
||||||
request.vars[passfield]):
|
self.messages.label_remember_me,
|
||||||
if not self in settings.login_methods:
|
_for="auth_user_remember",
|
||||||
# do not store password in db
|
)), "",
|
||||||
form.vars[passfield] = None
|
settings.formstyle,
|
||||||
user = self.get_or_create_user(
|
'auth_user_remember__row')
|
||||||
form.vars, settings.update_fields)
|
elif settings.formstyle == 'bootstrap':
|
||||||
break
|
addrow(form,
|
||||||
if not user:
|
"",
|
||||||
# alternates have failed, maybe because service inaccessible
|
LABEL(
|
||||||
if settings.login_methods[0] == self:
|
INPUT(_type='checkbox',
|
||||||
# try logging in locally using cached credentials
|
_id="auth_user_remember",
|
||||||
if form.vars.get(passfield, '') == temp_user[passfield]:
|
_name="remember"),
|
||||||
# success
|
self.messages.label_remember_me,
|
||||||
user = temp_user
|
_class="checkbox"),
|
||||||
else:
|
"",
|
||||||
# user not in db
|
settings.formstyle,
|
||||||
if not settings.alternate_requires_registration:
|
'auth_user_remember__row')
|
||||||
# we're allowed to auto-register users from external systems
|
|
||||||
|
captcha = settings.login_captcha or \
|
||||||
|
(settings.login_captcha != False and settings.captcha)
|
||||||
|
if captcha:
|
||||||
|
addrow(form, captcha.label, captcha, captcha.comment,
|
||||||
|
settings.formstyle, 'captcha__row')
|
||||||
|
accepted_form = False
|
||||||
|
|
||||||
|
if form.accepts(request, session if self.csrf_prevention else None,
|
||||||
|
formname='login', dbio=False,
|
||||||
|
onvalidation=onvalidation,
|
||||||
|
hideerror=settings.hideerror):
|
||||||
|
|
||||||
|
accepted_form = True
|
||||||
|
# check for username in db
|
||||||
|
entered_username = form.vars[username]
|
||||||
|
if multi_login and '@' in entered_username:
|
||||||
|
# if '@' in username check for email, not username
|
||||||
|
user = table_user(email = entered_username)
|
||||||
|
else:
|
||||||
|
user = table_user(**{username: entered_username})
|
||||||
|
if user:
|
||||||
|
# user in db, check if registration pending or disabled
|
||||||
|
temp_user = user
|
||||||
|
if temp_user.registration_key == 'pending':
|
||||||
|
response.flash = self.messages.registration_pending
|
||||||
|
return form
|
||||||
|
elif temp_user.registration_key in ('disabled', 'blocked'):
|
||||||
|
response.flash = self.messages.login_disabled
|
||||||
|
return form
|
||||||
|
elif (not temp_user.registration_key is None
|
||||||
|
and temp_user.registration_key.strip()):
|
||||||
|
response.flash = \
|
||||||
|
self.messages.registration_verifying
|
||||||
|
return form
|
||||||
|
# try alternate logins 1st as these have the
|
||||||
|
# current version of the password
|
||||||
|
user = None
|
||||||
for login_method in settings.login_methods:
|
for login_method in settings.login_methods:
|
||||||
if login_method != self and \
|
if login_method != self and \
|
||||||
login_method(request.vars[username],
|
login_method(request.vars[username],
|
||||||
@@ -2404,33 +2416,124 @@ class Auth(object):
|
|||||||
user = self.get_or_create_user(
|
user = self.get_or_create_user(
|
||||||
form.vars, settings.update_fields)
|
form.vars, settings.update_fields)
|
||||||
break
|
break
|
||||||
if not user:
|
if not user:
|
||||||
self.log_event(self.messages['login_failed_log'],
|
# alternates have failed, maybe because service inaccessible
|
||||||
request.post_vars)
|
if settings.login_methods[0] == self:
|
||||||
# invalid login
|
# try logging in locally using cached credentials
|
||||||
session.flash = self.messages.invalid_login
|
if form.vars.get(passfield, '') == temp_user[passfield]:
|
||||||
callback(onfail, None)
|
# success
|
||||||
redirect(
|
user = temp_user
|
||||||
self.url(args=request.args, vars=request.get_vars),
|
else:
|
||||||
client_side=settings.client_side)
|
# user not in db
|
||||||
|
if not settings.alternate_requires_registration:
|
||||||
|
# we're allowed to auto-register users from external systems
|
||||||
|
for login_method in settings.login_methods:
|
||||||
|
if login_method != self and \
|
||||||
|
login_method(request.vars[username],
|
||||||
|
request.vars[passfield]):
|
||||||
|
if not self in settings.login_methods:
|
||||||
|
# do not store password in db
|
||||||
|
form.vars[passfield] = None
|
||||||
|
user = self.get_or_create_user(
|
||||||
|
form.vars, settings.update_fields)
|
||||||
|
break
|
||||||
|
if not user:
|
||||||
|
self.log_event(self.messages['login_failed_log'],
|
||||||
|
request.post_vars)
|
||||||
|
# invalid login
|
||||||
|
session.flash = self.messages.invalid_login
|
||||||
|
callback(onfail, None)
|
||||||
|
redirect(
|
||||||
|
self.url(args=request.args, vars=request.get_vars),
|
||||||
|
client_side=settings.client_side)
|
||||||
|
|
||||||
|
else: # use a central authentication server
|
||||||
|
cas = settings.login_form
|
||||||
|
cas_user = cas.get_user()
|
||||||
|
|
||||||
|
if cas_user:
|
||||||
|
cas_user[passfield] = None
|
||||||
|
user = self.get_or_create_user(
|
||||||
|
table_user._filter_fields(cas_user),
|
||||||
|
settings.update_fields)
|
||||||
|
elif hasattr(cas, 'login_form'):
|
||||||
|
return cas.login_form()
|
||||||
|
else:
|
||||||
|
# we need to pass through login again before going on
|
||||||
|
next = self.url(settings.function, args='login')
|
||||||
|
redirect(cas.login_url(next),
|
||||||
|
client_side=settings.client_side)
|
||||||
|
|
||||||
else:
|
# Extra login logic for two-factor authentication
|
||||||
# use a central authentication server
|
#################################################
|
||||||
cas = settings.login_form
|
# If the 'user' variable has a value, this means that the first
|
||||||
cas_user = cas.get_user()
|
# authentication step was successful (i.e. user provided correct
|
||||||
|
# username and password at the first challenge).
|
||||||
if cas_user:
|
# Check if this user is signed up for two-factor authentication
|
||||||
cas_user[passfield] = None
|
# Default rule is that the user must be part of a group that is called
|
||||||
user = self.get_or_create_user(
|
# 'web2py Two-Step Authentication'
|
||||||
table_user._filter_fields(cas_user),
|
if user:
|
||||||
settings.update_fields)
|
memberships = self.db(self.table_membership().user_id == user.id).select()
|
||||||
elif hasattr(cas, 'login_form'):
|
session.auth_2_factor_enabled = 'web2py Two-Step Authentication' in [i.group_id for i in memberships.values()]
|
||||||
return cas.login_form()
|
# If user is signed up for two-factor authentication, present the second
|
||||||
|
# challenge
|
||||||
|
if session.auth_2_factor_enabled:
|
||||||
|
form = SQLFORM.factory(
|
||||||
|
Field('authentication_code',
|
||||||
|
required=True,
|
||||||
|
comment='This code was emailed to you and is required for login.'),
|
||||||
|
hidden=dict(_next=next),
|
||||||
|
formstyle=settings.formstyle,
|
||||||
|
separator=settings.label_separator
|
||||||
|
)
|
||||||
|
# accepted_form is used by some default web2py code later in the
|
||||||
|
# function that handles running specified functions before redirect
|
||||||
|
# Set it to False until the challenge form is accepted.
|
||||||
|
accepted_form = False
|
||||||
|
# Handle the case when a user has submitted the login/password
|
||||||
|
# form successfully, and the password has been validated, but
|
||||||
|
# the two-factor form has not been displayed or validated yet.
|
||||||
|
if session.auth_2_factor_user is None and user is not None:
|
||||||
|
session.auth_2_factor_user = user # store the validated user and associate with this session
|
||||||
|
session.auth_2_factor = random.randint(100000, 999999)
|
||||||
|
session.auth_2_factor_tries_left = 3 # Allow user to try up to 4 times
|
||||||
|
# TODO: Add some error checking to handle cases where email cannot be sent
|
||||||
|
self.settings.mailer.send(
|
||||||
|
to=user.email,
|
||||||
|
subject="FDSI Login Authentication Code",
|
||||||
|
message="Your temporary login code is {0}".format(session.auth_2_factor))
|
||||||
|
if form.accepts(request, session if self.csrf_prevention else None,
|
||||||
|
formname='login', dbio=False,
|
||||||
|
onvalidation=onvalidation,
|
||||||
|
hideerror=settings.hideerror):
|
||||||
|
accepted_form = True
|
||||||
|
if form.vars['authentication_code'] == str(session.auth_2_factor):
|
||||||
|
# Handle the case when the two-factor form has been successfully validated
|
||||||
|
# and the user was previously stored (the current user should be None because
|
||||||
|
# in this case, the previous username/password login form should not be displayed.
|
||||||
|
# This will allow the code after the 2-factor authentication block to proceed as
|
||||||
|
# normal.
|
||||||
|
if user is None or user == session.auth_2_factor_user:
|
||||||
|
user = session.auth_2_factor_user
|
||||||
|
# For security, because the username stored in the
|
||||||
|
# session somehow does not match the just validated
|
||||||
|
# user. Should not be possible without session stealing
|
||||||
|
# which is hard with SSL.
|
||||||
|
elif user != session.auth_2_factor_user:
|
||||||
|
user = None
|
||||||
|
# Either way, the user and code associated with this session should
|
||||||
|
# be removed. This handles cases where the session login may have
|
||||||
|
# expired but browser window is open, so the old session key and
|
||||||
|
# session usernamem will still exist
|
||||||
|
self._reset_2_factor_auth(session)
|
||||||
|
else:
|
||||||
|
# TODO: Limit the number of retries allowed.
|
||||||
|
response.flash = 'Incorrect code. {0} more attempt(s) remaining.'.format(session.auth_2_factor_tries_left)
|
||||||
|
session.auth_2_factor_tries_left -= 1
|
||||||
|
return form
|
||||||
else:
|
else:
|
||||||
# we need to pass through login again before going on
|
return form
|
||||||
next = self.url(settings.function, args='login')
|
# End login logic for two-factor authentication
|
||||||
redirect(cas.login_url(next),
|
|
||||||
client_side=settings.client_side)
|
|
||||||
|
|
||||||
# process authenticated users
|
# process authenticated users
|
||||||
if user:
|
if user:
|
||||||
@@ -2468,7 +2571,11 @@ class Auth(object):
|
|||||||
"""
|
"""
|
||||||
Logouts and redirects to login
|
Logouts and redirects to login
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
# Clear out 2-step authentication information if user logs
|
||||||
|
# out. This information is also cleared on successful login.
|
||||||
|
self._reset_2_factor_auth(current.session)
|
||||||
|
|
||||||
if next is DEFAULT:
|
if next is DEFAULT:
|
||||||
next = self.get_vars_next() or self.settings.logout_next
|
next = self.get_vars_next() or self.settings.logout_next
|
||||||
if onlogout is DEFAULT:
|
if onlogout is DEFAULT:
|
||||||
|
|||||||
Reference in New Issue
Block a user