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

@@ -1,358 +1,390 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-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 time
 import collections
 import datetime
 import formencode
 import logging
 import urlparse
 
 from pylons import url
 from pyramid.httpexceptions import HTTPFound
 from pyramid.view import view_config
 from recaptcha.client.captcha import submit
 
 from rhodecode.authentication.base import authenticate, HTTP_TYPE
 from rhodecode.events import UserRegistered
 from rhodecode.lib import helpers as h
 from rhodecode.lib.auth import (
     AuthUser, HasPermissionAnyDecorator, CSRFRequired)
 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 _
 
 
 log = logging.getLogger(__name__)
 
 CaptchaData = collections.namedtuple(
     'CaptchaData', 'active, private_key, public_key')
 
 
 def _store_user_in_session(session, username, remember=False):
     user = User.get_by_username(username, case_insensitive=True)
     auth_user = AuthUser(user.user_id)
     auth_user.set_authenticated()
     cs = auth_user.get_cookie_store()
     session['rhodecode_user'] = cs
     user.update_lastlogin()
     Session().commit()
 
     # If they want to be remembered, update the cookie
     if remember:
         _year = (datetime.datetime.now() +
                  datetime.timedelta(seconds=60 * 60 * 24 * 365))
         session._set_cookie_expires(_year)
 
     session.save()
 
     safe_cs = cs.copy()
     safe_cs['password'] = '****'
     log.info('user %s is now authenticated and stored in '
              'session, session attrs %s', username, safe_cs)
 
     # dumps session attrs back to cookie
     session._update_cookie_out()
     # we set new cookie
     headers = None
     if session.request['set_cookie']:
         # send set-cookie headers back to response to update cookie
         headers = [('Set-Cookie', session.request['cookie_out'])]
     return headers
 
 
 def get_came_from(request):
     came_from = safe_str(request.GET.get('came_from', ''))
     parsed = urlparse.urlparse(came_from)
     allowed_schemes = ['http', 'https']
     if parsed.scheme and parsed.scheme not in allowed_schemes:
         log.error('Suspicious URL scheme detected %s for url %s' %
                   (parsed.scheme, parsed))
         came_from = url('home')
     elif parsed.netloc and request.host != parsed.netloc:
         log.error('Suspicious NETLOC detected %s for url %s server url '
                   'is: %s' % (parsed.netloc, parsed, request.host))
         came_from = url('home')
     elif any(bad_str in parsed.path for bad_str in ('\r', '\n')):
         log.error('Header injection detected `%s` for url %s server url ' %
                   (parsed.path, parsed))
         came_from = url('home')
 
     return came_from or url('home')
 
 
 class LoginView(object):
 
     def __init__(self, context, request):
         self.request = request
         self.context = context
         self.session = request.session
         self._rhodecode_user = request.user
 
     def _get_template_context(self):
         return {
             'came_from': get_came_from(self.request),
             'defaults': {},
             'errors': {},
         }
 
     def _get_captcha_data(self):
         settings = SettingsModel().get_all_settings()
         private_key = settings.get('rhodecode_captcha_private_key')
         public_key = settings.get('rhodecode_captcha_public_key')
         active = bool(private_key)
         return CaptchaData(
             active=active, private_key=private_key, public_key=public_key)
 
     @view_config(
         route_name='login', request_method='GET',
         renderer='rhodecode:templates/login.mako')
     def login(self):
         came_from = get_came_from(self.request)
         user = self.request.user
 
         # redirect if already logged in
         if user.is_authenticated and not user.is_default and user.ip_allowed:
             raise HTTPFound(came_from)
 
         # check if we use headers plugin, and try to login using it.
         try:
             log.debug('Running PRE-AUTH for headers based authentication')
             auth_info = authenticate(
                 '', '', self.request.environ, HTTP_TYPE, skip_missing=True)
             if auth_info:
                 headers = _store_user_in_session(
                     self.session, auth_info.get('username'))
                 raise HTTPFound(came_from, headers=headers)
         except UserCreationError as e:
             log.error(e)
             self.session.flash(e, queue='error')
 
         return self._get_template_context()
 
     @view_config(
         route_name='login', request_method='POST',
         renderer='rhodecode:templates/login.mako')
     def login_post(self):
         came_from = get_came_from(self.request)
 
         login_form = LoginForm()()
 
         try:
             self.session.invalidate()
             form_result = login_form.to_python(self.request.params)
             # form checks for username/password, now we're authenticated
             headers = _store_user_in_session(
                 self.session,
                 username=form_result['username'],
                 remember=form_result['remember'])
             log.debug('Redirecting to "%s" after login.', came_from)
             raise HTTPFound(came_from, headers=headers)
         except formencode.Invalid as errors:
             defaults = errors.value
             # remove password from filling in form again
             defaults.pop('password', None)
             render_ctx = self._get_template_context()
             render_ctx.update({
                 'errors': errors.error_dict,
                 'defaults': defaults,
             })
             return render_ctx
 
         except UserCreationError as e:
             # headers auth or other auth functions that create users on
             # the fly can throw this exception signaling that there's issue
             # with user creation, explanation should be provided in
             # Exception itself
             self.session.flash(e, queue='error')
             return self._get_template_context()
 
     @CSRFRequired()
     @view_config(route_name='logout', request_method='POST')
     def logout(self):
         user = self.request.user
         log.info('Deleting session for user: `%s`', user)
         self.session.delete()
         return HTTPFound(url('home'))
 
     @HasPermissionAnyDecorator(
         'hg.admin', 'hg.register.auto_activate', 'hg.register.manual_activate')
     @view_config(
         route_name='register', request_method='GET',
         renderer='rhodecode:templates/register.mako',)
     def register(self, defaults=None, errors=None):
         defaults = defaults or {}
         errors = errors or {}
 
         settings = SettingsModel().get_all_settings()
         register_message = settings.get('rhodecode_register_message') or ''
         captcha = self._get_captcha_data()
         auto_active = 'hg.register.auto_activate' in User.get_default_user()\
             .AuthUser.permissions['global']
 
         render_ctx = self._get_template_context()
         render_ctx.update({
             'defaults': defaults,
             'errors': errors,
             'auto_active': auto_active,
             'captcha_active': captcha.active,
             'captcha_public_key': captcha.public_key,
             'register_message': register_message,
         })
         return render_ctx
 
     @HasPermissionAnyDecorator(
         'hg.admin', 'hg.register.auto_activate', 'hg.register.manual_activate')
     @view_config(
         route_name='register', request_method='POST',
         renderer='rhodecode:templates/register.mako')
     def register_post(self):
         captcha = self._get_captcha_data()
         auto_active = 'hg.register.auto_activate' in User.get_default_user()\
             .AuthUser.permissions['global']
 
         register_form = RegisterForm()()
         try:
             form_result = register_form.to_python(self.request.params)
             form_result['active'] = auto_active
 
             if captcha.active:
                 response = submit(
                     self.request.params.get('recaptcha_challenge_field'),
                     self.request.params.get('recaptcha_response_field'),
                     private_key=captcha.private_key,
                     remoteip=get_ip_addr(self.request.environ))
                 if not response.is_valid:
                     _value = form_result
                     _msg = _('Bad captcha')
                     error_dict = {'recaptcha_field': _msg}
                     raise formencode.Invalid(_msg, _value, None,
                                              error_dict=error_dict)
 
             new_user = UserModel().create_registration(form_result)
             event = UserRegistered(user=new_user, session=self.session)
             self.request.registry.notify(event)
             self.session.flash(
                 _('You have successfully registered with RhodeCode'),
                 queue='success')
             Session().commit()
 
             redirect_ro = self.request.route_path('login')
             raise HTTPFound(redirect_ro)
 
         except formencode.Invalid as errors:
             errors.value.pop('password', None)
             errors.value.pop('password_confirmation', None)
             return self.register(
                 defaults=errors.value, errors=errors.error_dict)
 
         except UserCreationError as e:
             # container auth or other auth functions that create users on
             # the fly can throw this exception signaling that there's issue
             # with user creation, explanation should be provided in
             # Exception itself
             self.session.flash(e, queue='error')
             return self.register()
 
     @view_config(
         route_name='reset_password', request_method=('GET', 'POST'),
         renderer='rhodecode:templates/password_reset.mako')
     def password_reset(self):
         captcha = self._get_captcha_data()
 
         render_ctx = {
             'captcha_active': captcha.active,
             'captcha_public_key': captcha.public_key,
             'defaults': {},
             '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'),
                         self.request.params.get('recaptcha_response_field'),
                         private_key=captcha.private_key,
                         remoteip=get_ip_addr(self.request.environ))
                     if not response.is_valid:
                         _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'),
                     queue='success')
             except Exception as e:
                 log.error(e)
                 return HTTPFound(self.request.route_path('reset_password'))
 
         return HTTPFound(self.request.route_path('login'))

rhodecode/model/auth_token.py

@@ -1,87 +1,97 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2013-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/
 
 """
 authentication tokens model for RhodeCode
 """
 
 import time
 import logging
 import traceback
 from sqlalchemy import or_
 
 from rhodecode.model import BaseModel
 from rhodecode.model.db import UserApiKeys
 from rhodecode.model.meta import Session
 
 log = logging.getLogger(__name__)
 
 
 class AuthTokenModel(BaseModel):
     cls = UserApiKeys
 
     def create(self, user, description, lifetime=-1, role=UserApiKeys.ROLE_ALL):
         """
         :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
 
         user = self._get_user(user)
 
         new_auth_token = UserApiKeys()
         new_auth_token.api_key = generate_auth_token(user.username)
         new_auth_token.user_id = user.user_id
         new_auth_token.description = description
         new_auth_token.role = role
         new_auth_token.expires = time.time() + (lifetime * 60) if lifetime != -1 else -1
         Session().add(new_auth_token)
 
         return new_auth_token
 
     def delete(self, api_key, user=None):
         """
         Deletes given api_key, if user is set it also filters the object for
         deletion by given user.
         """
         api_key = UserApiKeys.query().filter(UserApiKeys.api_key == api_key)
 
         if user:
             user = self._get_user(user)
             api_key = api_key.filter(UserApiKeys.user_id == user.user_id)
 
         api_key = api_key.scalar()
         try:
             Session().delete(api_key)
         except Exception:
             log.error(traceback.format_exc())
             raise
 
     def get_auth_tokens(self, user, show_expired=True):
         user = self._get_user(user)
         user_auth_tokens = UserApiKeys.query()\
             .filter(UserApiKeys.user_id == user.user_id)
         if not show_expired:
             user_auth_tokens = user_auth_tokens\
                 .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/user.py

@@ -1,845 +1,852 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/
 
 """
 users model for RhodeCode
 """
 
 import logging
 import traceback
 
 import datetime
 from pylons.i18n.translation import _
 
 import ipaddress
 from sqlalchemy.exc import DatabaseError
 from sqlalchemy.sql.expression import true, false
 
 from rhodecode import events
 from rhodecode.lib.utils2 import (
     safe_unicode, get_current_rhodecode_user, action_logger_generic,
     AttributeDict, str2bool)
 from rhodecode.lib.caching_query import FromCache
 from rhodecode.model import BaseModel
 from rhodecode.model.auth_token import AuthTokenModel
 from rhodecode.model.db import (
     User, UserToPerm, UserEmailMap, UserIpMap)
 from rhodecode.lib.exceptions import (
     DefaultUserException, UserOwnsReposException, UserOwnsRepoGroupsException,
     UserOwnsUserGroupsException, NotAllowedToCreateUserError)
 from rhodecode.model.meta import Session
 from rhodecode.model.repo_group import RepoGroupModel
 
 
 log = logging.getLogger(__name__)
 
 
 class UserModel(BaseModel):
     cls = User
 
     def get(self, user_id, cache=False):
         user = self.sa.query(User)
         if cache:
             user = user.options(FromCache("sql_cache_short",
                                           "get_user_%s" % user_id))
         return user.get(user_id)
 
     def get_user(self, user):
         return self._get_user(user)
 
     def get_by_username(self, username, cache=False, case_insensitive=False):
 
         if case_insensitive:
             user = self.sa.query(User).filter(User.username.ilike(username))
         else:
             user = self.sa.query(User)\
                 .filter(User.username == username)
         if cache:
             user = user.options(FromCache("sql_cache_short",
                                           "get_user_%s" % username))
         return user.scalar()
 
     def get_by_email(self, email, cache=False, case_insensitive=False):
         return User.get_by_email(email, case_insensitive, cache)
 
     def get_by_auth_token(self, auth_token, cache=False):
         return User.get_by_auth_token(auth_token, cache)
 
     def get_active_user_count(self, cache=False):
         return User.query().filter(
             User.active == True).filter(
                 User.username != User.DEFAULT_USER).count()
 
     def create(self, form_data, cur_user=None):
         if not cur_user:
             cur_user = getattr(get_current_rhodecode_user(), 'username', None)
 
         user_data = {
             'username': form_data['username'],
             'password': form_data['password'],
             'email': form_data['email'],
             'firstname': form_data['firstname'],
             'lastname': form_data['lastname'],
             'active': form_data['active'],
             'extern_type': form_data['extern_type'],
             'extern_name': form_data['extern_name'],
             'admin': False,
             'cur_user': cur_user
         }
 
         if 'create_repo_group' in form_data:
             user_data['create_repo_group'] = str2bool(
                 form_data.get('create_repo_group'))
 
         try:
             if form_data.get('password_change'):
                 user_data['force_password_change'] = True
             return UserModel().create_or_update(**user_data)
         except Exception:
             log.error(traceback.format_exc())
             raise
 
     def update_user(self, user, skip_attrs=None, **kwargs):
         from rhodecode.lib.auth import get_crypt_password
 
         user = self._get_user(user)
         if user.username == User.DEFAULT_USER:
             raise DefaultUserException(
                 _("You can't Edit this user since it's"
                   " crucial for entire application"))
 
         # first store only defaults
         user_attrs = {
             'updating_user_id': user.user_id,
             'username': user.username,
             'password': user.password,
             'email': user.email,
             'firstname': user.name,
             'lastname': user.lastname,
             'active': user.active,
             'admin': user.admin,
             'extern_name': user.extern_name,
             'extern_type': user.extern_type,
             'language': user.user_data.get('language')
         }
 
         # in case there's new_password, that comes from form, use it to
         # store password
         if kwargs.get('new_password'):
             kwargs['password'] = kwargs['new_password']
 
         # cleanups, my_account password change form
         kwargs.pop('current_password', None)
         kwargs.pop('new_password', None)
 
         # cleanups, user edit password change form
         kwargs.pop('password_confirmation', None)
         kwargs.pop('password_change', None)
 
         # create repo group on user creation
         kwargs.pop('create_repo_group', None)
 
         # legacy forms send name, which is the firstname
         firstname = kwargs.pop('name', None)
         if firstname:
             kwargs['firstname'] = firstname
 
         for k, v in kwargs.items():
             # skip if we don't want to update this
             if skip_attrs and k in skip_attrs:
                 continue
 
             user_attrs[k] = v
 
         try:
             return self.create_or_update(**user_attrs)
         except Exception:
             log.error(traceback.format_exc())
             raise
 
     def create_or_update(
             self, username, password, email, firstname='', lastname='',
             active=True, admin=False, extern_type=None, extern_name=None,
             cur_user=None, plugin=None, force_password_change=False,
             allow_to_create_user=True, create_repo_group=None,
             updating_user_id=None, language=None, strict_creation_check=True):
         """
         Creates a new instance if not found, or updates current one
 
         :param username:
         :param password:
         :param email:
         :param firstname:
         :param lastname:
         :param active:
         :param admin:
         :param extern_type:
         :param extern_name:
         :param cur_user:
         :param plugin: optional plugin this method was called from
         :param force_password_change: toggles new or existing user flag
             for password change
         :param allow_to_create_user: Defines if the method can actually create
             new users
         :param create_repo_group: Defines if the method should also
             create an repo group with user name, and owner
         :param updating_user_id: if we set it up this is the user we want to
             update this allows to editing username.
         :param language: language of user from interface.
 
         :returns: new User object with injected `is_new_user` attribute.
         """
         if not cur_user:
             cur_user = getattr(get_current_rhodecode_user(), 'username', None)
 
         from rhodecode.lib.auth import (
             get_crypt_password, check_password, generate_auth_token)
         from rhodecode.lib.hooks_base import (
             log_create_user, check_allowed_create_user)
 
         def _password_change(new_user, password):
             # empty password
             if not new_user.password:
                 return False
 
             # password check is only needed for RhodeCode internal auth calls
             # in case it's a plugin we don't care
             if not plugin:
 
                 # first check if we gave crypted password back, and if it
                 # matches it's not password change
                 if new_user.password == password:
                     return False
 
                 password_match = check_password(password, new_user.password)
                 if not password_match:
                     return True
 
             return False
 
         # read settings on default personal repo group creation
         if create_repo_group is None:
             default_create_repo_group = RepoGroupModel()\
                 .get_default_create_personal_repo_group()
             create_repo_group = default_create_repo_group
 
         user_data = {
             'username': username,
             'password': password,
             'email': email,
             'firstname': firstname,
             'lastname': lastname,
             'active': active,
             'admin': admin
         }
 
         if updating_user_id:
             log.debug('Checking for existing account in RhodeCode '
                       'database with user_id `%s` ' % (updating_user_id,))
             user = User.get(updating_user_id)
         else:
             log.debug('Checking for existing account in RhodeCode '
                       'database with username `%s` ' % (username,))
             user = User.get_by_username(username, case_insensitive=True)
 
         if user is None:
             # we check internal flag if this method is actually allowed to
             # create new user
             if not allow_to_create_user:
                 msg = ('Method wants to create new user, but it is not '
                        'allowed to do so')
                 log.warning(msg)
                 raise NotAllowedToCreateUserError(msg)
 
             log.debug('Creating new user %s', username)
 
             # only if we create user that is active
             new_active_user = active
             if new_active_user and strict_creation_check:
                 # raises UserCreationError if it's not allowed for any reason to
                 # create new active user, this also executes pre-create hooks
                 check_allowed_create_user(user_data, cur_user, strict_check=True)
             events.trigger(events.UserPreCreate(user_data))
             new_user = User()
             edit = False
         else:
             log.debug('updating user %s', username)
             events.trigger(events.UserPreUpdate(user, user_data))
             new_user = user
             edit = True
 
             # we're not allowed to edit default user
             if user.username == User.DEFAULT_USER:
                 raise DefaultUserException(
                     _("You can't edit this user (`%(username)s`) since it's "
                       "crucial for entire application") % {'username': user.username})
 
         # inject special attribute that will tell us if User is new or old
         new_user.is_new_user = not edit
         # for users that didn's specify auth type, we use RhodeCode built in
         from rhodecode.authentication.plugins import auth_rhodecode
         extern_name = extern_name or auth_rhodecode.RhodeCodeAuthPlugin.name
         extern_type = extern_type or auth_rhodecode.RhodeCodeAuthPlugin.name
 
         try:
             new_user.username = username
             new_user.admin = admin
             new_user.email = email
             new_user.active = active
             new_user.extern_name = safe_unicode(extern_name)
             new_user.extern_type = safe_unicode(extern_type)
             new_user.name = firstname
             new_user.lastname = lastname
 
             if not edit:
                 new_user.api_key = generate_auth_token(username)
 
             # set password only if creating an user or password is changed
             if not edit or _password_change(new_user, password):
                 reason = 'new password' if edit else 'new user'
                 log.debug('Updating password reason=>%s', reason)
                 new_user.password = get_crypt_password(password) if password else None
 
             if force_password_change:
                 new_user.update_userdata(force_password_change=True)
             if language:
                 new_user.update_userdata(language=language)
             new_user.update_userdata(notification_status=True)
 
             self.sa.add(new_user)
 
             if not edit and create_repo_group:
                 RepoGroupModel().create_personal_repo_group(
                     new_user, commit_early=False)
 
             if not edit:
                 # add the RSS token
                 AuthTokenModel().create(username,
                                         description='Generated feed token',
                                         role=AuthTokenModel.cls.ROLE_FEED)
                 log_create_user(created_by=cur_user, **new_user.get_dict())
                 events.trigger(events.UserPostCreate(user_data))
             return new_user
         except (DatabaseError,):
             log.error(traceback.format_exc())
             raise
 
     def create_registration(self, form_data):
         from rhodecode.model.notification import NotificationModel
         from rhodecode.model.notification import EmailNotificationModel
 
         try:
             form_data['admin'] = False
             form_data['extern_name'] = 'rhodecode'
             form_data['extern_type'] = 'rhodecode'
             new_user = self.create(form_data)
 
             self.sa.add(new_user)
             self.sa.flush()
 
             user_data = new_user.get_dict()
             kwargs = {
                 # use SQLALCHEMY safe dump of user data
                 'user': AttributeDict(user_data),
                 'date': datetime.datetime.now()
             }
             notification_type = EmailNotificationModel.TYPE_REGISTRATION
             # pre-generate the subject for notification itself
             (subject,
              _h, _e,  # we don't care about those
              body_plaintext) = EmailNotificationModel().render_email(
                 notification_type, **kwargs)
 
             # create notification objects, and emails
             NotificationModel().create(
                 created_by=new_user,
                 notification_subject=subject,
                 notification_body=body_plaintext,
                 notification_type=notification_type,
                 recipients=None,  # all admins
                 email_kwargs=kwargs,
             )
 
             return new_user
         except Exception:
             log.error(traceback.format_exc())
             raise
 
     def _handle_user_repos(self, username, repositories, handle_mode=None):
         _superadmin = self.cls.get_first_super_admin()
         left_overs = True
 
         from rhodecode.model.repo import RepoModel
 
         if handle_mode == 'detach':
             for obj in repositories:
                 obj.user = _superadmin
                 # set description we know why we super admin now owns
                 # additional repositories that were orphaned !
                 obj.description += '  \n::detached repository from deleted user: %s' % (username,)
                 self.sa.add(obj)
             left_overs = False
         elif handle_mode == 'delete':
             for obj in repositories:
                 RepoModel().delete(obj, forks='detach')
             left_overs = False
 
         # if nothing is done we have left overs left
         return left_overs
 
     def _handle_user_repo_groups(self, username, repository_groups,
                                  handle_mode=None):
         _superadmin = self.cls.get_first_super_admin()
         left_overs = True
 
         from rhodecode.model.repo_group import RepoGroupModel
 
         if handle_mode == 'detach':
             for r in repository_groups:
                 r.user = _superadmin
                 # set description we know why we super admin now owns
                 # additional repositories that were orphaned !
                 r.group_description += '  \n::detached repository group from deleted user: %s' % (username,)
                 self.sa.add(r)
             left_overs = False
         elif handle_mode == 'delete':
             for r in repository_groups:
                 RepoGroupModel().delete(r)
             left_overs = False
 
         # if nothing is done we have left overs left
         return left_overs
 
     def _handle_user_user_groups(self, username, user_groups, handle_mode=None):
         _superadmin = self.cls.get_first_super_admin()
         left_overs = True
 
         from rhodecode.model.user_group import UserGroupModel
 
         if handle_mode == 'detach':
             for r in user_groups:
                 for user_user_group_to_perm in r.user_user_group_to_perm:
                     if user_user_group_to_perm.user.username == username:
                         user_user_group_to_perm.user = _superadmin
                 r.user = _superadmin
                 # set description we know why we super admin now owns
                 # additional repositories that were orphaned !
                 r.user_group_description += '  \n::detached user group from deleted user: %s' % (username,)
                 self.sa.add(r)
             left_overs = False
         elif handle_mode == 'delete':
             for r in user_groups:
                 UserGroupModel().delete(r)
             left_overs = False
 
         # if nothing is done we have left overs left
         return left_overs
 
     def delete(self, user, cur_user=None, handle_repos=None,
                handle_repo_groups=None, handle_user_groups=None):
         if not cur_user:
             cur_user = getattr(get_current_rhodecode_user(), 'username', None)
         user = self._get_user(user)
 
         try:
             if user.username == User.DEFAULT_USER:
                 raise DefaultUserException(
                     _(u"You can't remove this user since it's"
                       u" crucial for entire application"))
 
             left_overs = self._handle_user_repos(
                 user.username, user.repositories, handle_repos)
             if left_overs and user.repositories:
                 repos = [x.repo_name for x in user.repositories]
                 raise UserOwnsReposException(
                     _(u'user "%s" still owns %s repositories and cannot be '
                       u'removed. Switch owners or remove those repositories:%s')
                     % (user.username, len(repos), ', '.join(repos)))
 
             left_overs = self._handle_user_repo_groups(
                 user.username, user.repository_groups, handle_repo_groups)
             if left_overs and user.repository_groups:
                 repo_groups = [x.group_name for x in user.repository_groups]
                 raise UserOwnsRepoGroupsException(
                     _(u'user "%s" still owns %s repository groups and cannot be '
                       u'removed. Switch owners or remove those repository groups:%s')
                     % (user.username, len(repo_groups), ', '.join(repo_groups)))
 
             left_overs = self._handle_user_user_groups(
                 user.username, user.user_groups, handle_user_groups)
             if left_overs and user.user_groups:
                 user_groups = [x.users_group_name for x in user.user_groups]
                 raise UserOwnsUserGroupsException(
                     _(u'user "%s" still owns %s user groups and cannot be '
                       u'removed. Switch owners or remove those user groups:%s')
                     % (user.username, len(user_groups), ', '.join(user_groups)))
 
             # we might change the user data with detach/delete, make sure
             # the object is marked as expired before actually deleting !
             self.sa.expire(user)
             self.sa.delete(user)
             from rhodecode.lib.hooks_base import log_delete_user
             log_delete_user(deleted_by=cur_user, **user.get_dict())
         except Exception:
             log.error(traceback.format_exc())
             raise
 
     def reset_password_link(self, data, pwd_reset_url):
         from rhodecode.lib.celerylib import tasks, run_task
         from rhodecode.model.notification import EmailNotificationModel
         user_email = data['email']
         try:
             user = User.get_by_email(user_email)
             if user:
                 log.debug('password reset user found %s', user)
 
                 email_kwargs = {
                     'password_reset_url': pwd_reset_url,
                     'user': user,
                     'email': user_email,
                     'date': datetime.datetime.now()
                 }
 
                 (subject, headers, email_body,
                  email_body_plaintext) = EmailNotificationModel().render_email(
                     EmailNotificationModel.TYPE_PASSWORD_RESET, **email_kwargs)
 
                 recipients = [user_email]
 
                 action_logger_generic(
                     'sending password reset email to user: {}'.format(
                         user), namespace='security.password_reset')
 
                 run_task(tasks.send_email, recipients, subject,
                          email_body_plaintext, email_body)
 
             else:
                 log.debug("password reset email %s not found", user_email)
         except Exception:
             log.error(traceback.format_exc())
             return False
 
         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
         user_email = data['email']
         pre_db = True
         try:
             user = User.get_by_email(user_email)
             new_passwd = auth.PasswordGenerator().gen_password(
                 12, auth.PasswordGenerator.ALPHABETS_BIG_SMALL)
             if user:
                 user.password = auth.get_crypt_password(new_passwd)
                 # also force this user to reset his password !
                 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')
 
             pre_db = False
 
             email_kwargs = {
                 'new_password': new_passwd,
-                'password_reset_url': pwd_reset_url,
                 'user': user,
                 'email': user_email,
                 'date': datetime.datetime.now()
             }
 
             (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]
 
             action_logger_generic(
                 'sent new password to user: {} with email: {}'.format(
                     user, user_email), namespace='security.password_reset')
 
             run_task(tasks.send_email, recipients, subject,
                      email_body_plaintext, email_body)
 
         except Exception:
             log.error('Failed to update user password')
             log.error(traceback.format_exc())
             if pre_db:
                 # we rollback only if local db stuff fails. If it goes into
                 # run_task, we're pass rollback state this wouldn't work then
                 Session().rollback()
 
         return True
 
     def fill_data(self, auth_user, user_id=None, api_key=None, username=None):
         """
         Fetches auth_user by user_id,or api_key if present.
         Fills auth_user attributes with those taken from database.
         Additionally set's is_authenitated if lookup fails
         present in database
 
         :param auth_user: instance of user to set attributes
         :param user_id: user id to fetch by
         :param api_key: api key to fetch by
         :param username: username to fetch by
         """
         if user_id is None and api_key is None and username is None:
             raise Exception('You need to pass user_id, api_key or username')
 
         log.debug(
             'doing fill data based on: user_id:%s api_key:%s username:%s',
             user_id, api_key, username)
         try:
             dbuser = None
             if user_id:
                 dbuser = self.get(user_id)
             elif api_key:
                 dbuser = self.get_by_auth_token(api_key)
             elif username:
                 dbuser = self.get_by_username(username)
 
             if not dbuser:
                 log.warning(
                     'Unable to lookup user by id:%s api_key:%s username:%s',
                     user_id, api_key, username)
                 return False
             if not dbuser.active:
                 log.debug('User `%s` is inactive, skipping fill data', username)
                 return False
 
             log.debug('filling user:%s data', dbuser)
 
             # TODO: johbo: Think about this and find a clean solution
             user_data = dbuser.get_dict()
             user_data.update(dbuser.get_api_data(include_secrets=True))
 
             for k, v in user_data.iteritems():
                 # properties of auth user we dont update
                 if k not in ['auth_tokens', 'permissions']:
                     setattr(auth_user, k, v)
 
             # few extras
             setattr(auth_user, 'feed_token', dbuser.feed_token)
         except Exception:
             log.error(traceback.format_exc())
             auth_user.is_authenticated = False
             return False
 
         return True
 
     def has_perm(self, user, perm):
         perm = self._get_perm(perm)
         user = self._get_user(user)
 
         return UserToPerm.query().filter(UserToPerm.user == user)\
             .filter(UserToPerm.permission == perm).scalar() is not None
 
     def grant_perm(self, user, perm):
         """
         Grant user global permissions
 
         :param user:
         :param perm:
         """
         user = self._get_user(user)
         perm = self._get_perm(perm)
         # if this permission is already granted skip it
         _perm = UserToPerm.query()\
             .filter(UserToPerm.user == user)\
             .filter(UserToPerm.permission == perm)\
             .scalar()
         if _perm:
             return
         new = UserToPerm()
         new.user = user
         new.permission = perm
         self.sa.add(new)
         return new
 
     def revoke_perm(self, user, perm):
         """
         Revoke users global permissions
 
         :param user:
         :param perm:
         """
         user = self._get_user(user)
         perm = self._get_perm(perm)
 
         obj = UserToPerm.query()\
                 .filter(UserToPerm.user == user)\
                 .filter(UserToPerm.permission == perm)\
                 .scalar()
         if obj:
             self.sa.delete(obj)
 
     def add_extra_email(self, user, email):
         """
         Adds email address to UserEmailMap
 
         :param user:
         :param email:
         """
         from rhodecode.model import forms
         form = forms.UserExtraEmailForm()()
         data = form.to_python({'email': email})
         user = self._get_user(user)
 
         obj = UserEmailMap()
         obj.user = user
         obj.email = data['email']
         self.sa.add(obj)
         return obj
 
     def delete_extra_email(self, user, email_id):
         """
         Removes email address from UserEmailMap
 
         :param user:
         :param email_id:
         """
         user = self._get_user(user)
         obj = UserEmailMap.query().get(email_id)
         if obj:
             self.sa.delete(obj)
 
     def parse_ip_range(self, ip_range):
         ip_list = []
         def make_unique(value):
             seen = []
             return [c for c in value if not (c in seen or seen.append(c))]
 
         # firsts split by commas
         for ip_range in ip_range.split(','):
             if not ip_range:
                 continue
             ip_range = ip_range.strip()
             if '-' in ip_range:
                 start_ip, end_ip = ip_range.split('-', 1)
                 start_ip = ipaddress.ip_address(start_ip.strip())
                 end_ip = ipaddress.ip_address(end_ip.strip())
                 parsed_ip_range = []
 
                 for index in xrange(int(start_ip), int(end_ip) + 1):
                     new_ip = ipaddress.ip_address(index)
                     parsed_ip_range.append(str(new_ip))
                 ip_list.extend(parsed_ip_range)
             else:
                 ip_list.append(ip_range)
 
         return make_unique(ip_list)
 
     def add_extra_ip(self, user, ip, description=None):
         """
         Adds ip address to UserIpMap
 
         :param user:
         :param ip:
         """
         from rhodecode.model import forms
         form = forms.UserExtraIpForm()()
         data = form.to_python({'ip': ip})
         user = self._get_user(user)
 
         obj = UserIpMap()
         obj.user = user
         obj.ip_addr = data['ip']
         obj.description = description
         self.sa.add(obj)
         return obj
 
     def delete_extra_ip(self, user, ip_id):
         """
         Removes ip address from UserIpMap
 
         :param user:
         :param ip_id:
         """
         user = self._get_user(user)
         obj = UserIpMap.query().get(ip_id)
         if obj:
             self.sa.delete(obj)
 
     def get_accounts_in_creation_order(self, current_user=None):
         """
         Get accounts in order of creation for deactivation for license limits
 
         pick currently logged in user, and append to the list in position 0
         pick all super-admins in order of creation date and add it to the list
         pick all other accounts in order of creation and add it to the list.
 
         Based on that list, the last accounts can be disabled as they are
         created at the end and don't include any of the super admins as well
         as the current user.
 
         :param current_user: optionally current user running this operation
         """
 
         if not current_user:
             current_user = get_current_rhodecode_user()
         active_super_admins = [
             x.user_id for x in User.query()
             .filter(User.user_id != current_user.user_id)
             .filter(User.active == true())
             .filter(User.admin == true())
             .order_by(User.created_on.asc())]
 
         active_regular_users = [
             x.user_id for x in User.query()
             .filter(User.user_id != current_user.user_id)
             .filter(User.active == true())
             .filter(User.admin == false())
             .order_by(User.created_on.asc())]
 
         list_of_accounts = [current_user.user_id]
         list_of_accounts += active_super_admins
         list_of_accounts += active_regular_users
 
         return list_of_accounts
 
     def deactivate_last_users(self, expected_users):
         """
         Deactivate accounts that are over the license limits.
         Algorithm of which accounts to disabled is based on the formula:
 
         Get current user, then super admins in creation order, then regular
         active users in creation order.
 
         Using that list we mark all accounts from the end of it as inactive.
         This way we block only latest created accounts.
 
         :param expected_users: list of users in special order, we deactivate
             the end N ammoun of users from that list
         """
 
         list_of_accounts = self.get_accounts_in_creation_order()
 
         for acc_id in list_of_accounts[expected_users + 1:]:
             user = User.get(acc_id)
             log.info('Deactivating account %s for license unlock', user)
             user.active = False
             Session().add(user)
             Session().commit()
 
         return

rhodecode/templates/email_templates/password_reset.mako

@@ -1,31 +1,33 b''
 ## -*- coding: utf-8 -*-
 <%inherit file="base.mako"/>
 
 <%def name="subject()" filter="n,trim">
 RhodeCode Password reset
 </%def>
 
 ## plain text version of the email. Empty by default
 <%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)}
 
 *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}
 
+This link will be active for 10 minutes.
 ${self.plaintext_footer()}
 </%def>
 
 ## BODY GOES BELOW
 <p>
 Hello ${user.username},
 </p><p>
 There was a request to reset your password using the email address ${email} on ${h.format_date(date)}
 <br/>
 <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

@@ -1,30 +1,29 b''
 ## -*- coding: utf-8 -*-
 <%inherit file="base.mako"/>
 
 <%def name="subject()" filter="n,trim">
 Your new RhodeCode password
 </%def>
 
 ## plain text version of the email. Empty by default
 <%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>
 
 ## BODY GOES BELOW
 <p>
 Hello ${user.username},
 </p><p>
 Below is your new access password for RhodeCode.
 <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

@@ -1,591 +1,510 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 urlparse
 
 import mock
 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()
 
 # 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'
 
 
 @pytest.mark.usefixtures('app')
-class TestLoginController:
+class TestLoginController(object):
     destroy_users = set()
 
     @classmethod
     def teardown_class(cls):
         fixture.destroy_users(cls.destroy_users)
 
     def teardown_method(self, method):
         for n in Notification.query().all():
             Session().delete(n)
 
         Session().commit()
         assert Notification.query().all() == []
 
     def test_index(self):
         response = self.app.get(login_url)
         assert response.status == '200 OK'
         # Test response...
 
     def test_login_admin_ok(self):
         response = self.app.post(login_url,
                                  {'username': 'test_admin',
                                   'password': 'test12'})
         assert response.status == '302 Found'
         session = get_session_from_response(response)
         username = session['rhodecode_user'].get('username')
         assert username == 'test_admin'
         response = response.follow()
         response.mustcontain('/%s' % HG_REPO)
 
     def test_login_regular_ok(self):
         response = self.app.post(login_url,
                                  {'username': 'test_regular',
                                   'password': 'test12'})
 
         assert response.status == '302 Found'
         session = get_session_from_response(response)
         username = session['rhodecode_user'].get('username')
         assert username == 'test_regular'
         response = response.follow()
         response.mustcontain('/%s' % HG_REPO)
 
     def test_login_ok_came_from(self):
         test_came_from = '/_admin/users?branch=stable'
         _url = '{}?came_from={}'.format(login_url, test_came_from)
         response = self.app.post(
             _url, {'username': 'test_admin', 'password': 'test12'})
         assert response.status == '302 Found'
         assert 'branch=stable' in response.location
         response = response.follow()
 
         assert response.status == '200 OK'
         response.mustcontain('Users administration')
 
     def test_redirect_to_login_with_get_args(self):
         with fixture.anon_access(False):
             kwargs = {'branch': 'stable'}
             response = self.app.get(
                 url('summary_home', repo_name=HG_REPO, **kwargs))
             assert response.status == '302 Found'
             response_query = urlparse.parse_qsl(response.location)
             assert 'branch=stable' in response_query[0][1]
 
     def test_login_form_with_get_args(self):
         _url = '{}?came_from=/_admin/users,branch=stable'.format(login_url)
         response = self.app.get(_url)
         assert 'branch%3Dstable' in response.form.action
 
     @pytest.mark.parametrize("url_came_from", [
         'data:text/html,<script>window.alert("xss")</script>',
         'mailto:[e-mail]',
         'file:///etc/passwd',
         'ftp://some.ftp.server',
         'http://other.domain',
         '/\r\nX-Forwarded-Host: http://example.org',
     ])
     def test_login_bad_came_froms(self, url_came_from):
         _url = '{}?came_from={}'.format(login_url, url_came_from)
         response = self.app.post(
             _url,
             {'username': 'test_admin', 'password': 'test12'})
         assert response.status == '302 Found'
         response = response.follow()
         assert response.status == '200 OK'
         assert response.request.path == '/'
 
     def test_login_short_password(self):
         response = self.app.post(login_url,
                                  {'username': 'test_admin',
                                   'password': 'as'})
         assert response.status == '200 OK'
 
         response.mustcontain('Enter 3 characters or more')
 
     def test_login_wrong_non_ascii_password(self, user_regular):
         response = self.app.post(
             login_url,
             {'username': user_regular.username,
              'password': u'invalid-non-asci\xe4'.encode('utf8')})
 
         response.mustcontain('invalid user name')
         response.mustcontain('invalid password')
 
     def test_login_with_non_ascii_password(self, user_util):
         password = u'valid-non-ascii\xe4'
         user = user_util.create_user(password=password)
         response = self.app.post(
             login_url,
             {'username': user.username,
              'password': password.encode('utf-8')})
         assert response.status_code == 302
 
     def test_login_wrong_username_password(self):
         response = self.app.post(login_url,
                                  {'username': 'error',
                                   'password': 'test12'})
 
         response.mustcontain('invalid user name')
         response.mustcontain('invalid password')
 
     def test_login_admin_ok_password_migration(self, real_crypto_backend):
         from rhodecode.lib import auth
 
         # create new user, with sha256 password
         temp_user = 'test_admin_sha256'
         user = fixture.create_user(temp_user)
         user.password = auth._RhodeCodeCryptoSha256().hash_create(
             b'test123')
         Session().add(user)
         Session().commit()
         self.destroy_users.add(temp_user)
         response = self.app.post(login_url,
                                  {'username': temp_user,
                                   'password': 'test123'})
 
         assert response.status == '302 Found'
         session = get_session_from_response(response)
         username = session['rhodecode_user'].get('username')
         assert username == temp_user
         response = response.follow()
         response.mustcontain('/%s' % HG_REPO)
 
         # new password should be bcrypted, after log-in and transfer
         user = User.get_by_username(temp_user)
         assert user.password.startswith('$')
 
     # REGISTRATIONS
     def test_register(self):
         response = self.app.get(register_url)
         response.mustcontain('Create an Account')
 
     def test_register_err_same_username(self):
         uname = 'test_admin'
         response = self.app.post(
             register_url,
             {
                 'username': uname,
                 'password': 'test12',
                 'password_confirmation': 'test12',
                 'email': '[e-mail]',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
 
         assertr = AssertResponse(response)
         msg = validators.ValidUsername()._messages['username_exists']
         msg = msg % {'username': uname}
         assertr.element_contains('#username+.error-message', msg)
 
     def test_register_err_same_email(self):
         response = self.app.post(
             register_url,
             {
                 'username': 'test_admin_0',
                 'password': 'test12',
                 'password_confirmation': 'test12',
                 'email': '[e-mail]',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
 
         assertr = AssertResponse(response)
         msg = validators.UniqSystemEmail()()._messages['email_taken']
         assertr.element_contains('#email+.error-message', msg)
 
     def test_register_err_same_email_case_sensitive(self):
         response = self.app.post(
             register_url,
             {
                 'username': 'test_admin_1',
                 'password': 'test12',
                 'password_confirmation': 'test12',
                 'email': '[e-mail]',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
         assertr = AssertResponse(response)
         msg = validators.UniqSystemEmail()()._messages['email_taken']
         assertr.element_contains('#email+.error-message', msg)
 
     def test_register_err_wrong_data(self):
         response = self.app.post(
             register_url,
             {
                 'username': 'xs',
                 'password': 'test',
                 'password_confirmation': 'test',
                 'email': 'goodmailm',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
         assert response.status == '200 OK'
         response.mustcontain('An email address must contain a single @')
         response.mustcontain('Enter a value 6 characters long or more')
 
     def test_register_err_username(self):
         response = self.app.post(
             register_url,
             {
                 'username': 'error user',
                 'password': 'test12',
                 'password_confirmation': 'test12',
                 'email': 'goodmailm',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
 
         response.mustcontain('An email address must contain a single @')
         response.mustcontain(
             'Username may only contain '
             'alphanumeric characters underscores, '
             'periods or dashes and must begin with '
             'alphanumeric character')
 
     def test_register_err_case_sensitive(self):
         usr = 'Test_Admin'
         response = self.app.post(
             register_url,
             {
                 'username': usr,
                 'password': 'test12',
                 'password_confirmation': 'test12',
                 'email': 'goodmailm',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
 
         assertr = AssertResponse(response)
         msg = validators.ValidUsername()._messages['username_exists']
         msg = msg % {'username': usr}
         assertr.element_contains('#username+.error-message', msg)
 
     def test_register_special_chars(self):
         response = self.app.post(
             register_url,
             {
                 'username': 'xxxaxn',
                 'password': 'ąćźżąśśśś',
                 'password_confirmation': 'ąćźżąśśśś',
                 'email': '[e-mail]',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
 
         msg = validators.ValidPassword()._messages['invalid_password']
         response.mustcontain(msg)
 
     def test_register_password_mismatch(self):
         response = self.app.post(
             register_url,
             {
                 'username': 'xs',
                 'password': '123qwe',
                 'password_confirmation': 'qwe123',
                 'email': '[e-mail]',
                 'firstname': 'test',
                 'lastname': 'test'
             }
         )
         msg = validators.ValidPasswordsMatch()._messages['password_mismatch']
         response.mustcontain(msg)
 
     def test_register_ok(self):
         username = 'test_regular4'
         password = 'qweqwe'
         email = '[e-mail]'
         name = 'testname'
         lastname = 'testlastname'
 
         response = self.app.post(
             register_url,
             {
                 'username': username,
                 'password': password,
                 'password_confirmation': password,
                 'email': email,
                 'firstname': name,
                 'lastname': lastname,
                 'admin': True
             }
         )  # This should be overriden
         assert response.status == '302 Found'
         assert_session_flash(
             response, 'You have successfully registered with RhodeCode')
 
         ret = Session().query(User).filter(
             User.username == 'test_regular4').one()
         assert ret.username == username
         assert check_password(password, ret.password)
         assert ret.email == email
         assert ret.name == name
         assert ret.lastname == lastname
         assert ret.api_key is not None
         assert not ret.admin
 
     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)
 
         assert_session_flash(
             response,
             '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 []}
         return config
 
     @pytest.mark.parametrize("test_name, auth_token", [
         ('none', None),
         ('empty_string', ''),
         ('fake_number', '123456'),
         ('proper_auth_token', None)
     ])
     def test_access_not_whitelisted_page_via_auth_token(
             self, test_name, auth_token, user_admin):
 
         whitelist = self._get_api_whitelist([])
         with mock.patch.dict('rhodecode.CONFIG', whitelist):
             assert [] == whitelist['api_access_controllers_whitelist']
             if test_name == 'proper_auth_token':
                 # use builtin if api_key is None
                 auth_token = user_admin.api_key
 
             with fixture.anon_access(False):
                 self.app.get(url(controller='changeset',
                                  action='changeset_raw',
                                  repo_name=HG_REPO, revision='tip',
                                  api_key=auth_token),
                              status=302)
 
     @pytest.mark.parametrize("test_name, auth_token, code", [
         ('none', None, 302),
         ('empty_string', '', 302),
         ('fake_number', '123456', 302),
         ('proper_auth_token', None, 200)
     ])
     def test_access_whitelisted_page_via_auth_token(
             self, test_name, auth_token, code, user_admin):
 
         whitelist_entry = ['ChangesetController:changeset_raw']
         whitelist = self._get_api_whitelist(whitelist_entry)
 
         with mock.patch.dict('rhodecode.CONFIG', whitelist):
             assert whitelist_entry == whitelist['api_access_controllers_whitelist']
 
             if test_name == 'proper_auth_token':
                 auth_token = user_admin.api_key
 
             with fixture.anon_access(False):
                 self.app.get(url(controller='changeset',
                                  action='changeset_raw',
                                  repo_name=HG_REPO, revision='tip',
                                  api_key=auth_token),
                              status=code)
 
     def test_access_page_via_extra_auth_token(self):
         whitelist = self._get_api_whitelist(
             ['ChangesetController:changeset_raw'])
         with mock.patch.dict('rhodecode.CONFIG', whitelist):
             assert ['ChangesetController:changeset_raw'] == \
                 whitelist['api_access_controllers_whitelist']
 
             new_auth_token = AuthTokenModel().create(
                 TEST_USER_ADMIN_LOGIN, 'test')
             Session().commit()
             with fixture.anon_access(False):
                 self.app.get(url(controller='changeset',
                                  action='changeset_raw',
                                  repo_name=HG_REPO, revision='tip',
                                  api_key=new_auth_token.api_key),
                              status=200)
 
     def test_access_page_via_expired_auth_token(self):
         whitelist = self._get_api_whitelist(
             ['ChangesetController:changeset_raw'])
         with mock.patch.dict('rhodecode.CONFIG', whitelist):
             assert ['ChangesetController:changeset_raw'] == \
                 whitelist['api_access_controllers_whitelist']
 
             new_auth_token = AuthTokenModel().create(
                 TEST_USER_ADMIN_LOGIN, 'test')
             Session().commit()
             # patch the api key and make it expired
             new_auth_token.expires = 0
             Session().add(new_auth_token)
             Session().commit()
             with fixture.anon_access(False):
                 self.app.get(url(controller='changeset',
                                  action='changeset_raw',
                                  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