mirror of
https://github.com/wassname/flask-security.git
synced 2026-07-07 00:06:15 +08:00
Forgo redirecting authentication endpoint so that login form errors can be displayed
This commit is contained in:
@@ -35,8 +35,7 @@ _default_config = {
|
||||
'FLASH_MESSAGES': True,
|
||||
'PASSWORD_HASH': 'plaintext',
|
||||
'PASSWORD_HMAC': False,
|
||||
'PASSWORD_SALT': None,
|
||||
'AUTH_URL': '/auth',
|
||||
'PASSWORD_HMAC_SALT': None,
|
||||
'LOGIN_URL': '/login',
|
||||
'LOGOUT_URL': '/logout',
|
||||
'REGISTER_URL': '/register',
|
||||
|
||||
@@ -89,7 +89,7 @@ class ForgotPasswordForm(Form, UserEmailFormMixin):
|
||||
return dict(email=self.email.data)
|
||||
|
||||
|
||||
class PasswordlessLoginForm(Form, EmailFormMixin):
|
||||
class PasswordlessLoginForm(Form, UserEmailFormMixin):
|
||||
"""The passwordless login form"""
|
||||
|
||||
next = HiddenField()
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
{% from "security/_macros.html" import render_field_with_errors, render_field %}
|
||||
{% include "security/_messages.html" %}
|
||||
<h1>Login</h1>
|
||||
<form action="{{ url_for_security('authenticate') }}" method="POST" name="login_form">
|
||||
<form action="{{ url_for_security('login') }}" method="POST" name="login_form">
|
||||
{{ login_form.hidden_tag() }}
|
||||
{{ render_field_with_errors(login_form.email) }}
|
||||
{{ render_field_with_errors(login_form.password) }}
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
{% from "security/_macros.html" import render_field_with_errors, render_field %}
|
||||
{% include "security/_messages.html" %}
|
||||
<h1>Login</h1>
|
||||
<form action="{{ url_for_security('send_login') }}" method="POST" name="login_form">
|
||||
<form action="{{ url_for_security('login') }}" method="POST" name="login_form">
|
||||
{{ login_form.hidden_tag() }}
|
||||
{{ render_field_with_errors(login_form.email) }}
|
||||
{{ render_field(login_form.next) }}
|
||||
|
||||
+50
-57
@@ -74,51 +74,44 @@ def _ctx(endpoint):
|
||||
return _security._run_ctx_processor(endpoint)
|
||||
|
||||
|
||||
def authenticate():
|
||||
"""View function which handles an authentication request."""
|
||||
|
||||
form = LoginForm(request.form)
|
||||
user, msg, confirm_url = None, None, None
|
||||
|
||||
if request.json:
|
||||
form = LoginForm(MultiDict(request.json))
|
||||
|
||||
try:
|
||||
user = _security.auth_provider.authenticate(form)
|
||||
except ConfirmationError, e:
|
||||
msg = str(e)
|
||||
confirm_url = url_for('send_confirmation', email=e.user.email)
|
||||
except BadCredentialsError, e:
|
||||
msg = str(e)
|
||||
|
||||
if user:
|
||||
if login_user(user, remember=form.remember.data):
|
||||
after_this_request(_commit)
|
||||
if request.json:
|
||||
return _json_auth_ok(user)
|
||||
return redirect(get_post_login_redirect())
|
||||
msg = get_message('DISABLED_ACCOUNT')[0]
|
||||
|
||||
_logger.debug('Unsuccessful authentication attempt: %s' % msg)
|
||||
|
||||
if request.json:
|
||||
return _json_auth_error(msg)
|
||||
|
||||
do_flash(msg, 'error')
|
||||
return redirect(confirm_url or url_for('login'))
|
||||
|
||||
|
||||
@anonymous_user_required
|
||||
def login():
|
||||
"""View function for login view"""
|
||||
form = LoginForm(request.form, csrf_enabled=not app.testing)
|
||||
user, msg, confirm_url = None, None, None
|
||||
|
||||
tmp, form = '', LoginForm
|
||||
if request.json:
|
||||
form = LoginForm(MultiDict(request.json), csrf_enabled=not app.testing)
|
||||
|
||||
if _security.passwordless:
|
||||
tmp, form = 'send_', PasswordlessLoginForm
|
||||
if form.validate_on_submit():
|
||||
try:
|
||||
user = _security.auth_provider.authenticate(form)
|
||||
except ConfirmationError, e:
|
||||
msg = str(e)
|
||||
confirm_url = url_for('send_confirmation', email=e.user.email)
|
||||
except BadCredentialsError, e:
|
||||
msg = str(e)
|
||||
|
||||
return render_template('security/%slogin.html' % tmp,
|
||||
login_form=form(),
|
||||
if user:
|
||||
if login_user(user, remember=form.remember.data):
|
||||
after_this_request(_commit)
|
||||
if request.json:
|
||||
return _json_auth_ok(user)
|
||||
return redirect(get_post_login_redirect())
|
||||
msg = get_message('DISABLED_ACCOUNT')[0]
|
||||
|
||||
_logger.debug('Unsuccessful authentication attempt: %s' % msg)
|
||||
|
||||
if request.json:
|
||||
return _json_auth_error(msg)
|
||||
|
||||
do_flash(msg, 'error')
|
||||
|
||||
if confirm_url:
|
||||
return redirect(confirm_url)
|
||||
|
||||
return render_template('security/login.html',
|
||||
login_form=form,
|
||||
**_ctx('login'))
|
||||
|
||||
|
||||
@@ -171,16 +164,19 @@ def register():
|
||||
def send_login():
|
||||
"""View function that sends login instructions for passwordless login"""
|
||||
|
||||
form = PasswordlessLoginForm()
|
||||
user = _datastore.find_user(**form.to_dict())
|
||||
form = PasswordlessLoginForm(csrf_enabled=not app.testing)
|
||||
|
||||
if user.is_active():
|
||||
send_login_instructions(user, form.next.data)
|
||||
msg = get_message('LOGIN_EMAIL_SENT', email=user.email)
|
||||
else:
|
||||
msg = get_message('DISABLED_ACCOUNT')
|
||||
if form.validate_on_submit():
|
||||
user = _datastore.find_user(**form.to_dict())
|
||||
|
||||
if user.is_active():
|
||||
send_login_instructions(user, form.next.data)
|
||||
msg = get_message('LOGIN_EMAIL_SENT', email=user.email)
|
||||
else:
|
||||
msg = get_message('DISABLED_ACCOUNT')
|
||||
|
||||
do_flash(*msg)
|
||||
|
||||
do_flash(*msg)
|
||||
return render_template('security/send_login.html',
|
||||
login_form=form,
|
||||
**_ctx('send_login'))
|
||||
@@ -305,20 +301,17 @@ def create_blueprint(app, name, import_name, **kwargs):
|
||||
bp = Blueprint(name, import_name, **kwargs)
|
||||
|
||||
if config_value('PASSWORDLESS', app=app):
|
||||
bp.route(config_value('AUTH_URL', app=app),
|
||||
methods=['POST'],
|
||||
endpoint='send_login')(send_login)
|
||||
bp.route(config_value('LOGIN_URL', app=app),
|
||||
methods=['GET', 'POST'],
|
||||
endpoint='login')(send_login)
|
||||
|
||||
bp.route(config_value('AUTH_URL', app=app) + '/<token>',
|
||||
bp.route(config_value('LOGIN_URL', app=app) + '/<token>',
|
||||
methods=['GET'],
|
||||
endpoint='token_login')(token_login)
|
||||
else:
|
||||
bp.route(config_value('AUTH_URL', app=app),
|
||||
methods=['POST'],
|
||||
endpoint='authenticate')(authenticate)
|
||||
|
||||
bp.route(config_value('LOGIN_URL', app=app),
|
||||
endpoint='login')(login)
|
||||
bp.route(config_value('LOGIN_URL', app=app),
|
||||
methods=['GET', 'POST'],
|
||||
endpoint='login')(login)
|
||||
|
||||
bp.route(config_value('LOGOUT_URL', app=app),
|
||||
endpoint='logout')(logout)
|
||||
|
||||
+2
-2
@@ -35,7 +35,7 @@ class SecurityTest(TestCase):
|
||||
|
||||
def authenticate(self, email="matt@lp.com", password="password", endpoint=None, **kwargs):
|
||||
data = dict(email=email, password=password, remember='y')
|
||||
r = self._post(endpoint or '/auth', data=data, **kwargs)
|
||||
r = self._post(endpoint or '/login', data=data, **kwargs)
|
||||
return r
|
||||
|
||||
def json_authenticate(self, email="matt@lp.com", password="password", endpoint=None):
|
||||
@@ -45,7 +45,7 @@ class SecurityTest(TestCase):
|
||||
"password": "%s"
|
||||
}
|
||||
"""
|
||||
return self._post(endpoint or '/auth',
|
||||
return self._post(endpoint or '/login',
|
||||
content_type="application/json",
|
||||
data=data % (email, password))
|
||||
|
||||
|
||||
+14
-14
@@ -110,9 +110,10 @@ class DefaultSecurityTests(SecurityTest):
|
||||
self.authenticate(user)
|
||||
r = self._get("/admin_and_editor", follow_redirects=True)
|
||||
self.assertIsHomePage(r.data)
|
||||
self._get('/logout')
|
||||
|
||||
self.authenticate('dave@lp.com')
|
||||
r = self._get("/admin_and_editor")
|
||||
r = self._get("/admin_and_editor", follow_redirects=True)
|
||||
self.assertIn('Admin and Editor Page', r.data)
|
||||
|
||||
def test_ok_json_auth(self):
|
||||
@@ -206,7 +207,6 @@ class ConfiguredSecurityTests(SecurityTest):
|
||||
'SECURITY_PASSWORD_HMAC_SALT': 'so-salty',
|
||||
'SECURITY_PASSWORD_HMAC': True,
|
||||
'SECURITY_REGISTERABLE': True,
|
||||
'SECURITY_AUTH_URL': '/custom_auth',
|
||||
'SECURITY_LOGOUT_URL': '/custom_logout',
|
||||
'SECURITY_LOGIN_URL': '/custom_login',
|
||||
'SECURITY_POST_LOGIN_VIEW': '/post_login',
|
||||
@@ -221,11 +221,11 @@ class ConfiguredSecurityTests(SecurityTest):
|
||||
self.assertIn("<h1>Login</h1>", r.data)
|
||||
|
||||
def test_authenticate(self):
|
||||
r = self.authenticate(endpoint="/custom_auth")
|
||||
r = self.authenticate(endpoint="/custom_login")
|
||||
self.assertIn('Post Login', r.data)
|
||||
|
||||
def test_logout(self):
|
||||
self.authenticate(endpoint="/custom_auth")
|
||||
self.authenticate(endpoint="/custom_login")
|
||||
r = self.logout(endpoint="/custom_logout")
|
||||
self.assertIn('Post Logout', r.data)
|
||||
|
||||
@@ -484,9 +484,9 @@ class PasswordlessTests(SecurityTest):
|
||||
'SECURITY_PASSWORDLESS': True,
|
||||
}
|
||||
|
||||
def test_login_requset_for_inactive_user(self):
|
||||
def test_login_request_for_inactive_user(self):
|
||||
msg = self.app.config['SECURITY_MSG_DISABLED_ACCOUNT'][0]
|
||||
r = self.client.post('/auth', data=dict(email='tiya@lp.com'), follow_redirects=True)
|
||||
r = self.client.post('/login', data=dict(email='tiya@lp.com'), follow_redirects=True)
|
||||
self.assertIn(msg, r.data)
|
||||
|
||||
def test_request_login_token_sends_email_and_can_login(self):
|
||||
@@ -495,7 +495,7 @@ class PasswordlessTests(SecurityTest):
|
||||
|
||||
with capture_passwordless_login_requests() as requests:
|
||||
with self.app.mail.record_messages() as outbox:
|
||||
r = self.client.post('/auth', data=dict(email=e), follow_redirects=True)
|
||||
r = self.client.post('/login', data=dict(email=e), follow_redirects=True)
|
||||
|
||||
self.assertEqual(len(outbox), 1)
|
||||
|
||||
@@ -509,7 +509,7 @@ class PasswordlessTests(SecurityTest):
|
||||
msg = self.app.config['SECURITY_MSG_LOGIN_EMAIL_SENT'][0] % dict(email=user.email)
|
||||
self.assertIn(msg, r.data)
|
||||
|
||||
r = self.client.get('/auth/' + token, follow_redirects=True)
|
||||
r = self.client.get('/login/' + token, follow_redirects=True)
|
||||
self.assertIn(self.get_message('PASSWORDLESS_LOGIN_SUCCESSFUL'), r.data)
|
||||
|
||||
r = self.client.get('/profile')
|
||||
@@ -517,18 +517,18 @@ class PasswordlessTests(SecurityTest):
|
||||
|
||||
def test_invalid_login_token(self):
|
||||
msg = self.app.config['SECURITY_MSG_INVALID_LOGIN_TOKEN'][0]
|
||||
r = self._get('/auth/bogus', follow_redirects=True)
|
||||
r = self._get('/login/bogus', follow_redirects=True)
|
||||
self.assertIn(msg, r.data)
|
||||
|
||||
def test_token_login_forwards_to_post_login_view_when_already_authenticated(self):
|
||||
with capture_passwordless_login_requests() as requests:
|
||||
self.client.post('/auth', data=dict(email='matt@lp.com'), follow_redirects=True)
|
||||
self.client.post('/login', data=dict(email='matt@lp.com'), follow_redirects=True)
|
||||
token = requests[0]['login_token']
|
||||
|
||||
r = self.client.get('/auth/' + token, follow_redirects=True)
|
||||
r = self.client.get('/login/' + token, follow_redirects=True)
|
||||
self.assertIn(self.get_message('PASSWORDLESS_LOGIN_SUCCESSFUL'), r.data)
|
||||
|
||||
r = self.client.get('/auth/' + token, follow_redirects=True)
|
||||
r = self.client.get('/login/' + token, follow_redirects=True)
|
||||
self.assertNotIn(self.get_message('PASSWORDLESS_LOGIN_SUCCESSFUL'), r.data)
|
||||
|
||||
|
||||
@@ -543,13 +543,13 @@ class ExpiredLoginTokenTests(SecurityTest):
|
||||
e = 'matt@lp.com'
|
||||
|
||||
with capture_passwordless_login_requests() as requests:
|
||||
self.client.post('/auth', data=dict(email=e), follow_redirects=True)
|
||||
self.client.post('/login', data=dict(email=e), follow_redirects=True)
|
||||
token = requests[0]['login_token']
|
||||
|
||||
time.sleep(3)
|
||||
|
||||
with self.app.mail.record_messages() as outbox:
|
||||
r = self.client.get('/auth/' + token, follow_redirects=True)
|
||||
r = self.client.get('/login/' + token, follow_redirects=True)
|
||||
|
||||
self.assertEqual(len(outbox), 1)
|
||||
self.assertIn(e, outbox[0].html)
|
||||
|
||||
Reference in New Issue
Block a user