1196

pending completion

Build # 1196

Build Type

Pull #347

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/7d9b15866c7d03efa09d58bcae2fba9bf4cf4a43">7d9b15866">&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/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/inveniosoftware/invenio-accounts/commit/7d9b15866c7d03efa09d58bcae2fba9bf4cf4a43">7d9b15866</a><a href="https://github.com/inveniosoftware/invenio-accounts/commit/7d9b15866c7d03efa09d58bcae2fba9bf4cf4a43">&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/7d9b15866c7d03efa09d58bcae2fba9bf4cf4a43&quot;&gt;7d9b15866&lt;/a&gt;&lt;a href=&quot;https://github.com/inveniosoftware/invenio-accounts/commit/7d9b15866c7d03efa09d58bcae2fba9bf4cf4a43&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com... (continued)

Pull Request Pull Request #347: rest: add get method for logout

Run Details

7 of 7 new or added lines in 2 files covered. (100.0%)

1070 of 1117 relevant lines covered (95.79%)

25.64 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, validate=[validate.Length(min=6, max=128)])
    }

    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(
206	required=True, validate=[validate.Length(min=6, max=128)])
207	}
208
209	def success_response(self, user):	27✔
210	"""Return a successful login response."""
211	return jsonify(default_user_payload(user))	27✔
212
213	def verify_login(self, user, password=None, **kwargs):	27✔
214	"""Verify the login via password."""
215	if not user.password:	27✔
216	_abort(get_message('PASSWORD_NOT_SET')[0], 'password')	27✔
217	if not verify_and_update_password(password, user):	27✔
218	_abort(get_message('INVALID_PASSWORD')[0], 'password')	×
219	if requires_confirmation(user):	27✔
220	_abort(get_message('CONFIRMATION_REQUIRED')[0])	×
221	if not user.is_active:	27✔
222	_abort(get_message('DISABLED_ACCOUNT')[0])	27✔
223
224	def login_user(self, user):	27✔
225	"""Perform any login actions."""
226	return login_user(user)	27✔
227
228	@use_kwargs(post_args)	27✔
229	def post(self, **kwargs):	9✔
230	"""Verify and login a user."""
231	user = self.get_user(**kwargs)	27✔
232	self.verify_login(user, **kwargs)	27✔
233	self.login_user(user)	27✔
234	return self.success_response(user)	27✔
235
236
237	class UserInfoView(MethodView):	27✔
238	"""View to fetch info from current user."""
239
240	decorators = [login_required]	27✔
241
242	def success_response(self, user):	27✔
243	"""Return a successful user info response."""
244	return jsonify(default_user_payload(user))	27✔
245
246	def get(self):	27✔
247	"""Return user info."""
248	return self.success_response(current_user)	27✔
249
250
251	class LogoutView(MethodView):	27✔
252	"""View to logout a user."""
253
254	def logout_user_without_post_redirect(self):	27✔
255	"""Perform any logout actions."""
256	if current_user.is_authenticated:	27✔
257	logout_user()	27✔
258
259	def success_response(self):	27✔
260	"""Return a successful logout response."""
261	return jsonify({'message': 'User logged out.'})	27✔
262
263	def post(self):	27✔
264	"""Logout a user."""
265	self.logout_user_without_post_redirect()	27✔
266	return self.success_response()	27✔
267
268	def get(self):	27✔
269	"""Logout user."""
270	return logout()	×
271
272
273	class RegisterView(MethodView):	27✔
274	"""View to register a new user."""
275
276	decorators = [user_already_authenticated]	27✔
277
278	post_args = {	27✔
279	'email': fields.Email(required=True, validate=[unique_user_email]),
280	'password': fields.String(
281	required=True, validate=[validate.Length(min=6, max=128)])
282	}
283
284	def login_user(self, user):	27✔
285	"""Perform any login actions."""
286	if not current_security.confirmable or \	27✔
287	current_security.login_without_confirmation:
288	after_this_request(_commit)	27✔
289	login_user(user)	27✔
290
291	def success_response(self, user):	27✔
292	"""Return a successful register response."""
293	return jsonify(default_user_payload(user))	27✔
294
295	@use_kwargs(post_args)	27✔
296	def post(self, **kwargs):	9✔
297	"""Register a user."""
298	user = register_user(**kwargs)	27✔
299	self.login_user(user)	27✔
300	return self.success_response(user)	27✔
301
302
303	class ForgotPasswordView(MethodView, UserViewMixin):	27✔
304	"""View to get a link to reset the user password."""
305
306	decorators = [user_already_authenticated]	27✔
307
308	reset_password_link_func = default_reset_password_link_func	27✔
309
310	post_args = {	27✔
311	'email': fields.Email(required=True, validate=[user_exists]),
312	}
313
314	@classmethod	27✔
315	def send_reset_password_instructions(cls, user):	9✔
316	"""Send email containing instructions to reset password."""
317	token, reset_link = cls.reset_password_link_func(user)	27✔
318	if config_value('SEND_PASSWORD_RESET_EMAIL'):	27✔
319	send_mail(config_value('EMAIL_SUBJECT_PASSWORD_RESET'), user.email,	27✔
320	'reset_instructions', user=user, reset_link=reset_link)
321	reset_password_instructions_sent.send(	27✔
322	current_app._get_current_object(), user=user, token=token)
323
324	def success_response(self, user):	27✔
325	"""Return a response containing reset password instructions."""
326	return jsonify({'message': get_message(	27✔
327	'PASSWORD_RESET_REQUEST', email=user.email)[0]})
328
329	@use_kwargs(post_args)	27✔
330	def post(self, **kwargs):	9✔
331	"""Send reset password instructions."""
332	user = self.get_user(**kwargs)	27✔
333	self.send_reset_password_instructions(user)	27✔
334	return self.success_response(user)	27✔
335
336
337	class ResetPasswordView(MethodView):	27✔
338	"""View to reset the user password."""
339
340	decorators = [user_already_authenticated]	27✔
341
342	post_args = {	27✔
343	'token': fields.String(required=True),
344	'password': fields.String(
345	required=True, validate=[validate.Length(min=6, max=128)]),
346	}
347
348	def get_user(self, token=None, **kwargs):	27✔
349	"""Retrieve a user by the provided arguments."""
350	# Verify the token
351	expired, invalid, user = reset_password_token_status(token)	27✔
352	if invalid:	27✔
353	_abort(get_message('INVALID_RESET_PASSWORD_TOKEN')[0])	×
354	if expired:	27✔
355	_abort(get_message(	×
356	'PASSWORD_RESET_EXPIRED',
357	email=user.email,
358	within=current_security.reset_password_within)[0])
359	return user	27✔
360
361	def success_response(self, user):	27✔
362	"""Return a successful reset password response."""
363	return jsonify({'message': get_message('PASSWORD_RESET')[0]})	27✔
364
365	@use_kwargs(post_args)	27✔
366	def post(self, **kwargs):	9✔
367	"""Reset user password."""
368	user = self.get_user(**kwargs)	27✔
369	after_this_request(_commit)	27✔
370	update_password(user, kwargs['password'])	27✔
371	login_user(user)	27✔
372	return self.success_response(user)	27✔
373
374
375	class ChangePasswordView(MethodView):	27✔
376	"""View to change the user password."""
377
378	decorators = [login_required]	27✔
379
380	post_args = {	27✔
381	'password': fields.String(
382	required=True, validate=[validate.Length(min=6, max=128)]),
383	'new_password': fields.String(
384	required=True, validate=[validate.Length(min=6, max=128)]),
385	}
386
387	def verify_password(self, password=None, new_password=None, **kwargs):	27✔
388	"""Verify password is not invalid."""
389	if not verify_and_update_password(password, current_user):	27✔
390	_abort(get_message('INVALID_PASSWORD')[0], 'password')	×
391	if password == new_password:	27✔
392	_abort(get_message('PASSWORD_IS_THE_SAME')[0], 'password')	27✔
393
394	def change_password(self, new_password=None, **kwargs):	27✔
395	"""Perform any change password actions."""
396	after_this_request(_commit)	27✔
397	change_user_password(user=current_user._get_current_object(),	27✔
398	password=new_password)
399
400	def success_response(self):	27✔
401	"""Return a successful change password response."""
402	return jsonify({'message': get_message('PASSWORD_CHANGE')[0]})	27✔
403
404	@use_kwargs(post_args)	27✔
405	def post(self, **kwargs):	9✔
406	"""Change user password."""
407	self.verify_password(**kwargs)	27✔
408	self.change_password(**kwargs)	27✔
409	return self.success_response()	27✔
410
411
412	class SendConfirmationEmailView(MethodView, UserViewMixin):	27✔
413	"""View function which sends confirmation instructions."""
414
415	decorators = [login_required]	27✔
416
417	confirmation_link_func = default_confirmation_link_func	27✔
418
419	post_args = {	27✔
420	'email': fields.Email(required=True, validate=[user_exists]),
421	}
422
423	def verify(self, user):	27✔
424	"""Verify that user is not confirmed."""
425	if user.confirmed_at is not None:	27✔
426	_abort(get_message('ALREADY_CONFIRMED')[0])	27✔
427
428	@classmethod	27✔
429	def send_confirmation_link(cls, user):	9✔
430	"""Send confirmation link."""
431	send_email_enabled = current_security.confirmable and \	27✔
432	config_value('SEND_REGISTER_EMAIL')
433	if send_email_enabled:	27✔
434	token, confirmation_link = cls.confirmation_link_func(user)	27✔
435	send_mail(	27✔
436	config_value('EMAIL_SUBJECT_REGISTER'), user.email,
437	'confirmation_instructions', user=user,
438	confirmation_link=confirmation_link)
439	return token	27✔
440
441	def success_response(self, user):	27✔
442	"""Return a successful confirmation email sent response."""
443	return jsonify({	27✔
444	'message': get_message('CONFIRMATION_REQUEST', email=user.email)[0]
445	})
446
447	@use_kwargs(post_args)	27✔
448	def post(self, **kwargs):	9✔
449	"""Send confirmation email."""
450	user = self.get_user(**kwargs)	27✔
451	self.verify(user)	27✔
452	self.send_confirmation_link(user)	27✔
453	return self.success_response(user)	27✔
454
455
456	class ConfirmEmailView(MethodView):	27✔
457	"""View that handles a email confirmation request."""
458
459	post_args = {	27✔
460	'token': fields.String(required=True),
461	}
462
463	def get_user(self, token=None, **kwargs):	27✔
464	"""Retrieve a user by the provided arguments."""
465	expired, invalid, user = confirm_email_token_status(token)	27✔
466
467	if not user or invalid:	27✔
468	_abort(get_message('INVALID_CONFIRMATION_TOKEN'))	27✔
469
470	already_confirmed = user is not None and user.confirmed_at is not None	27✔
471	if expired and not already_confirmed:	27✔
472	_abort(get_message(	×
473	'CONFIRMATION_EXPIRED',
474	email=user.email,
475	within=current_security.confirm_email_within))
476	return user	27✔
477
478	@use_kwargs(post_args)	27✔
479	def post(self, **kwargs):	9✔
480	"""Confirm user email."""
481	user = self.get_user(**kwargs)	27✔
482
483	if user != current_user:	27✔
484	logout_user()	27✔
485
486	if confirm_user(user):	27✔
487	after_this_request(_commit)	27✔
488	return jsonify({'message': get_message('EMAIL_CONFIRMED')[0]})	27✔
489	else:
490	return jsonify({'message': get_message('ALREADY_CONFIRMED')[0]})	27✔
491
492
493	class SessionsListView(MethodView):	27✔
494	"""View that returns the list of user sessions."""
495
496	decorators = [login_required]	27✔
497
498	def get(self, **kwargs):	27✔
499	"""Return user sessions info."""
500	sessions = SessionActivity.query_by_user(	27✔
501	user_id=current_user.get_id())
502	results = [{	27✔
503	'created': s.created,
504	'current': SessionActivity.is_current(s.sid_s),
505	'browser': s.browser,
506	'browser_version': s.browser_version,
507	'os': s.os,
508	'device': s.device,
509	'country': s.country} for s in sessions]
510
511	return jsonify({'total': sessions.count(), 'results': results})	27✔
512
513
514	class SessionsItemView(MethodView):	27✔
515	"""View for operations related to user session."""
516
517	decorators = [login_required]	27✔
518
519	def delete(self, sid_s=None, **kwargs):	27✔
520	"""Revoke the given user session."""
521	if SessionActivity.query_by_user(current_user.get_id()) \	27✔
522	.filter_by(sid_s=sid_s).count() == 1:
523	delete_session(sid_s=sid_s)	27✔
524	db.session.commit()	27✔
525	message = 'Session {0} successfully removed. {1}.'	27✔
526	if SessionActivity.is_current(sid_s=sid_s):	27✔
527	message = message.format(sid_s, "Logged out")	27✔
528	else:
529	message = message.format(sid_s, "Revoked")	27✔
530	return jsonify({'message': message})	27✔
531	else:
532	return jsonify({	27✔
533	'message': 'Unable to remove session {0}.'.format(sid_s)}), 400

inveniosoftware / invenio-accounts / 1196

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