Changeset - dffb92224edf
[Not reviewed]
beta
0 2 0
Marcin Kuzminski - 13 years ago 2012-07-31 12:15:54
marcin@python-works.com
removed ftp from allowed schemas
- added tests for the schemas fix
- moved parsing url if we only have came_from present
2 files changed with 33 insertions and 15 deletions:
0 comments (0 inline, 0 general)
rhodecode/controllers/login.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
"""
 
    rhodecode.controllers.login
 
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

	
 
    Login controller for rhodeocode
 

	
 
    :created_on: Apr 22, 2010
 
    :author: marcink
 
    :copyright: (C) 2010-2012 Marcin Kuzminski <marcin@python-works.com>
 
    :license: GPLv3, see COPYING for more details.
 
"""
 
# 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/>.
 

	
 
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 abort, redirect
 
from pylons import request, response, session, tmpl_context as c, url
 

	
 
import rhodecode.lib.helpers as h
 
from rhodecode.lib.auth import AuthUser, HasPermissionAnyDecorator
 
from rhodecode.lib.base import BaseController, render
 
from rhodecode.model.db import User
 
from rhodecode.model.forms import LoginForm, RegisterForm, PasswordResetForm
 
from rhodecode.model.user import UserModel
 
from rhodecode.model.meta import Session
 

	
 

	
 

	
 
log = logging.getLogger(__name__)
 

	
 

	
 
class LoginController(BaseController):
 

	
 
    def __before__(self):
 
        super(LoginController, self).__before__()
 

	
 
    def index(self):
 
        # redirect if already logged in
 
        c.came_from = request.GET.get('came_from', None)
 
        c.came_from = request.GET.get('came_from')
 

	
 
        if self.rhodecode_user.is_authenticated \
 
                            and self.rhodecode_user.username != 'default':
 

	
 
            return redirect(url('home'))
 

	
 
        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
 
                username = c.form_result['username']
 
                user = User.get_by_username(username, case_insensitive=True)
 
                auth_user = AuthUser(user.user_id)
 
                auth_user.set_authenticated()
 
                cs = auth_user.get_cookie_store()
 
                session['rhodecode_user'] = cs
 
                user.update_lastlogin()
 
                Session().commit()
 

	
 
                # If they want to be remembered, update the cookie
 
                if c.form_result['remember'] is not False:
 
                    _year = (datetime.datetime.now() +
 
                             datetime.timedelta(seconds=60 * 60 * 24 * 365))
 
                    session._set_cookie_expires(_year)
 

	
 
                session.save()
 

	
 
                log.info('user %s is now authenticated and stored in '
 
                         'session, session attrs %s' % (username, cs))
 

	
 
                # dumps session attrs back to cookie
 
                session._update_cookie_out()
 

	
 
                # we set new cookie
 
                headers = None
 
                if session.request['set_cookie']:
 
                    # send set-cookie headers back to response to update cookie
 
                    headers = [('Set-Cookie', session.request['cookie_out'])]
 

	
 
                allowed_schemes = ['http', 'https', 'ftp']
 
                parsed = urlparse.urlparse(c.came_from)
 
                server_parsed = urlparse.urlparse(url.current())
 

	
 
                if parsed.scheme and parsed.scheme not in allowed_schemes:
 
                    log.error('Suspicious URL scheme detected %s for url %s' %
 
                              (parsed.scheme, parsed))
 
                    c.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))
 
                    c.came_from = url('home')
 
                allowed_schemes = ['http', 'https']
 
                if c.came_from:
 
                    parsed = urlparse.urlparse(c.came_from)
 
                    server_parsed = urlparse.urlparse(url.current())
 
                    if parsed.scheme and parsed.scheme not in allowed_schemes:
 
                        log.error(
 
                            'Suspicious URL scheme detected %s for url %s' %
 
                            (parsed.scheme, parsed))
 
                        c.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))
 
                        c.came_from = url('home')
 
                    raise HTTPFound(location=c.came_from, headers=headers)
 
                else:
 
                    raise HTTPFound(location=url('home'), headers=headers)
 

	
 
            except formencode.Invalid, errors:
 
                return htmlfill.render(
 
                    render('/login.html'),
 
                    defaults=errors.value,
 
                    errors=errors.error_dict or {},
 
                    prefix_error=False,
 
                    encoding="UTF-8")
 

	
 
        return render('/login.html')
 

	
 
    @HasPermissionAnyDecorator('hg.admin', 'hg.register.auto_activate',
 
                               'hg.register.manual_activate')
 
    def register(self):
 
        c.auto_active = False
 
        for perm in User.get_by_username('default').user_perms:
 
            if perm.permission.permission_name == 'hg.register.auto_activate':
 
                c.auto_active = True
 
                break
 

	
 
        if request.POST:
 

	
 
            register_form = RegisterForm()()
 
            try:
 
                form_result = register_form.to_python(dict(request.POST))
 
                form_result['active'] = c.auto_active
 
                UserModel().create_registration(form_result)
 
                h.flash(_('You have successfully registered into rhodecode'),
 
                            category='success')
 
                Session().commit()
 
                return redirect(url('login_home'))
 

	
 
            except formencode.Invalid, errors:
 
                return htmlfill.render(
 
                    render('/register.html'),
 
                    defaults=errors.value,
 
                    errors=errors.error_dict or {},
 
                    prefix_error=False,
 
                    encoding="UTF-8")
 

	
 
        return render('/register.html')
 

	
 
    def password_reset(self):
 
        if request.POST:
 
            password_reset_form = PasswordResetForm()()
 
            try:
 
                form_result = password_reset_form.to_python(dict(request.POST))
 
                UserModel().reset_password_link(form_result)
 
                h.flash(_('Your password reset link was sent'),
 
                            category='success')
 
                return redirect(url('login_home'))
 

	
 
            except formencode.Invalid, errors:
 
                return htmlfill.render(
 
                    render('/password_reset.html'),
 
                    defaults=errors.value,
 
                    errors=errors.error_dict or {},
 
                    prefix_error=False,
 
                    encoding="UTF-8")
 

	
 
        return render('/password_reset.html')
 

	
 
    def password_reset_confirmation(self):
 
        if request.GET and request.GET.get('key'):
 
            try:
 
                user = User.get_by_api_key(request.GET.get('key'))
 
                data = dict(email=user.email)
 
                UserModel().reset_password(data)
 
                h.flash(_('Your password reset was successful, '
 
                          'new password has been sent to your email'),
 
                            category='success')
 
            except Exception, e:
 
                log.error(e)
 
                return redirect(url('reset_password'))
 

	
 
        return redirect(url('login_home'))
 

	
 
    def logout(self):
 
        session.delete()
 
        log.info('Logging out and deleting session for user')
 
        redirect(url('home'))
rhodecode/tests/functional/test_login.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
from rhodecode.tests import *
 
from rhodecode.model.db import User, Notification
 
from rhodecode.lib.utils2 import generate_api_key
 
from rhodecode.lib.auth import check_password
 
from rhodecode.lib import helpers as h
 
from rhodecode.model import validators
 

	
 

	
 
class TestLoginController(TestController):
 

	
 
    def tearDown(self):
 
        for n in Notification.query().all():
 
            self.Session().delete(n)
 

	
 
        self.Session().commit()
 
        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_admin',
 
                                  'password': 'test12'})
 
        self.assertEqual(response.status, '302 Found')
 
        self.assertEqual(response.session['rhodecode_user'].get('username'),
 
                         'test_admin')
 
        response = response.follow()
 
        self.assertTrue('%s repository' % HG_REPO in response.body)
 

	
 
    def test_login_regular_ok(self):
 
        response = self.app.post(url(controller='login', action='index'),
 
                                 {'username': 'test_regular',
 
                                  'password': 'test12'})
 

	
 
        self.assertEqual(response.status, '302 Found')
 
        self.assertEqual(response.session['rhodecode_user'].get('username'),
 
                         'test_regular')
 
        response = response.follow()
 
        self.assertTrue('%s repository' % HG_REPO in response.body)
 
        self.assertTrue('<a title="Admin" href="/_admin">' not in response.body)
 

	
 
    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')
 
        self.assertTrue('Users administration' in response.body)
 

	
 
    @parameterized.expand([
 
          ('data:text/html,<script>window.alert("xss")</script>',),
 
          ('mailto:test@rhodecode.org',),
 
          ('file:///etc/passwd',),
 
          ('ftp://some.ftp.server',),
 
          ('http://other.domain',),
 
    ])
 
    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')
 

	
 
        self.assertTrue('Enter 3 characters or more' in response.body)
 

	
 
    def test_login_wrong_username_password(self):
 
        response = self.app.post(url(controller='login', action='index'),
 
                                 {'username': 'error',
 
                                  'password': 'test12'})
 

	
 
        self.assertTrue('invalid user name' in response.body)
 
        self.assertTrue('invalid password' in response.body)
 

	
 
    #==========================================================================
 
    # REGISTRATIONS
 
    #==========================================================================
 
    def test_register(self):
 
        response = self.app.get(url(controller='login', action='register'))
 
        self.assertTrue('Sign Up to RhodeCode' in response.body)
 

	
 
    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',
 
                                             'password': 'test12',
 
                                             'password_confirmation': 'test12',
 
                                             'email': 'test_admin@mail.com',
 
                                             'firstname': 'test',
 
                                             'lastname': 'test'})
 

	
 
        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_Admin@mail.COM',
 
                                             'firstname': 'test',
 
                                             'lastname': 'test'})
 
        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'})
 
        self.assertEqual(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'})
 

	
 
        response.mustcontain('An email address must contain a single @')
 
        response.mustcontain('Username may only contain '
 
                'alphanumeric characters underscores, '
 
                'periods or dashes and must begin with '
 
                'alphanumeric character')
 

	
 
    def test_register_err_case_sensitive(self):
 
        usr = 'Test_Admin'
 
        response = self.app.post(url(controller='login', action='register'),
 
                                            {'username': usr,
 
                                             'password': 'test12',
 
                                             'password_confirmation': 'test12',
 
                                             'email': 'goodmailm',
 
                                             'firstname': 'test',
 
                                             'lastname': 'test'})
0 comments (0 inline, 0 general)