Changeset - 4c5c59b96adc
[Not reviewed]
default
0 4 0
Thomas De Schampheleire - 11 years ago 2015-05-19 21:50:35
thomas.de.schampheleire@gmail.com
login: preserve GET arguments throughout login redirection (issue #104)

When redirecting a user to the login page and while handling this login and
redirecting to the original page, the GET arguments passed to the original
URL are lost through the login redirection process.

For example, when creating a pull request for a specific revision from the
repository changelog, there are rev_start and rev_end arguments passed in
the URL. Through the login redirection, they are lost.

Fix the issue by passing along the GET arguments to the login page, in the
login form action, and when redirecting back to the original page.
Tests are added to cover these cases, including tests with unicode GET
arguments (in URL encoding).
4 files changed with 75 insertions and 7 deletions:
0 comments (0 inline, 0 general)
kallithea/controllers/login.py
Show inline comments
 
@@ -21,48 +21,49 @@ This file was forked by the Kallithea pr
 
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 formencode
 
import datetime
 
import urlparse
 

	
 
from formencode import htmlfill
 
from webob.exc import HTTPFound
 
from pylons.i18n.translation import _
 
from pylons.controllers.util import redirect
 
from pylons import request, session, tmpl_context as c, url
 

	
 
import kallithea.lib.helpers as h
 
from kallithea.lib.auth import AuthUser, HasPermissionAnyDecorator
 
from kallithea.lib.auth_modules import importplugin
 
from kallithea.lib.base import BaseController, 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, PasswordResetForm
 
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 _store_user_in_session(self, username, remember=False):
 
        user = User.get_by_username(username, case_insensitive=True)
 
        auth_user = AuthUser(user.user_id)
 
        auth_user.set_authenticated()
 
        cs = auth_user.get_cookie_store()
 
        session['authuser'] = cs
 
        user.update_lastlogin()
 
        Session().commit()
 

	
 
        # If they want to be remembered, update the cookie
 
@@ -81,104 +82,111 @@ class LoginController(BaseController):
 
        # we set new cookie
 
        headers = None
 
        if session.request['set_cookie']:
 
            # send set-cookie headers back to response to update cookie
 
            headers = [('Set-Cookie', session.request['cookie_out'])]
 
        return headers
 

	
 
    def _validate_came_from(self, came_from):
 
        if not came_from:
 
            return came_from
 

	
 
        parsed = urlparse.urlparse(came_from)
 
        server_parsed = urlparse.urlparse(url.current())
 
        allowed_schemes = ['http', 'https']
 
        if parsed.scheme and parsed.scheme not in allowed_schemes:
 
            log.error('Suspicious URL scheme detected %s for url %s' %
 
                     (parsed.scheme, parsed))
 
            came_from = url('home')
 
        elif server_parsed.netloc != parsed.netloc:
 
            log.error('Suspicious NETLOC detected %s for url %s server url '
 
                      'is: %s' % (parsed.netloc, parsed, server_parsed))
 
            came_from = url('home')
 
        return came_from
 

	
 
    def _redirect_to_origin(self, origin, headers=None):
 
        '''redirect to the original page, preserving any get arguments given'''
 
        request.GET.pop('came_from', None)
 
        raise HTTPFound(location=url(origin, **request.GET), headers=headers)
 

	
 
    def index(self):
 
        _default_came_from = url('home')
 
        came_from = self._validate_came_from(request.GET.get('came_from'))
 
        came_from = self._validate_came_from(safe_str(request.GET.get('came_from', '')))
 
        c.came_from = came_from or _default_came_from
 

	
 
        not_default = self.authuser.username != User.DEFAULT_USER
 
        ip_allowed = self.authuser.ip_allowed
 

	
 
        # redirect if already logged in
 
        if self.authuser.is_authenticated and not_default and ip_allowed:
 
            raise HTTPFound(location=c.came_from)
 
            return self._redirect_to_origin(c.came_from)
 

	
 
        if request.POST:
 
            # import Login Form validator class
 
            login_form = LoginForm()
 
            try:
 
                session.invalidate()
 
                c.form_result = login_form.to_python(dict(request.POST))
 
                # form checks for username/password, now we're authenticated
 
                headers = self._store_user_in_session(
 
                                        username=c.form_result['username'],
 
                                        remember=c.form_result['remember'])
 
                raise HTTPFound(location=c.came_from, headers=headers)
 
                return self._redirect_to_origin(c.came_from, headers)
 

	
 
            except formencode.Invalid, 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, 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')
 

	
 
        # check if we use container plugin, and try to login using it.
 
        auth_plugins = Setting.get_auth_plugins()
 
        if any((importplugin(name).is_container_auth for name in auth_plugins)):
 
            from kallithea.lib import auth_modules
 
            try:
 
                auth_info = auth_modules.authenticate('', '', request.environ)
 
            except UserCreationError, e:
 
                log.error(e)
 
                h.flash(e, 'error')
 
                # render login, with flash message about limit
 
                return render('/login.html')
 

	
 
            if auth_info:
 
                headers = self._store_user_in_session(auth_info.get('username'))
 
                raise HTTPFound(location=c.came_from, headers=headers)
 
                return self._redirect_to_origin(c.came_from, headers)
 

	
 
        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,
kallithea/lib/auth.py
Show inline comments
 
@@ -690,49 +690,49 @@ def set_available_permissions(config):
 
    ie. to decorate new views with the newly created permission
 

	
 
    :param config: current pylons config instance
 

	
 
    """
 
    log.info('getting information about all available permissions')
 
    try:
 
        sa = meta.Session
 
        all_perms = sa.query(Permission).all()
 
        config['available_permissions'] = [x.permission_name for x in all_perms]
 
    finally:
 
        meta.Session.remove()
 

	
 

	
 
#==============================================================================
 
# CHECK DECORATORS
 
#==============================================================================
 

	
 
def redirect_to_login(message=None):
 
    from kallithea.lib import helpers as h
 
    p = url.current()
 
    if message:
 
        h.flash(h.literal(message), category='warning')
 
    log.debug('Redirecting to login page, origin: %s' % p)
 
    return redirect(url('login_home', came_from=p))
 
    return redirect(url('login_home', came_from=p, **request.GET))
 

	
 
class LoginRequired(object):
 
    """
 
    Must be logged in to execute this function else
 
    redirect to login page
 

	
 
    :param api_access: if enabled this checks only for valid auth token
 
        and grants access based on valid token
 
    """
 

	
 
    def __init__(self, api_access=False):
 
        self.api_access = api_access
 

	
 
    def __call__(self, func):
 
        return decorator(self.__wrapper, func)
 

	
 
    def __wrapper(self, func, *fargs, **fkwargs):
 
        cls = fargs[0]
 
        user = cls.authuser
 
        loc = "%s:%s" % (cls.__class__.__name__, func.__name__)
 
        log.debug('Checking access for user %s @ %s' % (user, loc))
 

	
 
        # check if our IP is allowed
 
        if not user.ip_allowed:
kallithea/templates/login.html
Show inline comments
 
## -*- coding: utf-8 -*-
 
<%inherit file="base/root.html"/>
 

	
 
<%block name="title">
 
    ${_('Log In')}
 
</%block>
 

	
 
<div id="login" class="panel panel-default">
 
    <%include file="/base/flash_msg.html"/>
 
    <!-- login -->
 
    <div class="panel-heading title withlogo">
 
        %if c.site_name:
 
            <h5>${_('Log In to %s') % c.site_name}</h5>
 
        %else:
 
            <h5>${_('Log In')}</h5>
 
        %endif
 
    </div>
 
    <div class="panel-body inner">
 
        ${h.form(h.url.current(came_from=c.came_from))}
 
        ${h.form(h.url.current(**request.GET))}
 
        <div class="form">
 
            <i class="icon-lock"></i>
 
            <!-- fields -->
 

	
 
            <div class="form-horizontal">
 
                <div class="form-group">
 
                    <label class="control-label col-sm-5" for="username">${_('Username')}:</label>
 
                    <div class="input col-sm-7">
 
                        ${h.text('username',class_='form-control focus large')}
 
                    </div>
 

	
 
                </div>
 
                <div class="form-group">
 
                    <label class="control-label col-sm-5" for="password">${_('Password')}:</label>
 
                    <div class="input col-sm-7">
 
                        ${h.password('password',class_='form-control focus large')}
 
                    </div>
 

	
 
                </div>
 
                <div class="form-group">
 
                    <div class="col-sm-offset-5 col-sm-7">
 
                        <div class="checkbox">
 
                            <label for="remember">
 
                                <input type="checkbox" id="remember" name="remember"/>
kallithea/tests/functional/test_login.py
Show inline comments
 
@@ -45,78 +45,138 @@ class TestLoginController(TestController
 

	
 
        self.assertEqual(response.status, '302 Found')
 
        self.assertEqual(response.session['authuser'].get('username'),
 
                         'test_regular')
 
        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_admin',
 
                                  'password': 'test12'})
 
        self.assertEqual(response.status, '302 Found')
 
        response = response.follow()
 

	
 
        self.assertEqual(response.status, '200 OK')
 
        response.mustcontain('Users Administration')
 

	
 
    @parameterized.expand([
 
          ('data:text/html,<script>window.alert("xss")</script>',),
 
          ('mailto:test@example.com',),
 
          ('file:///etc/passwd',),
 
          ('ftp://some.ftp.server',),
 
          ('http://other.domain',),
 
          ('http://other.domain/bl%C3%A5b%C3%A6rgr%C3%B8d',),
 
    ])
 
    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_admin',
 
                                  'password': 'test12'})
 
        self.assertEqual(response.status, '302 Found')
 
        self.assertEqual(response._environ['paste.testing_variables']
 
                         ['tmpl_context'].came_from, '/')
 
        response = response.follow()
 

	
 
        self.assertEqual(response.status, '200 OK')
 

	
 
    def test_login_short_password(self):
 
        response = self.app.post(url(controller='login', action='index'),
 
                                 {'username': 'test_admin',
 
                                  '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 user name')
 
        response.mustcontain('invalid 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=bl%C3%A5', 'green=gr%C3%B8n')),
 
    ])
 
    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')
 
            for encoded in args_encoded:
 
                self.assertIn(encoded, response.location)
 

	
 
    @parameterized.expand([
 
        ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
 
        ({'blue': u'blå'.encode('utf-8'), '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 = '/_admin/users',
 
                                    **args))
 
        for encoded in args_encoded:
 
            self.assertIn(encoded, response.form.action)
 

	
 
    @parameterized.expand([
 
        ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
 
        ({'blue': u'blå'.encode('utf-8'), '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 = '/_admin/users',
 
                                     **args),
 
                                 {'username': 'test_admin',
 
                                  'password': 'test12'})
 
        self.assertEqual(response.status, '302 Found')
 
        for encoded in args_encoded:
 
            self.assertIn(encoded, response.location)
 

	
 
    @parameterized.expand([
 
        ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
 
        ({'blue': u'blå'.encode('utf-8'), '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 = '/_admin/users',
 
                                     **args),
 
                                 {'username': 'error',
 
                                  'password': 'test12'})
 

	
 
        response.mustcontain('invalid user name')
 
        response.mustcontain('invalid password')
 
        for encoded in args_encoded:
 
            self.assertIn(encoded, response.form.action)
 

	
 
    #==========================================================================
 
    # 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_admin'
 
        response = self.app.post(url(controller='login', action='register'),
 
                                            {'username': uname,
 
                                             'password': 'test12',
 
                                             'password_confirmation': 'test12',
 
                                             'email': 'goodmail@domain.com',
 
                                             'firstname': 'test',
 
                                             'lastname': 'test'})
 

	
 
        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',
0 comments (0 inline, 0 general)