Compare commits

...
47 Commits
Author SHA1 Message Date
Matt Wright f387759cee Bump version number to 1.7.2 2014-05-06 14:05:27 -04:00
Matt Wright faa9f65596 Update CHANGES 2014-05-06 14:04:44 -04:00
Matt Wright 3d5b37cef2 Polish for static analysis 2014-05-06 13:14:25 -04:00
Matt Wright d185407ac4 Add mock X-Forwarded-For header in trackable tests 2014-05-06 13:14:14 -04:00
Matt Wright 58b7fa8e2e Check X-Forwarded-For header value when tracking IP addresses. Fixes #234 2014-05-06 12:35:50 -04:00
Matt Wright 721b985bc4 Something about a bug... 2014-05-06 11:59:28 -04:00
Matt Wright b08d7c2085 Clean up tests with some pytest marking 2014-05-06 11:58:39 -04:00
Matt Wright fff71e3e04 Polish 2014-05-06 07:57:00 -04:00
Matt Wright 546680a94b Polish 2014-03-14 15:33:25 -04:00
Matt Wright 00e121c869 Remove duplicated line 2014-03-14 15:32:39 -04:00
Matt Wright 7098a6757c Update build files 2014-03-14 15:30:30 -04:00
Matt Wright 510d1356a2 A bunch of adjustments to satisfy existing tests and even some new ones 2014-03-14 15:26:53 -04:00
Matt Wright e5111dbb0c Add moar tests! 2014-03-14 15:26:20 -04:00
Matt Wright 44b91a351a Update gitignore 2014-03-13 18:33:22 -04:00
Matt Wright 4b58a00abd Update tox.ini and MANIFEST.in 2014-03-13 18:32:39 -04:00
Matt Wright f3014d01df Convert all tests to use pytest. Phew! 2014-03-13 18:28:25 -04:00
Matt Wright ed645b51f2 Fix confirmation endpoint to flash an "already confirmed" message if user attempts to confirm after already being confirmed 2014-03-13 17:18:39 -04:00
Matt Wright bc60c021a3 Fix password_changed signal to behave like the other signals. Fixes #222 2014-03-13 09:35:54 -04:00
Matt Wright 9f00613906 Merge pull request #225 from nfvs/fix_emails_with_celery_link
Fixed Celery link in "Emails with Celery".
2014-03-13 09:09:39 -04:00
Nuno Santos 0f872af03a Fixed Celery link in "Emails with Celery". 2014-02-27 16:45:25 +01:00
Matt Wright 1d3a75d77b Merge pull request #223 from ahtik/issue-verify_password
verify_password in 1.7.0 was no longer behaving like in 1.6.9
2014-02-20 10:10:44 -05:00
Ahti Kitsik 1395df334e Changing verify_password so it works like verify_and_update_password. Currently verify_password was not only creating a hmac hash but also encrypting (encrypt_password is first hmac-signing and then encrypting).
Removed unneccessary and wrong tests.
2014-02-20 16:46:49 +02:00
Matt Wright 0268a2d568 Merge pull request #217 from swarmer/logout-fix
Make logout view do nothing if the user is already logged out
2014-02-04 07:24:41 -08:00
Anton Barkovsky 09f6f64e5b Make logout view do nothing if the user is already logged out 2014-02-04 16:51:00 +03:00
Matt Wright 8e8d581f0d Add images to README.rst 2014-01-24 14:17:07 -05:00
Matt Wright 65b454ca7e Merge branch 'develop' of github.com:mattupstate/flask-security into develop 2014-01-24 14:09:46 -05:00
Matt Wright 2006d2ae3d Add coverage and coveralls to .travis.yml 2014-01-24 14:09:39 -05:00
Matt Wright c0673c1239 Merge pull request #180 from pythonhub/develop
Allow overwritten on render_template to use Flask-Themes based app #179
2014-01-24 08:15:00 -08:00
Bruno Rocha 922a038d28 Merge https://github.com/mattupstate/flask-security into develop 2014-01-24 14:06:03 -02:00
Matt Wright 668b4ee651 Merge branch 'develop' of github.com:mattupstate/flask-security into develop 2014-01-24 10:32:09 -05:00
Matt Wright f854c24094 Merge pull request #205 from HereLabsInc/catch_decoding_errors
catch possible TypeError and ValueError from serializer
2014-01-24 07:31:51 -08:00
Matt Wright d314e54680 Merge branch 'develop' of github.com:mattupstate/flask-security into develop 2014-01-24 10:25:01 -05:00
Matt Wright 9400032a1f Merge pull request #214 from kleink/develop
Update script.py for Python 2's print statement vs. Python 3's function.
2014-01-24 07:24:55 -08:00
Matt Wright 2e57734d1f Add @anonymous_user_required to register endpoint. Fixes #212 2014-01-24 10:23:42 -05:00
Klaus Klein c658ee4500 Update script.py for Python 2's print statement vs. Python 3's function. 2014-01-24 10:21:41 +01:00
Jameson 45c8951877 passes tests 2014-01-24 04:41:41 +00:00
Jameson b4d1a7c921 update to make test run red 2014-01-23 23:33:11 +00:00
Jameson f47fce9365 add test to show TypeError can occur 2014-01-23 23:26:18 +00:00
Jameson b2174bf035 catch possible TypeError and ValueError from serializer 2014-01-23 23:26:18 +00:00
Matt Wright 316d945d96 Bump version number to 1.7.1 2014-01-14 10:41:53 -05:00
Matt Wright 6eb77c06ad Update CHANGES 2014-01-14 10:41:38 -05:00
Matt Wright 35fd08772b Add configured password hash test back and fix bug with checking passwords 2014-01-14 10:34:57 -05:00
Bruno Rocha 01a4199a16 Merge https://github.com/mattupstate/flask-security into develop
Conflicts:
	flask_security/views.py
2013-12-23 13:55:13 -02:00
Bruno Rocha 910d80d511 Fix version to be able to merge 2013-12-23 13:46:14 -02:00
Bruno Rocha 6494275c8c typo 2013-11-16 22:01:49 -02:00
Bruno Rocha aa80198047 Try to fix a problem with version installed 2013-11-16 21:58:27 -02:00
Bruno Rocha 23b128bcbd Allow overwritten on render_template to use Flask-Themes based app 2013-11-07 04:57:54 -02:00
58 changed files with 2180 additions and 2174 deletions
+1
View File
@@ -33,3 +33,4 @@ env/
*~
*.db
*cache*
+7 -4
View File
@@ -7,14 +7,17 @@ python:
- "pypy"
install:
- pip install . --quiet
- "if [[ $TRAVIS_PYTHON_VERSION == '2.6' ]]; then pip install importlib --quiet --use-mirrors; fi"
- pip install nose simplejson Flask-SQLAlchemy Flask-MongoEngine Flask-Peewee bcrypt --quiet
- pip install -r requirements.txt -r requirements-dev.txt -e .
- pip install coverage coveralls
- "if [[ $TRAVIS_PYTHON_VERSION == '2.6' ]]; then pip install importlib; fi"
services:
- mongodb
script: nosetests
script: py.test --cov flask_security --cov-report term-missing --pep8 --flakes
after_script:
- coveralls
branches:
only:
+18
View File
@@ -3,6 +3,24 @@ Flask-Security Changelog
Here you can see the full list of changes between each Flask-Security release.
Version 1.7.2
-------------
Released May 6th 2014
- Updated IP tracking to check for `X-Forwarded-For` header
- Fixed a bug regarding the re-hashing of passwords with a new algorithm
- Fixed a bug regarding the `password_changed` signal.
Version 1.7.1
-------------
Released January 14th 2014
- Fixed a bug where passwords would fail to verify when specifying a password hash algorithm
Version 1.7.0
-------------
+6 -2
View File
@@ -1,3 +1,7 @@
include CHANGES LICENSE
recursive-include tests *.py
include README.rst
include requirements.txt
include requirements-dev.txt
include tox.ini
recursive-include flask_security/templates *.*
recursive-include tests *
recursive-exclude tests/__pycache__ *
+16 -1
View File
@@ -3,6 +3,21 @@ Flask-Security
.. image:: https://secure.travis-ci.org/mattupstate/flask-security.png?branch=develop
.. image:: https://pypip.in/v/Flask-Security/badge.png
:target: https://pypi.python.org/pypi/Flask-Security/
:alt: Latest Version
.. image:: https://coveralls.io/repos/mattupstate/flask-security/badge.png?branch=develop
:target: https://coveralls.io/r/mattupstate/flask-security
.. image:: https://pypip.in/d/Flask-Security/badge.png
:target: https://pypi.python.org/pypi//Flask-Security/
:alt: Downloads
.. image:: https://pypip.in/license/Flask-Security/badge.png
:target: https://pypi.python.org/pypi/Flask-Security/
:alt: License
Flask-Security quickly adds security features to your Flask application.
Resources
@@ -12,4 +27,4 @@ Resources
- `Issue Tracker <http://github.com/mattupstate/flask-security/issues>`_
- `Code <http://github.com/mattupstate/flask-security/>`_
- `Development Version
<http://github.com/mattupstate/flask-security/zipball/develop#egg=Flask-Security-dev>`_
<http://github.com/mattupstate/flask-security/zipball/develop#egg=Flask-Security-dev>`_
+1 -1
View File
@@ -49,7 +49,7 @@ copyright = u'2012, Matt Wright'
# built documents.
#
# The short X.Y version.
version = '1.7.0'
version = '1.7.2'
# The full version, including alpha/beta/rc tags.
release = version
+4 -3
View File
@@ -145,9 +145,9 @@ templates you can specify an email context processor with the
Emails with Celery
------------------
Sometimes it makes sense to send emails via a task queue, such as
`Celery<http://www.celeryproject.org/>`_. To delay the sending of emails you can
use the ``@security.send_mail_task`` decorator like so::
Sometimes it makes sense to send emails via a task queue, such as `Celery`_.
To delay the sending of emails you can use the ``@security.send_mail_task``
decorator like so::
# Setup the task
@celery.task
@@ -160,3 +160,4 @@ use the ``@security.send_mail_task`` decorator like so::
def delay_security_email(msg):
send_security_email.delay(msg)
.. _Celery: http://www.celeryproject.org/
+4 -4
View File
@@ -10,14 +10,14 @@
:license: MIT, see LICENSE for more details.
"""
__version__ = '1.7.0'
__version__ = '1.7.2'
from .core import Security, RoleMixin, UserMixin, AnonymousUser, current_user
from .datastore import SQLAlchemyUserDatastore, MongoEngineUserDatastore, PeeweeUserDatastore
from .decorators import auth_token_required, http_auth_required, \
login_required, roles_accepted, roles_required, auth_required
login_required, roles_accepted, roles_required, auth_required
from .forms import ForgotPasswordForm, LoginForm, RegisterForm, \
ResetPasswordForm, PasswordlessLoginForm, ConfirmRegisterForm
ResetPasswordForm, PasswordlessLoginForm, ConfirmRegisterForm
from .signals import confirm_instructions_sent, password_reset, \
reset_password_instructions_sent, user_confirmed, user_registered
reset_password_instructions_sent, user_confirmed, user_registered
from .utils import login_user, logout_user, url_for_security
+3 -3
View File
@@ -29,8 +29,8 @@ def send_password_changed_notice(user):
:param user: The user to send the notice to
"""
if config_value('SEND_PASSWORD_CHANGE_EMAIL'):
send_mail(config_value('EMAIL_SUBJECT_PASSWORD_CHANGE_NOTICE'), user.email,
'change_notice', user=user)
subject = config_value('EMAIL_SUBJECT_PASSWORD_CHANGE_NOTICE')
send_mail(subject, user.email, 'change_notice', user=user)
def change_user_password(user, password):
@@ -42,4 +42,4 @@ def change_user_password(user, password):
user.password = encrypt_password(password)
_datastore.put(user)
send_password_changed_notice(user)
password_changed.send(user, app=app._get_current_object())
password_changed.send(app._get_current_object(), user=user)
+6 -1
View File
@@ -58,7 +58,9 @@ def generate_confirmation_token(user):
def requires_confirmation(user):
"""Returns `True` if the user requires confirmation."""
return _security.confirmable and not _security.login_without_confirmation and user.confirmed_at == None
return (_security.confirmable and
not _security.login_without_confirmation and
user.confirmed_at is None)
def confirm_email_token_status(token):
@@ -77,6 +79,9 @@ def confirm_user(user):
:param user: The user to confirm
"""
if user.confirmed_at is not None:
return False
user.confirmed_at = datetime.utcnow()
_datastore.put(user)
user_confirmed.send(app._get_current_object(), user=user)
return True
+94 -52
View File
@@ -9,7 +9,7 @@
:license: MIT, see LICENSE for more details.
"""
from flask import current_app
from flask import current_app, render_template
from flask.ext.login import AnonymousUserMixin, UserMixin as BaseUserMixin, \
LoginManager, current_user
from flask.ext.principal import Principal, RoleNeed, UserNeed, Identity, \
@@ -87,56 +87,92 @@ _default_config = {
'EMAIL_SUBJECT_PASSWORD_NOTICE': 'Your password has been reset',
'EMAIL_SUBJECT_PASSWORD_CHANGE_NOTICE': 'Your password has been changed',
'EMAIL_SUBJECT_PASSWORD_RESET': 'Password reset instructions',
'USER_IDENTITY_ATTRIBUTES': ['email']
'USER_IDENTITY_ATTRIBUTES': ['email'],
'PASSWORD_SCHEMES': [
'bcrypt',
'des_crypt',
'pbkdf2_sha256',
'pbkdf2_sha512',
'sha256_crypt',
'sha512_crypt',
# And always last one...
'plaintext'
],
'DEPRECATED_PASSWORD_SCHEMES': ['auto']
}
#: Default Flask-Security messages
_default_messages = {
'UNAUTHORIZED': ('You do not have permission to view this resource.', 'error'),
'CONFIRM_REGISTRATION': ('Thank you. Confirmation instructions have been sent to %(email)s.', 'success'),
'EMAIL_CONFIRMED': ('Thank you. Your email has been confirmed.', 'success'),
'ALREADY_CONFIRMED': ('Your email has already been confirmed.', 'info'),
'INVALID_CONFIRMATION_TOKEN': ('Invalid confirmation token.', 'error'),
'EMAIL_ALREADY_ASSOCIATED': ('%(email)s is already associated with an account.', 'error'),
'PASSWORD_MISMATCH': ('Password does not match', 'error'),
'RETYPE_PASSWORD_MISMATCH': ('Passwords do not match', 'error'),
'INVALID_REDIRECT': ('Redirections outside the domain are forbidden', 'error'),
'PASSWORD_RESET_REQUEST': ('Instructions to reset your password have been sent to %(email)s.', 'info'),
'PASSWORD_RESET_EXPIRED': ('You did not reset your password within %(within)s. New instructions have been sent to %(email)s.', 'error'),
'INVALID_RESET_PASSWORD_TOKEN': ('Invalid reset password token.', 'error'),
'CONFIRMATION_REQUIRED': ('Email requires confirmation.', 'error'),
'CONFIRMATION_REQUEST': ('Confirmation instructions have been sent to %(email)s.', 'info'),
'CONFIRMATION_EXPIRED': ('You did not confirm your email within %(within)s. New instructions to confirm your email have been sent to %(email)s.', 'error'),
'LOGIN_EXPIRED': ('You did not login within %(within)s. New instructions to login have been sent to %(email)s.', 'error'),
'LOGIN_EMAIL_SENT': ('Instructions to login have been sent to %(email)s.', 'success'),
'INVALID_LOGIN_TOKEN': ('Invalid login token.', 'error'),
'DISABLED_ACCOUNT': ('Account is disabled.', 'error'),
'EMAIL_NOT_PROVIDED': ('Email not provided', 'error'),
'INVALID_EMAIL_ADDRESS': ('Invalid email address', 'error'),
'PASSWORD_NOT_PROVIDED': ('Password not provided', 'error'),
'PASSWORD_NOT_SET': ('No password is set for this user', 'error'),
'PASSWORD_INVALID_LENGTH': ('Password must be at least 6 characters', 'error'),
'USER_DOES_NOT_EXIST': ('Specified user does not exist', 'error'),
'INVALID_PASSWORD': ('Invalid password', 'error'),
'PASSWORDLESS_LOGIN_SUCCESSFUL': ('You have successfuly logged in.', 'success'),
'PASSWORD_RESET': ('You successfully reset your password and you have been logged in automatically.', 'success'),
'PASSWORD_IS_THE_SAME': ('Your new password must be different than your previous password.', 'error'),
'PASSWORD_CHANGE': ('You successfully changed your password.', 'success'),
'LOGIN': ('Please log in to access this page.', 'info'),
'REFRESH': ('Please reauthenticate to access this page.', 'info'),
'UNAUTHORIZED': (
'You do not have permission to view this resource.', 'error'),
'CONFIRM_REGISTRATION': (
'Thank you. Confirmation instructions have been sent to %(email)s.', 'success'),
'EMAIL_CONFIRMED': (
'Thank you. Your email has been confirmed.', 'success'),
'ALREADY_CONFIRMED': (
'Your email has already been confirmed.', 'info'),
'INVALID_CONFIRMATION_TOKEN': (
'Invalid confirmation token.', 'error'),
'EMAIL_ALREADY_ASSOCIATED': (
'%(email)s is already associated with an account.', 'error'),
'PASSWORD_MISMATCH': (
'Password does not match', 'error'),
'RETYPE_PASSWORD_MISMATCH': (
'Passwords do not match', 'error'),
'INVALID_REDIRECT': (
'Redirections outside the domain are forbidden', 'error'),
'PASSWORD_RESET_REQUEST': (
'Instructions to reset your password have been sent to %(email)s.', 'info'),
'PASSWORD_RESET_EXPIRED': (
'You did not reset your password within %(within)s. New instructions have been sent '
'to %(email)s.', 'error'),
'INVALID_RESET_PASSWORD_TOKEN': (
'Invalid reset password token.', 'error'),
'CONFIRMATION_REQUIRED': (
'Email requires confirmation.', 'error'),
'CONFIRMATION_REQUEST': (
'Confirmation instructions have been sent to %(email)s.', 'info'),
'CONFIRMATION_EXPIRED': (
'You did not confirm your email within %(within)s. New instructions to confirm your email '
'have been sent to %(email)s.', 'error'),
'LOGIN_EXPIRED': (
'You did not login within %(within)s. New instructions to login have been sent to '
'%(email)s.', 'error'),
'LOGIN_EMAIL_SENT': (
'Instructions to login have been sent to %(email)s.', 'success'),
'INVALID_LOGIN_TOKEN': (
'Invalid login token.', 'error'),
'DISABLED_ACCOUNT': (
'Account is disabled.', 'error'),
'EMAIL_NOT_PROVIDED': (
'Email not provided', 'error'),
'INVALID_EMAIL_ADDRESS': (
'Invalid email address', 'error'),
'PASSWORD_NOT_PROVIDED': (
'Password not provided', 'error'),
'PASSWORD_NOT_SET': (
'No password is set for this user', 'error'),
'PASSWORD_INVALID_LENGTH': (
'Password must be at least 6 characters', 'error'),
'USER_DOES_NOT_EXIST': (
'Specified user does not exist', 'error'),
'INVALID_PASSWORD': (
'Invalid password', 'error'),
'PASSWORDLESS_LOGIN_SUCCESSFUL': (
'You have successfuly logged in.', 'success'),
'PASSWORD_RESET': (
'You successfully reset your password and you have been logged in automatically.',
'success'),
'PASSWORD_IS_THE_SAME': (
'Your new password must be different than your previous password.', 'error'),
'PASSWORD_CHANGE': (
'You successfully changed your password.', 'success'),
'LOGIN': (
'Please log in to access this page.', 'info'),
'REFRESH': (
'Please reauthenticate to access this page.', 'info'),
}
_allowed_password_hash_schemes = [
'bcrypt',
'des_crypt',
'pbkdf2_sha256',
'pbkdf2_sha512',
'sha256_crypt',
'sha512_crypt',
# And always last one...
'plaintext'
]
_default_forms = {
'login_form': LoginForm,
'confirm_register_form': ConfirmRegisterForm,
@@ -206,10 +242,12 @@ def _get_principal(app):
def _get_pwd_context(app):
pw_hash = cv('PASSWORD_HASH', app=app)
if pw_hash not in _allowed_password_hash_schemes:
allowed = ', '.join(_allowed_password_hash_schemes[:-1]) + ' and ' + _allowed_password_hash_schemes[-1]
schemes = cv('PASSWORD_SCHEMES', app=app)
deprecated = cv('DEPRECATED_PASSWORD_SCHEMES', app=app)
if pw_hash not in schemes:
allowed = (', '.join(schemes[:-1]) + ' and ' + schemes[-1])
raise ValueError("Invalid hash scheme %r. Allowed values are %s" % (pw_hash, allowed))
return CryptContext(schemes=_allowed_password_hash_schemes, default=pw_hash)
return CryptContext(schemes=schemes, default=pw_hash, deprecated=deprecated)
def _get_serializer(app, name):
@@ -257,6 +295,9 @@ class RoleMixin(object):
def __ne__(self, other):
return not self.__eq__(other)
def __hash__(self):
return hash(self.name)
class UserMixin(BaseUserMixin):
"""Mixin for `User` model definitions"""
@@ -302,15 +343,12 @@ class _SecurityState(object):
fn not in group and group.append(fn)
def _run_ctx_processor(self, endpoint):
rv, fns = {}, []
rv = {}
for g in [None, endpoint]:
for fn in self._context_processors.setdefault(g, []):
rv.update(fn())
return rv
def context_processor(self, fn):
self._add_ctx_processor(None, fn)
def forgot_password_context_processor(self, fn):
self._add_ctx_processor('forgot_password', fn)
@@ -388,9 +426,13 @@ class Security(object):
app.register_blueprint(create_blueprint(state, __name__))
app.context_processor(_context_processor)
state.render_template = self.render_template
app.extensions['security'] = state
return state
def render_template(self, *args, **kwargs):
return render_template(*args, **kwargs)
def __getattr__(self, name):
return getattr(self._state, name, None)
+9 -7
View File
@@ -237,17 +237,18 @@ class MongoEngineUserDatastore(MongoEngineDatastore, UserDatastore):
query = QCombination(QCombination.AND, queries)
try:
return self.user_model.objects(query).first()
except ValidationError:
except ValidationError: # pragma: no cover
return None
def find_role(self, role):
return self.role_model.objects(name=role).first()
def add_role_to_user(self, user, role):
rv = super(MongoEngineUserDatastore, self).add_role_to_user(user, role)
if rv:
self.put(user)
return rv
# TODO: Not sure why this was added but tests pass without it
# def add_role_to_user(self, user, role):
# rv = super(MongoEngineUserDatastore, self).add_role_to_user(user, role)
# if rv:
# self.put(user)
# return rv
class PeeweeUserDatastore(PeeweeDatastore, UserDatastore):
@@ -295,6 +296,7 @@ class PeeweeUserDatastore(PeeweeDatastore, UserDatastore):
user = self.put(user)
for role in roles:
self.add_role_to_user(user, role)
self.put(user)
return user
def add_role_to_user(self, user, role):
@@ -309,7 +311,7 @@ class PeeweeUserDatastore(PeeweeDatastore, UserDatastore):
if result.count():
return False
else:
self.UserRole.create(user=user.id, role=role.id)
self.put(self.UserRole.create(user=user.id, role=role.id))
return True
def remove_role_from_user(self, user, role):
+1 -1
View File
@@ -13,7 +13,7 @@ from collections import namedtuple
from functools import wraps
from flask import current_app, Response, request, redirect, _request_ctx_stack
from flask.ext.login import current_user, login_required
from flask.ext.login import current_user, login_required # pragma: no flakes
from flask.ext.principal import RoleNeed, Permission, Identity, identity_changed
from werkzeug.local import LocalProxy
+8 -16
View File
@@ -10,12 +10,8 @@
"""
import inspect
try:
from urlparse import urlsplit
except ImportError:
from urllib.parse import urlsplit
from flask import request, current_app
from flask import request, current_app, flash
from flask_wtf import Form as BaseForm
from wtforms import TextField, PasswordField, validators, \
SubmitField, HiddenField, BooleanField, ValidationError, Field
@@ -23,7 +19,7 @@ from flask_login import current_user
from werkzeug.local import LocalProxy
from .confirmable import requires_confirmation
from .utils import verify_and_update_password, get_message, config_value
from .utils import verify_and_update_password, get_message, config_value, validate_redirect_url
# Convenient reference
_datastore = LocalProxy(lambda: current_app.extensions['security'].datastore)
@@ -137,12 +133,10 @@ class NextFormMixin():
next = HiddenField()
def validate_next(self, field):
if field.data:
url_next = urlsplit(field.data)
url_base = urlsplit(request.host_url)
if url_next.netloc and url_next.netloc != url_base.netloc:
field.data = ''
raise ValidationError(get_message('INVALID_REDIRECT')[0])
if field.data and not validate_redirect_url(field.data):
field.data = ''
flash(*get_message('INVALID_REDIRECT'))
raise ValidationError(get_message('INVALID_REDIRECT')[0])
class RegisterFormMixin():
@@ -209,6 +203,8 @@ class LoginForm(Form, NextFormMixin):
def __init__(self, *args, **kwargs):
super(LoginForm, self).__init__(*args, **kwargs)
if not self.next.data:
self.next.data = request.args.get('next', '')
self.remember.default = config_value('DEFAULT_REMEMBER_ME')
def validate(self):
@@ -223,7 +219,6 @@ class LoginForm(Form, NextFormMixin):
self.password.errors.append(get_message('PASSWORD_NOT_PROVIDED')[0])
return False
self.user = _datastore.get_user(self.email.data)
if self.user is None:
@@ -276,9 +271,6 @@ class ChangePasswordForm(Form, PasswordFormMixin):
if not super(ChangePasswordForm, self).validate():
return False
if self.password.data.strip() == '':
self.password.errors.append(get_message('PASSWORD_NOT_PROVIDED')[0])
return False
if not verify_and_update_password(self.password.data, current_user):
self.password.errors.append(get_message('INVALID_PASSWORD')[0])
return False
+12 -10
View File
@@ -8,6 +8,8 @@
:copyright: (c) 2012 by Matt Wright.
:license: MIT, see LICENSE for more details.
"""
from __future__ import print_function
try:
import simplejson as json
except ImportError:
@@ -26,7 +28,7 @@ _datastore = LocalProxy(lambda: current_app.extensions['security'].datastore)
def pprint(obj):
print json.dumps(obj, sort_keys=True, indent=4)
print(json.dumps(obj, sort_keys=True, indent=4))
def commit(fn):
@@ -40,9 +42,9 @@ class CreateUserCommand(Command):
"""Create a user"""
option_list = (
Option('-e', '--email', dest='email', default=None),
Option('-e', '--email', dest='email', default=None),
Option('-p', '--password', dest='password', default=None),
Option('-a', '--active', dest='active', default=''),
Option('-a', '--active', dest='active', default=''),
)
@commit
@@ -59,11 +61,11 @@ class CreateUserCommand(Command):
if form.validate():
kwargs['password'] = encrypt_password(kwargs['password'])
_datastore.create_user(**kwargs)
print 'User created successfully.'
print('User created successfully.')
kwargs['password'] = '****'
pprint(kwargs)
else:
print 'Error creating user'
print('Error creating user')
pprint(form.errors)
@@ -78,7 +80,7 @@ class CreateRoleCommand(Command):
@commit
def run(self, **kwargs):
_datastore.create_role(**kwargs)
print 'Role "%(name)s" created successfully.' % kwargs
print('Role "%(name)s" created successfully.' % kwargs)
class _RoleCommand(Command):
@@ -94,7 +96,7 @@ class AddRoleCommand(_RoleCommand):
@commit
def run(self, user_identifier, role_name):
_datastore.add_role_to_user(user_identifier, role_name)
print "Role '%s' added to user '%s' successfully" % (role_name, user_identifier)
print("Role '%s' added to user '%s' successfully" % (role_name, user_identifier))
class RemoveRoleCommand(_RoleCommand):
@@ -103,7 +105,7 @@ class RemoveRoleCommand(_RoleCommand):
@commit
def run(self, user_identifier, role_name):
_datastore.remove_role_from_user(user_identifier, role_name)
print "Role '%s' removed from user '%s' successfully" % (role_name, user_identifier)
print("Role '%s' removed from user '%s' successfully" % (role_name, user_identifier))
class _ToggleActiveCommand(Command):
@@ -118,7 +120,7 @@ class DeactivateUserCommand(_ToggleActiveCommand):
@commit
def run(self, user_identifier):
_datastore.deactivate_user(user_identifier)
print "User '%s' has been deactivated" % user_identifier
print("User '%s' has been deactivated" % user_identifier)
class ActivateUserCommand(_ToggleActiveCommand):
@@ -127,4 +129,4 @@ class ActivateUserCommand(_ToggleActiveCommand):
@commit
def run(self, user_identifier):
_datastore.activate_user(user_identifier)
print "User '%s' has been activated" % user_identifier
print("User '%s' has been activated" % user_identifier)
+53 -79
View File
@@ -10,26 +10,26 @@
"""
import base64
import blinker
import functools
import hashlib
import hmac
import sys
try:
from urlparse import urlsplit
except ImportError: # pragma: no cover
from urllib.parse import urlsplit
from contextlib import contextmanager
from datetime import datetime, timedelta
from flask import url_for, flash, current_app, request, session, render_template
from flask.ext.login import login_user as _login_user, \
logout_user as _logout_user
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 .signals import user_registered, user_confirmed, \
confirm_instructions_sent, login_instructions_sent, \
password_reset, password_changed, reset_password_instructions_sent
from .signals import user_registered, login_instructions_sent, reset_password_instructions_sent
# Convenient references
_security = LocalProxy(lambda: current_app.extensions['security'])
@@ -40,12 +40,12 @@ _pwd_context = LocalProxy(lambda: _security.pwd_context)
PY3 = sys.version_info[0] == 3
if PY3:
string_types = str,
text_type = str
else:
string_types = basestring,
text_type = unicode
if PY3: # pragma: no cover
string_types = str, # pragma: no flakes
text_type = str # pragma: no flakes
else: # pragma: no cover
string_types = basestring, # pragma: no flakes
text_type = unicode # pragma: no flakes
def login_user(user, remember=None):
@@ -58,12 +58,16 @@ def login_user(user, remember=None):
if remember is None:
remember = config_value('DEFAULT_REMEMBER_ME')
if not _login_user(user, remember):
if not _login_user(user, remember): # pragma: no cover
return False
if _security.trackable:
if 'X-Forwarded-For' not in request.headers:
remote_addr = request.remote_addr or 'untrackable'
else:
remote_addr = request.headers.getlist("X-Forwarded-For")[0]
old_current_login, new_current_login = user.current_login_at, datetime.utcnow()
remote_addr = request.remote_addr or 'untrackable'
old_current_ip, new_current_ip = user.current_login_ip, remote_addr
user.last_login_at = old_current_login or new_current_login
@@ -111,7 +115,10 @@ def verify_password(password, password_hash):
:param password: A plaintext password to verify
:param password_hash: The expected hash value of the password (usually form your database)
"""
return _pwd_context.verify(encrypt_password(password), password_hash)
if _security.password_hash != 'plaintext':
password = get_hmac(password)
return _pwd_context.verify(password, password_hash)
def verify_and_update_password(password, user):
@@ -121,7 +128,10 @@ def verify_and_update_password(password, user):
:param password: A plaintext password to verify
:param user: The user to verify against
"""
verified, new_password = _pwd_context.verify_and_update(encrypt_password(password), user.password)
if _pwd_context.identify(user.password) != 'plaintext':
password = get_hmac(password)
verified, new_password = _pwd_context.verify_and_update(password, user.password)
if verified and new_password:
user.password = new_password
_datastore.put(user)
@@ -135,8 +145,8 @@ def encrypt_password(password):
"""
if _security.password_hash == 'plaintext':
return password
signed = get_hmac(password)
return _pwd_context.encrypt(signed.decode('ascii'))
signed = get_hmac(password).decode('ascii')
return _pwd_context.encrypt(signed)
def md5(data):
@@ -185,14 +195,31 @@ def url_for_security(endpoint, **values):
return url_for(endpoint, **values)
def get_post_action_redirect(config_key):
return (get_url(request.args.get('next')) or
get_url(request.form.get('next')) or
find_redirect(config_key))
def validate_redirect_url(url):
if url is None:
return False
url_next = urlsplit(url)
url_base = urlsplit(request.host_url)
if url_next.netloc and url_next.netloc != url_base.netloc:
return False
return True
def get_post_login_redirect():
return get_post_action_redirect('SECURITY_POST_LOGIN_VIEW')
def get_post_action_redirect(config_key, declared=None):
urls = [
get_url(request.args.get('next')),
get_url(request.form.get('next')),
find_redirect(config_key)
]
if declared:
urls.insert(0, declared)
for url in urls:
if validate_redirect_url(url):
return url
def get_post_login_redirect(declared=None):
return get_post_action_redirect('SECURITY_POST_LOGIN_VIEW', declared)
def get_post_register_redirect():
@@ -313,7 +340,7 @@ def get_token_status(token, serializer, max_age=None):
except SignatureExpired:
d, data = serializer.loads_unsafe(token)
expired = True
except BadSignature:
except (BadSignature, TypeError, ValueError):
invalid = True
if data:
@@ -386,56 +413,3 @@ def capture_reset_password_requests(reset_password_sent_at=None):
yield reset_requests
finally:
reset_password_instructions_sent.disconnect(_on)
class CaptureSignals(object):
"""Testing utility for capturing blinker signals.
Context manager which mocks out selected signals and registers which are `sent` on and what
arguments were sent. Instantiate with a list of blinker `NamedSignals` to patch. Each signal
has it's `send` mocked out.
"""
def __init__(self, signals):
"""Patch all given signals and make them available as attributes.
:param signals: list of signals
"""
self._records = {}
self._receivers = {}
for signal in signals:
self._records[signal] = []
self._receivers[signal] = functools.partial(self._record, signal)
def __getitem__(self, signal):
"""All captured signals are available via `ctxt[signal]`.
"""
if isinstance(signal, blinker.base.NamedSignal):
return self._records[signal]
else:
super(CaptureSignals, self).__setitem__(signal)
def _record(self, signal, *args, **kwargs):
self._records[signal].append((args, kwargs))
def __enter__(self):
for signal, receiver in self._receivers.items():
signal.connect(receiver)
return self
def __exit__(self, type, value, traceback):
for signal, receiver in self._receivers.items():
signal.disconnect(receiver)
def signals_sent(self):
"""Return a set of the signals sent.
:rtype: list of blinker `NamedSignals`.
"""
return set([signal for signal, _ in self._records.items() if self._records[signal]])
def capture_signals():
"""Factory method that creates a `CaptureSignals` with all the flask_security signals."""
return CaptureSignals([user_registered, user_confirmed,
confirm_instructions_sent, login_instructions_sent,
password_reset, password_changed,
reset_password_instructions_sent])
+34 -32
View File
@@ -9,7 +9,7 @@
:license: MIT, see LICENSE for more details.
"""
from flask import current_app, redirect, request, render_template, jsonify, \
from flask import current_app, redirect, request, jsonify, \
after_this_request, Blueprint
from flask_login import current_user
from werkzeug.datastructures import MultiDict
@@ -75,29 +75,27 @@ def login():
after_this_request(_commit)
if not request.json:
return redirect(get_post_login_redirect())
form.next.data = get_url(request.args.get('next')) \
or get_url(request.form.get('next')) or ''
return redirect(get_post_login_redirect(form.next.data))
if request.json:
return _render_json(form, True)
return render_template(config_value('LOGIN_USER_TEMPLATE'),
login_user_form=form,
**_ctx('login'))
return _security.render_template(config_value('LOGIN_USER_TEMPLATE'),
login_user_form=form,
**_ctx('login'))
@login_required
def logout():
"""View function which handles a logout request."""
logout_user()
if current_user.is_authenticated():
logout_user()
return redirect(request.args.get('next', None) or
get_url(_security.post_logout_view))
@anonymous_user_required
def register():
"""View function which handles a registration request."""
@@ -128,9 +126,9 @@ def register():
if request.json:
return _render_json(form)
return render_template(config_value('REGISTER_USER_TEMPLATE'),
register_user_form=form,
**_ctx('register'))
return _security.render_template(config_value('REGISTER_USER_TEMPLATE'),
register_user_form=form,
**_ctx('register'))
def send_login():
@@ -151,9 +149,9 @@ def send_login():
if request.json:
return _render_json(form)
return render_template(config_value('SEND_LOGIN_TEMPLATE'),
send_login_form=form,
**_ctx('send_login'))
return _security.render_template(config_value('SEND_LOGIN_TEMPLATE'),
send_login_form=form,
**_ctx('send_login'))
@anonymous_user_required
@@ -196,9 +194,9 @@ def send_confirmation():
if request.json:
return _render_json(form)
return render_template(config_value('SEND_CONFIRMATION_TEMPLATE'),
send_confirmation_form=form,
**_ctx('send_confirmation'))
return _security.render_template(config_value('SEND_CONFIRMATION_TEMPLATE'),
send_confirmation_form=form,
**_ctx('send_confirmation'))
def confirm_email(token):
@@ -221,9 +219,13 @@ def confirm_email(token):
logout_user()
login_user(user)
confirm_user(user)
after_this_request(_commit)
do_flash(*get_message('EMAIL_CONFIRMED'))
if confirm_user(user):
after_this_request(_commit)
msg = 'EMAIL_CONFIRMED'
else:
msg = 'ALREADY_CONFIRMED'
do_flash(*get_message(msg))
return redirect(get_url(_security.post_confirm_view) or
get_url(_security.post_login_view))
@@ -247,9 +249,9 @@ def forgot_password():
if request.json:
return _render_json(form)
return render_template(config_value('FORGOT_PASSWORD_TEMPLATE'),
forgot_password_form=form,
**_ctx('forgot_password'))
return _security.render_template(config_value('FORGOT_PASSWORD_TEMPLATE'),
forgot_password_form=form,
**_ctx('forgot_password'))
@anonymous_user_required
@@ -276,10 +278,10 @@ def reset_password(token):
return redirect(get_url(_security.post_reset_view) or
get_url(_security.post_login_view))
return render_template(config_value('RESET_PASSWORD_TEMPLATE'),
reset_password_form=form,
reset_password_token=token,
**_ctx('reset_password'))
return _security.render_template(config_value('RESET_PASSWORD_TEMPLATE'),
reset_password_form=form,
reset_password_token=token,
**_ctx('reset_password'))
@login_required
@@ -305,9 +307,9 @@ def change_password():
form.user = current_user
return _render_json(form)
return render_template(config_value('CHANGE_PASSWORD_TEMPLATE'),
change_password_form=form,
**_ctx('change_password'))
return _security.render_template(config_value('CHANGE_PASSWORD_TEMPLATE'),
change_password_form=form,
**_ctx('change_password'))
def create_blueprint(state, import_name):
+10
View File
@@ -0,0 +1,10 @@
Flask-SQLAlchemy>=1.0
bcrypt>=1.0.2
flask-mongoengine>=0.7.0
flask-peewee>=0.6.5
pytest>=2.5.2
pytest-cache>=1.0
pytest-cov>=1.6
pytest-flakes>=0.2
pytest-pep8>=1.0.5
tox>=1.7.0
+14 -1
View File
@@ -3,4 +3,17 @@ source-dir = docs/
build-dir = docs/_build
[upload_sphinx]
upload-dir = docs/_build/html
upload-dir = docs/_build/html
[pytest]
pep8maxlinelength = 99
pep8ignore =
docs/* ALL
scripts/* ALL
flakes-ignore =
ImportStarUsed
flask_security/__init__.py UnusedImport
docs/* ALL
scripts/* ALL
+44 -36
View File
@@ -1,55 +1,63 @@
"""
Flask-Security
==============
Flask-Security is a Flask extension that aims to add quick and simple security
to your Flask applications.
Resources
---------
* `Documentation <http://packages.python.org/Flask-Security/>`_
* `Issue Tracker <https://github.com/mattupstate/flask-security/issues>`_
* `Source <https://github.com/mattupstate/flask-security>`_
* `Development Version
<https://github.com/mattupstate/flask-security/raw/develop#egg=Flask-Security-dev>`_
"""
from setuptools import setup
import multiprocessing # pragma: no flakes
import sys
from setuptools import setup, find_packages
from setuptools.command.test import test as TestCommand
def get_requirements(suffix=''):
with open('requirements%s.txt' % suffix) as f:
rv = f.read().splitlines()
return rv
def get_long_description():
with open('README.rst') as f:
rv = f.read()
return rv
class PyTest(TestCommand):
def finalize_options(self):
TestCommand.finalize_options(self)
self.test_args = [
'-xrs',
'--cov', 'flask_security',
'--cov-report', 'term-missing',
'--pep8',
'--flakes',
'--clearcache'
]
self.test_suite = True
def run_tests(self):
import pytest
errno = pytest.main(self.test_args)
sys.exit(errno)
setup(
name='Flask-Security',
version='1.7.0',
version='1.7.2',
url='https://github.com/mattupstate/flask-security',
license='MIT',
author='Matt Wright',
author_email='matt@nobien.net',
description='Simple security for Flask apps',
long_description=__doc__,
packages=[
'flask_security'
],
long_description=get_long_description(),
packages=find_packages(),
zip_safe=False,
include_package_data=True,
platforms='any',
install_requires=[
'Flask>=0.10.1',
'Flask-Login>=0.2.9',
'Flask-Mail>=0.9.0',
'Flask-Principal>=0.4.0',
'Flask-WTF>=0.9.3',
'passlib>=1.6.2',
],
test_suite='nose.collector',
tests_require=[
'nose',
'Flask-SQLAlchemy',
'Flask-MongoEngine',
'Flask-Peewee',
'bcrypt',
'simplejson'
],
install_requires=get_requirements(),
tests_require=get_requirements('-dev'),
cmdclass={'test': PyTest},
classifiers=[
'Development Status :: 4 - Beta',
'Environment :: Web Environment',
-85
View File
@@ -1,85 +0,0 @@
# -*- coding: utf-8 -*-
import hmac
from hashlib import sha1
from unittest import TestCase
from tests.test_app.sqlalchemy import create_app
class SecurityTest(TestCase):
APP_KWARGS = {
'register_blueprint': True,
}
AUTH_CONFIG = None
def setUp(self):
super(SecurityTest, self).setUp()
app_kwargs = self.APP_KWARGS
app = self._create_app(self.AUTH_CONFIG or {}, **app_kwargs)
app.debug = False
app.config['TESTING'] = True
app.config['WTF_CSRF_ENABLED'] = False
self.app = app
self.client = app.test_client()
def _create_app(self, auth_config, **kwargs):
return create_app(auth_config, **kwargs)
def _get(self, route, content_type=None, follow_redirects=None, headers=None):
return self.client.get(route, follow_redirects=follow_redirects,
content_type=content_type or 'text/html',
headers=headers)
def _post(self, route, data=None, content_type=None, follow_redirects=True, headers=None):
content_type = content_type or 'application/x-www-form-urlencoded'
return self.client.post(route, data=data,
follow_redirects=follow_redirects,
content_type=content_type, headers=headers)
def register(self, email, password='password'):
data = dict(email=email, password=password)
return self.client.post('/register', data=data, follow_redirects=True)
def authenticate(self, email="matt@lp.com", password="password", endpoint=None, **kwargs):
data = dict(email=email, password=password, remember='y')
return self._post(endpoint or '/login', data=data, **kwargs)
def json_authenticate(self, email="matt@lp.com", password="password", endpoint=None):
data = """{
"email": "%s",
"password": "%s"
}"""
return self._post(endpoint or '/login', content_type="application/json",
data=data % (email, password))
def logout(self, endpoint=None):
return self._get(endpoint or '/logout', follow_redirects=True)
def assertIsHomePage(self, data):
self.assertIn(b'Home Page', data)
def assertIn(self, member, container, msg=None):
if hasattr(TestCase, 'assertIn'):
return TestCase.assertIn(self, member, container, msg)
return self.assertTrue(member in container)
def assertNotIn(self, member, container, msg=None):
if hasattr(TestCase, 'assertNotIn'):
return TestCase.assertNotIn(self, member, container, msg)
return self.assertFalse(member in container)
def assertIsNotNone(self, obj, msg=None):
if hasattr(TestCase, 'assertIsNotNone'):
return TestCase.assertIsNotNone(self, obj, msg)
return self.assertTrue(obj is not None)
def get_message(self, key, **kwargs):
return self.app.config['SECURITY_MSG_' + key][0] % kwargs
-854
View File
@@ -1,854 +0,0 @@
# -*- coding: utf-8 -*-
# from __future__ import with_statement
import base64
import time
import simplejson as json
import flask
from flask_security.utils import capture_registrations, \
capture_reset_password_requests, capture_passwordless_login_requests
from flask_security.forms import LoginForm, ConfirmRegisterForm, RegisterForm, \
ForgotPasswordForm, ResetPasswordForm, SendConfirmationForm, \
PasswordlessLoginForm
from flask_security.forms import TextField, SubmitField, valid_user_email
from flask_security.signals import user_registered
from tests import SecurityTest
# TODO: Wait for passlib + bcrypt python3 compatibility to be fixed
# class ConfiguredPasswordHashSecurityTests(SecurityTest):
# AUTH_CONFIG = {
# 'SECURITY_PASSWORD_HASH': 'bcrypt',
# 'SECURITY_PASSWORD_SALT': 'so-salty',
# 'USER_COUNT': 1
# }
# def test_authenticate(self):
# r = self.authenticate(endpoint="/login")
# self.assertIn(b'Home Page', r.data)
class ConfiguredSecurityTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_REGISTERABLE': True,
'SECURITY_LOGOUT_URL': '/custom_logout',
'SECURITY_LOGIN_URL': '/custom_login',
'SECURITY_POST_LOGIN_VIEW': '/post_login',
'SECURITY_POST_LOGOUT_VIEW': '/post_logout',
'SECURITY_POST_REGISTER_VIEW': '/post_register',
'SECURITY_UNAUTHORIZED_VIEW': '/unauthorized',
'SECURITY_DEFAULT_HTTP_AUTH_REALM': 'Custom Realm'
}
def test_login_view(self):
r = self._get('/custom_login')
self.assertIn(b"<h1>Login</h1>", r.data)
def test_authenticate(self):
r = self.authenticate(endpoint="/custom_login")
self.assertIn(b'Post Login', r.data)
def test_logout(self):
self.authenticate(endpoint="/custom_login")
r = self.logout(endpoint="/custom_logout")
self.assertIn(b'Post Logout', r.data)
def test_register_view(self):
r = self._get('/register')
self.assertIn(b'<h1>Register</h1>', r.data)
def test_register(self):
data = dict(email='dude@lp.com',
password='password',
password_confirm='password')
r = self._post('/register', data=data, follow_redirects=True)
self.assertIn(b'Post Register', r.data)
def test_register_with_next_querystring_argument(self):
data = dict(email='dude@lp.com',
password='password',
password_confirm='password')
r = self._post('/register?next=/page1', data=data, follow_redirects=True)
self.assertIn(b'Page 1', r.data)
def test_register_json(self):
data = '{ "email": "dude@lp.com", "password": "password"}'
r = self._post('/register', data=data, content_type='application/json')
data = json.loads(r.data)
self.assertEquals(data['meta']['code'], 200)
def test_register_existing_email(self):
data = dict(email='matt@lp.com',
password='password',
password_confirm='password')
r = self._post('/register', data=data, follow_redirects=True)
msg = b'matt@lp.com is already associated with an account'
self.assertIn(msg, r.data)
def test_unauthorized(self):
self.authenticate("joe@lp.com", endpoint="/custom_auth")
r = self._get("/admin", follow_redirects=True)
msg = b'You are not allowed to access the requested resouce'
self.assertIn(msg, r.data)
def test_default_http_auth_realm(self):
r = self._get('/http', headers={
'Authorization': 'Basic %s' % base64.b64encode(b"joe@lp.com:bogus")
})
self.assertIn(b'<h1>Unauthorized</h1>', r.data)
self.assertIn('WWW-Authenticate', r.headers)
self.assertEquals('Basic realm="Custom Realm"',
r.headers['WWW-Authenticate'])
class BadConfiguredSecurityTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_PASSWORD_HASH': 'bcrypt',
'USER_COUNT': 1
}
def test_bad_configuration_raises_runtimer_error(self):
self.assertRaises(RuntimeError, self.authenticate)
class DefaultTemplatePathTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_LOGIN_USER_TEMPLATE': 'custom_security/login_user.html',
}
def test_login_user_template(self):
r = self._get('/login')
self.assertIn(b'CUSTOM LOGIN USER', r.data)
class RegisterableTemplatePathTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_REGISTERABLE': True,
'SECURITY_REGISTER_USER_TEMPLATE': 'custom_security/register_user.html'
}
def test_register_user_template(self):
r = self._get('/register')
self.assertIn(b'CUSTOM REGISTER USER', r.data)
class RecoverableTemplatePathTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_RECOVERABLE': True,
'SECURITY_FORGOT_PASSWORD_TEMPLATE': 'custom_security/forgot_password.html',
'SECURITY_RESET_PASSWORD_TEMPLATE': 'custom_security/reset_password.html',
}
def test_forgot_password_template(self):
r = self._get('/reset')
self.assertIn(b'CUSTOM FORGOT PASSWORD', r.data)
def test_reset_password_template(self):
with capture_reset_password_requests() as requests:
r = self._post('/reset', data=dict(email='joe@lp.com'),
follow_redirects=True)
t = requests[0]['token']
r = self._get('/reset/' + t)
self.assertIn(b'CUSTOM RESET PASSWORD', r.data)
class ConfirmableTemplatePathTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_CONFIRMABLE': True,
'SECURITY_SEND_CONFIRMATION_TEMPLATE': 'custom_security/send_confirmation.html'
}
def test_send_confirmation_template(self):
r = self._get('/confirm')
self.assertIn(b'CUSTOM SEND CONFIRMATION', r.data)
class PasswordlessTemplatePathTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_PASSWORDLESS': True,
'SECURITY_SEND_LOGIN_TEMPLATE': 'custom_security/send_login.html'
}
def test_send_login_template(self):
r = self._get('/login')
self.assertIn(b'CUSTOM SEND LOGIN', r.data)
class RegisterableTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_REGISTERABLE': True,
'USER_COUNT': 1
}
def test_register_valid_user(self):
data = dict(email='dude@lp.com',
password='password',
password_confirm='password')
self._post('/register', data=data, follow_redirects=True)
r = self.authenticate('dude@lp.com')
self.assertIn(b'Hello dude@lp.com', r.data)
class ConfirmableTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_CONFIRMABLE': True,
'SECURITY_REGISTERABLE': True,
'SECURITY_EMAIL_SUBJECT_REGISTER': 'Custom welcome subject',
'USER_COUNT': 1
}
def test_login_before_confirmation(self):
e = 'dude@lp.com'
self.register(e)
r = self.authenticate(email=e)
self.assertIn(self.get_message('CONFIRMATION_REQUIRED').encode('utf-8'), r.data)
def test_send_confirmation_of_already_confirmed_account(self):
e = 'dude@lp.com'
with capture_registrations() as registrations:
r = self.register(e)
token = registrations[0]['confirm_token']
self.client.get('/confirm/' + token, follow_redirects=True)
self.logout()
r = self._post('/confirm', data=dict(email=e))
m = self.get_message('ALREADY_CONFIRMED')
self.assertIn(m.encode('utf-8'), r.data)
def test_register_sends_confirmation_email(self):
e = 'dude@lp.com'
with self.app.extensions['mail'].record_messages() as outbox:
self.register(e)
self.assertEqual(len(outbox), 1)
self.assertIn(e, outbox[0].html)
self.assertEqual('Custom welcome subject', outbox[0].subject)
def test_confirm_email(self):
e = 'dude@lp.com'
tokens = []
def on_registered(sender, **kwargs):
tokens.append(kwargs['confirm_token'])
user_registered.connect(on_registered, self.app)
r = self.register(e)
self.assertEqual(len(tokens), 1)
r = self.client.get('/confirm/' + tokens[0], follow_redirects=True)
msg = self.app.config['SECURITY_MSG_EMAIL_CONFIRMED'][0]
self.assertIn(msg.encode('utf-8'), r.data)
def test_invalid_token_when_confirming_email(self):
r = self.client.get('/confirm/bogus', follow_redirects=True)
msg = self.app.config['SECURITY_MSG_INVALID_CONFIRMATION_TOKEN'][0]
self.assertIn(msg.encode('utf-8'), r.data)
def test_send_confirmation_json(self):
r = self._post('/confirm', data='{"email": "matt@lp.com"}',
content_type='application/json')
self.assertEquals(r.status_code, 200)
def test_send_confirmation_with_invalid_email(self):
r = self._post('/confirm', data=dict(email='bogus@bogus.com'))
msg = self.app.config['SECURITY_MSG_USER_DOES_NOT_EXIST'][0]
self.assertIn(msg.encode('utf-8'), r.data)
def test_resend_confirmation(self):
e = 'dude@lp.com'
self.register(e)
r = self._post('/confirm', data={'email': e})
msg = self.get_message('CONFIRMATION_REQUEST', email=e).encode('utf-8')
self.assertIn(msg, r.data)
def test_user_deleted_before_confirmation(self):
e = 'dude@lp.com'
with capture_registrations() as registrations:
self.register(e)
user = registrations[0]['user']
token = registrations[0]['confirm_token']
with self.app.app_context():
from flask_security.core import _security
_security.datastore.delete(user)
_security.datastore.commit()
r = self.client.get('/confirm/' + token, follow_redirects=True)
msg = self.app.config['SECURITY_MSG_INVALID_CONFIRMATION_TOKEN'][0]
self.assertIn(msg.encode('utf-8'), r.data)
class ExpiredConfirmationTest(SecurityTest):
AUTH_CONFIG = {
'SECURITY_CONFIRMABLE': True,
'SECURITY_REGISTERABLE': True,
'SECURITY_CONFIRM_EMAIL_WITHIN': '1 milliseconds',
'USER_COUNT': 1
}
def test_expired_confirmation_token_sends_email(self):
e = 'dude@lp.com'
with capture_registrations() as registrations:
self.register(e)
token = registrations[0]['confirm_token']
time.sleep(1.25)
with self.app.extensions['mail'].record_messages() as outbox:
r = self.client.get('/confirm/' + token, follow_redirects=True)
self.assertEqual(len(outbox), 1)
self.assertNotIn(token, outbox[0].html)
expire_text = self.AUTH_CONFIG['SECURITY_CONFIRM_EMAIL_WITHIN']
msg = self.app.config['SECURITY_MSG_CONFIRMATION_EXPIRED'][0]
msg = msg % dict(within=expire_text, email=e)
self.assertIn(msg.encode('utf-8'), r.data)
class LoginWithoutImmediateConfirmTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_CONFIRMABLE': True,
'SECURITY_REGISTERABLE': True,
'SECURITY_LOGIN_WITHOUT_CONFIRMATION': True,
'USER_COUNT': 1
}
def test_register_valid_user_automatically_signs_in(self):
e = 'dude@lp.com'
p = 'password'
data = dict(email=e, password=p, password_confirm=p)
r = self._post('/register', data=data, follow_redirects=True)
self.assertIn(e.encode('utf-8'), r.data)
def test_confirm_email_of_user_different_than_current_user(self):
e1 = 'dude@lp.com'
e2 = 'lady@lp.com'
with capture_registrations() as registrations:
self.register(e1)
self.register(e2)
token1 = registrations[0]['confirm_token']
token2 = registrations[1]['confirm_token']
self.client.get('/confirm/' + token1, follow_redirects=True)
self.client.get('/logout')
self.authenticate(email=e1)
r = self.client.get('/confirm/' + token2, follow_redirects=True)
m = self.app.config['SECURITY_MSG_EMAIL_CONFIRMED'][0]
self.assertIn(m.encode('utf-8'), r.data)
self.assertIn(b'Hello lady@lp.com', r.data)
def test_login_unconfirmed_user_when_login_without_confirmation_is_true(self):
e = 'dude@lp.com'
p = 'password'
data = dict(email=e, password=p, password_confirm=p)
r = self._post('/register', data=data, follow_redirects=True)
self.assertIn(e.encode('utf-8'), r.data)
self.client.get('/logout')
r = self.authenticate(email=e)
self.assertIn(e.encode('utf-8'), r.data)
class RecoverableTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_RECOVERABLE': True,
'SECURITY_RESET_PASSWORD_ERROR_VIEW': '/',
'SECURITY_POST_FORGOT_VIEW': '/'
}
def test_reset_view(self):
with capture_reset_password_requests() as requests:
r = self._post('/reset', data=dict(email='joe@lp.com'),
follow_redirects=True)
t = requests[0]['token']
r = self._get('/reset/' + t)
self.assertIn(b'<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:
self._post('/reset', data=dict(email='joe@lp.com'))
self.assertEqual(len(outbox), 1)
def test_forgot_password_json(self):
r = self._post('/reset', data='{"email": "matt@lp.com"}',
content_type="application/json")
self.assertEquals(r.status_code, 200)
def test_forgot_password_invalid_email(self):
r = self._post('/reset', data=dict(email='larry@lp.com'),
follow_redirects=True)
self.assertIn(b"Specified user does not exist", r.data)
def test_reset_password_with_valid_token(self):
with capture_reset_password_requests() as requests:
r = self._post('/reset', data=dict(email='joe@lp.com'),
follow_redirects=True)
t = requests[0]['token']
r = self._post('/reset/' + t, data={
'password': 'newpassword',
'password_confirm': 'newpassword'
}, follow_redirects=True)
r = self.logout()
r = self.authenticate('joe@lp.com', 'newpassword')
self.assertIn(b'Hello joe@lp.com', r.data)
def test_reset_password_with_invalid_token(self):
r = self._post('/reset/bogus', data={
'password': 'newpassword',
'password_confirm': 'newpassword'
}, follow_redirects=True)
m = self.get_message('INVALID_RESET_PASSWORD_TOKEN')
self.assertIn(m.encode('utf-8'), r.data)
class ExpiredResetPasswordTest(SecurityTest):
AUTH_CONFIG = {
'SECURITY_RECOVERABLE': True,
'SECURITY_RESET_PASSWORD_WITHIN': '1 milliseconds'
}
def test_reset_password_with_expired_token(self):
with capture_reset_password_requests() as requests:
r = self._post('/reset', data=dict(email='joe@lp.com'),
follow_redirects=True)
t = requests[0]['token']
time.sleep(1)
r = self._post('/reset/' + t, data={
'password': 'newpassword',
'password_confirm': 'newpassword'
}, follow_redirects=True)
self.assertIn(b'You did not reset your password within', r.data)
class ChangePasswordTest(SecurityTest):
AUTH_CONFIG = {
'SECURITY_RECOVERABLE': True,
'SECURITY_CHANGEABLE': True,
}
def test_change_password(self):
self.authenticate()
r = self.client.get('/change', follow_redirects=True)
self.assertIn(b'Change password', r.data)
def test_change_password_invalid(self):
self.authenticate()
r = self._post('/change', data={
'password': 'notpassword',
'new_password': 'newpassword',
'new_password_confirm': 'newpassword'
}, follow_redirects=True)
self.assertNotIn(b'You successfully changed your password', r.data)
self.assertIn(b'Invalid password', r.data)
def test_change_password_mismatch(self):
self.authenticate()
r = self._post('/change', data={
'password': 'password',
'new_password': 'newpassword',
'new_password_confirm': 'notnewpassword'
}, follow_redirects=True)
self.assertNotIn(b'You successfully changed your password', r.data)
self.assertIn(b'Passwords do not match', r.data)
def test_change_password_bad_password(self):
self.authenticate()
r = self._post('/change', data={
'password': 'password',
'new_password': 'a',
'new_password_confirm': 'a'
}, follow_redirects=True)
self.assertNotIn(b'You successfully changed your password', r.data)
self.assertIn(b'Password must be at least 6 characters', r.data)
def test_change_password_same_as_previous(self):
self.authenticate()
r = self._post('/change', data={
'password': 'password',
'new_password': 'password',
'new_password_confirm': 'password'
}, follow_redirects=True)
self.assertNotIn(b'You successfully changed your password', r.data)
self.assertIn(b'Your new password must be different than your previous password.', r.data)
def test_change_password_success(self):
data = {
'password': 'password',
'new_password': 'newpassword',
'new_password_confirm': 'newpassword'
}
self.authenticate()
with self.app.extensions['mail'].record_messages() as outbox:
r = self._post('/change', data=data, follow_redirects=True)
self.assertIn(b'You successfully changed your password', r.data)
self.assertIn(b'Home Page', r.data)
self.assertEqual(len(outbox), 1)
self.assertIn("Your password has been changed", outbox[0].html)
self.assertIn("/reset", outbox[0].html)
class EmailConfigTest(SecurityTest):
AUTH_CONFIG = {
'SECURITY_SEND_REGISTER_EMAIL': False,
'SECURITY_SEND_PASSWORD_CHANGE_EMAIL': False,
}
def test_change_password_success_email_option(self):
"""Test the change password email can be turned off w/ configuration."""
data = {
'password': 'password',
'new_password': 'newpassword',
'new_password_confirm': 'newpassword'
}
self.authenticate()
with self.app.extensions['mail'].record_messages() as outbox:
self._post('/change', data=data, follow_redirects=True)
self.assertEqual(len(outbox), 0)
class ChangePasswordPostViewTest(SecurityTest):
AUTH_CONFIG = {
'SECURITY_CHANGEABLE': True,
'SECURITY_POST_CHANGE_VIEW': '/profile',
}
def test_change_password_success(self):
data = {
'password': 'password',
'new_password': 'newpassword',
'new_password_confirm': 'newpassword'
}
self.authenticate()
r = self._post('/change', data=data, follow_redirects=True)
self.assertIn(b'Profile Page', r.data)
class ChangePasswordDisabledTest(SecurityTest):
AUTH_CONFIG = {
'SECURITY_CHANGEABLE': False,
}
def test_change_password_endpoint_is_404(self):
self.authenticate()
r = self.client.get('/change', follow_redirects=True)
self.assertEqual(404, r.status_code)
class TrackableTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_TRACKABLE': True,
'USER_COUNT': 1
}
def test_did_track(self):
e = 'matt@lp.com'
self.authenticate(email=e)
self.logout()
self.authenticate(email=e)
with self.app.test_request_context('/profile'):
user = self.app.security.datastore.find_user(email=e)
self.assertIsNotNone(user.last_login_at)
self.assertIsNotNone(user.current_login_at)
self.assertEquals('untrackable', user.last_login_ip)
self.assertEquals('untrackable', user.current_login_ip)
self.assertEquals(2, user.login_count)
class PasswordlessTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_PASSWORDLESS': True
}
def test_login_request_for_inactive_user(self):
msg = self.app.config['SECURITY_MSG_DISABLED_ACCOUNT'][0]
r = self._post('/login', data=dict(email='tiya@lp.com'),
follow_redirects=True)
self.assertIn(msg.encode('utf-8'), r.data)
def test_request_login_token_with_json_and_valid_email(self):
data = '{"email": "matt@lp.com", "password": "password"}'
r = self._post('/login', data=data, content_type='application/json')
self.assertEquals(r.status_code, 200)
self.assertNotIn(b'error', r.data)
def test_request_login_token_with_json_and_invalid_email(self):
data = '{"email": "nobody@lp.com", "password": "password"}'
r = self._post('/login', data=data, content_type='application/json')
self.assertIn(b'errors', r.data)
def test_request_login_token_sends_email_and_can_login(self):
e = 'matt@lp.com'
r, user, token = None, None, None
with capture_passwordless_login_requests() as requests:
with self.app.extensions['mail'].record_messages() as outbox:
r = self._post('/login', data=dict(email=e),
follow_redirects=True)
self.assertEqual(len(outbox), 1)
self.assertEquals(1, len(requests))
self.assertIn('user', requests[0])
self.assertIn('login_token', requests[0])
user = requests[0]['user']
token = requests[0]['login_token']
msg = self.app.config['SECURITY_MSG_LOGIN_EMAIL_SENT'][0]
msg = msg % dict(email=user.email)
self.assertIn(msg.encode('utf-8'), r.data)
r = self.client.get('/login/' + token, follow_redirects=True)
msg = self.get_message('PASSWORDLESS_LOGIN_SUCCESSFUL').encode('utf-8')
self.assertIn(msg, r.data)
r = self.client.get('/profile')
self.assertIn(b'Profile Page', r.data)
def test_invalid_login_token(self):
m = self.app.config['SECURITY_MSG_INVALID_LOGIN_TOKEN'][0]
r = self._get('/login/bogus', follow_redirects=True)
self.assertIn(m.encode('utf-8'), r.data)
def test_token_login_when_already_authenticated(self):
with capture_passwordless_login_requests() as requests:
self._post('/login', data=dict(email='matt@lp.com'),
follow_redirects=True)
token = requests[0]['login_token']
r = self.client.get('/login/' + token, follow_redirects=True)
msg = self.get_message('PASSWORDLESS_LOGIN_SUCCESSFUL')
self.assertIn(msg.encode('utf-8'), r.data)
r = self.client.get('/login/' + token, follow_redirects=True)
msg = self.get_message('PASSWORDLESS_LOGIN_SUCCESSFUL')
self.assertNotIn(msg.encode('utf-8'), r.data)
def test_send_login_with_invalid_email(self):
r = self._post('/login', data=dict(email='bogus@bogus.com'))
self.assertIn(b'Specified user does not exist', r.data)
class ExpiredLoginTokenTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_PASSWORDLESS': True,
'SECURITY_LOGIN_WITHIN': '1 milliseconds',
'USER_COUNT': 1
}
def test_expired_login_token_sends_email(self):
e = 'matt@lp.com'
with capture_passwordless_login_requests() as requests:
self._post('/login', data=dict(email=e), follow_redirects=True)
token = requests[0]['login_token']
time.sleep(1.25)
with self.app.extensions['mail'].record_messages() as outbox:
r = self.client.get('/login/' + token, follow_redirects=True)
expire_text = self.AUTH_CONFIG['SECURITY_LOGIN_WITHIN']
msg = self.app.config['SECURITY_MSG_LOGIN_EXPIRED'][0]
msg = msg % dict(within=expire_text, email=e)
self.assertIn(msg.encode('utf-8'), r.data)
self.assertEqual(len(outbox), 1)
self.assertIn(e, outbox[0].html)
self.assertNotIn(token, outbox[0].html)
class AsyncMailTaskTests(SecurityTest):
AUTH_CONFIG = {
'SECURITY_RECOVERABLE': True,
'USER_COUNT': 1
}
def setUp(self):
super(AsyncMailTaskTests, self).setUp()
self.mail_sent = False
def test_send_email_task_is_called(self):
@self.app.security.send_mail_task
def send_email(msg):
self.mail_sent = True
self._post('/reset', data=dict(email='matt@lp.com'))
self.assertTrue(self.mail_sent)
class NoBlueprintTests(SecurityTest):
APP_KWARGS = {
'register_blueprint': False,
}
AUTH_CONFIG = {
'USER_COUNT': 1
}
def test_login_endpoint_is_404(self):
r = self._get('/login')
self.assertEqual(404, r.status_code)
def test_http_auth_without_blueprint(self):
auth = base64.b64encode(b"matt@lp.com:password").decode('utf-8')
r = self._get('/http', headers={'Authorization': 'basic %s' % auth})
self.assertIn(b'HTTP Authentication', r.data)
class ExtendFormsTest(SecurityTest):
class MyLoginForm(LoginForm):
email = TextField('My Login Email Address Field')
class MyRegisterForm(RegisterForm):
email = TextField('My Register Email Address Field')
APP_KWARGS = {
'login_form': MyLoginForm,
'register_form': MyRegisterForm,
}
AUTH_CONFIG = {
'SECURITY_CONFIRMABLE': False,
'SECURITY_REGISTERABLE': True,
}
def test_login_view(self):
r = self._get('/login', follow_redirects=True)
self.assertIn(b"My Login Email Address Field", r.data)
def test_register(self):
r = self._get('/register', follow_redirects=True)
self.assertIn(b"My Register Email Address Field", r.data)
class RecoverableExtendFormsTest(SecurityTest):
class MyForgotPasswordForm(ForgotPasswordForm):
email = TextField('My Forgot Password Email Address Field',
validators=[valid_user_email])
class MyResetPasswordForm(ResetPasswordForm):
submit = SubmitField("My Reset Password Submit Field")
APP_KWARGS = {
'forgot_password_form': MyForgotPasswordForm,
'reset_password_form': MyResetPasswordForm,
}
AUTH_CONFIG = {
'SECURITY_RECOVERABLE': True,
}
def test_forgot_password(self):
r = self._get('/reset', follow_redirects=True)
self.assertIn(b"My Forgot Password Email Address Field", r.data)
def test_reset_password(self):
with capture_reset_password_requests() as requests:
self._post('/reset', data=dict(email='joe@lp.com'),
follow_redirects=True)
token = requests[0]['token']
r = self._get('/reset/' + token)
self.assertIn(b"My Reset Password Submit Field", r.data)
class PasswordlessExtendFormsTest(SecurityTest):
class MyPasswordlessLoginForm(PasswordlessLoginForm):
email = TextField('My Passwordless Login Email Address Field')
APP_KWARGS = {
'passwordless_login_form': MyPasswordlessLoginForm,
}
AUTH_CONFIG = {
'SECURITY_PASSWORDLESS': True,
}
def test_passwordless_login(self):
r = self._get('/login', follow_redirects=True)
self.assertIn(b"My Passwordless Login Email Address Field", r.data)
class ConfirmableExtendFormsTest(SecurityTest):
class MyConfirmRegisterForm(ConfirmRegisterForm):
email = TextField('My Confirm Register Email Address Field')
class MySendConfirmationForm(SendConfirmationForm):
email = TextField('My Send Confirmation Email Address Field')
APP_KWARGS = {
'confirm_register_form': MyConfirmRegisterForm,
'send_confirmation_form': MySendConfirmationForm,
}
AUTH_CONFIG = {
'SECURITY_CONFIRMABLE': True,
'SECURITY_REGISTERABLE': True,
}
def test_register(self):
r = self._get('/register', follow_redirects=True)
self.assertIn(b"My Confirm Register Email Address Field", r.data)
def test_send_confirmation(self):
r = self._get('/confirm', follow_redirects=True)
self.assertIn(b"My Send Confirmation Email Address Field", r.data)
class AdditionalUserIdentityAttributes(SecurityTest):
AUTH_CONFIG = {
'SECURITY_USER_IDENTITY_ATTRIBUTES': ('email', 'username')
}
def test_authenticate(self):
r = self.authenticate(email='matt')
self.assertIn(b'Hello matt@lp.com', r.data)
+287
View File
@@ -0,0 +1,287 @@
# -*- coding: utf-8 -*-
"""
conftest
~~~~~~~~
Test fixtures and what not
"""
import os
import tempfile
import time
import pytest
from flask import Flask, render_template
from flask_mail import Mail
from flask_security import Security, MongoEngineUserDatastore, SQLAlchemyUserDatastore, \
PeeweeUserDatastore, UserMixin, RoleMixin, http_auth_required, login_required, \
auth_token_required, auth_required, roles_required, roles_accepted
from utils import populate_data, Response
@pytest.fixture()
def app(request):
app = Flask(__name__)
app.response_class = Response
app.debug = True
app.config['SECRET_KEY'] = 'secret'
app.config['TESTING'] = True
app.config['LOGIN_DISABLED'] = False
app.config['WTF_CSRF_ENABLED'] = False
for opt in ['changeable', 'recoverable', 'registerable',
'trackable', 'passwordless', 'confirmable']:
app.config['SECURITY_' + opt.upper()] = opt in request.keywords
if 'settings' in request.keywords:
for key, value in request.keywords['settings'].kwargs.items():
app.config['SECURITY_' + key.upper()] = value
mail = Mail(app)
app.mail = mail
@app.route('/')
def index():
return render_template('index.html', content='Home Page')
@app.route('/profile')
@login_required
def profile():
return render_template('index.html', content='Profile Page')
@app.route('/post_login')
@login_required
def post_login():
return render_template('index.html', content='Post Login')
@app.route('/http')
@http_auth_required
def http():
return 'HTTP Authentication'
@app.route('/http_custom_realm')
@http_auth_required('My Realm')
def http_custom_realm():
return render_template('index.html', content='HTTP Authentication')
@app.route('/token', methods=['GET', 'POST'])
@auth_token_required
def token():
return render_template('index.html', content='Token Authentication')
@app.route('/multi_auth')
@auth_required('session', 'token', 'basic')
def multi_auth():
return render_template('index.html', content='Session, Token, Basic auth')
@app.route('/post_logout')
def post_logout():
return render_template('index.html', content='Post Logout')
@app.route('/post_register')
def post_register():
return render_template('index.html', content='Post Register')
@app.route('/admin')
@roles_required('admin')
def admin():
return render_template('index.html', content='Admin Page')
@app.route('/admin_and_editor')
@roles_required('admin', 'editor')
def admin_and_editor():
return render_template('index.html', content='Admin and Editor Page')
@app.route('/admin_or_editor')
@roles_accepted('admin', 'editor')
def admin_or_editor():
return render_template('index.html', content='Admin or Editor Page')
@app.route('/unauthorized')
def unauthorized():
return render_template('unauthorized.html')
@app.route('/page1')
def page_1():
return 'Page 1'
return app
@pytest.fixture()
def mongoengine_datastore(request, app):
from flask_mongoengine import MongoEngine
db_name = 'flask_security_test_%s' % str(time.time()).replace('.', '_')
app.config['MONGODB_SETTINGS'] = {
'db': db_name,
'host': 'localhost',
'port': 27017,
'alias': db_name
}
db = MongoEngine(app)
class Role(db.Document, RoleMixin):
name = db.StringField(required=True, unique=True, max_length=80)
description = db.StringField(max_length=255)
meta = {"db_alias": db_name}
class User(db.Document, UserMixin):
email = db.StringField(unique=True, max_length=255)
username = db.StringField(max_length=255)
password = db.StringField(required=False, max_length=255)
last_login_at = db.DateTimeField()
current_login_at = db.DateTimeField()
last_login_ip = db.StringField(max_length=100)
current_login_ip = db.StringField(max_length=100)
login_count = db.IntField()
active = db.BooleanField(default=True)
confirmed_at = db.DateTimeField()
roles = db.ListField(db.ReferenceField(Role), default=[])
meta = {"db_alias": db_name}
request.addfinalizer(lambda: db.connection.drop_database(db_name))
return MongoEngineUserDatastore(db, User, Role)
@pytest.fixture()
def sqlalchemy_datastore(request, app, tmpdir):
from flask_sqlalchemy import SQLAlchemy
f, path = tempfile.mkstemp(prefix='flask-security-test-db', suffix='.db', dir=str(tmpdir))
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + path
db = SQLAlchemy(app)
roles_users = db.Table(
'roles_users',
db.Column('user_id', db.Integer(), db.ForeignKey('user.id')),
db.Column('role_id', db.Integer(), db.ForeignKey('role.id')))
class Role(db.Model, RoleMixin):
id = db.Column(db.Integer(), primary_key=True)
name = db.Column(db.String(80), unique=True)
description = db.Column(db.String(255))
class User(db.Model, UserMixin):
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(255), unique=True)
username = db.Column(db.String(255))
password = db.Column(db.String(255))
last_login_at = db.Column(db.DateTime())
current_login_at = db.Column(db.DateTime())
last_login_ip = db.Column(db.String(100))
current_login_ip = db.Column(db.String(100))
login_count = db.Column(db.Integer)
active = db.Column(db.Boolean())
confirmed_at = db.Column(db.DateTime())
roles = db.relationship('Role', secondary=roles_users,
backref=db.backref('users', lazy='dynamic'))
with app.app_context():
db.create_all()
request.addfinalizer(lambda: os.remove(path))
return SQLAlchemyUserDatastore(db, User, Role)
@pytest.fixture()
def peewee_datastore(request, app, tmpdir):
from peewee import TextField, DateTimeField, IntegerField, BooleanField, ForeignKeyField
from flask_peewee.db import Database
f, path = tempfile.mkstemp(prefix='flask-security-test-db', suffix='.db', dir=str(tmpdir))
app.config['DATABASE'] = {
'name': path,
'engine': 'peewee.SqliteDatabase'
}
db = Database(app)
class Role(db.Model, RoleMixin):
name = TextField(unique=True)
description = TextField(null=True)
class User(db.Model, UserMixin):
email = TextField()
username = TextField()
password = TextField(null=True)
last_login_at = DateTimeField(null=True)
current_login_at = DateTimeField(null=True)
last_login_ip = TextField(null=True)
current_login_ip = TextField(null=True)
login_count = IntegerField(null=True)
active = BooleanField(default=True)
confirmed_at = DateTimeField(null=True)
class UserRoles(db.Model):
""" Peewee does not have built-in many-to-many support, so we have to
create this mapping class to link users to roles."""
user = ForeignKeyField(User, related_name='roles')
role = ForeignKeyField(Role, related_name='users')
name = property(lambda self: self.role.name)
description = property(lambda self: self.role.description)
with app.app_context():
for Model in (Role, User, UserRoles):
Model.create_table()
request.addfinalizer(lambda: os.remove(path))
return PeeweeUserDatastore(db, User, Role, UserRoles)
@pytest.fixture()
def sqlalchemy_app(app, sqlalchemy_datastore):
def create():
app.security = Security(app, datastore=sqlalchemy_datastore)
return app
return create
@pytest.fixture()
def peewee_app(app, peewee_datastore):
def create():
app.security = Security(app, datastore=peewee_datastore)
return app
return create
@pytest.fixture()
def mongoengine_app(app, mongoengine_datastore):
def create():
app.security = Security(app, datastore=mongoengine_datastore)
return app
return create
@pytest.fixture()
def client(request, sqlalchemy_app):
app = sqlalchemy_app()
populate_data(app)
return app.test_client()
@pytest.fixture()
def get_message(app):
def fn(key, **kwargs):
rv = app.config['SECURITY_MSG_' + key][0] % kwargs
return rv.encode('utf-8')
return fn
@pytest.fixture(params=['sqlalchemy', 'mongoengine', 'peewee'])
def datastore(request, sqlalchemy_datastore, mongoengine_datastore, peewee_datastore):
if request.param == 'sqlalchemy':
rv = sqlalchemy_datastore
elif request.param == 'mongoengine':
rv = mongoengine_datastore
elif request.param == 'peewee':
rv = peewee_datastore
return rv
-277
View File
@@ -1,277 +0,0 @@
# -*- coding: utf-8 -*-
from __future__ import with_statement
import base64
import simplejson as json
try:
from cookielib import Cookie
except ImportError:
from http.cookiejar import Cookie
from werkzeug.utils import parse_cookie
from tests import SecurityTest
def get_cookies(rv):
cookies = {}
for value in rv.headers.get_all("Set-Cookie"):
cookies.update(parse_cookie(value))
return cookies
class DefaultSecurityTests(SecurityTest):
def test_instance(self):
self.assertIsNotNone(self.app)
self.assertIsNotNone(self.app.security)
self.assertIsNotNone(self.app.security.pwd_context)
def test_login_view(self):
r = self._get('/login')
self.assertIn(b'<h1>Login</h1>', r.data)
def test_authenticate(self):
r = self.authenticate()
self.assertIn(b'Hello matt@lp.com', r.data)
def test_authenticate_case_insensitive_email(self):
r = self.authenticate(email='MATT@lp.com')
self.assertIn(b'Hello matt@lp.com', r.data)
def test_unprovided_username(self):
r = self.authenticate("")
self.assertIn(self.get_message('EMAIL_NOT_PROVIDED').encode('utf-8'), r.data)
def test_unprovided_password(self):
r = self.authenticate(password="")
self.assertIn(self.get_message('PASSWORD_NOT_PROVIDED').encode('utf-8'), r.data)
def test_invalid_user(self):
r = self.authenticate(email="bogus@bogus.com")
self.assertIn(self.get_message('USER_DOES_NOT_EXIST').encode('utf-8'), r.data)
def test_bad_password(self):
r = self.authenticate(password="bogus")
self.assertIn(self.get_message('INVALID_PASSWORD').encode('utf-8'), r.data)
def test_inactive_user(self):
r = self.authenticate("tiya@lp.com", "password")
self.assertIn(self.get_message('DISABLED_ACCOUNT').encode('utf-8'), r.data)
def test_logout(self):
self.authenticate()
r = self.logout()
self.assertIsHomePage(r.data)
def test_unauthorized_access(self):
self.logout()
r = self._get('/profile', follow_redirects=True)
self.assertIn(b'<li class="info">Please log in to access this page.</li>', r.data)
def test_authorized_access(self):
self.authenticate()
r = self._get("/profile")
self.assertIn(b'profile', r.data)
def test_valid_admin_role(self):
self.authenticate()
r = self._get("/admin")
self.assertIn(b'Admin Page', r.data)
def test_invalid_admin_role(self):
self.authenticate("joe@lp.com")
r = self._get("/admin", follow_redirects=True)
self.assertIsHomePage(r.data)
def test_roles_accepted(self):
for user in ("matt@lp.com", "joe@lp.com"):
self.authenticate(user)
r = self._get("/admin_or_editor")
self.assertIn(b'Admin or Editor Page', r.data)
self.logout()
self.authenticate("jill@lp.com")
r = self._get("/admin_or_editor", follow_redirects=True)
self.assertIsHomePage(r.data)
def test_unauthenticated_role_required(self):
r = self._get('/admin', follow_redirects=True)
self.assertIn(self.get_message('UNAUTHORIZED').encode('utf-8'), r.data)
def test_multiple_role_required(self):
for user in ("matt@lp.com", "joe@lp.com"):
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", follow_redirects=True)
self.assertIn(b'Admin and Editor Page', r.data)
def test_ok_json_auth(self):
r = self.json_authenticate()
data = json.loads(r.data)
self.assertEquals(data['meta']['code'], 200)
self.assertIn('authentication_token', data['response']['user'])
def test_invalid_json_auth(self):
r = self.json_authenticate(password='junk')
self.assertIn(b'"code": 400', r.data)
def test_token_auth_via_querystring_valid_token(self):
r = self.json_authenticate()
data = json.loads(r.data)
token = data['response']['user']['authentication_token']
r = self._get('/token?auth_token=' + token)
self.assertIn(b'Token Authentication', r.data)
def test_token_auth_via_header_valid_token(self):
r = self.json_authenticate()
data = json.loads(r.data)
token = data['response']['user']['authentication_token']
headers = {"Authentication-Token": token}
r = self._get('/token', headers=headers)
self.assertIn(b'Token Authentication', r.data)
def test_token_auth_via_querystring_invalid_token(self):
r = self._get('/token?auth_token=X')
self.assertEqual(401, r.status_code)
def test_token_auth_via_header_invalid_token(self):
r = self._get('/token', headers={"Authentication-Token": 'X'})
self.assertEqual(401, r.status_code)
def test_http_auth(self):
r = self._get('/http', headers={
'Authorization': 'Basic %s' % base64.b64encode(b"joe@lp.com:password").decode('utf-8')
})
self.assertIn(b'HTTP Authentication', r.data)
def test_http_auth_no_authorization(self):
r = self._get('/http', headers={})
self.assertIn(b'<h1>Unauthorized</h1>', r.data)
self.assertIn('WWW-Authenticate', r.headers)
self.assertEquals('Basic realm="Login Required"',
r.headers['WWW-Authenticate'])
def test_invalid_http_auth_invalid_username(self):
r = self._get('/http', headers={
'Authorization': 'Basic %s' % base64.b64encode(b"bogus:bogus").decode('utf-8')
})
self.assertIn(b'<h1>Unauthorized</h1>', r.data)
self.assertIn('WWW-Authenticate', r.headers)
self.assertEquals('Basic realm="Login Required"',
r.headers['WWW-Authenticate'])
def test_invalid_http_auth_bad_password(self):
r = self._get('/http', headers={
'Authorization': 'Basic %s' % base64.b64encode(b"joe@lp.com:bogus").decode('utf-8')
})
self.assertIn(b'<h1>Unauthorized</h1>', r.data)
self.assertIn('WWW-Authenticate', r.headers)
self.assertEquals('Basic realm="Login Required"',
r.headers['WWW-Authenticate'])
def test_custom_http_auth_realm(self):
r = self._get('/http_custom_realm', headers={
'Authorization': 'Basic %s' % base64.b64encode(b"joe@lp.com:bogus").decode('utf-8')
})
self.assertIn(b'<h1>Unauthorized</h1>', r.data)
self.assertIn('WWW-Authenticate', r.headers)
self.assertEquals('Basic realm="My Realm"',
r.headers['WWW-Authenticate'])
def test_multi_auth_basic(self):
r = self._get('/multi_auth', headers={
'Authorization': 'Basic %s' % base64.b64encode(b"joe@lp.com:password").decode('utf-8')
})
self.assertIn(b'Basic', r.data)
def test_multi_auth_token(self):
r = self.json_authenticate()
data = json.loads(r.data)
token = data['response']['user']['authentication_token']
r = self._get('/multi_auth?auth_token=' + token)
self.assertIn(b'Token', r.data)
def test_multi_auth_session(self):
self.authenticate()
r = self._get('/multi_auth')
self.assertIn(b'Session', r.data)
def test_user_deleted_during_session_reverts_to_anonymous_user(self):
self.authenticate()
with self.app.test_request_context('/'):
user = self.app.security.datastore.find_user(email='matt@lp.com')
self.app.security.datastore.delete_user(user)
self.app.security.datastore.commit()
r = self._get('/')
self.assertNotIn(b'Hello matt@lp.com', r.data)
def test_remember_token(self):
r = self.authenticate(follow_redirects=False)
self.client.cookie_jar.clear_session_cookies()
r = self._get('/profile')
self.assertIn(b'profile', r.data)
def test_token_loader_does_not_fail_with_invalid_token(self):
c = Cookie(version=0, name='remember_token', value='None', port=None,
port_specified=False, domain='www.example.com',
domain_specified=False, domain_initial_dot=False, path='/',
path_specified=True, secure=False, expires=None,
discard=True, comment=None, comment_url=None,
rest={'HttpOnly': None}, rfc2109=False)
self.client.cookie_jar.set_cookie(c)
r = self._get('/')
self.assertNotIn(b'BadSignature', r.data)
class MongoEngineSecurityTests(DefaultSecurityTests):
def _create_app(self, auth_config, **kwargs):
from tests.test_app.mongoengine import create_app
return create_app(auth_config, **kwargs)
class PeeweeSecurityTests(DefaultSecurityTests):
def _create_app(self, auth_config, **kwargs):
from tests.test_app.peewee_app import create_app
return create_app(auth_config, **kwargs)
class DefaultDatastoreTests(SecurityTest):
def test_add_role_to_user(self):
r = self._get('/coverage/add_role_to_user')
self.assertIn(b'success', r.data)
def test_remove_role_from_user(self):
r = self._get('/coverage/remove_role_from_user')
self.assertIn(b'success', r.data)
def test_activate_user(self):
r = self._get('/coverage/activate_user')
self.assertIn(b'success', r.data)
def test_deactivate_user(self):
r = self._get('/coverage/deactivate_user')
self.assertIn(b'success', r.data)
def test_invalid_role(self):
r = self._get('/coverage/invalid_role')
self.assertIn(b'success', r.data)
class MongoEngineDatastoreTests(DefaultDatastoreTests):
def _create_app(self, auth_config, **kwargs):
from tests.test_app.mongoengine import create_app
return create_app(auth_config, **kwargs)
-244
View File
@@ -1,244 +0,0 @@
# -*- coding: utf-8 -*-
from __future__ import with_statement
from flask_security.utils import capture_registrations, \
capture_reset_password_requests, capture_signals
from flask_security.signals import user_registered, user_confirmed, \
confirm_instructions_sent, login_instructions_sent, \
password_reset, password_changed, reset_password_instructions_sent
from tests import SecurityTest
def compare_user(a, b):
"""Helper to compare two users."""
return a.id == b.id and a.email == b.email and a.password == b.password
class SignalTest(SecurityTest):
def _create_app(self, auth_config, **kwargs):
from tests.test_app.mongoengine import create_app
return create_app(auth_config, **kwargs)
class RegisterableSignalsTests(SignalTest):
AUTH_CONFIG = {
'SECURITY_CONFIRMABLE': True,
'SECURITY_REGISTERABLE': True,
}
def test_register(self):
e = 'dude@lp.com'
with capture_signals() as mocks:
self.register(e)
user = self.app.security.datastore.find_user(email='dude@lp.com')
self.assertEqual(mocks.signals_sent(), set([user_registered]))
calls = mocks[user_registered]
self.assertEqual(len(calls), 1)
args, kwargs = calls[0]
self.assertTrue(compare_user(kwargs['user'], user))
self.assertIn('confirm_token', kwargs)
self.assertEqual(args[0], self.app)
def test_register_without_password(self):
e = 'dude@lp.com'
with capture_signals() as mocks:
self.register(e, password='')
self.assertEqual(mocks.signals_sent(), set())
class ConfirmableSignalsTests(SignalTest):
AUTH_CONFIG = {
'SECURITY_CONFIRMABLE': True,
'SECURITY_REGISTERABLE': True,
}
def test_confirm(self):
e = 'dude@lp.com'
with capture_registrations() as registrations:
self.register(e)
token = registrations[0]['confirm_token']
with capture_signals() as mocks:
self.client.get('/confirm/' + token, follow_redirects=True)
user = self.app.security.datastore.find_user(email='dude@lp.com')
self.assertTrue(mocks.signals_sent(), set([user_confirmed]))
calls = mocks[user_confirmed]
self.assertEqual(len(calls), 1)
args, kwargs = calls[0]
self.assertEqual(args[0], self.app)
self.assertTrue(compare_user(kwargs['user'], user))
def test_confirm_bad_token(self):
e = 'dude@lp.com'
with capture_registrations():
self.register(e)
with capture_signals() as mocks:
self.client.get('/confirm/bogus', follow_redirects=True)
self.assertEqual(mocks.signals_sent(), set())
def test_confirm_twice(self):
e = 'dude@lp.com'
with capture_registrations() as registrations:
self.register(e)
token = registrations[0]['confirm_token']
self.client.get('/confirm/' + token, follow_redirects=True)
self.logout()
with capture_signals() as mocks:
self.client.get('/confirm/' + token, follow_redirects=True)
self.assertEqual(mocks.signals_sent(), set([user_confirmed]))
# TODO: is that the desired behaviour?
def test_resend_confirmation(self):
e = 'dude@lp.com'
self.register(e)
with capture_signals() as mocks:
self._post('/confirm', data={'email': e})
user = self.app.security.datastore.find_user(email='dude@lp.com')
self.assertEqual(mocks.signals_sent(), set([confirm_instructions_sent]))
calls = mocks[confirm_instructions_sent]
self.assertEqual(len(calls), 1)
args, kwargs = calls[0]
self.assertTrue(compare_user(kwargs['user'], user))
self.assertEqual(args[0], self.app)
def test_send_confirmation_bad_email(self):
with capture_signals() as mocks:
self._post('/confirm', data=dict(email='bogus@bogus.com'))
self.assertEqual(mocks.signals_sent(), set())
class RecoverableSignalsTests(SignalTest):
AUTH_CONFIG = {
'SECURITY_RECOVERABLE': True,
'SECURITY_RESET_PASSWORD_ERROR_VIEW': '/',
'SECURITY_POST_FORGOT_VIEW': '/'
}
def test_reset_password_request(self):
with capture_signals() as mocks:
self._post('/reset', data=dict(email='joe@lp.com'),
follow_redirects=True)
self.assertEqual(mocks.signals_sent(), set([reset_password_instructions_sent]))
user = self.app.security.datastore.find_user(email='joe@lp.com')
calls = mocks[reset_password_instructions_sent]
self.assertEqual(len(calls), 1)
args, kwargs = calls[0]
self.assertTrue(compare_user(kwargs['user'], user))
self.assertIn('token', kwargs)
self.assertEqual(args[0], self.app)
def test_reset_password(self):
with capture_reset_password_requests() as requests:
self._post('/reset', data=dict(email='joe@lp.com'),
follow_redirects=True)
token = requests[0]['token']
with capture_signals() as mocks:
data = dict(password='newpassword', password_confirm='newpassword')
self._post('/reset/' + token, data, follow_redirects=True)
self.assertEqual(mocks.signals_sent(), set([password_reset]))
user = self.app.security.datastore.find_user(email='joe@lp.com')
calls = mocks[password_reset]
self.assertEqual(len(calls), 1)
args, kwargs = calls[0]
self.assertTrue(compare_user(kwargs['user'], user))
self.assertEqual(args[0], self.app)
def test_reset_password_invalid_emails(self):
with capture_signals() as mocks:
self._post('/reset', data=dict(email='nobody@lp.com'),
follow_redirects=True)
self.assertEqual(mocks.signals_sent(), set())
def test_reset_password_invalid_token(self):
with capture_signals() as mocks:
data = dict(password='newpassword', password_confirm='newpassword')
self._post('/reset/bogus', data, follow_redirects=True)
self.assertEqual(mocks.signals_sent(), set())
class ChangeableSignalsTests(SignalTest):
AUTH_CONFIG = {
'SECURITY_CHANGEABLE': True,
}
def test_change_password(self):
self.authenticate('joe@lp.com')
with capture_signals() as mocks:
with self.client as client:
client.post('/change',
data=dict(password='password',
new_password='newpassword',
new_password_confirm='newpassword'))
self.assertEqual(mocks.signals_sent(), set([password_changed]))
user = self.app.security.datastore.find_user(email='joe@lp.com')
calls = mocks[password_changed]
self.assertEqual(len(calls), 1)
args, kwargs = calls[0]
self.assertTrue(compare_user(args[0], user))
self.assertEqual(kwargs['app'], self.app)
def test_change_password_invalid_password(self):
with capture_signals() as mocks:
self.client.post('/change',
data=dict(password='notpassword',
new_password='newpassword',
new_password_confirm='newpassword'),
follow_redirects=True)
self.assertEqual(mocks.signals_sent(), set())
def test_change_password_bad_password(self):
with capture_signals() as mocks:
self.client.post('/change',
data=dict(password='notpassword',
new_password='a',
new_password_confirm='a'),
follow_redirects=True)
self.assertEqual(mocks.signals_sent(), set())
def test_change_password_mismatch_password(self):
with capture_signals() as mocks:
self.client.post('/change',
data=dict(password='password',
new_password='newpassword',
new_password_confirm='notnewpassword'),
follow_redirects=True)
self.assertEqual(mocks.signals_sent(), set())
class PasswordlessTests(SignalTest):
AUTH_CONFIG = {
'SECURITY_PASSWORDLESS': True
}
def test_login_request_for_inactive_user(self):
with capture_signals() as mocks:
self._post('/login', data=dict(email='tiya@lp.com'),
follow_redirects=True)
self.assertEqual(mocks.signals_sent(), set())
def test_login_request_for_invalid_email(self):
with capture_signals() as mocks:
self._post('/login', data=dict(email='nobody@lp.com'),
follow_redirects=True)
self.assertEqual(mocks.signals_sent(), set())
def test_request_login_token_sends_email_and_can_login(self):
e = 'matt@lp.com'
with capture_signals() as mocks:
self._post('/login', data=dict(email=e), follow_redirects=True)
self.assertEqual(mocks.signals_sent(), set([login_instructions_sent]))
user = self.app.security.datastore.find_user(email='matt@lp.com')
calls = mocks[login_instructions_sent]
self.assertEqual(len(calls), 1)
args, kwargs = calls[0]
self.assertTrue(compare_user(kwargs['user'], user))
self.assertIn('login_token', kwargs)
self.assertEqual(args[0], self.app)
@@ -0,0 +1,3 @@
CUSTOM CHANGE PASSWORD
{{ foo }}
@@ -0,0 +1,3 @@
CUSTOM RESET INSTRUCTIONS
{{ foo }}
-185
View File
@@ -1,185 +0,0 @@
# -*- coding: utf-8 -*-
from flask import Flask, render_template, current_app
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, auth_required
from flask.ext.security.utils import encrypt_password
from werkzeug.local import LocalProxy
ds = LocalProxy(lambda: current_app.extensions['security'].datastore)
def create_app(config):
app = Flask(__name__)
app.debug = True
app.config['SECRET_KEY'] = 'secret'
app.config['TESTING'] = True
app.config['LOGIN_DISABLED'] = False
for key, value in config.items():
app.config[key] = value
mail = Mail(app)
app.extensions['mail'] = mail
@app.route('/')
def index():
return render_template('index.html', content='Home Page')
@app.route('/profile')
@login_required
def profile():
return render_template('index.html', content='Profile Page')
@app.route('/post_login')
@login_required
def post_login():
return render_template('index.html', content='Post Login')
@app.route('/http')
@http_auth_required
def http():
return 'HTTP Authentication'
@app.route('/http_custom_realm')
@http_auth_required('My Realm')
def http_custom_realm():
return render_template('index.html', content='HTTP Authentication')
@app.route('/token')
@auth_token_required
def token():
return render_template('index.html', content='Token Authentication')
@app.route('/multi_auth')
@auth_required('session', 'token', 'basic')
def multi_auth():
return render_template('index.html', content='Session, Token, Basic auth')
@app.route('/post_logout')
def post_logout():
return render_template('index.html', content='Post Logout')
@app.route('/post_register')
def post_register():
return render_template('index.html', content='Post Register')
@app.route('/admin')
@roles_required('admin')
def admin():
return render_template('index.html', content='Admin Page')
@app.route('/admin_and_editor')
@roles_required('admin', 'editor')
def admin_and_editor():
return render_template('index.html', content='Admin and Editor Page')
@app.route('/admin_or_editor')
@roles_accepted('admin', 'editor')
def admin_or_editor():
return render_template('index.html', content='Admin or Editor Page')
@app.route('/unauthorized')
def unauthorized():
return render_template('unauthorized.html')
@app.route('/coverage/add_role_to_user')
def add_role_to_user():
u = ds.find_user(email='joe@lp.com')
r = ds.find_role('admin')
ds.add_role_to_user(u, r)
return 'success'
@app.route('/coverage/remove_role_from_user')
def remove_role_from_user():
u = ds.find_user(email='matt@lp.com')
ds.remove_role_from_user(u, 'admin')
return 'success'
@app.route('/coverage/deactivate_user')
def deactivate_user():
u = ds.find_user(email='matt@lp.com')
ds.deactivate_user(u)
return 'success'
@app.route('/coverage/activate_user')
def activate_user():
u = ds.find_user(email='tiya@lp.com')
ds.activate_user(u)
return 'success'
@app.route('/coverage/invalid_role')
def invalid_role():
return 'success' if ds.find_role('bogus') is None else 'failure'
@app.route('/page1')
def page_1():
return 'Page 1'
return app
def create_roles():
for role in ('admin', 'editor', 'author'):
ds.create_role(name=role)
ds.commit()
def create_users(count=None):
users = [('matt@lp.com', 'matt', 'password', ['admin'], True),
('joe@lp.com', 'joe', 'password', ['editor'], True),
('dave@lp.com', 'dave', 'password', ['admin', 'editor'], True),
('jill@lp.com', 'jill', 'password', ['author'], True),
('tiya@lp.com', 'tiya', 'password', [], False)]
count = count or len(users)
for u in users[:count]:
pw = encrypt_password(u[2])
roles = [ds.find_or_create_role(rn) for rn in u[3]]
ds.commit()
user = ds.create_user(email=u[0], username=u[1], password=pw, active=u[4])
ds.commit()
for role in roles:
ds.add_role_to_user(user, role)
ds.commit()
def populate_data(user_count=None):
create_roles()
create_users(user_count)
def add_context_processors(s):
@s.context_processor
def for_all():
return dict()
@s.forgot_password_context_processor
def forgot_password():
return dict()
@s.login_context_processor
def login():
return dict()
@s.register_context_processor
def register():
return dict()
@s.reset_password_context_processor
def reset_password():
return dict()
@s.send_confirmation_context_processor
def send_confirmation():
return dict()
@s.send_login_context_processor
def send_login():
return dict()
@s.mail_context_processor
def mail():
return dict()
-57
View File
@@ -1,57 +0,0 @@
# -*- coding: utf-8 -*-
import sys
import os
sys.path.pop(0)
sys.path.insert(0, os.getcwd())
from flask.ext.mongoengine import MongoEngine
from flask.ext.security import Security, UserMixin, RoleMixin, \
MongoEngineUserDatastore
from tests.test_app import create_app as create_base_app, populate_data, \
add_context_processors
def create_app(config, **kwargs):
app = create_base_app(config)
app.config['MONGODB_SETTINGS'] = dict(
db='flask_security_test',
host='localhost',
port=27017
)
db = MongoEngine(app)
class Role(db.Document, RoleMixin):
name = db.StringField(required=True, unique=True, max_length=80)
description = db.StringField(max_length=255)
class User(db.Document, UserMixin):
email = db.StringField(unique=True, max_length=255)
username = db.StringField(max_length=255)
password = db.StringField(required=True, max_length=255)
last_login_at = db.DateTimeField()
current_login_at = db.DateTimeField()
last_login_ip = db.StringField(max_length=100)
current_login_ip = db.StringField(max_length=100)
login_count = db.IntField()
active = db.BooleanField(default=True)
confirmed_at = db.DateTimeField()
roles = db.ListField(db.ReferenceField(Role), default=[])
@app.before_first_request
def before_first_request():
User.drop_collection()
Role.drop_collection()
populate_data(app.config.get('USER_COUNT', None))
app.security = Security(app, datastore=MongoEngineUserDatastore(db, User, Role), **kwargs)
add_context_processors(app.security)
return app
if __name__ == '__main__':
create_app({}).run()
-64
View File
@@ -1,64 +0,0 @@
# -*- coding: utf-8 -*-
import sys
import os
sys.path.pop(0)
sys.path.insert(0, os.getcwd())
from flask_peewee.db import Database
from peewee import *
from flask.ext.security import Security, UserMixin, RoleMixin, \
PeeweeUserDatastore
from tests.test_app import create_app as create_base_app, populate_data, \
add_context_processors
def create_app(config, **kwargs):
app = create_base_app(config)
app.config['DATABASE'] = {
'name': 'peewee.db',
'engine': 'peewee.SqliteDatabase'
}
db = Database(app)
class Role(db.Model, RoleMixin):
name = TextField(unique=True)
description = TextField(null=True)
class User(db.Model, UserMixin):
email = TextField()
username = TextField()
password = TextField()
last_login_at = DateTimeField(null=True)
current_login_at = DateTimeField(null=True)
last_login_ip = TextField(null=True)
current_login_ip = TextField(null=True)
login_count = IntegerField(null=True)
active = BooleanField(default=True)
confirmed_at = DateTimeField(null=True)
class UserRoles(db.Model):
""" Peewee does not have built-in many-to-many support, so we have to
create this mapping class to link users to roles."""
user = ForeignKeyField(User, related_name='roles')
role = ForeignKeyField(Role, related_name='users')
name = property(lambda self: self.role.name)
description = property(lambda self: self.role.description)
@app.before_first_request
def before_first_request():
for Model in (Role, User, UserRoles):
Model.drop_table(fail_silently=True)
Model.create_table()
populate_data(app.config.get('USER_COUNT', None))
app.security = Security(app, datastore=PeeweeUserDatastore(db, User, Role, UserRoles), **kwargs)
add_context_processors(app.security)
return app
if __name__ == '__main__':
create_app({}).run()
-60
View File
@@ -1,60 +0,0 @@
# -*- coding: utf-8 -*-
import sys
import os
sys.path.pop(0)
sys.path.insert(0, os.getcwd())
from flask.ext.sqlalchemy import SQLAlchemy
from flask.ext.security import Security, UserMixin, RoleMixin, \
SQLAlchemyUserDatastore
from tests.test_app import create_app as create_base_app, populate_data, \
add_context_processors
def create_app(config, **kwargs):
app = create_base_app(config)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite://'
db = SQLAlchemy(app)
roles_users = db.Table('roles_users',
db.Column('user_id', db.Integer(), db.ForeignKey('user.id')),
db.Column('role_id', db.Integer(), db.ForeignKey('role.id')))
class Role(db.Model, RoleMixin):
id = db.Column(db.Integer(), primary_key=True)
name = db.Column(db.String(80), unique=True)
description = db.Column(db.String(255))
class User(db.Model, UserMixin):
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(255), unique=True)
username = db.Column(db.String(255))
password = db.Column(db.String(255))
last_login_at = db.Column(db.DateTime())
current_login_at = db.Column(db.DateTime())
last_login_ip = db.Column(db.String(100))
current_login_ip = db.Column(db.String(100))
login_count = db.Column(db.Integer)
active = db.Column(db.Boolean())
confirmed_at = db.Column(db.DateTime())
roles = db.relationship('Role', secondary=roles_users,
backref=db.backref('users', lazy='dynamic'))
@app.before_first_request
def before_first_request():
db.drop_all()
db.create_all()
populate_data(app.config.get('USER_COUNT', None))
app.security = Security(app, datastore=SQLAlchemyUserDatastore(db, User, Role), **kwargs)
add_context_processors(app.security)
return app
if __name__ == '__main__':
create_app({}).run()
+127
View File
@@ -0,0 +1,127 @@
# -*- coding: utf-8 -*-
"""
test_changeable
~~~~~~~~~~~~~~~
Changeable tests
"""
import pytest
from flask_security.signals import password_changed
from utils import authenticate
pytestmark = pytest.mark.changeable()
def test_recoverable_flag(app, client, get_message):
recorded = []
@password_changed.connect_via(app)
def on_password_changed(app, user):
recorded.append(user)
authenticate(client)
# Test change view
response = client.get('/change', follow_redirects=True)
assert b'Change password' in response.data
# Test wrong original password
response = client.post('/change', data={
'password': 'notpassword',
'new_password': 'newpassword',
'new_password_confirm': 'newpassword'
}, follow_redirects=True)
assert get_message('INVALID_PASSWORD') in response.data
# Test mismatch
response = client.post('/change', data={
'password': 'password',
'new_password': 'newpassword',
'new_password_confirm': 'notnewpassword'
}, follow_redirects=True)
assert get_message('RETYPE_PASSWORD_MISMATCH') in response.data
# Test missing password
response = client.post('/change', data={
'password': ' ',
'new_password': '',
'new_password_confirm': ''
}, follow_redirects=True)
assert get_message('PASSWORD_NOT_PROVIDED') in response.data
# Test bad password
response = client.post('/change', data={
'password': 'password',
'new_password': 'a',
'new_password_confirm': 'a'
}, follow_redirects=True)
assert get_message('PASSWORD_INVALID_LENGTH') in response.data
# Test same as previous
response = client.post('/change', data={
'password': 'password',
'new_password': 'password',
'new_password_confirm': 'password'
}, follow_redirects=True)
assert get_message('PASSWORD_IS_THE_SAME') in response.data
# Test successful submit sends email notification
with app.mail.record_messages() as outbox:
response = client.post('/change', data={
'password': 'password',
'new_password': 'newpassword',
'new_password_confirm': 'newpassword'
}, follow_redirects=True)
assert get_message('PASSWORD_CHANGE') in response.data
assert b'Home Page' in response.data
assert len(recorded) == 1
assert len(outbox) == 1
assert "Your password has been changed" in outbox[0].html
# Test JSON
data = ('{"password": "newpassword", "new_password": "newpassword2", '
'"new_password_confirm": "newpassword2"}')
response = client.post('/change', data=data, headers={'Content-Type': 'application/json'})
assert response.status_code == 200
assert response.headers['Content-Type'] == 'application/json'
@pytest.mark.settings(change_url='/custom_change')
def test_custom_change_url(client):
authenticate(client)
response = client.get('/custom_change')
assert response.status_code == 200
@pytest.mark.settings(change_password_template='custom_security/change_password.html')
def test_custom_change_template(client):
authenticate(client)
response = client.get('/change')
assert b'CUSTOM CHANGE PASSWORD' in response.data
@pytest.mark.settings(send_password_change_email=False)
def test_disable_change_emails(app, client):
with app.mail.record_messages() as outbox:
client.post('/change', data={
'password': 'password',
'new_password': 'newpassword',
'new_password_confirm': 'newpassword'
}, follow_redirects=True)
assert len(outbox) == 0
@pytest.mark.settings(post_change_view='/profile')
def test_custom_post_change_view(client):
authenticate(client)
response = client.post('/change', data={
'password': 'password',
'new_password': 'newpassword',
'new_password_confirm': 'newpassword'
}, follow_redirects=True)
assert b'Profile Page' in response.data
+273
View File
@@ -0,0 +1,273 @@
# -*- coding: utf-8 -*-
"""
test_common
~~~~~~~~~~~
Test common functionality
"""
import base64
try:
from cookielib import Cookie
except ImportError:
from http.cookiejar import Cookie
from utils import authenticate, json_authenticate, logout
def test_login_view(client):
response = client.get('/login')
assert b'<h1>Login</h1>' in response.data
def test_authenticate(client):
response = authenticate(client)
assert response.status_code == 302
response = authenticate(client, follow_redirects=True)
assert b'Hello matt@lp.com' in response.data
def test_authenticate_with_next(client):
data = dict(email='matt@lp.com', password='password')
response = client.post('/login?next=/page1', data=data, follow_redirects=True)
assert b'Page 1' in response.data
def test_authenticate_with_invalid_next(client, get_message):
data = dict(email='matt@lp.com', password='password')
response = client.post('/login?next=http://google.com', data=data)
assert get_message('INVALID_REDIRECT') in response.data
def test_authenticate_case_insensitive_email(app, client):
response = authenticate(client, 'MATT@lp.com', follow_redirects=True)
assert b'Hello matt@lp.com' in response.data
def test_unprovided_username(client, get_message):
response = authenticate(client, "")
assert get_message('EMAIL_NOT_PROVIDED') in response.data
def test_unprovided_password(client, get_message):
response = authenticate(client, password="")
assert get_message('PASSWORD_NOT_PROVIDED') in response.data
def test_invalid_user(client, get_message):
response = authenticate(client, email="bogus@bogus.com")
assert get_message('USER_DOES_NOT_EXIST') in response.data
def test_bad_password(client, get_message):
response = authenticate(client, password="bogus")
assert get_message('INVALID_PASSWORD') in response.data
def test_inactive_user(client, get_message):
response = authenticate(client, "tiya@lp.com", "password")
assert get_message('DISABLED_ACCOUNT') in response.data
def test_unset_password(client, get_message):
response = authenticate(client, "jess@lp.com", "password")
assert get_message('PASSWORD_NOT_SET') in response.data
def test_logout(client):
authenticate(client)
response = logout(client, follow_redirects=True)
assert b'Home Page' in response.data
def test_missing_session_access(client, get_message):
response = client.get('/profile', follow_redirects=True)
assert get_message('LOGIN') in response.data
def test_has_session_access(client):
authenticate(client)
response = client.get("/profile", follow_redirects=True)
assert b'profile' in response.data
def test_authorized_access(client):
authenticate(client)
response = client.get("/admin")
assert b'Admin Page' in response.data
def test_unauthorized_access(client, get_message):
authenticate(client, "joe@lp.com")
response = client.get("/admin", follow_redirects=True)
assert get_message('UNAUTHORIZED') in response.data
def test_roles_accepted(client):
for user in ("matt@lp.com", "joe@lp.com"):
authenticate(client, user)
response = client.get("/admin_or_editor")
assert b'Admin or Editor Page' in response.data
logout(client)
authenticate(client, "jill@lp.com")
response = client.get("/admin_or_editor", follow_redirects=True)
assert b'Home Page' in response.data
def test_unauthenticated_role_required(client, get_message):
response = client.get('/admin', follow_redirects=True)
assert get_message('UNAUTHORIZED') in response.data
def test_multiple_role_required(client):
for user in ("matt@lp.com", "joe@lp.com"):
authenticate(client, user)
response = client.get("/admin_and_editor", follow_redirects=True)
assert b'Home Page' in response.data
client.get('/logout')
authenticate(client, 'dave@lp.com')
response = client.get("/admin_and_editor", follow_redirects=True)
assert b'Admin and Editor Page' in response.data
def test_ok_json_auth(client):
response = json_authenticate(client)
assert response.jdata['meta']['code'] == 200
assert 'authentication_token' in response.jdata['response']['user']
def test_invalid_json_auth(client):
response = json_authenticate(client, password='junk')
assert b'"code": 400' in response.data
def test_token_auth_via_querystring_valid_token(client):
response = json_authenticate(client)
token = response.jdata['response']['user']['authentication_token']
response = client.get('/token?auth_token=' + token)
assert b'Token Authentication' in response.data
def test_token_auth_via_header_valid_token(client):
response = json_authenticate(client)
token = response.jdata['response']['user']['authentication_token']
headers = {"Authentication-Token": token}
response = client.get('/token', headers=headers)
assert b'Token Authentication' in response.data
def test_token_auth_via_querystring_invalid_token(client):
response = client.get('/token?auth_token=X')
assert 401 == response.status_code
def test_token_auth_via_header_invalid_token(client):
response = client.get('/token', headers={"Authentication-Token": 'X'})
assert 401 == response.status_code
def test_http_auth(client):
response = client.get('/http', headers={
'Authorization': 'Basic %s' % base64.b64encode(b"joe@lp.com:password").decode('utf-8')
})
assert b'HTTP Authentication' in response.data
def test_http_auth_no_authorization(client):
response = client.get('/http', headers={})
assert b'<h1>Unauthorized</h1>' in response.data
assert 'WWW-Authenticate' in response.headers
assert 'Basic realm="Login Required"' == response.headers['WWW-Authenticate']
def test_invalid_http_auth_invalid_username(client):
response = client.get('/http', headers={
'Authorization': 'Basic %s' % base64.b64encode(b"bogus:bogus").decode('utf-8')
})
assert b'<h1>Unauthorized</h1>' in response.data
assert 'WWW-Authenticate' in response.headers
assert 'Basic realm="Login Required"' == response.headers['WWW-Authenticate']
def test_invalid_http_auth_bad_password(client):
response = client.get('/http', headers={
'Authorization': 'Basic %s' % base64.b64encode(b"joe@lp.com:bogus").decode('utf-8')
})
assert b'<h1>Unauthorized</h1>' in response.data
assert 'WWW-Authenticate' in response.headers
assert 'Basic realm="Login Required"' == response.headers['WWW-Authenticate']
def test_custom_http_auth_realm(client):
response = client.get('/http_custom_realm', headers={
'Authorization': 'Basic %s' % base64.b64encode(b"joe@lp.com:bogus").decode('utf-8')
})
assert b'<h1>Unauthorized</h1>' in response.data
assert 'WWW-Authenticate' in response.headers
assert 'Basic realm="My Realm"' == response.headers['WWW-Authenticate']
def test_multi_auth_basic(client):
response = client.get('/multi_auth', headers={
'Authorization': 'Basic %s' % base64.b64encode(b"joe@lp.com:password").decode('utf-8')
})
assert b'Basic' in response.data
response = client.get('/multi_auth')
assert response.status_code == 401
def test_multi_auth_token(client):
response = json_authenticate(client)
token = response.jdata['response']['user']['authentication_token']
response = client.get('/multi_auth?auth_token=' + token)
assert b'Token' in response.data
def test_multi_auth_session(client):
authenticate(client, )
response = client.get('/multi_auth')
assert b'Session' in response.data
def test_user_deleted_during_session_reverts_to_anonymous_user(app, client):
authenticate(client)
with app.test_request_context('/'):
user = app.security.datastore.find_user(email='matt@lp.com')
app.security.datastore.delete_user(user)
app.security.datastore.commit()
response = client.get('/')
assert b'Hello matt@lp.com' not in response.data
def test_remember_token(client):
response = authenticate(client, follow_redirects=False)
client.cookie_jar.clear_session_cookies()
response = client.get('/profile')
assert b'profile' in response.data
def test_token_loader_does_not_fail_with_invalid_token(client):
c = Cookie(version=0, name='remember_token', value='None', port=None,
port_specified=False, domain='www.example.com',
domain_specified=False, domain_initial_dot=False, path='/',
path_specified=True, secure=False, expires=None,
discard=True, comment=None, comment_url=None,
rest={'HttpOnly': None}, rfc2109=False)
client.cookie_jar.set_cookie(c)
response = client.get('/')
assert b'BadSignature' not in response.data
def test_sending_auth_token_with_json(client):
response = json_authenticate(client)
token = response.jdata['response']['user']['authentication_token']
data = '{"auth_token": "%s"}' % token
response = client.post('/token', data=data, headers={'Content-Type': 'application/json'})
assert b'Token Authentication' in response.data
+43
View File
@@ -0,0 +1,43 @@
# -*- coding: utf-8 -*-
"""
test_configuration
~~~~~~~~~~~~~~~~~~
Basic configuration tests
"""
import base64
import pytest
from utils import authenticate, logout
@pytest.mark.settings(
logout_url='/custom_logout',
login_url='/custom_login',
post_login_view='/post_login',
post_logout_view='/post_logout',
default_http_auth_realm='Custom Realm')
def test_view_configuration(client):
response = client.get('/custom_login')
assert b"<h1>Login</h1>" in response.data
response = authenticate(client, endpoint='/custom_login', follow_redirects=True)
assert b'Post Login' in response.data
response = logout(client, endpoint='/custom_logout', follow_redirects=True)
assert b'Post Logout' in response.data
response = client.get('/http', headers={
'Authorization': 'Basic %s' % base64.b64encode(b"joe@lp.com:bogus")
})
assert b'<h1>Unauthorized</h1>' in response.data
assert 'WWW-Authenticate' in response.headers
assert 'Basic realm="Custom Realm"' == response.headers['WWW-Authenticate']
@pytest.mark.settings(login_user_template='custom_security/login_user.html')
def test_template_configuration(client):
response = client.get('/login')
assert b'CUSTOM LOGIN USER' in response.data
+146
View File
@@ -0,0 +1,146 @@
# -*- coding: utf-8 -*-
"""
test_confirmable
~~~~~~~~~~~~~~~~
Confirmable tests
"""
import time
import pytest
from flask_security.signals import user_confirmed, confirm_instructions_sent
from flask_security.utils import capture_registrations
from utils import authenticate, logout
pytestmark = pytest.mark.confirmable()
@pytest.mark.registerable()
def test_confirmable_flag(app, client, sqlalchemy_datastore, get_message):
recorded_confirms = []
recorded_instructions_sent = []
@user_confirmed.connect_via(app)
def on_confirmed(app, user):
recorded_confirms.append(user)
@confirm_instructions_sent.connect_via(app)
def on_instructions_sent(app, user):
recorded_instructions_sent.append(user)
# Test login before confirmation
email = 'dude@lp.com'
with capture_registrations() as registrations:
response = client.post('/register', data=dict(email=email, password='password'))
assert response.status_code == 302
response = authenticate(client, email=email)
assert get_message('CONFIRMATION_REQUIRED') in response.data
# Test invalid token
response = client.get('/confirm/bogus', follow_redirects=True)
assert get_message('INVALID_CONFIRMATION_TOKEN') in response.data
# Test JSON
response = client.post('/confirm', data='{"email": "matt@lp.com"}', headers={
'Content-Type': 'application/json'
})
assert response.status_code == 200
assert response.headers['Content-Type'] == 'application/json'
assert 'user' in response.jdata['response']
assert len(recorded_instructions_sent) == 1
# Test ask for instructions with invalid email
response = client.post('/confirm', data=dict(email='bogus@bogus.com'))
assert get_message('USER_DOES_NOT_EXIST') in response.data
# Test resend instructions
response = client.post('/confirm', data=dict(email=email))
assert get_message('CONFIRMATION_REQUEST', email=email) in response.data
assert len(recorded_instructions_sent) == 2
# Test confirm
token = registrations[0]['confirm_token']
response = client.get('/confirm/' + token, follow_redirects=True)
assert get_message('EMAIL_CONFIRMED') in response.data
assert len(recorded_confirms) == 1
# Test already confirmed
response = client.get('/confirm/' + token, follow_redirects=True)
assert get_message('ALREADY_CONFIRMED') in response.data
# Test already confirmed when asking for confirmation instructions
logout(client)
response = client.get('/confirm')
assert response.status_code == 200
response = client.post('/confirm', data=dict(email=email))
assert get_message('ALREADY_CONFIRMED') in response.data
# Test user was deleted before confirmation
with capture_registrations() as registrations:
client.post('/register', data=dict(email='mary@lp.com', password='password'))
user = registrations[0]['user']
token = registrations[0]['confirm_token']
with app.app_context():
sqlalchemy_datastore.delete(user)
sqlalchemy_datastore.commit()
response = client.get('/confirm/' + token, follow_redirects=True)
assert get_message('INVALID_CONFIRMATION_TOKEN') in response.data
@pytest.mark.registerable()
@pytest.mark.settings(confirm_email_within='1 milliseconds')
def test_expired_confirmation_token(client, get_message):
with capture_registrations() as registrations:
data = dict(email='mary@lp.com', password='password')
client.post('/register', data=data, follow_redirects=True)
user = registrations[0]['user']
token = registrations[0]['confirm_token']
time.sleep(1)
response = client.get('/confirm/' + token, follow_redirects=True)
msg = get_message('CONFIRMATION_EXPIRED', within='1 milliseconds', email=user.email)
assert msg in response.data
@pytest.mark.registerable()
@pytest.mark.settings(login_without_confirmation=True)
def test_login_when_unconfirmed(client, get_message):
data = dict(email='mary@lp.com', password='password')
response = client.post('/register', data=data, follow_redirects=True)
assert b'mary@lp.com' in response.data
@pytest.mark.registerable()
@pytest.mark.settings(login_without_confirmation=True)
def test_confirmation_different_user_when_logged_in(client, get_message):
e1 = 'dude@lp.com'
e2 = 'lady@lp.com'
with capture_registrations() as registrations:
for e in e1, e2:
client.post('/register', data=dict(email=e, password='password'))
logout(client)
token1 = registrations[0]['confirm_token']
token2 = registrations[1]['confirm_token']
client.get('/confirm/' + token1, follow_redirects=True)
logout(client)
authenticate(client, email=e1)
response = client.get('/confirm/' + token2, follow_redirects=True)
assert get_message('EMAIL_CONFIRMED') in response.data
assert b'Hello lady@lp.com' in response.data
+89
View File
@@ -0,0 +1,89 @@
# -*- coding: utf-8 -*-
"""
test_context_processors
~~~~~~~~~~~~~~~~~~~~~~~
Context processor tests
"""
import pytest
from utils import authenticate
@pytest.mark.recoverable()
@pytest.mark.registerable()
@pytest.mark.confirmable()
@pytest.mark.changeable()
@pytest.mark.settings(
login_without_confirmation=True,
change_password_template='custom_security/change_password.html',
login_user_template='custom_security/login_user.html',
reset_password_template='custom_security/reset_password.html',
forgot_password_template='custom_security/forgot_password.html',
send_confirmation_template='custom_security/send_confirmation.html',
register_user_template='custom_security/register_user.html')
def test_context_processors(client, app):
@app.security.forgot_password_context_processor
def forgot_password():
return {'foo': 'bar'}
response = client.get('/reset')
assert b'bar' in response.data
@app.security.login_context_processor
def login():
return {'foo': 'bar'}
response = client.get('/login')
assert b'bar' in response.data
@app.security.register_context_processor
def register():
return {'foo': 'bar'}
response = client.get('/register')
assert b'bar' in response.data
@app.security.reset_password_context_processor
def reset_password():
return {'foo': 'bar'}
response = client.get('/reset')
assert b'bar' in response.data
@app.security.change_password_context_processor
def change_password():
return {'foo': 'bar'}
authenticate(client)
response = client.get('/change')
assert b'bar' in response.data
@app.security.send_confirmation_context_processor
def send_confirmation():
return {'foo': 'bar'}
response = client.get('/confirm')
assert b'bar' in response.data
@app.security.mail_context_processor
def mail():
return {'foo': 'bar'}
with app.mail.record_messages() as outbox:
client.post('/reset', data=dict(email='matt@lp.com'))
email = outbox[0]
assert 'bar' in email.html
@pytest.mark.passwordless()
@pytest.mark.settings(send_login_template='custom_security/send_login.html')
def test_passwordless_login_context_processor(app, client):
@app.security.send_login_context_processor
def send_login():
return {'foo': 'bar'}
response = client.get('/login')
assert b'bar' in response.data
+153
View File
@@ -0,0 +1,153 @@
# -*- coding: utf-8 -*-
"""
test_datastore
~~~~~~~~~~~~~~
Datastore tests
"""
from pytest import raises
from flask_security import UserMixin, RoleMixin
from flask_security.datastore import Datastore, UserDatastore
from utils import init_app_with_options
class User(UserMixin):
pass
class Role(RoleMixin):
pass
def test_unimplemented_datastore_methods():
datastore = Datastore(None)
assert datastore.db is None
with raises(NotImplementedError):
datastore.put(None)
with raises(NotImplementedError):
datastore.delete(None)
assert not datastore.commit()
def test_unimplemented_user_datastore_methods():
datastore = UserDatastore(None, None)
with raises(NotImplementedError):
datastore.find_user(None)
with raises(NotImplementedError):
datastore.find_role(None)
with raises(NotImplementedError):
datastore.get_user(None)
def test_toggle_active():
datastore = UserDatastore(None, None)
user = User()
user.active = True
assert datastore.toggle_active(user) is True
assert not user.active
assert datastore.toggle_active(user) is True
assert user.active is True
def test_deactivate_user():
datastore = UserDatastore(None, None)
user = User()
user.active = True
assert datastore.deactivate_user(user) is True
assert not user.active
def test_activate_user():
datastore = UserDatastore(None, None)
user = User()
user.active = False
assert datastore.activate_user(user) is True
assert user.active is True
def test_deactivate_returns_false_if_already_false():
datastore = UserDatastore(None, None)
user = User()
user.active = False
assert not datastore.deactivate_user(user)
def test_activate_returns_false_if_already_true():
datastore = UserDatastore(None, None)
user = User()
user.active = True
assert not datastore.activate_user(user)
def test_get_user(app, datastore):
init_app_with_options(app, datastore, **{
'SECURITY_USER_IDENTITY_ATTRIBUTES': ('email', 'username')
})
with app.app_context():
user_id = datastore.find_user(email='matt@lp.com').id
user = datastore.get_user(user_id)
assert user is not None
user = datastore.get_user('matt@lp.com')
assert user is not None
user = datastore.get_user('matt')
assert user is not None
def test_find_role(app, datastore):
init_app_with_options(app, datastore)
role = datastore.find_role('admin')
assert role is not None
role = datastore.find_role('bogus')
assert role is None
def test_add_role_to_user(app, datastore):
init_app_with_options(app, datastore)
# Test with user object
user = datastore.find_user(email='matt@lp.com')
assert user.has_role('editor') is False
assert datastore.add_role_to_user(user, 'editor') is True
assert datastore.add_role_to_user(user, 'editor') is False
assert user.has_role('editor') is True
# Test with email
assert datastore.add_role_to_user('jill@lp.com', 'editor') is True
user = datastore.find_user(email='jill@lp.com')
assert user.has_role('editor') is True
# Test remove role
assert datastore.remove_role_from_user(user, 'editor') is True
assert datastore.remove_role_from_user(user, 'editor') is False
def test_create_user_with_roles(app, datastore):
init_app_with_options(app, datastore)
role = datastore.find_role('admin')
datastore.commit()
user = datastore.create_user(email='dude@lp.com', username='dude',
password='password', roles=[role])
datastore.commit()
user = datastore.find_user(email='dude@lp.com')
assert user.has_role('admin') is True
def test_delete_user(app, datastore):
init_app_with_options(app, datastore)
user = datastore.find_user(email='matt@lp.com')
datastore.delete_user(user)
datastore.commit()
user = datastore.find_user(email='matt@lp.com')
assert user is None
+46
View File
@@ -0,0 +1,46 @@
# -*- coding: utf-8 -*-
"""
test_entities
~~~~~~~~~~~~~
Entity tests
"""
from flask_security import RoleMixin, UserMixin, AnonymousUser
class Role(RoleMixin):
def __init__(self, name):
self.name = name
class User(UserMixin):
def __init__(self, roles):
self.roles = roles
def test_role_mixin_equal():
admin1 = Role('admin')
admin2 = Role('admin')
assert admin1 == admin2
def test_role_mixin_not_equal():
admin = Role('admin')
editor = Role('editor')
assert admin != editor
def test_user_mixin_has_role_with_string():
admin = Role('admin')
editor = Role('editor')
user = User([admin, editor])
assert user.has_role('admin') is True
assert user.has_role('editor') is True
assert user.has_role(admin) is True
assert user.has_role(editor) is True
def test_anonymous_user_has_no_roles():
user = AnonymousUser()
assert not user.has_role('admin')
+38
View File
@@ -0,0 +1,38 @@
# -*- coding: utf-8 -*-
"""
test_hashing
~~~~~~~~~~~~
hashing tests
"""
from pytest import raises
from flask_security.utils import verify_password, encrypt_password
from utils import authenticate, init_app_with_options
def test_verify_password_bcrypt(app, sqlalchemy_datastore):
init_app_with_options(app, sqlalchemy_datastore, **{
'SECURITY_PASSWORD_HASH': 'bcrypt',
'SECURITY_PASSWORD_SALT': 'salty'
})
with app.app_context():
assert verify_password('pass', encrypt_password('pass'))
def test_login_with_bcrypt_enabled(app, sqlalchemy_datastore):
init_app_with_options(app, sqlalchemy_datastore, **{
'SECURITY_PASSWORD_HASH': 'bcrypt',
'SECURITY_PASSWORD_SALT': 'salty'
})
response = authenticate(app.test_client(), follow_redirects=True)
assert b'Home Page' in response.data
def test_missing_hash_salt_option(app, sqlalchemy_datastore):
with raises(RuntimeError):
init_app_with_options(app, sqlalchemy_datastore, **{
'SECURITY_PASSWORD_HASH': 'bcrypt',
})
+172
View File
@@ -0,0 +1,172 @@
# -*- coding: utf-8 -*-
"""
test_emails
~~~~~~~~~~~
Email functionality tests
"""
import pytest
from flask_security import Security
from flask_security.forms import LoginForm, RegisterForm, ConfirmRegisterForm, \
SendConfirmationForm, PasswordlessLoginForm, ForgotPasswordForm, ResetPasswordForm, \
ChangePasswordForm, TextField, PasswordField, email_required, email_validator, valid_user_email
from flask_security.utils import capture_reset_password_requests
from utils import authenticate, init_app_with_options, populate_data
@pytest.mark.recoverable()
def test_async_email_task(app, client):
app.mail_sent = False
@app.security.send_mail_task
def send_email(msg):
app.mail_sent = True
client.post('/reset', data=dict(email='matt@lp.com'))
assert app.mail_sent is True
def test_register_blueprint_flag(app, sqlalchemy_datastore):
app.security = Security(app, datastore=Security, register_blueprint=False)
client = app.test_client()
response = client.get('/login')
assert response.status_code == 404
@pytest.mark.registerable()
@pytest.mark.recoverable()
@pytest.mark.changeable()
def test_basic_custom_forms(app, sqlalchemy_datastore):
class MyLoginForm(LoginForm):
email = TextField('My Login Email Address Field')
class MyRegisterForm(RegisterForm):
email = TextField('My Register Email Address Field')
class MyForgotPasswordForm(ForgotPasswordForm):
email = TextField('My Forgot Email Address Field',
validators=[email_required, email_validator, valid_user_email])
class MyResetPasswordForm(ResetPasswordForm):
password = TextField('My Reset Password Field')
class MyChangePasswordForm(ChangePasswordForm):
password = PasswordField('My Change Password Field')
app.security = Security(app,
datastore=sqlalchemy_datastore,
login_form=MyLoginForm,
register_form=MyRegisterForm,
forgot_password_form=MyForgotPasswordForm,
reset_password_form=MyResetPasswordForm,
change_password_form=MyChangePasswordForm)
populate_data(app)
client = app.test_client()
response = client.get('/login')
assert b'My Login Email Address Field' in response.data
response = client.get('/register')
assert b'My Register Email Address Field' in response.data
response = client.get('/reset')
assert b'My Forgot Email Address Field' in response.data
with capture_reset_password_requests() as requests:
response = client.post('/reset', data=dict(email='matt@lp.com'))
token = requests[0]['token']
response = client.get('/reset/' + token)
assert b'My Reset Password Field' in response.data
authenticate(client)
response = client.get('/change')
assert b'My Change Password Field' in response.data
@pytest.mark.registerable()
@pytest.mark.confirmable()
def test_confirmable_custom_form(app, sqlalchemy_datastore):
app.config['SECURITY_REGISTERABLE'] = True
app.config['SECURITY_CONFIRMABLE'] = True
class MyRegisterForm(ConfirmRegisterForm):
email = TextField('My Register Email Address Field')
class MySendConfirmationForm(SendConfirmationForm):
email = TextField('My Send Confirmation Email Address Field')
app.security = Security(app,
datastore=sqlalchemy_datastore,
send_confirmation_form=MySendConfirmationForm,
confirm_register_form=MyRegisterForm)
client = app.test_client()
response = client.get('/register')
assert b'My Register Email Address Field' in response.data
response = client.get('/confirm')
assert b'My Send Confirmation Email Address Field' in response.data
def test_passwordless_custom_form(app, sqlalchemy_datastore):
app.config['SECURITY_PASSWORDLESS'] = True
class MyPasswordlessLoginForm(PasswordlessLoginForm):
email = TextField('My Passwordless Email Address Field')
app.security = Security(app,
datastore=sqlalchemy_datastore,
passwordless_login_form=MyPasswordlessLoginForm)
client = app.test_client()
response = client.get('/login')
assert b'My Passwordless Email Address Field' in response.data
def test_addition_identity_attributes(app, sqlalchemy_datastore):
init_app_with_options(app, sqlalchemy_datastore, **{
'SECURITY_USER_IDENTITY_ATTRIBUTES': ('email', 'username')
})
client = app.test_client()
response = authenticate(client, email='matt', follow_redirects=True)
assert b'Hello matt@lp.com' in response.data
def test_flash_messages_off(app, sqlalchemy_datastore, get_message):
init_app_with_options(app, sqlalchemy_datastore, **{
'SECURITY_FLASH_MESSAGES': False
})
client = app.test_client()
response = client.get('/profile')
assert get_message('LOGIN') not in response.data
def test_invalid_hash_scheme(app, sqlalchemy_datastore, get_message):
with pytest.raises(ValueError):
init_app_with_options(app, sqlalchemy_datastore, **{
'SECURITY_PASSWORD_HASH': 'bogus'
})
def test_change_hash_type(app, sqlalchemy_datastore):
init_app_with_options(app, sqlalchemy_datastore, **{
'SECURITY_PASSWORD_SCHEMES': ['bcrypt', 'plaintext']
})
app.config['SECURITY_PASSWORD_HASH'] = 'bcrypt'
app.config['SECURITY_PASSWORD_SALT'] = 'salty'
app.security = Security(app, datastore=sqlalchemy_datastore, register_blueprint=False)
client = app.test_client()
response = client.post('/login', data=dict(email='matt@lp.com', password='password'))
assert response.status_code == 302
+89
View File
@@ -0,0 +1,89 @@
# -*- coding: utf-8 -*-
"""
test_passwordless
~~~~~~~~~~~~~~~~~
Passwordless tests
"""
import time
import pytest
from flask_security.signals import login_instructions_sent
from flask_security.utils import capture_passwordless_login_requests
from utils import logout
pytestmark = pytest.mark.passwordless()
def test_trackable_flag(app, client, get_message):
recorded = []
@login_instructions_sent.connect_via(app)
def on_instructions_sent(app, user, login_token):
recorded.append(user)
# Test disabled account
response = client.post('/login', data=dict(email='tiya@lp.com'), follow_redirects=True)
assert get_message('DISABLED_ACCOUNT') in response.data
# Test login with json and valid email
data = '{"email": "matt@lp.com", "password": "password"}'
response = client.post('/login', data=data, headers={'Content-Type': 'application/json'})
assert response.status_code == 200
assert len(recorded) == 1
# Test login with json and invalid email
data = '{"email": "nobody@lp.com", "password": "password"}'
response = client.post('/login', data=data, headers={'Content-Type': 'application/json'})
assert b'errors' in response.data
# Test sends email and shows appropriate response
with capture_passwordless_login_requests() as requests:
with app.mail.record_messages() as outbox:
response = client.post('/login', data=dict(email='matt@lp.com'), follow_redirects=True)
assert len(recorded) == 2
assert len(requests) == 1
assert len(outbox) == 1
assert 'user' in requests[0]
assert 'login_token' in requests[0]
user = requests[0]['user']
assert get_message('LOGIN_EMAIL_SENT', email=user.email) in response.data
token = requests[0]['login_token']
response = client.get('/login/' + token, follow_redirects=True)
assert get_message('PASSWORDLESS_LOGIN_SUCCESSFUL') in response.data
# Test already authenticated
response = client.get('/login/' + token, follow_redirects=True)
assert get_message('PASSWORDLESS_LOGIN_SUCCESSFUL') not in response.data
logout(client)
# Test invalid token
response = client.get('/login/bogus', follow_redirects=True)
assert get_message('INVALID_LOGIN_TOKEN') in response.data
# Test login request with invalid email
response = client.post('/login', data=dict(email='bogus@bogus.com'))
assert get_message('USER_DOES_NOT_EXIST') in response.data
@pytest.mark.settings(login_within='1 milliseconds')
def test_expired_login_token(client, app, get_message):
e = 'matt@lp.com'
with capture_passwordless_login_requests() as requests:
client.post('/login', data=dict(email=e), follow_redirects=True)
token = requests[0]['login_token']
user = requests[0]['user']
time.sleep(1)
response = client.get('/login/' + token, follow_redirects=True)
assert get_message('LOGIN_EXPIRED', within='1 milliseconds', email=user.email) in response.data
+137
View File
@@ -0,0 +1,137 @@
# -*- coding: utf-8 -*-
"""
test_recoverable
~~~~~~~~~~~~~~~~
Recoverable functionality tests
"""
import time
import pytest
from flask_security.signals import reset_password_instructions_sent, password_reset
from flask_security.utils import capture_reset_password_requests
from utils import authenticate, logout
pytestmark = pytest.mark.recoverable()
def test_recoverable_flag(app, client, get_message):
recorded_resets = []
recorded_instructions_sent = []
@password_reset.connect_via(app)
def on_password_reset(app, user):
recorded_resets.append(user)
@reset_password_instructions_sent.connect_via(app)
def on_instructions_sent(app, user, token):
recorded_instructions_sent.append(user)
# Test the reset view
response = client.get('/reset')
assert b'<h1>Send password reset instructions</h1>' in response.data
# Test submitting email to reset password creates a token and sends email
with capture_reset_password_requests() as requests:
with app.mail.record_messages() as outbox:
response = client.post('/reset', data=dict(email='joe@lp.com'), follow_redirects=True)
assert len(recorded_instructions_sent) == 1
assert len(outbox) == 1
assert response.status_code == 200
assert get_message('PASSWORD_RESET_REQUEST', email='joe@lp.com') in response.data
token = requests[0]['token']
# Test view for reset token
response = client.get('/reset/' + token)
assert b'<h1>Reset password</h1>' in response.data
# Test submitting a new password
response = client.post('/reset/' + token, data={
'password': 'newpassword',
'password_confirm': 'newpassword'
}, follow_redirects=True)
assert get_message('PASSWORD_RESET') in response.data
assert len(recorded_resets) == 1
logout(client)
# Test logging in with the new password
response = authenticate(client, 'joe@lp.com', 'newpassword', follow_redirects=True)
assert b'Hello joe@lp.com' in response.data
logout(client)
# Test submitting JSON
response = client.post('/reset', data='{"email": "joe@lp.com"}', headers={
'Content-Type': 'application/json'
})
assert response.headers['Content-Type'] == 'application/json'
assert 'user' in response.jdata['response']
logout(client)
# Test invalid email
response = client.post('/reset', data=dict(email='bogus@lp.com'), follow_redirects=True)
assert get_message('USER_DOES_NOT_EXIST') in response.data
logout(client)
# Test invalid token
response = client.post('/reset/bogus', data={
'password': 'newpassword',
'password_confirm': 'newpassword'
}, follow_redirects=True)
assert get_message('INVALID_RESET_PASSWORD_TOKEN') in response.data
# Test mangled token
token = ("WyIxNjQ2MzYiLCIxMzQ1YzBlZmVhM2VhZjYwODgwMDhhZGU2YzU0MzZjMiJd.BZEw_Q.lQyo3npdPZtcJ"
"_sNHVHP103syjM&url_id=fbb89a8328e58c181ea7d064c2987874bc54a23d")
response = client.post('/reset/' + token, data={
'password': 'newpassword',
'password_confirm': 'newpassword'
}, follow_redirects=True)
assert get_message('INVALID_RESET_PASSWORD_TOKEN') in response.data
@pytest.mark.settings(reset_password_within='1 milliseconds')
def test_expired_reset_token(client, get_message):
with capture_reset_password_requests() as requests:
client.post('/reset', data=dict(email='joe@lp.com'), follow_redirects=True)
user = requests[0]['user']
token = requests[0]['token']
time.sleep(1)
response = client.post('/reset/' + token, data={
'password': 'newpassword',
'password_confirm': 'newpassword'
}, follow_redirects=True)
msg = get_message('PASSWORD_RESET_EXPIRED', within='1 milliseconds', email=user.email)
assert msg in response.data
@pytest.mark.settings(reset_url='/custom_reset')
def test_custom_reset_url(client):
response = client.get('/custom_reset')
assert response.status_code == 200
@pytest.mark.settings(reset_password_template='custom_security/reset_password.html',
forgot_password_template='custom_security/forgot_password.html')
def test_custom_reset_templates(client):
response = client.get('/reset')
assert b'CUSTOM FORGOT PASSWORD' in response.data
with capture_reset_password_requests() as requests:
client.post('/reset', data=dict(email='joe@lp.com'), follow_redirects=True)
token = requests[0]['token']
response = client.get('/reset/' + token)
assert b'CUSTOM RESET PASSWORD' in response.data
+101
View File
@@ -0,0 +1,101 @@
# -*- coding: utf-8 -*-
"""
test_registerable
~~~~~~~~~~~~~~~~~
Registerable tests
"""
import pytest
from flask_security.signals import user_registered
from utils import authenticate, logout
pytestmark = pytest.mark.registerable()
@pytest.mark.settings(post_register_view='/post_register')
def test_registerable_flag(client, app, get_message):
recorded = []
# Test the register view
response = client.get('/register')
assert b"<h1>Register</h1>" in response.data
# Test registering is successful, sends email, and fires signal
@user_registered.connect_via(app)
def on_user_registerd(app, user, confirm_token):
recorded.append(user)
data = dict(email='dude@lp.com', password='password', password_confirm='password')
with app.mail.record_messages() as outbox:
response = client.post('/register', data=data, follow_redirects=True)
assert len(recorded) == 1
assert len(outbox) == 1
assert b'Post Register' in response.data
logout(client)
# Test user can login after registering
response = authenticate(client, email='dude@lp.com', password='password')
assert response.status_code == 302
logout(client)
# Test registering with an existing email
data = dict(email='dude@lp.com', password='password', password_confirm='password')
response = client.post('/register', data=data, follow_redirects=True)
assert get_message('EMAIL_ALREADY_ASSOCIATED', email='dude@lp.com') in response.data
# Test registering with JSON
data = '{ "email": "dude2@lp.com", "password": "password"}'
response = client.post('/register', data=data, headers={'Content-Type': 'application/json'})
assert response.headers['content-type'] == 'application/json'
assert response.jdata['meta']['code'] == 200
logout(client)
# Test registering with invalid JSON
data = '{ "email": "bogus", "password": "password"}'
response = client.post('/register', data=data, headers={'Content-Type': 'application/json'})
assert response.headers['content-type'] == 'application/json'
assert response.jdata['meta']['code'] == 400
logout(client)
# Test ?next param
data = dict(email='dude3@lp.com',
password='password',
password_confirm='password')
response = client.post('/register?next=/page1', data=data, follow_redirects=True)
assert b'Page 1' in response.data
@pytest.mark.settings(register_url='/custom_register', post_register_view='/post_register')
def test_custom_register_url(client):
response = client.get('/custom_register')
assert b"<h1>Register</h1>" in response.data
data = dict(email='dude@lp.com',
password='password',
password_confirm='password')
response = client.post('/custom_register', data=data, follow_redirects=True)
assert b'Post Register' in response.data
@pytest.mark.settings(register_user_template='custom_security/register_user.html')
def test_custom_register_tempalate(client):
response = client.get('/register')
assert b'CUSTOM REGISTER USER' in response.data
@pytest.mark.settings(send_register_email=False)
def test_disable_register_emails(client, app):
data = dict(email='dude@lp.com', password='password', password_confirm='password')
with app.mail.record_messages() as outbox:
client.post('/register', data=data, follow_redirects=True)
assert len(outbox) == 0
+28
View File
@@ -0,0 +1,28 @@
# -*- coding: utf-8 -*-
"""
test_trackable
~~~~~~~~~~~~~~
Trackable tests
"""
import pytest
from utils import authenticate, logout
pytestmark = pytest.mark.trackable()
def test_trackable_flag(app, client):
e = 'matt@lp.com'
authenticate(client, email=e)
logout(client)
authenticate(client, email=e, headers={'X-Forwarded-For': '127.0.0.1'})
with app.app_context():
user = app.security.datastore.find_user(email=e)
assert user.last_login_at is not None
assert user.current_login_at is not None
assert user.last_login_ip == 'untrackable'
assert user.current_login_ip == '127.0.0.1'
assert user.login_count == 2
-88
View File
@@ -1,88 +0,0 @@
# -*- coding: utf-8 -*-
import unittest
from flask_security import RoleMixin, UserMixin, AnonymousUser
from flask_security.datastore import Datastore, UserDatastore
class Role(RoleMixin):
def __init__(self, name):
self.name = name
class User(UserMixin):
def __init__(self, email, roles):
self.email = email
self.roles = roles
admin = Role('admin')
admin2 = Role('admin')
editor = Role('editor')
user = User('matt@lp.com', [admin, editor])
class SecurityEntityTests(unittest.TestCase):
def test_role_mixin_equal(self):
self.assertEqual(admin, admin2)
def test_role_mixin_not_equal(self):
self.assertNotEqual(admin, editor)
def test_user_mixin_has_role_with_string(self):
self.assertTrue(user.has_role('admin'))
def test_user_mixin_has_role_with_role_obj(self):
self.assertTrue(user.has_role(Role('admin')))
def test_anonymous_user_has_no_roles(self):
au = AnonymousUser()
self.assertEqual(0, len(au.roles))
self.assertFalse(au.has_role('admin'))
class DatastoreTests(unittest.TestCase):
def setUp(self):
super(DatastoreTests, self).setUp()
self.ds = UserDatastore(None, None)
def test_unimplemented_datastore_methods(self):
ds = Datastore(None)
self.assertRaises(NotImplementedError, ds.put, None)
self.assertRaises(NotImplementedError, ds.delete, None)
def test_unimplemented_user_datastore_methods(self):
self.assertRaises(NotImplementedError, self.ds.find_user, None)
self.assertRaises(NotImplementedError, self.ds.find_role, None)
def test_toggle_active(self):
user.active = True
rv = self.ds.toggle_active(user)
self.assertTrue(rv)
self.assertFalse(user.active)
rv = self.ds.toggle_active(user)
self.assertTrue(rv)
self.assertTrue(user.active)
def test_deactivate_user(self):
user.active = True
rv = self.ds.deactivate_user(user)
self.assertTrue(rv)
self.assertFalse(user.active)
def test_activate_user(self):
ds = UserDatastore(None, None)
user.active = False
ds.activate_user(user)
self.assertTrue(user.active)
def test_deactivate_returns_false_if_already_false(self):
user.active = False
self.assertFalse(self.ds.deactivate_user(user))
def test_activate_returns_false_if_already_true(self):
user.active = True
self.assertFalse(self.ds.activate_user(user))
+84
View File
@@ -0,0 +1,84 @@
# -*- coding: utf-8 -*-
"""
utils
~~~~~
Test utils
"""
from flask import Response as BaseResponse, json
from flask_security import Security
from flask_security.utils import encrypt_password
_missing = object
def authenticate(client, email="matt@lp.com", password="password", endpoint=None, **kwargs):
data = dict(email=email, password=password, remember='y')
return client.post(endpoint or '/login', data=data, **kwargs)
def json_authenticate(client, email="matt@lp.com", password="password", endpoint=None):
data = '{"email": "%s", "password": "%s"}' % (email, password)
return client.post(endpoint or '/login', content_type="application/json", data=data)
def logout(client, endpoint=None, **kwargs):
return client.get(endpoint or '/logout', **kwargs)
def create_roles(ds):
for role in ('admin', 'editor', 'author'):
ds.create_role(name=role)
ds.commit()
def create_users(ds, count=None):
users = [('matt@lp.com', 'matt', 'password', ['admin'], True),
('joe@lp.com', 'joe', 'password', ['editor'], True),
('dave@lp.com', 'dave', 'password', ['admin', 'editor'], True),
('jill@lp.com', 'jill', 'password', ['author'], True),
('tiya@lp.com', 'tiya', 'password', [], False),
('jess@lp.com', 'jess', None, [], True)]
count = count or len(users)
for u in users[:count]:
pw = u[2]
if pw is not None:
pw = encrypt_password(pw)
roles = [ds.find_or_create_role(rn) for rn in u[3]]
ds.commit()
user = ds.create_user(email=u[0], username=u[1], password=pw, active=u[4])
ds.commit()
for role in roles:
ds.add_role_to_user(user, role)
ds.commit()
def populate_data(app, user_count=None):
ds = app.security.datastore
with app.app_context():
create_roles(ds)
create_users(ds, user_count)
class Response(BaseResponse): # pragma: no cover
@property
def jdata(self):
rv = getattr(self, '_cached_jdata', _missing)
if rv is not _missing:
return rv
try:
self._cached_jdata = json.loads(self.data)
except ValueError:
raise Exception('Invalid JSON response')
return self._cached_jdata
def init_app_with_options(app, datastore, **options):
security_args = options.pop('security_args', {})
app.config.update(**options)
app.security = Security(app, datastore=datastore, **security_args)
populate_data(app)
+4 -7
View File
@@ -3,11 +3,8 @@ envlist = py26, py27, py33, pypy
[testenv]
deps =
nose
simplejson
Flask-SQLAlchemy
Flask-MongoEngine
Flask-Peewee
bcrypt
-r{toxinidir}/requirements.txt
-r{toxinidir}/requirements-dev.txt
commands = nosetests -xs []
commands =
python setup.py test