# -*- 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 pylons.i18n.translation import _

from pylons import request, session, tmpl_context as c, url

from webob.exc import HTTPFound, HTTPBadRequest

import kallithea.lib.helpers as h

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 User, Setting

from kallithea.model.forms import \

    LoginForm, RegisterForm, PasswordResetRequestForm, PasswordResetConfirmationForm

from kallithea.model.user import UserModel

from kallithea.model.meta import Session

log = logging.getLogger(__name__)

class LoginController(BaseController):

    def __before__(self):

        super(LoginController, self).__before__()

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

        ip_allowed = AuthUser.check_ip_allowed(self.authuser, self.ip_addr)

        # redirect if already logged in

        if self.authuser.is_authenticated and ip_allowed:

            raise HTTPFound(location=c.came_from)

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

            except formencode.Invalid as errors:

                defaults = errors.value

                # remove password from filling in form again

                del defaults['password']

                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:

                log_in_user(user, c.form_result['remember'],

                    is_external_auth=False)

                raise HTTPFound(location=c.came_from)

        return render('/login.html')

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

                               'hg.register.manual_activate')

    def register(self):

        c.auto_active = 'hg.register.auto_activate' in User.get_default_user() \

            .AuthUser.permissions['global']

        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('recaptcha_challenge_field'),

                                      request.POST.get('recaptcha_response_field'),

                                      private_key=captcha_private_key,

                                      remoteip=self.ip_addr)

                    if c.captcha_active and 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 into 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('recaptcha_challenge_field'),

                                      request.POST.get('recaptcha_response_field'),

                                      private_key=captcha_private_key,

                                      remoteip=self.ip_addr)

                    if c.captcha_active and not response.is_valid:

                        _value = form_result

                        _msg = _('Bad captcha')

                )

            old_gname = None

            if edit:

                old_gname = RepoGroup.get(old_data.get('group_id')).group_name

            if old_gname != group_name or not edit:

                # check group

                gr = RepoGroup.query() \

                      .filter(RepoGroup.group_name == slug) \

                      .filter(RepoGroup.group_parent_id == group_parent_id) \

                      .scalar()

                if gr is not None:

                    msg = M(self, 'group_exists', state, group_name=slug)

                    raise formencode.Invalid(msg, value, state,

                            error_dict=dict(group_name=msg)

                    )

                # check for same repo

                repo = Repository.query() \

                      .filter(Repository.repo_name == slug) \

                      .scalar()

                if repo is not None:

                    msg = M(self, 'repo_exists', state, group_name=slug)

                    raise formencode.Invalid(msg, value, state,

                            error_dict=dict(group_name=msg)

                    )

    return _validator

def ValidPassword():

    class _validator(formencode.validators.FancyValidator):

        messages = {

            'invalid_password':

                _('Invalid characters (non-ascii) in password')

        }

        def validate_python(self, value, state):

            try:

                (value or '').decode('ascii')

            except UnicodeError:

                msg = M(self, 'invalid_password', state)

                raise formencode.Invalid(msg, value, state,)

    return _validator

def ValidOldPassword(username):

    class _validator(formencode.validators.FancyValidator):

        messages = {

            'invalid_password': _('Invalid old password')

        }

        def validate_python(self, value, state):

            from kallithea.lib import auth_modules

            if auth_modules.authenticate(username, value, '') is None:

                msg = M(self, 'invalid_password', state)

                raise formencode.Invalid(msg, value, state,

                    error_dict=dict(current_password=msg)

                )

    return _validator

def ValidPasswordsMatch(password_field, password_confirmation_field):

    class _validator(formencode.validators.FancyValidator):

        messages = {

            'password_mismatch': _('Passwords do not match'),

        }

        def validate_python(self, value, state):

            if value.get(password_field) != value[password_confirmation_field]:

                msg = M(self, 'password_mismatch', state)

                raise formencode.Invalid(msg, value, state,

                     error_dict={password_field:msg, password_confirmation_field: msg}

                )

    return _validator

def ValidAuth():

    class _validator(formencode.validators.FancyValidator):

        messages = {

            'invalid_auth': _(u'Invalid username or password'),

        }

        def validate_python(self, value, state):

            from kallithea.lib import auth_modules

            password = value['password']

            username = value['username']

            # authenticate returns unused dict but has called

            # plugin._authenticate which has create_or_update'ed the username user in db

            if auth_modules.authenticate(username, password) is None:

                if user and not user.active:

                    log.warning('user %s is disabled', username)

                    msg = M(self, 'invalid_auth', state)

                    raise formencode.Invalid(msg, value, state,

                        error_dict=dict(username=' ', password=msg)

                    )

                else:

                    log.warning('user %s failed to authenticate', username)

                    msg = M(self, 'invalid_auth', state)

                    raise formencode.Invalid(msg, value, state,

                        error_dict=dict(username=' ', password=msg)

                    )

    return _validator

def ValidAuthToken():

    class _validator(formencode.validators.FancyValidator):

        messages = {

            'invalid_token': _('Token mismatch')

        }

        def validate_python(self, value, state):

            if value != authentication_token():

                msg = M(self, 'invalid_token', state)

                raise formencode.Invalid(msg, value, state)

    return _validator

def ValidRepoName(edit=False, old_data=None):

    old_data = old_data or {}

    class _validator(formencode.validators.FancyValidator):

        messages = {

            'invalid_repo_name':

                _('Repository name %(repo)s is not allowed'),

            'repository_exists':

                _('Repository named %(repo)s already exists'),

            'repository_in_group_exists': _('Repository "%(repo)s" already '

                                            'exists in group "%(group)s"'),

            'same_group_exists': _('Repository group with name "%(repo)s" '

                                   'already exists')

        }

        def _to_python(self, value, state):

            repo_name = repo_name_slug(value.get('repo_name', ''))

            repo_group = value.get('repo_group')

            if repo_group:

                gr = RepoGroup.get(repo_group)

                group_path = gr.full_path

                group_name = gr.group_name

                # value needs to be aware of group name in order to check

                # db key This is an actual just the name to store in the

                # database

                repo_name_full = group_path + RepoGroup.url_sep() + repo_name

            else:

                group_name = group_path = ''

                repo_name_full = repo_name

            value['repo_name'] = repo_name

            value['repo_name_full'] = repo_name_full

            value['group_path'] = group_path

            value['group_name'] = group_name

            return value

        def validate_python(self, value, state):

            repo_name = value.get('repo_name')

            repo_name_full = value.get('repo_name_full')

            group_path = value.get('group_path')

            group_name = value.get('group_name')

            if repo_name in [ADMIN_PREFIX, '']:

                msg = M(self, 'invalid_repo_name', state, repo=repo_name)

                raise formencode.Invalid(msg, value, state,

                    error_dict=dict(repo_name=msg)

                )

            rename = old_data.get('repo_name') != repo_name_full

            create = not edit

            if rename or create:

                if group_path != '':

                    if Repository.get_by_repo_name(repo_name_full):

                        msg = M(self, 'repository_in_group_exists', state,

                                repo=repo_name, group=group_name)

                        raise formencode.Invalid(msg, value, state,

                            error_dict=dict(repo_name=msg)

                        )

                elif RepoGroup.get_by_group_name(repo_name_full):

                        msg = M(self, 'same_group_exists', state,

                                repo=repo_name)

                        raise formencode.Invalid(msg, value, state,

                            error_dict=dict(repo_name=msg)

                        )

                elif Repository.get_by_repo_name(repo_name_full):

                        msg = M(self, 'repository_exists', state,

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

import re

import time

import urlparse

import mock

from kallithea.tests 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, Notification

from kallithea.model.meta import Session

from kallithea.model.user import UserModel

fixture = Fixture()

class TestLoginController(TestController):

    def setUp(self):

        self.remove_all_notifications()

        self.assertEqual(Notification.query().all(), [])

    def test_index(self):

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

        self.assertEqual(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})

        self.assertEqual(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})

        self.assertEqual(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})

        self.assertEqual(response.status, '302 Found')

        response = response.follow()

        self.assertEqual(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})

        self.assertIn('Set-Cookie', response.headers)

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

            self.assertFalse(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})

        self.assertIn('Set-Cookie', response.headers)

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

            self.assertTrue(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})

        # Verify that a login session has been established.

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

        response = response.follow()

        self.assertIn('authuser', response.session)

        response.click('Log Out')

        # Verify that the login session has been terminated.

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

        self.assertNotIn('authuser', response.session)

    @parameterized.expand([

          ('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},

                                 status=400)

    def test_login_short_password(self):

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

                                 {'username': TEST_USER_ADMIN_LOGIN,

                                  'password': 'as'})

        self.assertEqual(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'})

        response.mustcontain('Invalid username or password')

    # verify that get arguments are correctly passed along login redirection

    @parameterized.expand([

        ({'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))

            self.assertEqual(response.status, '302 Found')