wassname/flask-security
1
0
Fork 0
mirror of https://github.com/wassname/flask-security.git synced 2026-07-10 00:30:24 +08:00
Code Issues Packages Projects Releases Wiki Activity

Major refactoring. Got rid of exceptions/errors in favor of using simple return values. Update tests to ensure full coverage according to nose coverage plugin

Browse Source
This commit is contained in:
Matt Wright
2012-08-23 17:58:33 -04:00
parent 5a4fb94be3
commit 6e754ed356
12 changed files with 174 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files
flask_security/confirmable.py
+17 -32
View File
@@ -11,12 +11,10 @@
from datetime import datetime
from itsdangerous import BadSignature, SignatureExpired
from flask import current_app as app, request
from werkzeug.local import LocalProxy
from .exceptions import ConfirmationError
from .utils import send_mail, get_max_age, md5, get_message, url_for_security
from .utils import send_mail, md5, url_for_security, get_token_status
from .signals import user_confirmed, confirm_instructions_sent
@@ -42,8 +40,8 @@ def send_confirmation_instructions(user):
confirmation_link, token = generate_confirmation_link(user)
send_mail('Please confirm your email', user.email,
'confirmation_instructions',
user=user, confirmation_link=confirmation_link)
'confirmation_instructions', user=user,
confirmation_link=confirmation_link)
confirm_instructions_sent.send(user, app=app._get_current_object())
return token
@@ -63,35 +61,22 @@ def requires_confirmation(user):
return user.confirmed_at == None and _security.confirmable
def confirm_by_token(token):
"""Confirm the user given the specified token. If the token is invalid or
the user is already confirmed a `ConfirmationError` error will be raised.
If the token is expired a `TokenExpiredError` error will be raised.
def confirm_email_token_status(token):
"""Returns the expired status, invalid status, and user of a confirmation
token. For example::
:param token: The user's confirmation token
expired, invalid, user = confirm_email_token_status('...')
:param token: The confirmation token
"""
serializer = _security.confirm_serializer
max_age = get_max_age('CONFIRM_EMAIL')
return get_token_status(token, 'confirm', 'CONFIRM_EMAIL')
try:
data = serializer.loads(token, max_age=max_age)
user = _datastore.find_user(id=data[0])
if user.confirmed_at:
raise ConfirmationError(get_message('ALREADY_CONFIRMED')[0])
def confirm_user(user):
"""Confirms the specified user
user.confirmed_at = datetime.utcnow()
_datastore._save_model(user)
user_confirmed.send(user, app=app._get_current_object())
return user
except SignatureExpired:
sig_okay, data = serializer.loads_unsafe(token)
user = _datastore.find_user(id=data[0])
msg = get_message('CONFIRMATION_EXPIRED',
within=_security.confirm_email_within,
email=user.email)[0]
raise ConfirmationError(msg, user=user)
except BadSignature:
raise ConfirmationError(get_message('INVALID_CONFIRMATION_TOKEN')[0])
:param user: The user to confirm
"""
user.confirmed_at = datetime.utcnow()
_datastore._save_model(user)
user_confirmed.send(user, app=app._get_current_object())
flask_security/core.py
+11 -13
View File
@@ -19,10 +19,8 @@ from passlib.context import CryptContext
from werkzeug.datastructures import ImmutableList
from werkzeug.local import LocalProxy
from . import views, exceptions
from .confirmable import requires_confirmation
from .utils import config_value as cv, get_config, verify_password, md5, \
url_for_security
from .utils import config_value as cv, get_config, md5, url_for_security
from .views import create_blueprint
# Convenient references
_security = LocalProxy(lambda: current_app.extensions['security'])
@@ -92,19 +90,19 @@ _default_messages = {
def _user_loader(user_id):
try:
return _security.datastore.find_user(id=user_id)
except:
return None
return _security.datastore.find_user(id=user_id)
def _token_loader(token):
try:
data = _security.remember_token_serializer.loads(token)
user = _security.datastore.find_user(id=data[0])
return user if md5(user.password) == data[1] else None
if user and md5(user.password) == data[1]:
return user
except:
return None
pass
return None
def _identity_loader():
@@ -294,9 +292,9 @@ class Security(object):
if register_blueprint:
name = cv('BLUEPRINT_NAME', app=app)
url_prefix = cv('URL_PREFIX', app=app)
bp = views.create_blueprint(app, name, __name__,
url_prefix=url_prefix,
template_folder='templates')
bp = create_blueprint(app, name, __name__,
url_prefix=url_prefix,
template_folder='templates')
app.register_blueprint(bp)
state = self._get_state(app, datastore, **kwargs)
flask_security/datastore.py
+2 -11
View File
@@ -9,9 +9,6 @@
:license: MIT, see LICENSE for more details.
"""
from . import exceptions
class UserDatastore(object):
"""Abstracted user datastore. Always extend this class and implement the
:attr:`_save_model`, :attr:`_delete_model`, :attr:`_do_find_user`, and
@@ -93,20 +90,14 @@ class UserDatastore(object):
:param user: User identifier, usually email address
"""
user = self._do_find_user(**kwargs)
if user:
return user
raise exceptions.UserNotFoundError('Parameters=%s' % kwargs)
return self._do_find_user(**kwargs)
def find_role(self, role):
"""Returns a role based on its name.
:param role: Role name
"""
role = self._do_find_role(role)
if role:
return role
raise exceptions.RoleNotFoundError()
return self._do_find_role(role)
def create_role(self, **kwargs):
"""Creates and returns a new role.
flask_security/decorators.py
+9 -14
View File
@@ -17,7 +17,6 @@ from flask.ext.principal import RoleNeed, Permission, Identity, identity_changed
from werkzeug.local import LocalProxy
from . import utils
from .exceptions import UserNotFoundError
# Convenient references
@@ -52,29 +51,25 @@ def _check_token():
header_token = request.headers.get(header_key, None)
token = request.args.get(args_key, header_token)
serializer = _security.remember_token_serializer
rv = False
try:
data = serializer.loads(token)
user = _security.datastore.find_user(id=data[0])
rv = utils.md5(user.password) == data[1]
return utils.md5(user.password) == data[1]
except:
pass
return rv
return False
def _check_http_auth():
auth = request.authorization or dict(username=None, password=None)
user = _security.datastore.find_user(email=auth.username)
try:
user = _security.datastore.find_user(email=auth.username)
if utils.verify_password(auth.password, user.password):
identity_changed.send(current_app._get_current_object(),
identity=Identity(user.id))
return True
except UserNotFoundError:
return False
if user and utils.verify_password(auth.password, user.password):
app = current_app._get_current_object()
identity_changed.send(app, identity=Identity(user.id))
return True
return False
def http_auth_required(realm):
flask_security/forms.py
+6 -25
View File
@@ -16,7 +16,6 @@ from flask.ext.wtf import Form, TextField, PasswordField, SubmitField, \
from werkzeug.local import LocalProxy
from .confirmable import requires_confirmation
from .exceptions import UserNotFoundError
from .utils import verify_password, get_message
# Convenient reference
@@ -30,17 +29,14 @@ password_required = Required(message="Password not provided")
def unique_user_email(form, field):
try:
_datastore.find_user(email=field.data)
raise ValidationError(field.data + ' is already associated with an account')
except UserNotFoundError:
pass
if _datastore.find_user(email=field.data) is not None:
raise ValidationError(field.data +
' is already associated with an account')
def valid_user_email(form, field):
try:
form.user = _datastore.find_user(email=field.data)
except UserNotFoundError:
form.user = _datastore.find_user(email=field.data)
if form.user is None:
raise ValidationError('Specified user does not exist')
@@ -106,28 +102,20 @@ class SendConfirmationForm(Form, UserEmailFormMixin):
return False
return True
def to_dict(self):
return dict(email=self.email.data)
class ForgotPasswordForm(Form, UserEmailFormMixin):
"""The default forgot password form"""
submit = SubmitField("Recover Password")
def to_dict(self):
return dict(email=self.email.data)
class PasswordlessLoginForm(Form, UserEmailFormMixin, NextFormMixin):
class PasswordlessLoginForm(Form, UserEmailFormMixin):
"""The passwordless login form"""
submit = SubmitField("Send Login Link")
def __init__(self, *args, **kwargs):
super(PasswordlessLoginForm, self).__init__(*args, **kwargs)
if request.method == 'GET':
self.next.data = request.args.get('next', None)
def validate(self):
if not super(PasswordlessLoginForm, self).validate():
@@ -137,9 +125,6 @@ class PasswordlessLoginForm(Form, UserEmailFormMixin, NextFormMixin):
return False
return True
def to_dict(self):
return dict(user=self.user, next=self.next.data)
class LoginForm(Form, UserEmailFormMixin, PasswordFormMixin, NextFormMixin):
"""The default login form"""
@@ -149,7 +134,6 @@ class LoginForm(Form, UserEmailFormMixin, PasswordFormMixin, NextFormMixin):
def __init__(self, *args, **kwargs):
super(LoginForm, self).__init__(*args, **kwargs)
self.next.data = request.args.get('next', None)
def validate(self):
if not super(LoginForm, self).validate():
@@ -181,6 +165,3 @@ class ResetPasswordForm(Form, NewPasswordFormMixin, PasswordConfirmFormMixin):
"""The default reset password form"""
submit = SubmitField("Reset Password")
def to_dict(self):
return dict(password=self.password.data)
flask_security/passwordless.py
+16 -29
View File
@@ -10,13 +10,10 @@
"""
from flask import request, current_app as app
from itsdangerous import SignatureExpired, BadSignature
from werkzeug.local import LocalProxy
from .exceptions import PasswordlessLoginError
from .signals import login_instructions_sent
from .utils import send_mail, md5, get_max_age, login_user, get_message, \
url_for_security, get_url
from .utils import send_mail, url_for_security, get_token_status
# Convenient references
@@ -25,13 +22,13 @@ _security = LocalProxy(lambda: app.extensions['security'])
_datastore = LocalProxy(lambda: _security.datastore)
def send_login_instructions(user, next):
def send_login_instructions(user):
"""Sends the login instructions email for the specified user.
:param user: The user to send the instructions to
:param token: The login token
"""
token = generate_login_token(user, next)
token = generate_login_token(user)
url = url_for_security('token_login', token=token)
login_link = request.url_root[:-1] + url
@@ -42,30 +39,20 @@ def send_login_instructions(user, next):
app=app._get_current_object())
def generate_login_token(user, next):
next = next or get_url(_security.post_login_view)
data = [user.id, md5(user.password), next]
return _security.login_serializer.dumps(data)
def generate_login_token(user):
"""Generates a unique login token for the specified user.
:param user: The user the token belongs to
"""
return _security.login_serializer.dumps([user.id])
def login_by_token(token):
serializer = _security.login_serializer
max_age = get_max_age('LOGIN')
def login_token_status(token):
"""Returns the expired status, invalid status, and user of a login token.
For example::
try:
user_id, pw, next = serializer.loads(token, max_age=max_age)
user = _datastore.find_user(id=user_id)
login_user(user, True)
return user, next
expired, invalid, user = login_token_status('...')
except SignatureExpired:
sig_okay, data = serializer.loads_unsafe(token)
user_id, pw, next = data
user = _datastore.find_user(id=data[0])
within = _security.login_within
msg = get_message('LOGIN_EXPIRED', within=within, email=user.email)
raise PasswordlessLoginError(msg[0], user=user, next=next)
except BadSignature:
msg = get_message('INVALID_LOGIN_TOKEN')
raise PasswordlessLoginError(msg[0])
:param token: The login token
"""
return get_token_status(token, 'login', 'LOGIN')
flask_security/recoverable.py
+18 -33
View File
@@ -9,14 +9,12 @@
:license: MIT, see LICENSE for more details.
"""
from itsdangerous import BadSignature, SignatureExpired
from flask import current_app as app, request
from werkzeug.local import LocalProxy
from .exceptions import ResetPasswordError
from .signals import password_reset, reset_password_instructions_sent
from .utils import send_mail, get_max_age, md5, get_message, encrypt_password, \
url_for_security
from .utils import send_mail, md5, encrypt_password, url_for_security, \
get_token_status
# Convenient references
@@ -60,36 +58,23 @@ def generate_reset_password_token(user):
return _security.reset_serializer.dumps(data)
def reset_by_token(token, password):
"""Resets the password of the user given the specified token, email and
password. If the token is invalid a `ResetPasswordError` error will be
raised. If the token is expired a `TokenExpiredError` error will be raised.
def reset_password_token_status(token):
"""Returns the expired status, invalid status, and user of a password reset
token. For example::
:param token: The user's reset password token
:param email: The user's email address
:param password: The user's new password
expired, invalid, user = reset_password_token_status('...')
:param token: The password reset token
"""
serializer = _security.reset_serializer
max_age = get_max_age('RESET_PASSWORD')
return get_token_status(token, 'reset', 'RESET_PASSWORD')
try:
data = serializer.loads(token, max_age=max_age)
user = _datastore.find_user(id=data[0])
def update_password(user, password):
"""Update the specified user's password
user.password = encrypt_password(password)
_datastore._save_model(user)
send_password_reset_notice(user)
password_reset.send(user, app=app._get_current_object())
return user
except SignatureExpired:
sig_okay, data = serializer.loads_unsafe(token)
user = _datastore.find_user(id=data[0])
msg = get_message('PASSWORD_RESET_EXPIRED',
within=_security.reset_password_within,
email=user.email)
raise ResetPasswordError(msg[0], user=user)
except BadSignature:
raise ResetPasswordError(get_message('INVALID_RESET_PASSWORD_TOKEN')[0])
:param user: The user to update_password
:param password: The unencrypted new password
"""
user.password = encrypt_password(password)
_datastore._save_model(user)
send_password_reset_notice(user)
password_reset.send(user, app=app._get_current_object())
flask_security/templates/security/send_login.html
-1
View File
@@ -4,7 +4,6 @@
<form action="{{ url_for_security('login') }}" method="POST" name="send_login_form">
{{ send_login_form.hidden_tag() }}
{{ render_field_with_errors(send_login_form.email) }}
{{ render_field(send_login_form.next) }}
{{ render_field(send_login_form.submit) }}
</form>
{% include "security/_menu.html" %}
flask_security/utils.py
+23 -4
View File
@@ -22,6 +22,7 @@ from flask.ext.login import login_user as _login_user, \
logout_user as _logout_user
from flask.ext.mail import Message
from flask.ext.principal import Identity, AnonymousIdentity, identity_changed
from itsdangerous import BadSignature, SignatureExpired
from werkzeug.local import LocalProxy
from .core import current_user
@@ -51,8 +52,7 @@ def anonymous_user_required(f):
def login_user(user, remember=True):
"""Performs the login and sends the appropriate signal."""
if not _login_user(user, remember):
return False
_login_user(user, remember)
if _security.trackable:
old_current, new_current = user.current_login_at, datetime.utcnow()
@@ -69,8 +69,6 @@ def login_user(user, remember=True):
_datastore._save_model(user)
identity_changed.send(current_app._get_current_object(),
identity=Identity(user.id))
_logger.debug('User %s logged in' % user)
return True
def logout_user():
@@ -259,6 +257,27 @@ def send_mail(subject, recipient, template, **context):
mail.send(msg)
def get_token_status(token, serializer, max_age=None):
serializer = getattr(_security, serializer + '_serializer')
max_age = get_max_age(max_age)
user, data = None, None
expired, invalid = False, False
try:
data = serializer.loads(token, max_age=max_age)
except SignatureExpired:
d, data = serializer.loads_unsafe(token)
expired = True
except BadSignature:
invalid = True
if data:
user = _datastore.find_user(id=data[0])
expired = expired and (user is not None)
return expired, invalid, user
@contextmanager
def capture_passwordless_login_requests():
login_requests = []
flask_security/views.py
+54 -47
View File
@@ -15,16 +15,15 @@ from werkzeug.datastructures import MultiDict
from werkzeug.local import LocalProxy
from flask_security.confirmable import send_confirmation_instructions, \
confirm_by_token
confirm_user, confirm_email_token_status
from flask_security.decorators import login_required
from flask_security.exceptions import ConfirmationError, ResetPasswordError, \
PasswordlessLoginError
from flask_security.forms import LoginForm, ConfirmRegisterForm, RegisterForm, \
ForgotPasswordForm, ResetPasswordForm, SendConfirmationForm, \
PasswordlessLoginForm
from flask_security.passwordless import send_login_instructions, login_by_token
from flask_security.recoverable import reset_by_token, \
send_reset_password_instructions
from flask_security.passwordless import send_login_instructions, \
login_token_status
from flask_security.recoverable import reset_password_token_status, \
send_reset_password_instructions, update_password
from flask_security.registerable import register_user
from flask_security.utils import get_url, get_post_login_redirect, do_flash, \
get_message, config_value, login_user, logout_user, \
@@ -64,10 +63,10 @@ def _ctx(endpoint):
def login():
"""View function for login view"""
form_data = request.form
if request.json:
form_data = MultiDict(request.json)
else:
form_data = request.form
form = LoginForm(form_data, csrf_enabled=not app.testing)
@@ -131,7 +130,7 @@ def send_login():
form = PasswordlessLoginForm(csrf_enabled=not app.testing)
if form.validate_on_submit():
send_login_instructions(**form.to_dict())
send_login_instructions(form.user)
do_flash(*get_message('LOGIN_EMAIL_SENT', email=form.user.email))
return render_template('security/send_login.html',
@@ -142,17 +141,22 @@ def send_login():
@anonymous_user_required
def token_login(token):
"""View function that handles passwordless login via a token"""
expired, invalid, user = login_token_status(token)
try:
user, next = login_by_token(token)
except PasswordlessLoginError, e:
if e.user:
send_login_instructions(e.user, e.next)
do_flash(str(e), 'error')
return redirect(request.referrer or url_for('login'))
if invalid:
do_flash(*get_message('INVALID_LOGIN_TOKEN'))
if expired:
send_login_instructions(user)
do_flash(*get_message('LOGIN_EXPIRED', email=user.email,
within=_security.login_within))
if invalid or expired:
return redirect(url_for('login'))
login_user(user, True)
after_this_request(_commit)
do_flash(*get_message('PASSWORDLESS_LOGIN_SUCCESSFUL'))
return redirect(next)
return redirect(get_post_login_redirect())
@anonymous_user_required
@@ -173,22 +177,26 @@ def send_confirmation():
@anonymous_user_required
def confirm_email(token):
"""View function which handles a email confirmation request."""
after_this_request(_commit)
try:
user = confirm_by_token(token)
except ConfirmationError, e:
if e.user:
send_confirmation_instructions(e.user)
do_flash(str(e), 'error')
confirm_error_url = get_url(_security.confirm_error_view)
return redirect(confirm_error_url or url_for('send_confirmation'))
expired, invalid, user = confirm_email_token_status(token)
do_flash(*get_message('EMAIL_CONFIRMED'))
if invalid:
do_flash(*get_message('INVALID_CONFIRMATION_TOKEN'))
if expired:
send_confirmation_instructions(user)
do_flash(*get_message('CONFIRMATION_EXPIRED', email=user.email,
within=_security.confirm_email_within))
if invalid or expired:
return redirect(get_url(_security.confirm_error_view) or
url_for('send_confirmation'))
confirm_user(user)
login_user(user, True)
post_confirm_url = get_url(_security.post_confirm_view)
post_login_url = get_url(_security.post_login_view)
return redirect(post_confirm_url or post_login_url)
after_this_request(_commit)
do_flash(*get_message('EMAIL_CONFIRMED'))
return redirect(get_url(_security.post_confirm_view) or
get_url(_security.post_login_view))
@anonymous_user_required
@@ -210,25 +218,24 @@ def forgot_password():
def reset_password(token):
"""View function that handles a reset password request."""
next = None
form = ResetPasswordForm(reset_token=token, csrf_enabled=not app.testing)
expired, invalid, user = reset_password_token_status(token)
if invalid:
do_flash(*get_message('INVALID_RESET_PASSWORD_TOKEN'))
if expired:
do_flash(*get_message('PASSWORD_RESET_EXPIRED', email=user.email,
within=_security.reset_password_within))
if invalid or expired:
return redirect(url_for('forgot_password'))
form = ResetPasswordForm(csrf_enabled=not app.testing)
if form.validate_on_submit():
try:
user = reset_by_token(token=token, **form.to_dict())
msg = get_message('PASSWORD_RESET')
next = (get_url(_security.post_reset_view) or
get_url(_security.post_login_view))
except ResetPasswordError, e:
msg = (str(e), 'error')
if e.user:
send_reset_password_instructions(e.user)
do_flash(*msg)
if next:
login_user(user)
return redirect(next)
update_password(user, form.password.data)
do_flash(*get_message('PASSWORD_RESET'))
login_user(user, True)
return redirect(get_url(_security.post_reset_view) or
get_url(_security.post_login_view))
return render_template('security/reset_password.html',
reset_password_form=form,
tests/functional_tests.py
+17
View File
@@ -321,6 +321,10 @@ class ConfirmableTests(SecurityTest):
r = self.client.get('/confirm/bogus', follow_redirects=True)
self.assertIn('Invalid confirmation token', r.data)
def test_send_confirmation_with_invalid_email(self):
r = self._post('/confirm', data=dict(email='bogus@bogus.com'))
self.assertIn('Specified user does not exist', r.data)
def test_resend_confirmation(self):
e = 'dude@lp.com'
self.register(e)
@@ -378,6 +382,15 @@ class RecoverableTests(SecurityTest):
'SECURITY_POST_FORGOT_VIEW': '/'
}
def test_reset_view(self):
with capture_reset_password_requests() as requests:
r = self.client.post('/reset',
data=dict(email='joe@lp.com'),
follow_redirects=True)
t = requests[0]['token']
r = self._get('/reset/' + t)
self.assertIn('<h1>Reset password</h1>', r.data)
def test_forgot_post_sends_email(self):
with capture_reset_password_requests():
with self.app.extensions['mail'].record_messages() as outbox:
@@ -512,6 +525,10 @@ class PasswordlessTests(SecurityTest):
r = self.client.get('/login/' + token, follow_redirects=True)
self.assertNotIn(self.get_message('PASSWORDLESS_LOGIN_SUCCESSFUL'), r.data)
def test_send_login_with_invalid_email(self):
r = self._post('/login', data=dict(email='bogus@bogus.com'))
self.assertIn('Specified user does not exist', r.data)
class ExpiredLoginTokenTests(SecurityTest):
tests/test_app/__init__.py
+1 -5
View File
@@ -5,7 +5,6 @@ from flask.ext.mail import Mail
from flask.ext.security import login_required, roles_required, roles_accepted
from flask.ext.security.decorators import http_auth_required, \
auth_token_required
from flask.ext.security.exceptions import RoleNotFoundError
from flask.ext.security.utils import encrypt_password
from werkzeug.local import LocalProxy
@@ -105,10 +104,7 @@ def create_app(config):
@app.route('/coverage/invalid_role')
def invalid_role():
try:
ds.find_role('bogus')
except RoleNotFoundError:
return 'success'
return 'success' if ds.find_role('bogus') is None else 'failure'
return app