# -*- coding: utf-8 -*-

# This program is free software: you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# 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 General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""

kallithea.controllers.login

~~~~~~~~~~~~~~~~~~~~~~~~~~~

Login controller for Kallithea

This file was forked by the Kallithea project in July 2014.

Original author and date, and relevant copyright and licensing information is below:

:created_on: Apr 22, 2010

:author: marcink

:copyright: (c) 2013 RhodeCode GmbH, and others.

:license: GPLv3, see LICENSE.md for more details.

"""

import logging

import re

import formencode

from formencode import htmlfill

from tg import request, session

from tg import tmpl_context as c

from tg.i18n import ugettext as _

from webob.exc import HTTPBadRequest, HTTPFound

import kallithea.lib.helpers as h

from kallithea.config.routing import url

from kallithea.lib.auth import AuthUser, HasPermissionAnyDecorator

from kallithea.lib.base import BaseController, log_in_user, render

from kallithea.lib.exceptions import UserCreationError

from kallithea.lib.utils2 import safe_str

from kallithea.model.db import Setting, User

from kallithea.model.forms import LoginForm, PasswordResetConfirmationForm, PasswordResetRequestForm, RegisterForm

from kallithea.model.meta import Session

from kallithea.model.user import UserModel

log = logging.getLogger(__name__)

class LoginController(BaseController):

    def _validate_came_from(self, came_from,

            _re=re.compile(r"/(?!/)[-!#$%&'()*+,./:;=?@_~0-9A-Za-z]*$")):

        """Return True if came_from is valid and can and should be used.

        Determines if a URI reference is valid and relative to the origin;

        or in RFC 3986 terms, whether it matches this production:

          origin-relative-ref = path-absolute [ "?" query ] [ "#" fragment ]

        with the exception that '%' escapes are not validated and '#' is

        allowed inside the fragment part.

        """

        return _re.match(came_from) is not None

    def index(self):

        c.came_from = safe_str(request.GET.get('came_from', ''))

        if c.came_from:

            if not self._validate_came_from(c.came_from):

                log.error('Invalid came_from (not server-relative): %r', c.came_from)

                raise HTTPBadRequest()

        else:

            c.came_from = url('home')

        if request.POST:

            # import Login Form validator class

            login_form = LoginForm()()

            try:

                c.form_result = login_form.to_python(dict(request.POST))

                # form checks for username/password, now we're authenticated

                username = c.form_result['username']

                user = User.get_by_username_or_email(username, case_insensitive=True)

            except formencode.Invalid as errors:

                defaults = errors.value

                # remove password from filling in form again

                defaults.pop('password', None)

                return htmlfill.render(

                    render('/login.html'),

                    defaults=errors.value,

                    errors=errors.error_dict or {},

                    prefix_error=False,

                    encoding="UTF-8",

                    force_defaults=False)

            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

                h.flash(e, 'error')

            else:

                auth_user = log_in_user(user, c.form_result['remember'], is_external_auth=False, ip_addr=request.ip_addr)

                # TODO: handle auth_user is None as failed authentication?

                raise HTTPFound(location=c.came_from)

        else:

            # redirect if already logged in

            if not request.authuser.is_anonymous:

                raise HTTPFound(location=c.came_from)

            # continue to show login to default user

        return render('/login.html')

    @HasPermissionAnyDecorator('hg.admin', 'hg.register.auto_activate',

                               'hg.register.manual_activate')

    def register(self):

        def_user_perms = AuthUser(dbuser=User.get_default_user()).permissions['global']

        c.auto_active = 'hg.register.auto_activate' in def_user_perms

        settings = Setting.get_app_settings()

        captcha_private_key = settings.get('captcha_private_key')

        c.captcha_active = bool(captcha_private_key)

        c.captcha_public_key = settings.get('captcha_public_key')

        if request.POST:

            register_form = RegisterForm()()

            try:

                form_result = register_form.to_python(dict(request.POST))

                form_result['active'] = c.auto_active

                if c.captcha_active:

                    from kallithea.lib.recaptcha import submit

                    response = submit(request.POST.get('g-recaptcha-response'),

                                      private_key=captcha_private_key,

                                      remoteip=request.ip_addr)

                    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)

                UserModel().create_registration(form_result)

                h.flash(_('You have successfully registered with %s') % (c.site_name or 'Kallithea'),

                        category='success')

                Session().commit()

                raise HTTPFound(location=url('login_home'))

            except formencode.Invalid as errors:

                return htmlfill.render(

                    render('/register.html'),

                    defaults=errors.value,

                    errors=errors.error_dict or {},

                    prefix_error=False,

                    encoding="UTF-8",

                    force_defaults=False)

            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

                h.flash(e, 'error')

        return render('/register.html')

    def password_reset(self):

        settings = Setting.get_app_settings()

        captcha_private_key = settings.get('captcha_private_key')

        c.captcha_active = bool(captcha_private_key)

        c.captcha_public_key = settings.get('captcha_public_key')

        if request.POST:

            password_reset_form = PasswordResetRequestForm()()

            try:

                form_result = password_reset_form.to_python(dict(request.POST))

                if c.captcha_active:

                    from kallithea.lib.recaptcha import submit

                    response = submit(request.POST.get('g-recaptcha-response'),

                                      private_key=captcha_private_key,

                                      remoteip=request.ip_addr)

                    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)

                redirect_link = UserModel().send_reset_password_email(form_result)

                h.flash(_('A password reset confirmation code has been sent'),

                            category='success')

                raise HTTPFound(location=redirect_link)

            except formencode.Invalid as errors:

                return htmlfill.render(

                    render('/password_reset.html'),

                    defaults=errors.value,

                    errors=errors.error_dict or {},

                    prefix_error=False,

                    encoding="UTF-8",

                    force_defaults=False)

        return render('/password_reset.html')

    def password_reset_confirmation(self):

        # This controller handles both GET and POST requests, though we

        # only ever perform the actual password change on POST (since

        # GET requests are not allowed to have side effects, and do not

        # receive automatic CSRF protection).

        # The template needs the email address outside of the form.

        c.email = request.params.get('email')

        if not request.POST:

        form = PasswordResetConfirmationForm()()

        try:

            form_result = form.to_python(dict(request.POST))

        except formencode.Invalid as errors:

            return htmlfill.render(

                render('/password_reset_confirmation.html'),

                defaults=errors.value,

                errors=errors.error_dict or {},

                prefix_error=False,

                encoding='UTF-8')

        if not UserModel().verify_reset_password_token(

            form_result['email'],

            form_result['timestamp'],

            form_result['token'],

        ):

            return htmlfill.render(

                render('/password_reset_confirmation.html'),

                defaults=form_result,

                errors={'token': _('Invalid password reset token')},

                prefix_error=False,

                encoding='UTF-8')

        UserModel().reset_password(form_result['email'], form_result['password'])

        h.flash(_('Successfully updated password'), category='success')

        raise HTTPFound(location=url('login_home'))

    def logout(self):

        session.delete()

        log.info('Logging out and deleting session for user')

        raise HTTPFound(location=url('home'))

    def session_csrf_secret_token(self):

        """Return the CSRF protection token for the session - just like it

        could have been screen scraped from a page with a form.

        Only intended for testing but might also be useful for other kinds

        of automation.

        """

        return h.session_csrf_secret_token()

## -*- coding: utf-8 -*-

<%inherit file="base/root.html"/>

<%block name="title">

    ${_('Reset Your Password')}

</%block>

<%include file="/base/flash_msg.html"/>

<div class="container">

<div class="row">

<div class="centered-column">

<div id="register" class="panel panel-primary">

    <div class="panel-heading">

        %if c.site_name:

            <h5>${_('Reset Your Password to %s') % c.site_name}</h5>

        %else:

            <h5>${_('Reset Your Password')}</h5>

        %endif

    </div>

    <div class="panel-body">

        ${h.form(h.url('reset_password_confirmation'), method='post')}

        <p>${_('You are about to set a new password for the email address %s.') % c.email}</p>

        <p>${_('Note that you must use the same browser session for this as the one used to request the password reset.')}</p>

        <div class="form">

                <div class="form-group">

                    <label class="control-label" for="token">${_('Code you received in the email')}:</label>

                    <div>

                    </div>

                </div>

                <div class="form-group">

                    <label class="control-label" for="password">${_('New Password')}:</label>

                    <div>

                        ${h.password('password',class_='form-control')}

                    </div>

                </div>

                <div class="form-group">

                    <label class="control-label" for="password_confirm">${_('Confirm New Password')}:</label>

                    <div>

                        ${h.password('password_confirm',class_='form-control')}

                    </div>

                </div>

                <div class="form-group">

                    <div class="buttons">

                        ${h.submit('send',_('Confirm'),class_="btn btn-default")}

                    </div>

                </div>

        </div>

        ${h.end_form()}

    </div>

   </div>

</div>

</div>

</div>

        response.mustcontain('Users Administration')

    def test_login_do_not_remember(self):

        response = self.app.post(url(controller='login', action='index'),

                                 {'username': TEST_USER_REGULAR_LOGIN,

                                  'password': TEST_USER_REGULAR_PASS,

                                  'remember': False,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        assert 'Set-Cookie' in response.headers

        for cookie in response.headers.getall('Set-Cookie'):

            assert not re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE), 'Cookie %r has expiration date, but should be a session cookie' % cookie

    def test_login_remember(self):

        response = self.app.post(url(controller='login', action='index'),

                                 {'username': TEST_USER_REGULAR_LOGIN,

                                  'password': TEST_USER_REGULAR_PASS,

                                  'remember': True,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        assert 'Set-Cookie' in response.headers

        for cookie in response.headers.getall('Set-Cookie'):

            assert re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE), 'Cookie %r should have expiration date, but is a session cookie' % cookie

    def test_logout(self):

        response = self.app.post(url(controller='login', action='index'),

                                 {'username': TEST_USER_REGULAR_LOGIN,

                                  'password': TEST_USER_REGULAR_PASS,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        # Verify that a login session has been established.

        response = self.app.get(url(controller='login', action='index'))

        response = response.follow()

        assert 'authuser' in response.session

        response.click('Log Out')

        # Verify that the login session has been terminated.

        response = self.app.get(url(controller='login', action='index'))

        assert 'authuser' not in response.session

    @parametrize('url_came_from', [

          ('data:text/html,<script>window.alert("xss")</script>',),

          ('mailto:test@example.com',),

          ('file:///etc/passwd',),

          ('ftp://ftp.example.com',),

          ('http://other.example.com/bl%C3%A5b%C3%A6rgr%C3%B8d',),

          ('//evil.example.com/',),

          ('/\r\nX-Header-Injection: boo',),

          ('/invälid_url_bytes',),

          ('non-absolute-path',),

    ])

    def test_login_bad_came_froms(self, url_came_from):

        response = self.app.post(url(controller='login', action='index',

                                     came_from=url_came_from),

                                 {'username': TEST_USER_ADMIN_LOGIN,

                                  'password': TEST_USER_ADMIN_PASS,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()},

                                 status=400)

    def test_login_short_password(self):

        response = self.app.post(url(controller='login', action='index'),

                                 {'username': TEST_USER_ADMIN_LOGIN,

                                  'password': 'as',

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        assert response.status == '200 OK'

        response.mustcontain('Enter 3 characters or more')

    def test_login_wrong_username_password(self):

        response = self.app.post(url(controller='login', action='index'),

                                 {'username': 'error',

                                  'password': 'test12',

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        response.mustcontain('Invalid username or password')

    def test_login_non_ascii(self):

        response = self.app.post(url(controller='login', action='index'),

                                 {'username': TEST_USER_REGULAR_LOGIN,

                                  'password': 'blåbærgrød',

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        response.mustcontain('>Invalid username or password<')

    # verify that get arguments are correctly passed along login redirection

    @parametrize('args,args_encoded', [

        ({'foo':'one', 'bar':'two'}, (('foo', 'one'), ('bar', 'two'))),

        ({'blue': u'blå'.encode('utf-8'), 'green':u'grøn'},

             (('blue', u'blå'.encode('utf-8')), ('green', u'grøn'.encode('utf-8')))),

    ])

    def test_redirection_to_login_form_preserves_get_args(self, args, args_encoded):

        with fixture.anon_access(False):

            response = self.app.get(url(controller='summary', action='index',

                                        repo_name=HG_REPO,

                                        **args))

            assert response.status == '302 Found'

            came_from = urlparse.parse_qs(urlparse.urlparse(response.location).query)['came_from'][0]

            came_from_qs = urlparse.parse_qsl(urlparse.urlparse(came_from).query)

            for encoded in args_encoded:

                assert encoded in came_from_qs

    @parametrize('args,args_encoded', [

        ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),

        ({'blue': u'blå', 'green':u'grøn'},

             ('blue=bl%C3%A5', 'green=gr%C3%B8n')),

    ])

    def test_login_form_preserves_get_args(self, args, args_encoded):

        response = self.app.get(url(controller='login', action='index',

                                    came_from=url('/_admin/users', **args)))

        came_from = urlparse.parse_qs(urlparse.urlparse(response.form.action).query)['came_from'][0]

        for encoded in args_encoded:

            assert encoded in came_from

    @parametrize('args,args_encoded', [

        ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),

        ({'blue': u'blå', 'green':u'grøn'},

             ('blue=bl%C3%A5', 'green=gr%C3%B8n')),

    ])

    def test_redirection_after_successful_login_preserves_get_args(self, args, args_encoded):

        response = self.app.post(url(controller='login', action='index',

                                     came_from=url('/_admin/users', **args)),

                                 {'username': TEST_USER_ADMIN_LOGIN,

                                  'password': TEST_USER_ADMIN_PASS,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        assert response.status == '302 Found'

        for encoded in args_encoded:

            assert encoded in response.location

    @parametrize('args,args_encoded', [

        ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),

        ({'blue': u'blå', 'green':u'grøn'},

             ('blue=bl%C3%A5', 'green=gr%C3%B8n')),

    ])

    def test_login_form_after_incorrect_login_preserves_get_args(self, args, args_encoded):

        response = self.app.post(url(controller='login', action='index',

                                     came_from=url('/_admin/users', **args)),

                                 {'username': 'error',

                                  'password': 'test12',

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        response.mustcontain('Invalid username or password')

        came_from = urlparse.parse_qs(urlparse.urlparse(response.form.action).query)['came_from'][0]

        for encoded in args_encoded:

            assert encoded in came_from

    #==========================================================================

    # REGISTRATIONS

    #==========================================================================

    def test_register(self):

        response = self.app.get(url(controller='login', action='register'))

        response.mustcontain('Sign Up')

    def test_register_err_same_username(self):

        uname = TEST_USER_ADMIN_LOGIN

        response = self.app.post(url(controller='login', action='register'),

                                            {'username': uname,

                                             'password': 'test12',

                                             'password_confirmation': 'test12',

                                             'email': 'goodmail@example.com',

                                             'firstname': 'test',

                                             'lastname': 'test',

                                             '_session_csrf_secret_token': self.session_csrf_secret_token()})

        with test_context(self.app):

            msg = validators.ValidUsername()._messages['username_exists']

        msg = h.html_escape(msg % {'username': uname})

        response.mustcontain(msg)

    def test_register_err_same_email(self):

        response = self.app.post(url(controller='login', action='register'),

                                            {'username': 'test_admin_0',

                                             'password': 'test12',

                                             'password_confirmation': 'test12',

                                             'email': TEST_USER_ADMIN_EMAIL,

                                             'firstname': 'test',

                                             'lastname': 'test',

                                             '_session_csrf_secret_token': self.session_csrf_secret_token()})

        with test_context(self.app):

            msg = validators.UniqSystemEmail()()._messages['email_taken']

        response.mustcontain(msg)

    def test_register_err_same_email_case_sensitive(self):

        response = self.app.post(url(controller='login', action='register'),

                                            {'username': 'test_admin_1',

                                             'password': 'test12',

                                             'password_confirmation': 'test12',

                                             'email': TEST_USER_ADMIN_EMAIL.title(),

                                             'firstname': 'test',

                                             'lastname': 'test',

                                             '_session_csrf_secret_token': self.session_csrf_secret_token()})

        with test_context(self.app):

            msg = validators.UniqSystemEmail()()._messages['email_taken']

        response.mustcontain(msg)

    def test_register_err_wrong_data(self):

        response = self.app.post(url(controller='login', action='register'),

                                            {'username': 'xs',

                                             'password': 'test',

                                             'password_confirmation': 'test',

                                             'email': 'goodmailm',

                                             'firstname': 'test',

                                             'lastname': 'test',

                                             '_session_csrf_secret_token': self.session_csrf_secret_token()})

        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(url(controller='login', action='register'),

                                            {'username': 'error user',

                                             'password': 'test12',

                                             'password_confirmation': 'test12',

                                             'email': 'goodmailm',

                                             'firstname': 'test',

                                             'lastname': 'test',

                                             '_session_csrf_secret_token': self.session_csrf_secret_token()})

        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 an '

                'alphanumeric character')

    def test_register_err_case_sensitive(self):

        usr = TEST_USER_ADMIN_LOGIN.title()

        response = self.app.post(url(controller='login', action='register'),

                                            {'username': usr,

                                             'password': 'test12',

                                             'password_confirmation': 'test12',

                                             'email': 'goodmailm',

                                             'firstname': 'test',

                                             'lastname': 'test',

                                             '_session_csrf_secret_token': self.session_csrf_secret_token()})

        response.mustcontain('An email address must contain a single @')

        with test_context(self.app):

            msg = validators.ValidUsername()._messages['username_exists']

        msg = h.html_escape(msg % {'username': usr})

        response.mustcontain(msg)

    def test_register_special_chars(self):

        response = self.app.post(url(controller='login', action='register'),

                                        {'username': 'xxxaxn',

                                         'password': 'ąćźżąśśśś',

                                         'password_confirmation': 'ąćźżąśśśś',

                                         'email': 'goodmailm@test.plx',

                                         'firstname': 'test',

                                         'lastname': 'test',

                                         '_session_csrf_secret_token': self.session_csrf_secret_token()})

        with test_context(self.app):

            msg = validators.ValidPassword()._messages['invalid_password']

        response.mustcontain(msg)

    def test_register_password_mismatch(self):

        response = self.app.post(url(controller='login', action='register'),

                                            {'username': 'xs',

                                             'password': '123qwe',

                                             'password_confirmation': 'qwe123',

                                             'email': 'goodmailm@test.plxa',

                                             'firstname': 'test',

                                             'lastname': 'test',

                                             '_session_csrf_secret_token': self.session_csrf_secret_token()})

        with test_context(self.app):

            msg = validators.ValidPasswordsMatch('password', 'password_confirmation')._messages['password_mismatch']

        response.mustcontain(msg)

    def test_register_ok(self):

        username = 'test_regular4'

        password = 'qweqwe'

        email = 'user4@example.com'

        name = 'testname'

        lastname = 'testlastname'

        response = self.app.post(url(controller='login', action='register'),

                                            {'username': username,

                                             'password': password,

                                             'password_confirmation': password,

                                             'email': email,

                                             'firstname': name,

                                             'lastname': lastname,

                                             'admin': True,

                                             '_session_csrf_secret_token': self.session_csrf_secret_token()})  # This should be overridden

        assert response.status == '302 Found'

        self.checkSessionFlash(response, 'You have successfully registered with Kallithea')

        ret = Session().query(User).filter(User.username == 'test_regular4').one()

        assert ret.username == username

        assert check_password(password, ret.password) == True

        assert ret.email == email

        assert ret.name == name

        assert ret.lastname == lastname

        assert ret.api_key is not None

        assert ret.admin == False

    #==========================================================================

    # PASSWORD RESET

    #==========================================================================

    def test_forgot_password_wrong_mail(self):

        bad_email = 'username%wrongmail.org'

        response = self.app.post(

                        url(controller='login', action='password_reset'),

                            {'email': bad_email,

                             '_session_csrf_secret_token': self.session_csrf_secret_token()})

        response.mustcontain('An email address must contain a single @')

    def test_forgot_password(self):

        response = self.app.get(url(controller='login',

                                    action='password_reset'))

        assert response.status == '200 OK'

        username = 'test_password_reset_1'

        password = 'qweqwe'

        email = 'username@example.com'

        name = u'passwd'

        lastname = u'reset'

        timestamp = int(time.time())

        new = User()

        new.username = username

        new.password = password

        new.email = email

        new.name = name

        new.lastname = lastname

        new.api_key = generate_api_key()

        Session().add(new)

        Session().commit()

        token = UserModel().get_reset_password_token(

            User.get_by_username(username), timestamp, self.session_csrf_secret_token())

        collected = []

        def mock_send_email(recipients, subject, body='', html_body='', headers=None, author=None):

            collected.append((recipients, subject, body, html_body))

        with mock.patch.object(kallithea.lib.celerylib.tasks, 'send_email', mock_send_email):

            response = self.app.post(url(controller='login',

                                         action='password_reset'),

                                     {'email': email,

                                      '_session_csrf_secret_token': self.session_csrf_secret_token()})

        self.checkSessionFlash(response, 'A password reset confirmation code has been sent')

        ((recipients, subject, body, html_body),) = collected

        assert recipients == ['username@example.com']

        assert subject == 'Password reset link'

        assert '\n%s\n' % token in body

        (confirmation_url,) = (line for line in body.splitlines() if line.startswith('http://'))

        assert ' href="%s"' % confirmation_url.replace('&', '&').replace('@', '%40') in html_body

        d = urlparse.parse_qs(urlparse.urlparse(confirmation_url).query)

        assert d['token'] == [token]

        assert d['timestamp'] == [str(timestamp)]

        assert d['email'] == [email]

        response = response.follow()

        # BAD TOKEN

        bad_token = "bad"

        response = self.app.post(url(controller='login',

                                     action='password_reset_confirmation'),

                                 {'email': email,

                                  'timestamp': timestamp,

                                  'password': "p@ssw0rd",

                                  'password_confirm': "p@ssw0rd",

                                  'token': bad_token,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token(),

                                 })

        assert response.status == '200 OK'

        response.mustcontain('Invalid password reset token')

        # GOOD TOKEN

        response = self.app.get(confirmation_url)

        assert response.status == '200 OK'

        response.mustcontain("You are about to set a new password for the email address %s" % email)

        response.mustcontain('<form action="%s" method="post">' % url(controller='login', action='password_reset_confirmation'))

        response.mustcontain('value="%s"' % token)

        response.mustcontain('value="%s"' % timestamp)

        response.mustcontain('value="username@example.com"')

        response = self.app.post(url(controller='login',

                                     action='password_reset_confirmation'),

                                 {'email': email,

                                  'timestamp': timestamp,

                                  'password': "p@ssw0rd",

                                  'password_confirm': "p@ssw0rd",

                                  'token': token,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token(),

                                 })

        assert response.status == '302 Found'

        self.checkSessionFlash(response, 'Successfully updated password')

        response = response.follow()

    #==========================================================================

    # API

    #==========================================================================

    def _api_key_test(self, api_key, status):

        """Verifies HTTP status code for accessing an auth-requiring page,

        using the given api_key URL parameter as well as using the API key

        with bearer authentication.

        If api_key is None, no api_key is passed at all. If api_key is True,

        a real, working API key is used.

        """

        with fixture.anon_access(False):

            if api_key is None:

                params = {}

                headers = {}

            else:

                if api_key is True:

                    api_key = User.get_first_admin().api_key

                params = {'api_key': api_key}

                headers = {'Authorization': 'Bearer ' + str(api_key)}

            self.app.get(url(controller='changeset', action='changeset_raw',

                             repo_name=HG_REPO, revision='tip', **params),

                         status=status)

            self.app.get(url(controller='changeset', action='changeset_raw',

                             repo_name=HG_REPO, revision='tip'),

                         headers=headers,

                         status=status)

    @parametrize('test_name,api_key,code', [

        ('none', None, 302),

        ('empty_string', '', 403),

        ('fake_number', '123456', 403),

        ('fake_not_alnum', 'a-z', 403),

        ('fake_api_key', '0123456789abcdef0123456789ABCDEF01234567', 403),

        ('proper_api_key', True, 200)

    ])

    def test_access_page_via_api_key(self, test_name, api_key, code):

        self._api_key_test(api_key, code)

    def test_access_page_via_extra_api_key(self):

        new_api_key = ApiKeyModel().create(TEST_USER_ADMIN_LOGIN, u'test')

        Session().commit()

        self._api_key_test(new_api_key.api_key, status=200)

    def test_access_page_via_expired_api_key(self):

        new_api_key = ApiKeyModel().create(TEST_USER_ADMIN_LOGIN, u'test')

        Session().commit()

        # patch the API key and make it expired

        new_api_key.expires = 0

        Session().commit()

        self._api_key_test(new_api_key.api_key, status=403)