1201

pending completion

Build # 1201

Build Type

Pull #348

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/inveniosoftware/invenio-accounts/commit/<a class=hub.com/inveniosoftware/invenio-accounts/commit/<a class="double-link" href="https://git"><a class=hub.com/inveniosoftware/invenio-accounts/commit/<a class="double-link" href="https://git"><a class=hub.com/inveniosoftware/invenio-accounts/commit/<a class="double-link" href="https://git"><a class=hub.com/inveniosoftware/invenio-accounts/commit/4aebf60d6bba84dc488fca213274675161fec026">4aebf60d6">&lt;a href=&quot;https://github.com/inveniosoftware/invenio-accounts/commit/</a><a class="double-link" href="https://github.com/inveniosoftware/invenio-accounts/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/inveniosoftware/invenio-accounts/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/inveniosoftware/invenio-accounts/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/inveniosoftware/invenio-accounts/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/inveniosoftware/invenio-accounts/commit/4aebf60d6bba84dc488fca213274675161fec026">4aebf60d6</a><a href="https://github.com/inveniosoftware/invenio-accounts/commit/4aebf60d6bba84dc488fca213274675161fec026">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/inveniosoftware/invenio-accounts/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/inveniosoftware/invenio-accounts/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/inveniosoftware/invenio-accounts/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/inveniosoftware/invenio-accounts/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/inveniosoftware/invenio-accounts/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/inveniosoftware/invenio-accounts/commit/4aebf60d6bba84dc488fca213274675161fec026&quot;&gt;4aebf60d6&lt;/a&gt;&lt;a href=&quot;https://github.com/inveniosoftware/invenio-accounts/commit/4aebf60d6bba84dc488fca213274675161fec026&quot;&gt;&amp;quot... (continued)

Pull Request Pull Request #348: update from maint-1.3.

Run Details

5 of 5 new or added lines in 1 file covered. (100.0%)

1077 of 1124 relevant lines covered (95.82%)

25.52 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.19

/invenio_accounts/views/rest.py

# -*- coding: utf-8 -*-
#
# This file is part of Invenio.
# Copyright (C) 2017-2019 CERN.
#
# Invenio is free software; you can redistribute it and/or modify it
# under the terms of the MIT License; see LICENSE file for more details.

"""REST API user management and authentication."""

from __future__ import absolute_import, print_function

from functools import wraps

from flask import Blueprint, after_this_request, current_app, jsonify
from flask.views import MethodView
from flask_login import login_required
from flask_security import current_user
from flask_security.confirmable import confirm_email_token_status, \
    confirm_user, requires_confirmation
from flask_security.recoverable import reset_password_token_status, \
    update_password
from flask_security.signals import reset_password_instructions_sent
from flask_security.utils import config_value, get_message, login_user, \
    logout_user, send_mail, verify_and_update_password
from flask_security.views import logout
from invenio_db import db
from invenio_rest.errors import FieldError, RESTValidationError
from webargs import ValidationError, fields, validate
from webargs.flaskparser import FlaskParser as FlaskParserBase

from invenio_accounts.models import SessionActivity
from invenio_accounts.sessions import delete_session

from ..proxies import current_datastore, current_security
from ..utils import change_user_password, default_confirmation_link_func, \
    default_reset_password_link_func, obj_or_import_string, register_user


def user_already_authenticated(f):
    """Return user if already authenticated."""
    @wraps(f)
    def wrapper(*args, **kwargs):
        if current_user.is_authenticated:
            return jsonify(default_user_payload(current_user))
        return f(*args, **kwargs)
    return wrapper


def role_to_dict(role):
    """Serialize a new role to dict.

    :param role: a new role to serialize into dict.
    :return: dict from role.
    :rtype: dict.
    """
    return dict(
        id=role.id,
        name=role.name,
        description=role.description,
    )


def create_blueprint(app):
    """Conditionally creates the blueprint."""
    blueprint = Blueprint('invenio_accounts_rest_auth', __name__)

    security_state = app.extensions['security']

    if app.config['ACCOUNTS_REST_AUTH_VIEWS']:
        # Resolve the view classes
        authentication_views = {
            k: obj_or_import_string(v)
            for k, v in app.config.get('ACCOUNTS_REST_AUTH_VIEWS', {}).items()
        }

        blueprint.add_url_rule(
            '/login',
            view_func=authentication_views['login'].as_view('login'))
        blueprint.add_url_rule(
            '/logout',
            view_func=authentication_views['logout'].as_view('logout'))
        blueprint.add_url_rule(
            '/me',
            view_func=authentication_views['user_info'].as_view('user_info'))

        if security_state.registerable:
            blueprint.add_url_rule(
                '/register',
                view_func=authentication_views['register'].as_view('register'))

        if security_state.changeable:
            blueprint.add_url_rule(
                '/change-password',
                view_func=authentication_views['change_password'].as_view(
                    'change_password'))

        if security_state.recoverable:
            blueprint.add_url_rule(
                '/forgot-password',
                view_func=authentication_views['forgot_password'].as_view(
                    'forgot_password'))
            blueprint.add_url_rule(
                '/reset-password',
                view_func=authentication_views['reset_password'].as_view(
                    'reset_password'))

        if security_state.confirmable:
            blueprint.add_url_rule(
                '/send-confirmation-email',
                view_func=authentication_views['send_confirmation'].as_view(
                    'send_confirmation'))

            blueprint.add_url_rule(
                '/confirm-email',
                view_func=authentication_views['confirm_email'].as_view(
                    'confirm_email'))

        if app.config['ACCOUNTS_SESSION_ACTIVITY_ENABLED']:
            blueprint.add_url_rule(
                '/sessions',
                view_func=authentication_views['sessions_list'].as_view(
                    'sessions_list'))

            blueprint.add_url_rule(
                '/sessions/<sid_s>',
                view_func=authentication_views['sessions_item'].as_view(
                    'sessions_item'))
    return blueprint


class FlaskParser(FlaskParserBase):
    """Parser to add FieldError to validation errors."""

    def handle_error(self, error, *args, **kwargs):
        """Handle errors during parsing."""
        if isinstance(error, ValidationError):
            _errors = []
            for field, messages in error.messages.items():
                _errors.extend([FieldError(field, msg) for msg in messages])
            raise RESTValidationError(errors=_errors)
        super(FlaskParser, self).handle_error(error, *args, **kwargs)


webargs_parser = FlaskParser()
use_args = webargs_parser.use_args
use_kwargs = webargs_parser.use_kwargs


#
# Field validators
#
def user_exists(email):
    """Validate that a user exists."""
    if not current_datastore.get_user(email):
        raise ValidationError(get_message('USER_DOES_NOT_EXIST')[0])


def unique_user_email(email):
    """Validate unique user email."""
    if current_datastore.get_user(email) is not None:
        raise ValidationError(
            get_message('EMAIL_ALREADY_ASSOCIATED', email=email)[0])


def default_user_payload(user):
    """Parse user payload."""
    return {
        'id': user.id,
        'email': user.email,
        'confirmed_at':
            user.confirmed_at.isoformat() if user.confirmed_at else None,
        'last_login_at':
            user.last_login_at.isoformat() if user.last_login_at else None,
        'roles': [role_to_dict(role) for role in user.roles],
    }


def _abort(message, field=None, status=None):
    if field:
        raise RESTValidationError([FieldError(field, message)])
    raise RESTValidationError(description=message)


def _commit(response=None):
    current_datastore.commit()
    return response


class UserViewMixin(object):
    """Mixin class for get user operations."""

    def get_user(self, email=None, **kwargs):
        """Retrieve a user by the provided arguments."""
        return current_datastore.get_user(email)


class LoginView(MethodView, UserViewMixin):
    """View to login a user."""

    decorators = [user_already_authenticated]

    post_args = {
        'email': fields.Email(required=True, validate=[user_exists]),
        'password': fields.String(required=True)
    }

    def success_response(self, user):
        """Return a successful login response."""
        return jsonify(default_user_payload(user))

    def verify_login(self, user, password=None, **kwargs):
        """Verify the login via password."""
        if not user.password:
            _abort(get_message('PASSWORD_NOT_SET')[0], 'password')
        if not verify_and_update_password(password, user):
            _abort(get_message('INVALID_PASSWORD')[0], 'password')
        if requires_confirmation(user):
            _abort(get_message('CONFIRMATION_REQUIRED')[0])
        if not user.is_active:
            _abort(get_message('DISABLED_ACCOUNT')[0])

    def login_user(self, user):
        """Perform any login actions."""
        return login_user(user)

    @use_kwargs(post_args)
    def post(self, **kwargs):
        """Verify and login a user."""
        user = self.get_user(**kwargs)
        self.verify_login(user, **kwargs)
        self.login_user(user)
        return self.success_response(user)


class UserInfoView(MethodView):
    """View to fetch info from current user."""

    decorators = [login_required]

    def success_response(self, user):
        """Return a successful user info response."""
        return jsonify(default_user_payload(user))

    def get(self):
        """Return user info."""
        return self.success_response(current_user)


class LogoutView(MethodView):
    """View to logout a user."""

    def logout_user_without_post_redirect(self):
        """Perform any logout actions."""
        if current_user.is_authenticated:
            logout_user()

    def success_response(self):
        """Return a successful logout response."""
        return jsonify({'message': 'User logged out.'})

    def post(self):
        """Logout a user."""
        self.logout_user_without_post_redirect()
        return self.success_response()

    def get(self):
        """Logout user."""
        return logout()


class RegisterView(MethodView):
    """View to register a new user."""

    decorators = [user_already_authenticated]

    post_args = {
        'email': fields.Email(required=True, validate=[unique_user_email]),
        'password': fields.String(
            required=True, validate=[validate.Length(min=6, max=128)])
    }

    def login_user(self, user):
        """Perform any login actions."""
        if not current_security.confirmable or \
                current_security.login_without_confirmation:
            after_this_request(_commit)
            login_user(user)

    def success_response(self, user):
        """Return a successful register response."""
        return jsonify(default_user_payload(user))

    @use_kwargs(post_args)
    def post(self, **kwargs):
        """Register a user."""
        user = register_user(**kwargs)
        self.login_user(user)
        return self.success_response(user)


class ForgotPasswordView(MethodView, UserViewMixin):
    """View to get a link to reset the user password."""

    decorators = [user_already_authenticated]

    reset_password_link_func = default_reset_password_link_func

    post_args = {
        'email': fields.Email(required=True, validate=[user_exists]),
    }

    @classmethod
    def send_reset_password_instructions(cls, user):
        """Send email containing instructions to reset password."""
        token, reset_link = cls.reset_password_link_func(user)
        if config_value('SEND_PASSWORD_RESET_EMAIL'):
            send_mail(config_value('EMAIL_SUBJECT_PASSWORD_RESET'), user.email,
                      'reset_instructions', user=user, reset_link=reset_link)
            reset_password_instructions_sent.send(
                current_app._get_current_object(), user=user, token=token)

    def success_response(self, user):
        """Return a response containing reset password instructions."""
        return jsonify({'message': get_message(
            'PASSWORD_RESET_REQUEST', email=user.email)[0]})

    @use_kwargs(post_args)
    def post(self, **kwargs):
        """Send reset password instructions."""
        user = self.get_user(**kwargs)
        self.send_reset_password_instructions(user)
        return self.success_response(user)


class ResetPasswordView(MethodView):
    """View to reset the user password."""

    decorators = [user_already_authenticated]

    post_args = {
        'token': fields.String(required=True),
        'password': fields.String(
            required=True, validate=[validate.Length(min=6, max=128)]),
    }

    def get_user(self, token=None, **kwargs):
        """Retrieve a user by the provided arguments."""
        # Verify the token
        expired, invalid, user = reset_password_token_status(token)
        if invalid:
            _abort(get_message('INVALID_RESET_PASSWORD_TOKEN')[0])
        if expired:
            _abort(get_message(
                'PASSWORD_RESET_EXPIRED',
                email=user.email,
                within=current_security.reset_password_within)[0])
        return user

    def success_response(self, user):
        """Return a successful reset password response."""
        return jsonify({'message': get_message('PASSWORD_RESET')[0]})

    @use_kwargs(post_args)
    def post(self, **kwargs):
        """Reset user password."""
        user = self.get_user(**kwargs)
        after_this_request(_commit)
        update_password(user, kwargs['password'])
        login_user(user)
        return self.success_response(user)


class ChangePasswordView(MethodView):
    """View to change the user password."""

    decorators = [login_required]

    post_args = {
        'password': fields.String(
            required=True, validate=[validate.Length(min=6, max=128)]),
        'new_password': fields.String(
            required=True, validate=[validate.Length(min=6, max=128)]),
    }

    def verify_password(self, password=None, new_password=None, **kwargs):
        """Verify password is not invalid."""
        if not verify_and_update_password(password, current_user):
            _abort(get_message('INVALID_PASSWORD')[0], 'password')
        if password == new_password:
            _abort(get_message('PASSWORD_IS_THE_SAME')[0], 'password')

    def change_password(self, new_password=None, **kwargs):
        """Perform any change password actions."""
        after_this_request(_commit)
        change_user_password(user=current_user._get_current_object(),
                             password=new_password)

    def success_response(self):
        """Return a successful change password response."""
        return jsonify({'message': get_message('PASSWORD_CHANGE')[0]})

    @use_kwargs(post_args)
    def post(self, **kwargs):
        """Change user password."""
        self.verify_password(**kwargs)
        self.change_password(**kwargs)
        return self.success_response()


class SendConfirmationEmailView(MethodView, UserViewMixin):
    """View function which sends confirmation instructions."""

    decorators = [login_required]

    confirmation_link_func = default_confirmation_link_func

    post_args = {
        'email': fields.Email(required=True, validate=[user_exists]),
    }

    def verify(self, user):
        """Verify that user is not confirmed."""
        if user.confirmed_at is not None:
            _abort(get_message('ALREADY_CONFIRMED')[0])

    @classmethod
    def send_confirmation_link(cls, user):
        """Send confirmation link."""
        send_email_enabled = current_security.confirmable and \
            config_value('SEND_REGISTER_EMAIL')
        if send_email_enabled:
            token, confirmation_link = cls.confirmation_link_func(user)
            send_mail(
                config_value('EMAIL_SUBJECT_REGISTER'), user.email,
                'confirmation_instructions', user=user,
                confirmation_link=confirmation_link)
            return token

    def success_response(self, user):
        """Return a successful confirmation email sent response."""
        return jsonify({
            'message': get_message('CONFIRMATION_REQUEST', email=user.email)[0]
        })

    @use_kwargs(post_args)
    def post(self, **kwargs):
        """Send confirmation email."""
        user = self.get_user(**kwargs)
        self.verify(user)
        self.send_confirmation_link(user)
        return self.success_response(user)


class ConfirmEmailView(MethodView):
    """View that handles a email confirmation request."""

    post_args = {
        'token': fields.String(required=True),
    }

    def get_user(self, token=None, **kwargs):
        """Retrieve a user by the provided arguments."""
        expired, invalid, user = confirm_email_token_status(token)

        if not user or invalid:
            _abort(get_message('INVALID_CONFIRMATION_TOKEN'))

        already_confirmed = user is not None and user.confirmed_at is not None
        if expired and not already_confirmed:
            _abort(get_message(
                'CONFIRMATION_EXPIRED',
                email=user.email,
                within=current_security.confirm_email_within))
        return user

    @use_kwargs(post_args)
    def post(self, **kwargs):
        """Confirm user email."""
        user = self.get_user(**kwargs)

        if user != current_user:
            logout_user()

        if confirm_user(user):
            after_this_request(_commit)
            return jsonify({'message': get_message('EMAIL_CONFIRMED')[0]})
        else:
            return jsonify({'message': get_message('ALREADY_CONFIRMED')[0]})


class SessionsListView(MethodView):
    """View that returns the list of user sessions."""

    decorators = [login_required]

    def get(self, **kwargs):
        """Return user sessions info."""
        sessions = SessionActivity.query_by_user(
            user_id=current_user.get_id())
        results = [{
            'created': s.created,
            'current': SessionActivity.is_current(s.sid_s),
            'browser': s.browser,
            'browser_version':  s.browser_version,
            'os': s.os,
            'device': s.device,
            'country': s.country} for s in sessions]

        return jsonify({'total': sessions.count(), 'results': results})


class SessionsItemView(MethodView):
    """View for operations related to user session."""

    decorators = [login_required]

    def delete(self, sid_s=None, **kwargs):
        """Revoke the given user session."""
        if SessionActivity.query_by_user(current_user.get_id()) \
                .filter_by(sid_s=sid_s).count() == 1:
            delete_session(sid_s=sid_s)
            db.session.commit()
            message = 'Session {0} successfully removed. {1}.'
            if SessionActivity.is_current(sid_s=sid_s):
                message = message.format(sid_s, "Logged out")
            else:
                message = message.format(sid_s, "Revoked")
            return jsonify({'message': message})
        else:
            return jsonify({
                'message': 'Unable to remove session {0}.'.format(sid_s)}), 400

1	# -- coding: utf-8 --
2	#
3	# This file is part of Invenio.
4	# Copyright (C) 2017-2019 CERN.
5	#
6	# Invenio is free software; you can redistribute it and/or modify it
7	# under the terms of the MIT License; see LICENSE file for more details.
8
9	"""REST API user management and authentication."""	27✔
10
11	from __future__ import absolute_import, print_function	27✔
12
13	from functools import wraps	27✔
14
15	from flask import Blueprint, after_this_request, current_app, jsonify	27✔
16	from flask.views import MethodView	27✔
17	from flask_login import login_required	27✔
18	from flask_security import current_user	27✔
19	from flask_security.confirmable import confirm_email_token_status, \	27✔
20	confirm_user, requires_confirmation
21	from flask_security.recoverable import reset_password_token_status, \	27✔
22	update_password
23	from flask_security.signals import reset_password_instructions_sent	27✔
24	from flask_security.utils import config_value, get_message, login_user, \	27✔
25	logout_user, send_mail, verify_and_update_password
26	from flask_security.views import logout	27✔
27	from invenio_db import db	27✔
28	from invenio_rest.errors import FieldError, RESTValidationError	27✔
29	from webargs import ValidationError, fields, validate	27✔
30	from webargs.flaskparser import FlaskParser as FlaskParserBase	27✔
31
32	from invenio_accounts.models import SessionActivity	27✔
33	from invenio_accounts.sessions import delete_session	27✔
34
35	from ..proxies import current_datastore, current_security	27✔
36	from ..utils import change_user_password, default_confirmation_link_func, \	27✔
37	default_reset_password_link_func, obj_or_import_string, register_user
38
39
40	def user_already_authenticated(f):	27✔
41	"""Return user if already authenticated."""
42	@wraps(f)	27✔
43	def wrapper(args, *kwargs):	9✔
44	if current_user.is_authenticated:	27✔
45	return jsonify(default_user_payload(current_user))	27✔
46	return f(args, *kwargs)	27✔
47	return wrapper	27✔
48
49
50	def role_to_dict(role):	27✔
51	"""Serialize a new role to dict.
52
53	:param role: a new role to serialize into dict.
54	:return: dict from role.
55	:rtype: dict.
56	"""
57	return dict(	×
58	id=role.id,
59	name=role.name,
60	description=role.description,
61	)
62
63
64	def create_blueprint(app):	27✔
65	"""Conditionally creates the blueprint."""
66	blueprint = Blueprint('invenio_accounts_rest_auth', __name__)	27✔
67
68	security_state = app.extensions['security']	27✔
69
70	if app.config['ACCOUNTS_REST_AUTH_VIEWS']:	27✔
71	# Resolve the view classes
72	authentication_views = {	27✔
73	k: obj_or_import_string(v)
74	for k, v in app.config.get('ACCOUNTS_REST_AUTH_VIEWS', {}).items()
75	}
76
77	blueprint.add_url_rule(	27✔
78	'/login',
79	view_func=authentication_views['login'].as_view('login'))
80	blueprint.add_url_rule(	27✔
81	'/logout',
82	view_func=authentication_views['logout'].as_view('logout'))
83	blueprint.add_url_rule(	27✔
84	'/me',
85	view_func=authentication_views['user_info'].as_view('user_info'))
86
87	if security_state.registerable:	27✔
88	blueprint.add_url_rule(	27✔
89	'/register',
90	view_func=authentication_views['register'].as_view('register'))
91
92	if security_state.changeable:	27✔
93	blueprint.add_url_rule(	27✔
94	'/change-password',
95	view_func=authentication_views['change_password'].as_view(
96	'change_password'))
97
98	if security_state.recoverable:	27✔
99	blueprint.add_url_rule(	27✔
100	'/forgot-password',
101	view_func=authentication_views['forgot_password'].as_view(
102	'forgot_password'))
103	blueprint.add_url_rule(	27✔
104	'/reset-password',
105	view_func=authentication_views['reset_password'].as_view(
106	'reset_password'))
107
108	if security_state.confirmable:	27✔
109	blueprint.add_url_rule(	27✔
110	'/send-confirmation-email',
111	view_func=authentication_views['send_confirmation'].as_view(
112	'send_confirmation'))
113
114	blueprint.add_url_rule(	27✔
115	'/confirm-email',
116	view_func=authentication_views['confirm_email'].as_view(
117	'confirm_email'))
118
119	if app.config['ACCOUNTS_SESSION_ACTIVITY_ENABLED']:	27✔
120	blueprint.add_url_rule(	27✔
121	'/sessions',
122	view_func=authentication_views['sessions_list'].as_view(
123	'sessions_list'))
124
125	blueprint.add_url_rule(	27✔
126	'/sessions/<sid_s>',
127	view_func=authentication_views['sessions_item'].as_view(
128	'sessions_item'))
129	return blueprint	27✔
130
131
132	class FlaskParser(FlaskParserBase):	27✔
133	"""Parser to add FieldError to validation errors."""
134
135	def handle_error(self, error, args, *kwargs):	27✔
136	"""Handle errors during parsing."""
137	if isinstance(error, ValidationError):	27✔
138	_errors = []	27✔
139	for field, messages in error.messages.items():	27✔
140	_errors.extend([FieldError(field, msg) for msg in messages])	27✔
141	raise RESTValidationError(errors=_errors)	27✔
142	super(FlaskParser, self).handle_error(error, args, *kwargs)	×
143
144
145	webargs_parser = FlaskParser()	27✔
146	use_args = webargs_parser.use_args	27✔
147	use_kwargs = webargs_parser.use_kwargs	27✔
148
149
150	#
151	# Field validators
152	#
153	def user_exists(email):	27✔
154	"""Validate that a user exists."""
155	if not current_datastore.get_user(email):	27✔
156	raise ValidationError(get_message('USER_DOES_NOT_EXIST')[0])	27✔
157
158
159	def unique_user_email(email):	27✔
160	"""Validate unique user email."""
161	if current_datastore.get_user(email) is not None:	27✔
162	raise ValidationError(	27✔
163	get_message('EMAIL_ALREADY_ASSOCIATED', email=email)[0])
164
165
166	def default_user_payload(user):	27✔
167	"""Parse user payload."""
168	return {	27✔
169	'id': user.id,
170	'email': user.email,
171	'confirmed_at':
172	user.confirmed_at.isoformat() if user.confirmed_at else None,
173	'last_login_at':
174	user.last_login_at.isoformat() if user.last_login_at else None,
175	'roles': [role_to_dict(role) for role in user.roles],
176	}
177
178
179	def _abort(message, field=None, status=None):	27✔
180	if field:	27✔
181	raise RESTValidationError([FieldError(field, message)])	27✔
182	raise RESTValidationError(description=message)	27✔
183
184
185	def _commit(response=None):	27✔
186	current_datastore.commit()	27✔
187	return response	27✔
188
189
190	class UserViewMixin(object):	27✔
191	"""Mixin class for get user operations."""
192
193	def get_user(self, email=None, **kwargs):	27✔
194	"""Retrieve a user by the provided arguments."""
195	return current_datastore.get_user(email)	27✔
196
197
198	class LoginView(MethodView, UserViewMixin):	27✔
199	"""View to login a user."""
200
201	decorators = [user_already_authenticated]	27✔
202
203	post_args = {	27✔
204	'email': fields.Email(required=True, validate=[user_exists]),
205	'password': fields.String(required=True)
206	}
207
208	def success_response(self, user):	27✔
209	"""Return a successful login response."""
210	return jsonify(default_user_payload(user))	27✔
211
212	def verify_login(self, user, password=None, **kwargs):	27✔
213	"""Verify the login via password."""
214	if not user.password:	27✔
215	_abort(get_message('PASSWORD_NOT_SET')[0], 'password')	27✔
216	if not verify_and_update_password(password, user):	27✔
217	_abort(get_message('INVALID_PASSWORD')[0], 'password')	×
218	if requires_confirmation(user):	27✔
219	_abort(get_message('CONFIRMATION_REQUIRED')[0])	×
220	if not user.is_active:	27✔
221	_abort(get_message('DISABLED_ACCOUNT')[0])	27✔
222
223	def login_user(self, user):	27✔
224	"""Perform any login actions."""
225	return login_user(user)	27✔
226
227	@use_kwargs(post_args)	27✔
228	def post(self, **kwargs):	9✔
229	"""Verify and login a user."""
230	user = self.get_user(**kwargs)	27✔
231	self.verify_login(user, **kwargs)	27✔
232	self.login_user(user)	27✔
233	return self.success_response(user)	27✔
234
235
236	class UserInfoView(MethodView):	27✔
237	"""View to fetch info from current user."""
238
239	decorators = [login_required]	27✔
240
241	def success_response(self, user):	27✔
242	"""Return a successful user info response."""
243	return jsonify(default_user_payload(user))	27✔
244
245	def get(self):	27✔
246	"""Return user info."""
247	return self.success_response(current_user)	27✔
248
249
250	class LogoutView(MethodView):	27✔
251	"""View to logout a user."""
252
253	def logout_user_without_post_redirect(self):	27✔
254	"""Perform any logout actions."""
255	if current_user.is_authenticated:	27✔
256	logout_user()	27✔
257
258	def success_response(self):	27✔
259	"""Return a successful logout response."""
260	return jsonify({'message': 'User logged out.'})	27✔
261
262	def post(self):	27✔
263	"""Logout a user."""
264	self.logout_user_without_post_redirect()	27✔
265	return self.success_response()	27✔
266
267	def get(self):	27✔
268	"""Logout user."""
269	return logout()	×
270
271
272	class RegisterView(MethodView):	27✔
273	"""View to register a new user."""
274
275	decorators = [user_already_authenticated]	27✔
276
277	post_args = {	27✔
278	'email': fields.Email(required=True, validate=[unique_user_email]),
279	'password': fields.String(
280	required=True, validate=[validate.Length(min=6, max=128)])
281	}
282
283	def login_user(self, user):	27✔
284	"""Perform any login actions."""
285	if not current_security.confirmable or \	27✔
286	current_security.login_without_confirmation:
287	after_this_request(_commit)	27✔
288	login_user(user)	27✔
289
290	def success_response(self, user):	27✔
291	"""Return a successful register response."""
292	return jsonify(default_user_payload(user))	27✔
293
294	@use_kwargs(post_args)	27✔
295	def post(self, **kwargs):	9✔
296	"""Register a user."""
297	user = register_user(**kwargs)	27✔
298	self.login_user(user)	27✔
299	return self.success_response(user)	27✔
300
301
302	class ForgotPasswordView(MethodView, UserViewMixin):	27✔
303	"""View to get a link to reset the user password."""
304
305	decorators = [user_already_authenticated]	27✔
306
307	reset_password_link_func = default_reset_password_link_func	27✔
308
309	post_args = {	27✔
310	'email': fields.Email(required=True, validate=[user_exists]),
311	}
312
313	@classmethod	27✔
314	def send_reset_password_instructions(cls, user):	9✔
315	"""Send email containing instructions to reset password."""
316	token, reset_link = cls.reset_password_link_func(user)	27✔
317	if config_value('SEND_PASSWORD_RESET_EMAIL'):	27✔
318	send_mail(config_value('EMAIL_SUBJECT_PASSWORD_RESET'), user.email,	27✔
319	'reset_instructions', user=user, reset_link=reset_link)
320	reset_password_instructions_sent.send(	27✔
321	current_app._get_current_object(), user=user, token=token)
322
323	def success_response(self, user):	27✔
324	"""Return a response containing reset password instructions."""
325	return jsonify({'message': get_message(	27✔
326	'PASSWORD_RESET_REQUEST', email=user.email)[0]})
327
328	@use_kwargs(post_args)	27✔
329	def post(self, **kwargs):	9✔
330	"""Send reset password instructions."""
331	user = self.get_user(**kwargs)	27✔
332	self.send_reset_password_instructions(user)	27✔
333	return self.success_response(user)	27✔
334
335
336	class ResetPasswordView(MethodView):	27✔
337	"""View to reset the user password."""
338
339	decorators = [user_already_authenticated]	27✔
340
341	post_args = {	27✔
342	'token': fields.String(required=True),
343	'password': fields.String(
344	required=True, validate=[validate.Length(min=6, max=128)]),
345	}
346
347	def get_user(self, token=None, **kwargs):	27✔
348	"""Retrieve a user by the provided arguments."""
349	# Verify the token
350	expired, invalid, user = reset_password_token_status(token)	27✔
351	if invalid:	27✔
352	_abort(get_message('INVALID_RESET_PASSWORD_TOKEN')[0])	×
353	if expired:	27✔
354	_abort(get_message(	×
355	'PASSWORD_RESET_EXPIRED',
356	email=user.email,
357	within=current_security.reset_password_within)[0])
358	return user	27✔
359
360	def success_response(self, user):	27✔
361	"""Return a successful reset password response."""
362	return jsonify({'message': get_message('PASSWORD_RESET')[0]})	27✔
363
364	@use_kwargs(post_args)	27✔
365	def post(self, **kwargs):	9✔
366	"""Reset user password."""
367	user = self.get_user(**kwargs)	27✔
368	after_this_request(_commit)	27✔
369	update_password(user, kwargs['password'])	27✔
370	login_user(user)	27✔
371	return self.success_response(user)	27✔
372
373
374	class ChangePasswordView(MethodView):	27✔
375	"""View to change the user password."""
376
377	decorators = [login_required]	27✔
378
379	post_args = {	27✔
380	'password': fields.String(
381	required=True, validate=[validate.Length(min=6, max=128)]),
382	'new_password': fields.String(
383	required=True, validate=[validate.Length(min=6, max=128)]),
384	}
385
386	def verify_password(self, password=None, new_password=None, **kwargs):	27✔
387	"""Verify password is not invalid."""
388	if not verify_and_update_password(password, current_user):	27✔
389	_abort(get_message('INVALID_PASSWORD')[0], 'password')	×
390	if password == new_password:	27✔
391	_abort(get_message('PASSWORD_IS_THE_SAME')[0], 'password')	27✔
392
393	def change_password(self, new_password=None, **kwargs):	27✔
394	"""Perform any change password actions."""
395	after_this_request(_commit)	27✔
396	change_user_password(user=current_user._get_current_object(),	27✔
397	password=new_password)
398
399	def success_response(self):	27✔
400	"""Return a successful change password response."""
401	return jsonify({'message': get_message('PASSWORD_CHANGE')[0]})	27✔
402
403	@use_kwargs(post_args)	27✔
404	def post(self, **kwargs):	9✔
405	"""Change user password."""
406	self.verify_password(**kwargs)	27✔
407	self.change_password(**kwargs)	27✔
408	return self.success_response()	27✔
409
410
411	class SendConfirmationEmailView(MethodView, UserViewMixin):	27✔
412	"""View function which sends confirmation instructions."""
413
414	decorators = [login_required]	27✔
415
416	confirmation_link_func = default_confirmation_link_func	27✔
417
418	post_args = {	27✔
419	'email': fields.Email(required=True, validate=[user_exists]),
420	}
421
422	def verify(self, user):	27✔
423	"""Verify that user is not confirmed."""
424	if user.confirmed_at is not None:	27✔
425	_abort(get_message('ALREADY_CONFIRMED')[0])	27✔
426
427	@classmethod	27✔
428	def send_confirmation_link(cls, user):	9✔
429	"""Send confirmation link."""
430	send_email_enabled = current_security.confirmable and \	27✔
431	config_value('SEND_REGISTER_EMAIL')
432	if send_email_enabled:	27✔
433	token, confirmation_link = cls.confirmation_link_func(user)	27✔
434	send_mail(	27✔
435	config_value('EMAIL_SUBJECT_REGISTER'), user.email,
436	'confirmation_instructions', user=user,
437	confirmation_link=confirmation_link)
438	return token	27✔
439
440	def success_response(self, user):	27✔
441	"""Return a successful confirmation email sent response."""
442	return jsonify({	27✔
443	'message': get_message('CONFIRMATION_REQUEST', email=user.email)[0]
444	})
445
446	@use_kwargs(post_args)	27✔
447	def post(self, **kwargs):	9✔
448	"""Send confirmation email."""
449	user = self.get_user(**kwargs)	27✔
450	self.verify(user)	27✔
451	self.send_confirmation_link(user)	27✔
452	return self.success_response(user)	27✔
453
454
455	class ConfirmEmailView(MethodView):	27✔
456	"""View that handles a email confirmation request."""
457
458	post_args = {	27✔
459	'token': fields.String(required=True),
460	}
461
462	def get_user(self, token=None, **kwargs):	27✔
463	"""Retrieve a user by the provided arguments."""
464	expired, invalid, user = confirm_email_token_status(token)	27✔
465
466	if not user or invalid:	27✔
467	_abort(get_message('INVALID_CONFIRMATION_TOKEN'))	27✔
468
469	already_confirmed = user is not None and user.confirmed_at is not None	27✔
470	if expired and not already_confirmed:	27✔
471	_abort(get_message(	×
472	'CONFIRMATION_EXPIRED',
473	email=user.email,
474	within=current_security.confirm_email_within))
475	return user	27✔
476
477	@use_kwargs(post_args)	27✔
478	def post(self, **kwargs):	9✔
479	"""Confirm user email."""
480	user = self.get_user(**kwargs)	27✔
481
482	if user != current_user:	27✔
483	logout_user()	27✔
484
485	if confirm_user(user):	27✔
486	after_this_request(_commit)	27✔
487	return jsonify({'message': get_message('EMAIL_CONFIRMED')[0]})	27✔
488	else:
489	return jsonify({'message': get_message('ALREADY_CONFIRMED')[0]})	27✔
490
491
492	class SessionsListView(MethodView):	27✔
493	"""View that returns the list of user sessions."""
494
495	decorators = [login_required]	27✔
496
497	def get(self, **kwargs):	27✔
498	"""Return user sessions info."""
499	sessions = SessionActivity.query_by_user(	27✔
500	user_id=current_user.get_id())
501	results = [{	27✔
502	'created': s.created,
503	'current': SessionActivity.is_current(s.sid_s),
504	'browser': s.browser,
505	'browser_version': s.browser_version,
506	'os': s.os,
507	'device': s.device,
508	'country': s.country} for s in sessions]
509
510	return jsonify({'total': sessions.count(), 'results': results})	27✔
511
512
513	class SessionsItemView(MethodView):	27✔
514	"""View for operations related to user session."""
515
516	decorators = [login_required]	27✔
517
518	def delete(self, sid_s=None, **kwargs):	27✔
519	"""Revoke the given user session."""
520	if SessionActivity.query_by_user(current_user.get_id()) \	27✔
521	.filter_by(sid_s=sid_s).count() == 1:
522	delete_session(sid_s=sid_s)	27✔
523	db.session.commit()	27✔
524	message = 'Session {0} successfully removed. {1}.'	27✔
525	if SessionActivity.is_current(sid_s=sid_s):	27✔
526	message = message.format(sid_s, "Logged out")	27✔
527	else:
528	message = message.format(sid_s, "Revoked")	27✔
529	return jsonify({'message': message})	27✔
530	else:
531	return jsonify({	27✔
532	'message': 'Unable to remove session {0}.'.format(sid_s)}), 400

inveniosoftware / invenio-accounts / 1201

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous