password-reset: strengthten security on password reset logic.

- generate token that has special password reset role
- set 10 minut expiration on the token
- add some sleep to prevent bruteforcing attacks
- use implicit messages to prevent user email discovery attacks

rhodecode/tests/functional/test_password_reset.py

@@ -0,0 +1,106 b''
+# -*- coding: utf-8 -*-
+
+# Copyright (C) 2010-2017 RhodeCode GmbH
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License, version 3
+# (only), as published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+# This program is dual-licensed. If you wish to learn more about the
+# RhodeCode Enterprise Edition, including its added features, Support services,
+# and proprietary license terms, please see https://rhodecode.com/licenses/
+
+import pytest
+
+from rhodecode.config.routing import ADMIN_PREFIX
+from rhodecode.tests import (
+    TestController, clear_all_caches, url,
+    TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS)
+from rhodecode.tests.fixture import Fixture
+from rhodecode.tests.utils import AssertResponse
+
+fixture = Fixture()
+
+# Hardcode URLs because we don't have a request object to use
+# pyramids URL generation methods.
+index_url = '/'
+login_url = ADMIN_PREFIX + '/login'
+logut_url = ADMIN_PREFIX + '/logout'
+register_url = ADMIN_PREFIX + '/register'
+pwd_reset_url = ADMIN_PREFIX + '/password_reset'
+pwd_reset_confirm_url = ADMIN_PREFIX + '/password_reset_confirmation'
+
+
+class TestPasswordReset(TestController):
+
+    @pytest.mark.parametrize(
+        'pwd_reset_setting, show_link, show_reset', [
+            ('hg.password_reset.enabled', True, True),
+            ('hg.password_reset.hidden', False, True),
+            ('hg.password_reset.disabled', False, False),
+        ])
+    def test_password_reset_settings(
+            self, pwd_reset_setting, show_link, show_reset):
+        clear_all_caches()
+        self.log_user(TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS)
+        params = {
+            'csrf_token': self.csrf_token,
+            'anonymous': 'True',
+            'default_register': 'hg.register.auto_activate',
+            'default_register_message': '',
+            'default_password_reset': pwd_reset_setting,
+            'default_extern_activate': 'hg.extern_activate.auto',
+        }
+        resp = self.app.post(url('admin_permissions_application'), params=params)
+        self.logout_user()
+
+        login_page = self.app.get(login_url)
+        asr_login = AssertResponse(login_page)
+        index_page = self.app.get(index_url)
+        asr_index = AssertResponse(index_page)
+
+        if show_link:
+            asr_login.one_element_exists('a.pwd_reset')
+            asr_index.one_element_exists('a.pwd_reset')
+        else:
+            asr_login.no_element_exists('a.pwd_reset')
+            asr_index.no_element_exists('a.pwd_reset')
+
+        response = self.app.get(pwd_reset_url)
+        
+        assert_response = AssertResponse(response)
+        if show_reset:
+            response.mustcontain('Send password reset email')
+            assert_response.one_element_exists('#email')
+            assert_response.one_element_exists('#send')
+        else:
+            response.mustcontain('Password reset is disabled.')
+            assert_response.no_element_exists('#email')
+            assert_response.no_element_exists('#send')
+
+    def test_password_form_disabled(self):
+        self.log_user(TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS)
+        params = {
+            'csrf_token': self.csrf_token,
+            'anonymous': 'True',
+            'default_register': 'hg.register.auto_activate',
+            'default_register_message': '',
+            'default_password_reset': 'hg.password_reset.disabled',
+            'default_extern_activate': 'hg.extern_activate.auto',
+        }
+        self.app.post(url('admin_permissions_application'), params=params)
+        self.logout_user()
+
+        response = self.app.post(
+            pwd_reset_url, {'email': '[e-mail]',}
+        )
+        response = response.follow()
+        response.mustcontain('Password reset is disabled.')

rhodecode/login/views.py

@@ -18,6 +18,7 b''
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
+import time
 import collections
 import datetime
 import formencode
@@ -37,9 +38,10 b' from rhodecode.lib.auth import ('
 from rhodecode.lib.base import get_ip_addr
 from rhodecode.lib.exceptions import UserCreationError
 from rhodecode.lib.utils2 import safe_str
-from rhodecode.model.db import User
+from rhodecode.model.db import User, UserApiKeys
 from rhodecode.model.forms import LoginForm, RegisterForm, PasswordResetForm
 from rhodecode.model.meta import Session
+from rhodecode.model.auth_token import AuthTokenModel
 from rhodecode.model.settings import SettingsModel
 from rhodecode.model.user import UserModel
 from rhodecode.translation import _
@@ -289,17 +291,24 b' class LoginView(object):'
             'errors': {},
         }
 
+        # always send implicit message to prevent from discovery of
+        # matching emails
+        msg = _('If such email exists, a password reset link was sent to it.')
+
         if self.request.POST:
+            if h.HasPermissionAny('hg.password_reset.disabled')():
+                _email = self.request.POST.get('email', '')
+                log.error('Failed attempt to reset password for `%s`.', _email)
+                self.session.flash(_('Password reset has been disabled.'),
+                                   queue='error')
+                return HTTPFound(self.request.route_path('reset_password'))
+
             password_reset_form = PasswordResetForm()()
             try:
                 form_result = password_reset_form.to_python(
                     self.request.params)
-                if h.HasPermissionAny('hg.password_reset.disabled')():
-                    log.error('Failed attempt to reset password for %s.', form_result['email'] )
-                    self.session.flash(
-                        _('Password reset has been disabled.'),
-                        queue='error')
-                    return HTTPFound(self.request.route_path('reset_password'))
+                user_email = form_result['email']
+
                 if captcha.active:
                     response = submit(
                         self.request.params.get('recaptcha_challenge_field'),
@@ -310,43 +319,66 b' class LoginView(object):'
                         _value = form_result
                         _msg = _('Bad captcha')
                         error_dict = {'recaptcha_field': _msg}
-                        raise formencode.Invalid(_msg, _value, None,
-                                                 error_dict=error_dict)
+                        raise formencode.Invalid(
+                            _msg, _value, None, error_dict=error_dict)
+                # Generate reset URL and send mail.
+                user = User.get_by_email(user_email)
 
-                # Generate reset URL and send mail.
-                user_email = form_result['email']
-                user = User.get_by_email(user_email)
+                # generate password reset token that expires in 10minutes
+                desc = 'Generated token for password reset from {}'.format(
+                    datetime.datetime.now().isoformat())
+                reset_token = AuthTokenModel().create(
+                    user, lifetime=10,
+                    description=desc,
+                    role=UserApiKeys.ROLE_PASSWORD_RESET)
+                Session().commit()
+
+                log.debug('Successfully created password recovery token')
                 password_reset_url = self.request.route_url(
                     'reset_password_confirmation',
-                    _query={'key': user.api_key})
+                    _query={'key': reset_token.api_key})
                 UserModel().reset_password_link(
                     form_result, password_reset_url)
-
                 # Display success message and redirect.
-                self.session.flash(
-                    _('Your password reset link was sent'),
-                    queue='success')
-                return HTTPFound(self.request.route_path('login'))
+                self.session.flash(msg, queue='success')
+                return HTTPFound(self.request.route_path('reset_password'))
 
             except formencode.Invalid as errors:
                 render_ctx.update({
                     'defaults': errors.value,
-                    'errors': errors.error_dict,
                 })
+            log.debug('faking response on invalid password reset')
+            # make this take 2s, to prevent brute forcing.
+            time.sleep(2)
+            self.session.flash(msg, queue='success')
+            return HTTPFound(self.request.route_path('reset_password'))
 
         return render_ctx
 
     @view_config(route_name='reset_password_confirmation',
                  request_method='GET')
     def password_reset_confirmation(self):
+
         if self.request.GET and self.request.GET.get('key'):
+            # make this take 2s, to prevent brute forcing.
+            time.sleep(2)
+
+            token = AuthTokenModel().get_auth_token(
+                self.request.GET.get('key'))
+
+            # verify token is the correct role
+            if token is None or token.role != UserApiKeys.ROLE_PASSWORD_RESET:
+                log.debug('Got token with role:%s expected is %s',
+                          getattr(token, 'role', 'EMPTY_TOKEN'),
+                          UserApiKeys.ROLE_PASSWORD_RESET)
+                self.session.flash(
+                    _('Given reset token is invalid'), queue='error')
+                return HTTPFound(self.request.route_path('reset_password'))
+
             try:
-                user = User.get_by_auth_token(self.request.GET.get('key'))
-                password_reset_url = self.request.route_url(
-                    'reset_password_confirmation',
-                    _query={'key': user.api_key})
-                data = {'email': user.email}
-                UserModel().reset_password(data, password_reset_url)
+                owner = token.user
+                data = {'email': owner.email, 'token': token.api_key}
+                UserModel().reset_password(data)
                 self.session.flash(
                     _('Your password reset was successful, '
                       'a new password has been sent to your email'),

rhodecode/model/auth_token.py

@@ -41,7 +41,7 b' class AuthTokenModel(BaseModel):'
         """
         :param user: user or user_id
         :param description: description of ApiKey
-        :param lifetime: expiration time in seconds
+        :param lifetime: expiration time in minutes
         :param role: role for the apikey
         """
         from rhodecode.lib.auth import generate_auth_token
@@ -85,3 +85,13 b' class AuthTokenModel(BaseModel):'
                 .filter(or_(UserApiKeys.expires == -1,
                             UserApiKeys.expires >= time.time()))
         return user_auth_tokens
+
+    def get_auth_token(self, auth_token):
+        auth_token = UserApiKeys.query().filter(
+            UserApiKeys.api_key == auth_token)
+        auth_token = auth_token \
+            .filter(or_(UserApiKeys.expires == -1,
+                        UserApiKeys.expires >= time.time()))\
+            .first()
+
+        return auth_token

rhodecode/model/db.py

@@ -943,6 +943,8 b' class UserApiKeys(Base, BaseModel):'
     ROLE_VCS = 'token_role_vcs'
     ROLE_API = 'token_role_api'
     ROLE_FEED = 'token_role_feed'
+    ROLE_PASSWORD_RESET = 'token_password_reset'
+
     ROLES = [ROLE_ALL, ROLE_HTTP, ROLE_VCS, ROLE_API, ROLE_FEED]
 
     user_api_key_id = Column("user_api_key_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)

rhodecode/model/user.py

@@ -538,7 +538,7 b' class UserModel(BaseModel):'
 
         return True
 
-    def reset_password(self, data, pwd_reset_url):
+    def reset_password(self, data):
         from rhodecode.lib.celerylib import tasks, run_task
         from rhodecode.model.notification import EmailNotificationModel
         from rhodecode.lib import auth
@@ -554,8 +554,15 b' class UserModel(BaseModel):'
                 user.update_userdata(force_password_change=True)
 
                 Session().add(user)
+
+                # now delete the token in question
+                UserApiKeys = AuthTokenModel.cls
+                UserApiKeys().query().filter(
+                    UserApiKeys.api_key == data['token']).delete()
+
                 Session().commit()
                 log.info('successfully reset password for `%s`', user_email)
+
             if new_passwd is None:
                 raise Exception('unable to generate new password')
 
@@ -563,7 +570,6 b' class UserModel(BaseModel):'
 
             email_kwargs = {
                 'new_password': new_passwd,
-                'password_reset_url': pwd_reset_url,
                 'user': user,
                 'email': user_email,
                 'date': datetime.datetime.now()
@@ -571,7 +577,8 b' class UserModel(BaseModel):'
 
             (subject, headers, email_body,
              email_body_plaintext) = EmailNotificationModel().render_email(
-                EmailNotificationModel.TYPE_PASSWORD_RESET_CONFIRMATION, **email_kwargs)
+                EmailNotificationModel.TYPE_PASSWORD_RESET_CONFIRMATION,
+                **email_kwargs)
 
             recipients = [user_email]
 

rhodecode/templates/email_templates/password_reset.mako

@@ -16,6 +16,7 b' There was a request to reset your passwo'
 You can continue, and generate new password by clicking following URL:
 ${password_reset_url}
 
+This link will be active for 10 minutes.
 ${self.plaintext_footer()}
 </%def>
 
@@ -28,4 +29,5 b' There was a request to reset your passwo'
 <strong>If you did not request a password reset, please contact your RhodeCode administrator.</strong>
 </p><p>
 <a href="${password_reset_url}">${_('Generate new password here')}.</a>
+This link will be active for 10 minutes.
 </p>

rhodecode/templates/email_templates/password_reset_confirmation.mako

@@ -9,12 +9,11 b' Your new RhodeCode password'
 <%def name="body_plaintext()" filter="n,trim">
 Hi ${user.username},
 
-There was a request to reset your password using the email address ${email} on ${h.format_date(date)}
+Below is your new access password for RhodeCode.
 
 *If you didn't do this, please contact your RhodeCode administrator.*
 
-You can continue, and generate new password by clicking following URL:
-${password_reset_url}
+password: ${new_password}
 
 ${self.plaintext_footer()}
 </%def>
@@ -27,4 +26,4 b' Below is your new access password for Rh'
 <br/>
 <strong>If you didn't request a new password, please contact your RhodeCode administrator.</strong>
 </p>
-<p>password: <input value='${new_password}'/></p>
+<p>password: <pre>${new_password}</pre>

rhodecode/tests/functional/test_login.py

@@ -25,15 +25,13 b' import pytest'
 
 from rhodecode.config.routing import ADMIN_PREFIX
 from rhodecode.tests import (
-    TestController, assert_session_flash, clear_all_caches, url,
-    HG_REPO, TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS)
+    assert_session_flash, url, HG_REPO, TEST_USER_ADMIN_LOGIN)
 from rhodecode.tests.fixture import Fixture
 from rhodecode.tests.utils import AssertResponse, get_session_from_response
-from rhodecode.lib.auth import check_password, generate_auth_token
-from rhodecode.lib import helpers as h
+from rhodecode.lib.auth import check_password
 from rhodecode.model.auth_token import AuthTokenModel
 from rhodecode.model import validators
-from rhodecode.model.db import User, Notification
+from rhodecode.model.db import User, Notification, UserApiKeys
 from rhodecode.model.meta import Session
 
 fixture = Fixture()
@@ -49,7 +47,7 b" pwd_reset_confirm_url = ADMIN_PREFIX + '"
 
 
 @pytest.mark.usefixtures('app')
-class TestLoginController:
+class TestLoginController(object):
     destroy_users = set()
 
     @classmethod
@@ -374,54 +372,42 b' class TestLoginController:'
     def test_forgot_password_wrong_mail(self):
         bad_email = '[e-mail]'
         response = self.app.post(
-            pwd_reset_url,
-            {'email': bad_email, }
+            pwd_reset_url, {'email': bad_email, }
         )
+        assert_session_flash(response,
+            'If such email exists, a password reset link was sent to it.')
 
-        msg = validators.ValidSystemEmail()._messages['non_existing_email']
-        msg = h.html_escape(msg % {'email': bad_email})
-        response.mustcontain()
-
-    def test_forgot_password(self):
+    def test_forgot_password(self, user_util):
         response = self.app.get(pwd_reset_url)
         assert response.status == '200 OK'
 
-        username = 'test_password_reset_1'
-        password = 'qweqwe'
-        email = '[e-mail]'
-        name = 'passwd'
-        lastname = 'reset'
+        user = user_util.create_user()
+        user_id = user.user_id
+        email = user.email
 
-        new = User()
-        new.username = username
-        new.password = password
-        new.email = email
-        new.name = name
-        new.lastname = lastname
-        new.api_key = generate_auth_token(username)
-        Session().add(new)
-        Session().commit()
+        response = self.app.post(pwd_reset_url, {'email': email, })
 
-        response = self.app.post(pwd_reset_url,
-                                 {'email': email, })
-
-        assert_session_flash(
-            response, 'Your password reset link was sent')
-
-        response = response.follow()
+        assert_session_flash(response,
+            'If such email exists, a password reset link was sent to it.')
 
         # BAD KEY
-
-        key = "bad"
-        confirm_url = '{}?key={}'.format(pwd_reset_confirm_url, key)
+        confirm_url = '{}?key={}'.format(pwd_reset_confirm_url, 'badkey')
         response = self.app.get(confirm_url)
         assert response.status == '302 Found'
         assert response.location.endswith(pwd_reset_url)
+        assert_session_flash(response, 'Given reset token is invalid')
+
+        response.follow()  # cleanup flash
 
         # GOOD KEY
+        key = UserApiKeys.query()\
+            .filter(UserApiKeys.user_id == user_id)\
+            .filter(UserApiKeys.role == UserApiKeys.ROLE_PASSWORD_RESET)\
+            .first()
 
-        key = User.get_by_username(username).api_key
-        confirm_url = '{}?key={}'.format(pwd_reset_confirm_url, key)
+        assert key
+
+        confirm_url = '{}?key={}'.format(pwd_reset_confirm_url, key.api_key)
         response = self.app.get(confirm_url)
         assert response.status == '302 Found'
         assert response.location.endswith(login_url)
@@ -431,7 +417,7 b' class TestLoginController:'
             'Your password reset was successful, '
             'a new password has been sent to your email')
 
-        response = response.follow()
+        response.follow()
 
     def _get_api_whitelist(self, values=None):
         config = {'api_access_controllers_whitelist': values or []}
@@ -522,70 +508,3 b' class TestLoginController:'
                                  repo_name=HG_REPO, revision='tip',
                                  api_key=new_auth_token.api_key),
                              status=302)
-
-
-class TestPasswordReset(TestController):
-
-    @pytest.mark.parametrize(
-        'pwd_reset_setting, show_link, show_reset', [
-            ('hg.password_reset.enabled', True, True),
-            ('hg.password_reset.hidden', False, True),
-            ('hg.password_reset.disabled', False, False),
-        ])
-    def test_password_reset_settings(
-            self, pwd_reset_setting, show_link, show_reset):
-        clear_all_caches()
-        self.log_user(TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS)
-        params = {
-            'csrf_token': self.csrf_token,
-            'anonymous': 'True',
-            'default_register': 'hg.register.auto_activate',
-            'default_register_message': '',
-            'default_password_reset': pwd_reset_setting,
-            'default_extern_activate': 'hg.extern_activate.auto',
-        }
-        resp = self.app.post(url('admin_permissions_application'), params=params)
-        self.logout_user()
-
-        login_page = self.app.get(login_url)
-        asr_login = AssertResponse(login_page)
-        index_page = self.app.get(index_url)
-        asr_index = AssertResponse(index_page)
-
-        if show_link:
-            asr_login.one_element_exists('a.pwd_reset')
-            asr_index.one_element_exists('a.pwd_reset')
-        else:
-            asr_login.no_element_exists('a.pwd_reset')
-            asr_index.no_element_exists('a.pwd_reset')
-
-        pwdreset_page = self.app.get(pwd_reset_url)
-        
-        asr_reset = AssertResponse(pwdreset_page)
-        if show_reset:
-            assert 'Send password reset email' in pwdreset_page
-            asr_reset.one_element_exists('#email')
-            asr_reset.one_element_exists('#send')
-        else:
-            assert 'Password reset is disabled.' in pwdreset_page
-            asr_reset.no_element_exists('#email')
-            asr_reset.no_element_exists('#send')
-
-    def test_password_form_disabled(self):
-        self.log_user(TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS)
-        params = {
-            'csrf_token': self.csrf_token,
-            'anonymous': 'True',
-            'default_register': 'hg.register.auto_activate',
-            'default_register_message': '',
-            'default_password_reset': 'hg.password_reset.disabled',
-            'default_extern_activate': 'hg.extern_activate.auto',
-        }
-        self.app.post(url('admin_permissions_application'), params=params)
-        self.logout_user()
-
-        pwdreset_page = self.app.post(
-            pwd_reset_url,
-            {'email': '[e-mail]',}
-        )
-        assert 'Password reset is disabled.' in pwdreset_page