Big code cleanup

This commit is contained in:
Matt Wright
2012-06-18 16:51:43 -04:00
parent 5fdfa544db
commit c20f244d66
11 changed files with 318 additions and 223 deletions
+9 -1
View File
@@ -10,4 +10,12 @@
:license: MIT, see LICENSE for more details.
"""
from .core import *
from .core import Security, RoleMixin, UserMixin, AnonymousUser, \
AuthenticationProvider
from .decorators import auth_token_required, http_auth_required, \
login_required, roles_accepted, roles_required
from .forms import ForgotPasswordForm, LoginForm, RegisterForm, \
ResetPasswordForm
from .signals import confirm_instructions_sent, password_reset, \
password_reset_requested, reset_instructions_sent, user_confirmed, \
user_registered
+28 -15
View File
@@ -1,22 +1,34 @@
# -*- coding: utf-8 -*-
"""
flask.ext.security.confirmable
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Flask-Security confirmable module
:copyright: (c) 2012 by Matt Wright.
:license: MIT, see LICENSE for more details.
"""
from datetime import datetime
from flask import current_app, request, url_for
from flask.ext.security.exceptions import UserNotFoundError, \
ConfirmationError, TokenExpiredError
from flask.ext.security.utils import generate_token, send_mail
from flask.ext.security.signals import user_confirmed, \
confirm_instructions_sent
from flask import current_app as app, request, url_for
from werkzeug.local import LocalProxy
security = LocalProxy(lambda: current_app.security)
logger = LocalProxy(lambda: current_app.logger)
from .exceptions import UserNotFoundError, ConfirmationError, TokenExpiredError
from .utils import generate_token, send_mail
from .signals import user_confirmed, confirm_instructions_sent
# Convenient references
_security = LocalProxy(lambda: app.security)
_datastore = LocalProxy(lambda: app.security.datastore)
def find_user_by_confirmation_token(token):
if not token:
raise ConfirmationError('Confirmation token required')
return security.datastore.find_user(confirmation_token=token)
return _datastore.find_user(confirmation_token=token)
def send_confirmation_instructions(user):
@@ -29,7 +41,7 @@ def send_confirmation_instructions(user):
'confirmation_instructions',
dict(user=user, confirmation_link=confirmation_link))
confirm_instructions_sent.send(user, app=current_app._get_current_object())
confirm_instructions_sent.send(user, app=app._get_current_object())
return True
@@ -56,7 +68,7 @@ def generate_confirmation_token(user):
def should_confirm_email(fn):
def wrapped(*args, **kwargs):
if security.confirm_email:
if _security.confirm_email:
return fn(*args, **kwargs)
return False
return wrapped
@@ -69,7 +81,7 @@ def requires_confirmation(user):
@should_confirm_email
def confirmation_token_is_expired(user):
token_expires = datetime.utcnow() - security.confirm_email_within
token_expires = datetime.utcnow() - _security.confirm_email_within
return user.confirmation_sent_at < token_expires
@@ -86,16 +98,17 @@ def confirm_by_token(token):
raise TokenExpiredError(message='Confirmation token is expired',
user=user)
# TODO: Clear confirmation_token after confirmation?
#user.confirmation_token = None
#user.confirmation_sent_at = None
user.confirmed_at = datetime.utcnow()
security.datastore._save_model(user)
_datastore._save_model(user)
user_confirmed.send(user, app=current_app._get_current_object())
user_confirmed.send(user, app=app._get_current_object())
return user
def reset_confirmation_token(user):
security.datastore._save_model(generate_confirmation_token(user))
_datastore._save_model(generate_confirmation_token(user))
send_confirmation_instructions(user)
+13 -141
View File
@@ -10,22 +10,20 @@
"""
from datetime import timedelta
from functools import wraps
from flask import current_app, Blueprint, redirect, request
from flask import current_app, Blueprint
from flask.ext.login import AnonymousUser as AnonymousUserBase, \
UserMixin as BaseUserMixin, LoginManager, login_required, \
current_user, login_url
from flask.ext.principal import Principal, RoleNeed, UserNeed, \
Permission, identity_loaded
from flask.ext.wtf import Form, TextField, PasswordField, SubmitField, \
HiddenField, Required, BooleanField, EqualTo, Email
from flask.ext.security import views, exceptions, utils
from flask.ext.security.confirmable import confirmation_token_is_expired, \
requires_confirmation, reset_confirmation_token
UserMixin as BaseUserMixin, LoginManager, current_user
from flask.ext.principal import Principal, RoleNeed, UserNeed, identity_loaded
from passlib.context import CryptContext
from werkzeug.datastructures import ImmutableList
from . import views, exceptions, utils
from .confirmable import confirmation_token_is_expired, requires_confirmation, \
reset_confirmation_token
from .decorators import login_required
from .forms import Form, LoginForm
#: Default Flask-Security configuration
_default_config = {
@@ -33,10 +31,10 @@ _default_config = {
'FLASH_MESSAGES': True,
'PASSWORD_HASH': 'plaintext',
'AUTH_PROVIDER': 'flask.ext.security::AuthenticationProvider',
'LOGIN_FORM': 'flask.ext.security::LoginForm',
'REGISTER_FORM': 'flask.ext.security::RegisterForm',
'RESET_PASSWORD_FORM': 'flask.ext.security::ResetPasswordForm',
'FORGOT_PASSWORD_FORM': 'flask.ext.security::ForgotPasswordForm',
'LOGIN_FORM': 'flask.ext.security.forms::LoginForm',
'REGISTER_FORM': 'flask.ext.security.forms::RegisterForm',
'RESET_PASSWORD_FORM': 'flask.ext.security.forms::ResetPasswordForm',
'FORGOT_PASSWORD_FORM': 'flask.ext.security.forms::ForgotPasswordForm',
'AUTH_URL': '/auth',
'LOGOUT_URL': '/logout',
'REGISTER_URL': '/register',
@@ -61,80 +59,6 @@ _default_config = {
}
def roles_required(*roles):
"""View decorator which specifies that a user must have all the specified
roles. Example::
@app.route('/dashboard')
@roles_required('admin', 'editor')
def dashboard():
return 'Dashboard'
The current user must have both the `admin` role and `editor` role in order
to view the page.
:param args: The required roles.
"""
perm = Permission(*[RoleNeed(role) for role in roles])
def wrapper(fn):
@wraps(fn)
def decorated_view(*args, **kwargs):
if not current_user.is_authenticated():
login_view = current_app.security.login_manager.login_view
return redirect(login_url(login_view, request.url))
if perm.can():
return fn(*args, **kwargs)
current_app.logger.debug('Identity does not provide the '
'roles: %s' % [r for r in roles])
return redirect(request.referrer or '/')
return decorated_view
return wrapper
def roles_accepted(*roles):
"""View decorator which specifies that a user must have at least one of the
specified roles. Example::
@app.route('/create_post')
@roles_accepted('editor', 'author')
def create_post():
return 'Create Post'
The current user must have either the `editor` role or `author` role in
order to view the page.
:param args: The possible roles.
"""
perms = [Permission(RoleNeed(role)) for role in roles]
def wrapper(fn):
@wraps(fn)
def decorated_view(*args, **kwargs):
if not current_user.is_authenticated():
login_view = current_app.security.login_manager.login_view
return redirect(login_url(login_view, request.url))
for perm in perms:
if perm.can():
return fn(*args, **kwargs)
r1 = [r for r in roles]
r2 = [r.name for r in current_user.roles]
current_app.logger.debug('Current user does not provide a '
'required role. Accepted: %s Provided: %s' % (r1, r2))
utils.do_flash('You do not have permission to '
'view this resource', 'error')
return redirect(request.referrer or '/')
return decorated_view
return wrapper
class RoleMixin(object):
"""Mixin for `Role` model definitions"""
def __eq__(self, other):
@@ -298,58 +222,6 @@ class Security(object):
endpoint='confirm')(views.confirm)
class ForgotPasswordForm(Form):
email = TextField("Email Address",
validators=[Required(message="Email not provided")])
def to_dict(self):
return dict(email=self.email.data)
class LoginForm(Form):
"""The default login form"""
email = TextField("Email Address",
validators=[Required(message="Email not provided")])
password = PasswordField("Password",
validators=[Required(message="Password not provided")])
remember = BooleanField("Remember Me")
next = HiddenField()
submit = SubmitField("Login")
def __init__(self, *args, **kwargs):
super(LoginForm, self).__init__(*args, **kwargs)
self.next.data = request.args.get('next', None)
class RegisterForm(Form):
"""The default register form"""
email = TextField("Email Address",
validators=[Required(message='Email not provided'), Email()])
password = PasswordField("Password",
validators=[Required(message="Password not provided")])
password_confirm = PasswordField("Retype Password",
validators=[EqualTo('password', message="Passwords do not match")])
def to_dict(self):
return dict(email=self.email.data, password=self.password.data)
class ResetPasswordForm(Form):
token = HiddenField(validators=[Required()])
email = HiddenField(validators=[Required()])
password = PasswordField("Password",
validators=[Required(message="Password not provided")])
password_confirm = PasswordField("Retype Password",
validators=[EqualTo('password', message="Passwords do not match")])
def to_dict(self):
return dict(token=self.token.data,
email=self.email.data,
password=self.password.data)
class AuthenticationProvider(object):
"""The default authentication provider implementation.
+2 -1
View File
@@ -10,7 +10,8 @@
"""
from flask import current_app
from flask.ext.security import exceptions, confirmable
from . import exceptions, confirmable
class UserDatastore(object):
+90 -1
View File
@@ -1,7 +1,22 @@
# -*- coding: utf-8 -*-
"""
flask.ext.security.decorators
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Flask-Security decorators module
:copyright: (c) 2012 by Matt Wright.
:license: MIT, see LICENSE for more details.
"""
from functools import wraps
from flask import current_app as app, Response, request, abort
from flask import current_app as app, Response, request, abort, redirect
from flask.ext.login import login_required, login_url, current_user
from flask.ext.principal import RoleNeed, Permission
from . import utils
_default_http_auth_msg = """
<h1>Unauthorized</h1>
@@ -62,3 +77,77 @@ def auth_token_required(fn):
abort(401)
return decorated
def roles_required(*roles):
"""View decorator which specifies that a user must have all the specified
roles. Example::
@app.route('/dashboard')
@roles_required('admin', 'editor')
def dashboard():
return 'Dashboard'
The current user must have both the `admin` role and `editor` role in order
to view the page.
:param args: The required roles.
"""
perm = Permission(*[RoleNeed(role) for role in roles])
def wrapper(fn):
@wraps(fn)
def decorated_view(*args, **kwargs):
if not current_user.is_authenticated():
login_view = app.security.login_manager.login_view
return redirect(login_url(login_view, request.url))
if perm.can():
return fn(*args, **kwargs)
app.logger.debug('Identity does not provide the '
'roles: %s' % [r for r in roles])
return redirect(request.referrer or '/')
return decorated_view
return wrapper
def roles_accepted(*roles):
"""View decorator which specifies that a user must have at least one of the
specified roles. Example::
@app.route('/create_post')
@roles_accepted('editor', 'author')
def create_post():
return 'Create Post'
The current user must have either the `editor` role or `author` role in
order to view the page.
:param args: The possible roles.
"""
perms = [Permission(RoleNeed(role)) for role in roles]
def wrapper(fn):
@wraps(fn)
def decorated_view(*args, **kwargs):
if not current_user.is_authenticated():
login_view = app.security.login_manager.login_view
return redirect(login_url(login_view, request.url))
for perm in perms:
if perm.can():
return fn(*args, **kwargs)
r1 = [r for r in roles]
r2 = [r.name for r in current_user.roles]
app.logger.debug('Current user does not provide a '
'required role. Accepted: %s Provided: %s' % (r1, r2))
utils.do_flash('You do not have permission to '
'view this resource', 'error')
return redirect(request.referrer or '/')
return decorated_view
return wrapper
+66
View File
@@ -0,0 +1,66 @@
# -*- coding: utf-8 -*-
"""
flask.ext.security.forms
~~~~~~~~~~~~~~~~~~~~~~~~
Flask-Security forms module
:copyright: (c) 2012 by Matt Wright.
:license: MIT, see LICENSE for more details.
"""
from flask import request
from flask.ext.wtf import Form, TextField, PasswordField, SubmitField, \
HiddenField, Required, BooleanField, EqualTo, Email
class ForgotPasswordForm(Form):
email = TextField("Email Address",
validators=[Required(message="Email not provided")])
def to_dict(self):
return dict(email=self.email.data)
class LoginForm(Form):
"""The default login form"""
email = TextField("Email Address",
validators=[Required(message="Email not provided")])
password = PasswordField("Password",
validators=[Required(message="Password not provided")])
remember = BooleanField("Remember Me")
next = HiddenField()
submit = SubmitField("Login")
def __init__(self, *args, **kwargs):
super(LoginForm, self).__init__(*args, **kwargs)
self.next.data = request.args.get('next', None)
class RegisterForm(Form):
"""The default register form"""
email = TextField("Email Address",
validators=[Required(message='Email not provided'), Email()])
password = PasswordField("Password",
validators=[Required(message="Password not provided")])
password_confirm = PasswordField("Retype Password",
validators=[EqualTo('password', message="Passwords do not match")])
def to_dict(self):
return dict(email=self.email.data, password=self.password.data)
class ResetPasswordForm(Form):
token = HiddenField(validators=[Required()])
email = HiddenField(validators=[Required()])
password = PasswordField("Password",
validators=[Required(message="Password not provided")])
password_confirm = PasswordField("Retype Password",
validators=[EqualTo('password', message="Passwords do not match")])
def to_dict(self):
return dict(token=self.token.data,
email=self.email.data,
password=self.password.data)
+30 -16
View File
@@ -1,22 +1,36 @@
# -*- coding: utf-8 -*-
"""
flask.ext.security.recoverable
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Flask-Security recoverable module
:copyright: (c) 2012 by Matt Wright.
:license: MIT, see LICENSE for more details.
"""
from datetime import datetime
from flask import current_app, request, url_for
from flask.ext.security.exceptions import ResetPasswordError, \
UserNotFoundError, TokenExpiredError
from flask.ext.security.signals import password_reset, \
password_reset_requested, confirm_instructions_sent
from flask.ext.security.utils import generate_token, send_mail
from flask import current_app as app, request, url_for
from werkzeug.local import LocalProxy
security = LocalProxy(lambda: current_app.security)
logger = LocalProxy(lambda: current_app.logger)
from .exceptions import ResetPasswordError, UserNotFoundError, \
TokenExpiredError
from .signals import password_reset, password_reset_requested, \
confirm_instructions_sent
from .utils import generate_token, send_mail
# Convenient references
_security = LocalProxy(lambda: app.security)
_datastore = LocalProxy(lambda: app.security.datastore)
def find_user_by_reset_token(token):
if not token:
raise ResetPasswordError('Reset password token required')
return security.datastore.find_user(reset_password_token=token)
return _datastore.find_user(reset_password_token=token)
def send_reset_password_instructions(user):
@@ -30,7 +44,7 @@ def send_reset_password_instructions(user):
'reset_instructions',
dict(user=user, reset_link=reset_link))
confirm_instructions_sent.send(user, app=current_app._get_current_object())
confirm_instructions_sent.send(user, app=app._get_current_object())
return True
@@ -56,7 +70,7 @@ def generate_reset_password_token(user):
def password_reset_token_is_expired(user):
token_expires = datetime.utcnow() - security.reset_password_within
token_expires = datetime.utcnow() - _security.reset_password_within
return user.reset_password_sent_at < token_expires
@@ -71,18 +85,18 @@ def reset_by_token(token, email, password):
user.reset_password_token = None
user.reset_password_sent_at = None
user.password = security.pwd_context.encrypt(password)
user.password = _security.pwd_context.encrypt(password)
security.datastore._save_model(user)
_datastore._save_model(user)
send_mail('Your password has been reset', user.email, 'reset_notice')
password_reset.send(user, app=current_app._get_current_object())
password_reset.send(user, app=app._get_current_object())
return user
def reset_password_reset_token(user):
security.datastore._save_model(generate_reset_password_token(user))
_datastore._save_model(generate_reset_password_token(user))
send_reset_password_instructions(user)
password_reset_requested.send(user, app=current_app._get_current_object())
password_reset_requested.send(user, app=app._get_current_object())
+12
View File
@@ -1,5 +1,17 @@
# -*- coding: utf-8 -*-
"""
flask.ext.security.signals
~~~~~~~~~~~~~~~~~~~~~~~~~~
Flask-Security signals module
:copyright: (c) 2012 by Matt Wright.
:license: MIT, see LICENSE for more details.
"""
import blinker
signals = blinker.Namespace()
user_registered = signals.signal("user-registered")
+19 -7
View File
@@ -1,19 +1,31 @@
# -*- coding: utf-8 -*-
"""
flask.ext.security.tokens
~~~~~~~~~~~~~~~~~~~~~~~~~
Flask-Security tokens module
:copyright: (c) 2012 by Matt Wright.
:license: MIT, see LICENSE for more details.
"""
from datetime import datetime
from flask import current_app
from flask.ext.security.exceptions import BadCredentialsError, \
UserNotFoundError
from flask.ext.security.utils import generate_token
from flask import current_app as app
from werkzeug.local import LocalProxy
datastore = LocalProxy(lambda: current_app.security.datastore)
from .exceptions import BadCredentialsError, UserNotFoundError
from .utils import generate_token
# Convenient references
_datastore = LocalProxy(lambda: app.security.datastore)
def find_user_by_authentication_token(token):
if not token:
raise BadCredentialsError('Authentication token required')
return datastore.find_user(authentication_token=token)
return _datastore.find_user(authentication_token=token)
def generate_authentication_token(user):
@@ -38,7 +50,7 @@ def generate_authentication_token(user):
def reset_authentication_token(user):
user = generate_authentication_token(user)
datastore._save_model(user)
_datastore._save_model(user)
return user.authentication_token
+2 -2
View File
@@ -11,12 +11,12 @@
import base64
import os
from contextlib import contextmanager
from importlib import import_module
from flask import url_for, flash, current_app, request, session, render_template
from flask.ext.security.signals import user_registered, password_reset_requested
from .signals import user_registered, password_reset_requested
def generate_token():
+47 -39
View File
@@ -9,31 +9,36 @@
:license: MIT, see LICENSE for more details.
"""
from flask import current_app, redirect, request, session, render_template
from flask import current_app as app, redirect, request, session, \
render_template
from flask.ext.login import login_user, logout_user
from flask.ext.principal import Identity, AnonymousIdentity, identity_changed
from flask.ext.security.confirmable import confirm_by_token, \
reset_confirmation_token, send_confirmation_instructions
from flask.ext.security.recoverable import reset_by_token, \
reset_password_reset_token
from flask.ext.security.exceptions import TokenExpiredError, UserNotFoundError, \
ConfirmationError, BadCredentialsError, ResetPasswordError
from flask.ext.security.utils import get_post_login_redirect, do_flash
from flask.ext.security.signals import user_registered
from werkzeug.local import LocalProxy
from .confirmable import confirm_by_token, \
reset_confirmation_token, send_confirmation_instructions
from .exceptions import TokenExpiredError, UserNotFoundError, \
ConfirmationError, BadCredentialsError, ResetPasswordError
from .recoverable import reset_by_token, \
reset_password_reset_token
from .signals import user_registered
from .utils import get_post_login_redirect, do_flash
security = LocalProxy(lambda: current_app.security)
datastore = LocalProxy(lambda: current_app.security.datastore)
logger = LocalProxy(lambda: current_app.logger)
# Convenient references
_security = LocalProxy(lambda: app.security)
_datastore = LocalProxy(lambda: app.security.datastore)
_logger = LocalProxy(lambda: app.logger)
def _do_login(user, remember=True):
if login_user(user, remember):
identity_changed.send(current_app._get_current_object(),
identity_changed.send(app._get_current_object(),
identity=Identity(user.id))
logger.debug('User %s logged in' % user)
_logger.debug('User %s logged in' % user)
return True
return False
@@ -46,10 +51,10 @@ def authenticate():
authenticate fails the user an appropriate error message is flashed and
the user is redirected to the referring page or the login view.
"""
form = security.LoginForm()
form = _security.LoginForm()
try:
user = security.auth_provider.authenticate(form)
user = _security.auth_provider.authenticate(form)
if _do_login(user, remember=form.remember.data):
return redirect(get_post_login_redirect())
@@ -63,9 +68,9 @@ def authenticate():
msg = 'Unknown authentication error'
do_flash(msg, 'error')
logger.debug('Unsuccessful authentication attempt: %s' % msg)
_logger.debug('Unsuccessful authentication attempt: %s' % msg)
return redirect(request.referrer or security.login_manager.login_view)
return redirect(request.referrer or _security.login_manager.login_view)
def logout():
@@ -76,14 +81,14 @@ def logout():
for key in ('identity.name', 'identity.auth_type'):
session.pop(key, None)
identity_changed.send(current_app._get_current_object(),
identity_changed.send(app._get_current_object(),
identity=AnonymousIdentity())
logout_user()
logger.debug('User logged out')
_logger.debug('User logged out')
return redirect(request.args.get('next', None) or
security.post_logout_view)
_security.post_logout_view)
def register():
@@ -94,29 +99,30 @@ def register():
Otherwise the user is redirected to the `SECURITY_POST_LOGIN_VIEW`
configuration value.
"""
form = security.RegisterForm(csrf_enabled=not current_app.testing)
form = _security.RegisterForm(csrf_enabled=not app.testing)
# Exit early if the form doesn't validate
if form.validate_on_submit():
# Create user and send signal
user = datastore.create_user(**form.to_dict())
app = current_app._get_current_object()
user = _datastore.create_user(**form.to_dict())
user_registered.send(user, app=app)
user_registered.send(user, app=app._get_current_object())
# Send confirmation instructions if necessary
if security.confirm_email:
if _security.confirm_email:
send_confirmation_instructions(user)
logger.debug('User %s registered' % user)
_logger.debug('User %s registered' % user)
# Login the user if allowed
if not security.confirm_email or security.login_without_confirmation:
if not _security.confirm_email or _security.login_without_confirmation:
_do_login(user)
return redirect(security.post_register_view or security.post_login_view)
return redirect(_security.post_register_view or
_security.post_login_view)
return redirect(request.referrer or security.register_url)
return redirect(request.referrer or
_security.register_url)
def confirm():
@@ -137,26 +143,27 @@ def confirm():
msg = 'You did not confirm your email within %s. ' \
'A new confirmation code has been sent to %s' % (
security.confirm_email_within_text, e.user.email)
_security.confirm_email_within_text, e.user.email)
do_flash(msg, 'error')
return redirect('/') # TODO: Don't redirect to root
logger.debug('User %s confirmed' % user)
_logger.debug('User %s confirmed' % user)
do_flash('Your email has been confirmed. You may now log in.', 'success')
return redirect(security.post_confirm_view or security.post_login_view)
return redirect(_security.post_confirm_view or
_security.post_login_view)
def forgot():
"""View function that handles the generation of a password reset token.
"""
form = security.ForgotPasswordForm(csrf_enabled=not current_app.testing)
form = _security.ForgotPasswordForm(csrf_enabled=not app.testing)
if form.validate_on_submit():
try:
user = datastore.find_user(**form.to_dict())
user = _datastore.find_user(**form.to_dict())
reset_password_reset_token(user)
@@ -166,7 +173,7 @@ def forgot():
except UserNotFoundError:
do_flash('The email you provided could not be found', 'error')
return redirect(security.post_forgot_view)
return redirect(_security.post_forgot_view)
return render_template('security/passwords/new.html',
forgot_password_form=form)
@@ -176,7 +183,7 @@ def reset():
"""View function that handles the reset of a user's password.
"""
form = security.ResetPasswordForm(csrf_enabled=not current_app.testing)
form = _security.ResetPasswordForm(csrf_enabled=not app.testing)
if form.validate_on_submit():
try:
@@ -187,6 +194,7 @@ def reset():
except TokenExpiredError, e:
do_flash('You did not reset your password within'
'%s.' % security.reset_password_within_text)
'%s.' % _security.reset_password_within_text)
return redirect(request.referrer or security.reset_password_error_view)
return redirect(request.referrer or
_security.reset_password_error_view)