Changeset - 9c2fc1390291
[Not reviewed]
default
0 2 0
Mads Kiilerich - 7 years ago 2019-01-04 03:42:17
mads@kiilerich.com
tests: prepare for adding CSRF protection on login forms

CSRF is about avoiding abuse of credentials by doing things in existing
sessions. The login form does not have any previous credentials, so there is
nothing to abuse and no real need for CSRF protection. But there is still an
unauth session, so we *can* have CSRF protection.

CSRF protection is currently in LoginRequired (which obviously isn't
applied to the login form), but let's prepare for changing that.
2 files changed with 52 insertions and 26 deletions:
0 comments (0 inline, 0 general)
kallithea/tests/base.py
Show inline comments
 
@@ -61,135 +61,136 @@ TEST_USER_ADMIN_EMAIL = 'test_admin@exam
 

	
 
TEST_USER_REGULAR_LOGIN = 'test_regular'
 
TEST_USER_REGULAR_PASS = 'test12'
 
TEST_USER_REGULAR_EMAIL = 'test_regular@example.com'
 

	
 
TEST_USER_REGULAR2_LOGIN = 'test_regular2'
 
TEST_USER_REGULAR2_PASS = 'test12'
 
TEST_USER_REGULAR2_EMAIL = 'test_regular2@example.com'
 

	
 
HG_REPO = u'vcs_test_hg'
 
GIT_REPO = u'vcs_test_git'
 

	
 
NEW_HG_REPO = u'vcs_test_hg_new'
 
NEW_GIT_REPO = u'vcs_test_git_new'
 

	
 
HG_FORK = u'vcs_test_hg_fork'
 
GIT_FORK = u'vcs_test_git_fork'
 

	
 
HG_TEST_REVISION = u"a53d9201d4bc278910d416d94941b7ea007ecd52"
 
GIT_TEST_REVISION = u"7ab37bc680b4aa72c34d07b230c866c28e9fc204"
 

	
 

	
 
## VCS
 
uniq_suffix = str(int(time.mktime(datetime.datetime.now().timetuple())))
 

	
 
GIT_REMOTE_REPO = os.path.join(TESTS_TMP_PATH, GIT_REPO)
 

	
 
TEST_GIT_REPO = os.path.join(TESTS_TMP_PATH, GIT_REPO)
 
TEST_GIT_REPO_CLONE = os.path.join(TESTS_TMP_PATH, 'vcs-git-clone-%s' % uniq_suffix)
 
TEST_GIT_REPO_PULL = os.path.join(TESTS_TMP_PATH, 'vcs-git-pull-%s' % uniq_suffix)
 

	
 
HG_REMOTE_REPO = os.path.join(TESTS_TMP_PATH, HG_REPO)
 

	
 
TEST_HG_REPO = os.path.join(TESTS_TMP_PATH, HG_REPO)
 
TEST_HG_REPO_CLONE = os.path.join(TESTS_TMP_PATH, 'vcs-hg-clone-%s' % uniq_suffix)
 
TEST_HG_REPO_PULL = os.path.join(TESTS_TMP_PATH, 'vcs-hg-pull-%s' % uniq_suffix)
 

	
 
# By default, some of the tests will utilise locally available
 
# repositories stored within tar.gz archives as source for
 
# cloning. Should you wish to use some other, remote archive, simply
 
# uncomment these entries and/or update the URLs to use.
 
#
 
# GIT_REMOTE_REPO = 'git://github.com/codeinn/vcs.git'
 
# HG_REMOTE_REPO = 'http://bitbucket.org/marcinkuzminski/vcs'
 

	
 
# skip ldap tests if LDAP lib is not installed
 
ldap_lib_installed = False
 
try:
 
    import ldap
 
    ldap.API_VERSION
 
    ldap_lib_installed = True
 
except ImportError:
 
    # means that python-ldap is not installed
 
    pass
 

	
 
try:
 
    import pam
 
    pam.PAM_TEXT_INFO
 
    pam_lib_installed = True
 
except ImportError:
 
    pam_lib_installed = False
 

	
 

	
 
def invalidate_all_caches():
 
    """Invalidate all beaker caches currently configured.
 
    Useful when manipulating IP permissions in a test and changes need to take
 
    effect immediately.
 
    Note: Any use of this function is probably a workaround - it should be
 
    replaced with a more specific cache invalidation in code or test."""
 
    from beaker.cache import cache_managers
 
    for cache in cache_managers.values():
 
        cache.clear()
 

	
 

	
 
class NullHandler(logging.Handler):
 
    def emit(self, record):
 
        pass
 

	
 

	
 
class TestController(object):
 
    """Pytest-style test controller"""
 

	
 
    # Note: pytest base classes cannot have an __init__ method
 

	
 
    @pytest.fixture(autouse=True)
 
    def app_fixture(self):
 
        h = NullHandler()
 
        logging.getLogger("kallithea").addHandler(h)
 
        self.app = TestApp(testapp)
 
        return self.app
 

	
 
    def log_user(self, username=TEST_USER_ADMIN_LOGIN,
 
                 password=TEST_USER_ADMIN_PASS):
 
        self._logged_username = username
 
        response = self.app.post(url(controller='login', action='index'),
 
                                 {'username': username,
 
                                  'password': password})
 
                                  'password': password,
 
                                  '_authentication_token': self.authentication_token()})
 

	
 
        if 'Invalid username or password' in response.body:
 
            pytest.fail('could not login using %s %s' % (username, password))
 

	
 
        assert response.status == '302 Found'
 
        self.assert_authenticated_user(response, username)
 

	
 
        response = response.follow()
 
        return response.session['authuser']
 

	
 
    def _get_logged_user(self):
 
        return User.get_by_username(self._logged_username)
 

	
 
    def assert_authenticated_user(self, response, expected_username):
 
        cookie = response.session.get('authuser')
 
        user = cookie and cookie.get('user_id')
 
        user = user and User.get(user)
 
        user = user and user.username
 
        assert user == expected_username
 

	
 
    def authentication_token(self):
 
        return self.app.get(url('authentication_token')).body
 

	
 
    def checkSessionFlash(self, response, msg=None, skip=0, _matcher=lambda msg, m: msg in m):
 
        if 'flash' not in response.session:
 
            pytest.fail(safe_str(u'msg `%s` not found - session has no flash:\n%s' % (msg, response)))
 
        try:
 
            level, m = response.session['flash'][-1 - skip]
 
            if _matcher(msg, m):
 
                return
 
        except IndexError:
 
            pass
 
        pytest.fail(safe_str(u'msg `%s` not found in session flash (skipping %s): %s' %
 
                           (msg, skip,
 
                            ', '.join('`%s`' % m for level, m in response.session['flash']))))
 

	
 
    def checkSessionFlashRegex(self, response, regex, skip=0):
 
        self.checkSessionFlash(response, regex, skip=skip, _matcher=re.search)
kallithea/tests/functional/test_login.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
import re
 
import time
 
import urlparse
 

	
 
import mock
 

	
 
from kallithea.tests.base import *
 
from kallithea.tests.fixture import Fixture
 
from kallithea.lib.utils2 import generate_api_key
 
from kallithea.lib.auth import check_password
 
from kallithea.lib import helpers as h
 
from kallithea.model.api_key import ApiKeyModel
 
from kallithea.model import validators
 
from kallithea.model.db import User
 
from kallithea.model.meta import Session
 
from kallithea.model.user import UserModel
 

	
 
from tg.util.webtest import test_context
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestLoginController(TestController):
 

	
 
    def test_index(self):
 
        response = self.app.get(url(controller='login', action='index'))
 
        assert response.status == '200 OK'
 
        # Test response...
 

	
 
    def test_login_admin_ok(self):
 
        response = self.app.post(url(controller='login', action='index'),
 
                                 {'username': TEST_USER_ADMIN_LOGIN,
 
                                  'password': TEST_USER_ADMIN_PASS})
 
                                  'password': TEST_USER_ADMIN_PASS,
 
                                  '_authentication_token': self.authentication_token()})
 
        assert response.status == '302 Found'
 
        self.assert_authenticated_user(response, TEST_USER_ADMIN_LOGIN)
 

	
 
        response = response.follow()
 
        response.mustcontain('/%s' % HG_REPO)
 

	
 
    def test_login_regular_ok(self):
 
        response = self.app.post(url(controller='login', action='index'),
 
                                 {'username': TEST_USER_REGULAR_LOGIN,
 
                                  'password': TEST_USER_REGULAR_PASS})
 
                                  'password': TEST_USER_REGULAR_PASS,
 
                                  '_authentication_token': self.authentication_token()})
 

	
 
        assert response.status == '302 Found'
 
        self.assert_authenticated_user(response, TEST_USER_REGULAR_LOGIN)
 

	
 
        response = response.follow()
 
        response.mustcontain('/%s' % HG_REPO)
 

	
 
    def test_login_regular_email_ok(self):
 
        response = self.app.post(url(controller='login', action='index'),
 
                                 {'username': TEST_USER_REGULAR_EMAIL,
 
                                  'password': TEST_USER_REGULAR_PASS})
 
                                  'password': TEST_USER_REGULAR_PASS,
 
                                  '_authentication_token': self.authentication_token()})
 

	
 
        assert response.status == '302 Found'
 
        self.assert_authenticated_user(response, TEST_USER_REGULAR_LOGIN)
 

	
 
        response = response.follow()
 
        response.mustcontain('/%s' % HG_REPO)
 

	
 
    def test_login_ok_came_from(self):
 
        test_came_from = '/_admin/users'
 
        response = self.app.post(url(controller='login', action='index',
 
                                     came_from=test_came_from),
 
                                 {'username': TEST_USER_ADMIN_LOGIN,
 
                                  'password': TEST_USER_ADMIN_PASS})
 
                                  'password': TEST_USER_ADMIN_PASS,
 
                                  '_authentication_token': self.authentication_token()})
 
        assert response.status == '302 Found'
 
        response = response.follow()
 

	
 
        assert response.status == '200 OK'
 
        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})
 
                                  'remember': False,
 
                                  '_authentication_token': self.authentication_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})
 
                                  'remember': True,
 
                                  '_authentication_token': self.authentication_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})
 
                                  'password': TEST_USER_REGULAR_PASS,
 
                                  '_authentication_token': self.authentication_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},
 
                                  'password': TEST_USER_ADMIN_PASS,
 
                                  '_authentication_token': self.authentication_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'})
 
                                  'password': 'as',
 
                                  '_authentication_token': self.authentication_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'})
 
                                  'password': 'test12',
 
                                  '_authentication_token': self.authentication_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'})
 
                                  'password': 'blåbærgrød',
 
                                  '_authentication_token': self.authentication_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})
 
                                  'password': TEST_USER_ADMIN_PASS,
 
                                  '_authentication_token': self.authentication_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'})
 
                                  'password': 'test12',
 
                                  '_authentication_token': self.authentication_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'})
 
                                             'lastname': 'test',
 
                                             '_authentication_token': self.authentication_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'})
 
                                             'lastname': 'test',
 
                                             '_authentication_token': self.authentication_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'})
 
                                             'lastname': 'test',
 
                                             '_authentication_token': self.authentication_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'})
 
                                             'lastname': 'test',
 
                                             '_authentication_token': self.authentication_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'})
 
                                             'lastname': 'test',
 
                                             '_authentication_token': self.authentication_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'})
 
                                             'lastname': 'test',
 
                                             '_authentication_token': self.authentication_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'})
 
                                         'lastname': 'test',
 
                                         '_authentication_token': self.authentication_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'})
 
                                             'lastname': 'test',
 
                                             '_authentication_token': self.authentication_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})  # This should be overridden
 
                                             'admin': True,
 
                                             '_authentication_token': self.authentication_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, }
 
        )
 
                            {'email': bad_email,
 
                             '_authentication_token': self.authentication_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()
 

	
 
        response = self.app.post(url(controller='login',
 
                                     action='password_reset'),
 
                                 {'email': email, })
 
                                 {'email': email,
 
                                  '_authentication_token': self.authentication_token()})
 

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

	
 
        response = response.follow()
 

	
 
        # BAD TOKEN
 

	
 
        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': token,
 
                                  '_authentication_token': self.authentication_token(),
 
                                 })
 
        assert response.status == '200 OK'
 
        response.mustcontain('Invalid password reset token')
 

	
 
        # GOOD TOKEN
 

	
 
        # TODO: The token should ideally be taken from the mail sent
 
        # above, instead of being recalculated.
 

	
 
        token = UserModel().get_reset_password_token(
 
            User.get_by_username(username), timestamp, self.authentication_token())
 

	
 
        response = self.app.get(url(controller='login',
 
                                    action='password_reset_confirmation',
 
                                    email=email,
 
                                    timestamp=timestamp,
 
                                    token=token))
 
        assert response.status == '200 OK'
 
        response.mustcontain("You are about to set a new password for the email address %s" % email)
 

	
 
        response = self.app.post(url(controller='login',
 
                                     action='password_reset_confirmation'),
 
                                 {'email': email,
 
                                  'timestamp': timestamp,
 
                                  'password': "p@ssw0rd",
 
                                  'password_confirm': "p@ssw0rd",
 
                                  'token': token,
 
                                  '_authentication_token': self.authentication_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)
0 comments (0 inline, 0 general)