kallithea Changeset - 23a86f1c33a1

Changeset - 23a86f1c33a1

Parent rev.

Child rev.

[Not reviewed]

default

0 4 0

Søren Løvborg - 10 years ago 2015-09-08 11:00:02
sorenl@unity3d.com

auth: note that we never emit authuser "cookies" for the default user

The only place where we set "authuser" in the session is in log_in_user,
which is called only by the internal auth system and by auth plugins.
The internal auth system cannot log a user in as the default user,
because the default user doesn't have a password (and cannot have a
password assigned). Auth plugins cannot log a user in as the default
user, because the user doesn't have the right extern_type. As such, it's
a bug if log_in_user is ever called with the default user (which this
commit documents with an assert).

This realization makes the is_authenticated field of the authuser cookie
redundant, as it's always True. It also emphasizes that is_default_user
and is_authenticated are mutually exclusive.

4 files changed with 8 insertions and 9 deletions:

kallithea/controllers/login.py

kallithea/lib/auth.py

kallithea/lib/base.py

kallithea/tests/__init__.py

0 comments (0 inline, 0 general)

kallithea/controllers/login.py

➞

Show inline comments

 # -*- 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 not self.authuser.is_default_user and ip_allowed:
         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']
                 user = User.get_by_username(username, case_insensitive=True)
             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')
                         error_dict = {'recaptcha_field': _msg}
                         raise formencode.Invalid(_msg, _value, None,
                                                  error_dict=error_dict)
                 redirect_link = UserModel().send_reset_password_email(form_result)
                 h.flash(_('A password reset confirmation code has been sent'),
                             category='success')
                 raise HTTPFound(location=redirect_link)
             except formencode.Invalid as errors:
                 return htmlfill.render(
                     render('/password_reset.html'),
                     defaults=errors.value,
                     errors=errors.error_dict or {},
                     prefix_error=False,
                     encoding="UTF-8",
                     force_defaults=False)
         return render('/password_reset.html')
     def password_reset_confirmation(self):
         # This controller handles both GET and POST requests, though we
         # only ever perform the actual password change on POST (since
         # GET requests are not allowed to have side effects, and do not
         # receive automatic CSRF protection).
         # The template needs the email address outside of the form.
         c.email = request.params.get('email')
         if not request.POST:
             return htmlfill.render(
                 render('/password_reset_confirmation.html'),
                 defaults=dict(request.params),
                 encoding='UTF-8')
         form = PasswordResetConfirmationForm()()
         try:
             form_result = form.to_python(dict(request.POST))
         except formencode.Invalid as errors:
             return htmlfill.render(
                 render('/password_reset_confirmation.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding='UTF-8')
         if not UserModel().verify_reset_password_token(
             form_result['email'],
             form_result['timestamp'],
             form_result['token'],
         ):
             return htmlfill.render(
                 render('/password_reset_confirmation.html'),
                 defaults=form_result,
                 errors={'token': _('Invalid password reset token')},
                 prefix_error=False,
                 encoding='UTF-8')
         UserModel().reset_password(form_result['email'], form_result['password'])
         h.flash(_('Successfully updated password'), category='success')
         raise HTTPFound(location=url('login_home'))
     def logout(self):
         session.delete()
         log.info('Logging out and deleting session for user')
         raise HTTPFound(location=url('home'))
     def authentication_token(self):
         """Return the CSRF protection token for the session - just like it
         could have been screen scraped from a page with a form.
         Only intended for testing but might also be useful for other kinds
         of automation.
         """
         return h.authentication_token()

kallithea/lib/auth.py

➞

Show inline comments

@@ @@ -437,401 +437,398 @@ def allowed_api_access(controller_name, @@
     if not whitelist:
         from kallithea import CONFIG
         whitelist = aslist(CONFIG.get('api_access_controllers_whitelist'),
                            sep=',')
         log.debug('whitelist of API access is: %s', whitelist)
     api_access_valid = controller_name in whitelist
     if api_access_valid:
         log.debug('controller:%s is in API whitelist', controller_name)
     else:
         msg = 'controller: %s is *NOT* in API whitelist' % (controller_name)
         if api_key:
             #if we use API key and don't have access it's a warning
             log.warning(msg)
         else:
             log.debug(msg)
     return api_access_valid
 class AuthUser(object):
     """
     Represents a Kallithea user, including various authentication and
     authorization information. Typically used to store the current user,
     but is also used as a generic user information data structure in
     parts of the code, e.g. user management.
     Constructed from a database `User` object, a user ID or cookie dict,
     it looks up the user (if needed) and copies all attributes to itself,
     adding various non-persistent data. If lookup fails but anonymous
     access to Kallithea is enabled, the default user is loaded instead.
     `AuthUser` does not by itself authenticate users and the constructor
     sets the `is_authenticated` field to False. It's up to other parts
     of the code to check e.g. if a supplied password is correct, and if
     so, set `is_authenticated` to True.
     However, `AuthUser` does refuse to load a user that is not `active`.
     """
     def __init__(self, user_id=None, dbuser=None,
             is_external_auth=False):
         self.is_authenticated = False
         self.is_external_auth = is_external_auth
         user_model = UserModel()
         self.anonymous_user = User.get_default_user(cache=True)
         # These attributes will be overriden by fill_data, below, unless the
         # requested user cannot be found and the default anonymous user is
         # not enabled.
         self.user_id = None
         self.username = None
         self.api_key = None
         self.name = ''
         self.lastname = ''
         self.email = ''
         self.admin = False
         self.inherit_default_permissions = False
         # Look up database user, if necessary.
         if user_id is not None:
             log.debug('Auth User lookup by USER ID %s', user_id)
             dbuser = user_model.get(user_id)
         else:
             # Note: dbuser is allowed to be None.
             log.debug('Auth User lookup by database user %s', dbuser)
         is_user_loaded = self._fill_data(dbuser)
         # If user cannot be found, try falling back to anonymous.
         if not is_user_loaded:
             is_user_loaded =  self._fill_data(self.anonymous_user)
         self.is_default_user = (self.user_id == self.anonymous_user.user_id)
         if not self.username:
             self.username = 'None'
         log.debug('Auth User is now %s', self)
     def _fill_data(self, dbuser):
         """
         Copies database fields from a `db.User` to this `AuthUser`. Does
         not copy `api_keys` and `permissions` attributes.
         Checks that `dbuser` is `active` (and not None) before copying;
         returns True on success.
         """
         if dbuser is not None and dbuser.active:
             log.debug('filling %s data', dbuser)
             for k, v in dbuser.get_dict().iteritems():
                 assert k not in ['api_keys', 'permissions']
                 setattr(self, k, v)
             return True
         return False
     @LazyProperty
     def permissions(self):
         return self.__get_perms(user=self, cache=False)
     @property
     def api_keys(self):
         return self._get_api_keys()
     def __get_perms(self, user, explicit=True, algo='higherwin', cache=False):
         """
         Fills user permission attribute with permissions taken from database
         works for permissions given for repositories, and for permissions that
         are granted to groups
         :param user: `AuthUser` instance
         :param explicit: In case there are permissions both for user and a group
             that user is part of, explicit flag will define if user will
             explicitly override permissions from group, if it's False it will
             make decision based on the algo
         :param algo: algorithm to decide what permission should be choose if
             it's multiple defined, eg user in two different groups. It also
             decides if explicit flag is turned off how to specify the permission
             for case when user is in a group + have defined separate permission
         """
         user_id = user.user_id
         user_is_admin = user.is_admin
         user_inherit_default_permissions = user.inherit_default_permissions
         log.debug('Getting PERMISSION tree')
         compute = conditional_cache('short_term', 'cache_desc',
                                     condition=cache, func=_cached_perms_data)
         return compute(user_id, user_is_admin,
                        user_inherit_default_permissions, explicit, algo)
     def _get_api_keys(self):
         api_keys = [self.api_key]
         for api_key in UserApiKeys.query()\
                 .filter(UserApiKeys.user_id == self.user_id)\
                 .filter(or_(UserApiKeys.expires == -1,
                             UserApiKeys.expires >= time.time())).all():
             api_keys.append(api_key.api_key)
         return api_keys
     @property
     def is_admin(self):
         return self.admin
     @property
     def repositories_admin(self):
         """
         Returns list of repositories you're an admin of
         """
         return [x[0] for x in self.permissions['repositories'].iteritems()
                 if x[1] == 'repository.admin']
     @property
     def repository_groups_admin(self):
         """
         Returns list of repository groups you're an admin of
         """
         return [x[0] for x in self.permissions['repositories_groups'].iteritems()
                 if x[1] == 'group.admin']
     @property
     def user_groups_admin(self):
         """
         Returns list of user groups you're an admin of
         """
         return [x[0] for x in self.permissions['user_groups'].iteritems()
                 if x[1] == 'usergroup.admin']
     @staticmethod
     def check_ip_allowed(user, ip_addr):
         """
         Check if the given IP address (a `str`) is allowed for the given
         user (an `AuthUser` or `db.User`).
         """
         allowed_ips = AuthUser.get_allowed_ips(user.user_id, cache=True,
             inherit_from_default=user.inherit_default_permissions)
         if check_ip_access(source_ip=ip_addr, allowed_ips=allowed_ips):
             log.debug('IP:%s is in range of %s', ip_addr, allowed_ips)
             return True
         else:
             log.info('Access for IP:%s forbidden, '
                      'not in %s' % (ip_addr, allowed_ips))
             return False
     def __repr__(self):
         return "<AuthUser('id:%s[%s] auth:%s')>"\
             % (self.user_id, self.username, (self.is_authenticated or self.is_default_user))
     def to_cookie(self):
         """ Serializes this login session to a cookie `dict`. """
         return {
             'user_id': self.user_id,
             'is_authenticated': self.is_authenticated or self.is_default_user,
             'is_external_auth': self.is_external_auth,
+        }
     @staticmethod
     def from_cookie(cookie):
         """
         Deserializes an `AuthUser` from a cookie `dict`.
         """
         au = AuthUser(
             user_id=cookie.get('user_id'),
             is_external_auth=cookie.get('is_external_auth', False),
+        )
         if not au.is_default_user and au.user_id is not None:
             # user is not authenticated and not empty
             au.is_authenticated = cookie.get('is_authenticated')
         au.is_authenticated = True
         return au
     @classmethod
     def get_allowed_ips(cls, user_id, cache=False, inherit_from_default=False):
         _set = set()
         if inherit_from_default:
             default_ips = UserIpMap.query().filter(UserIpMap.user ==
                                             User.get_default_user(cache=True))
             if cache:
                 default_ips = default_ips.options(FromCache("sql_cache_short",
                                                   "get_user_ips_default"))
             # populate from default user
             for ip in default_ips:
                 try:
                     _set.add(ip.ip_addr)
                 except ObjectDeletedError:
                     # since we use heavy caching sometimes it happens that we get
                     # deleted objects here, we just skip them
                     pass
         user_ips = UserIpMap.query().filter(UserIpMap.user_id == user_id)
         if cache:
             user_ips = user_ips.options(FromCache("sql_cache_short",
                                                   "get_user_ips_%s" % user_id))
         for ip in user_ips:
             try:
                 _set.add(ip.ip_addr)
             except ObjectDeletedError:
                 # since we use heavy caching sometimes it happens that we get
                 # deleted objects here, we just skip them
                 pass
         return _set or set(['0.0.0.0/0', '::/0'])
 def set_available_permissions(config):
     """
     This function will propagate pylons globals with all available defined
     permission given in db. We don't want to check each time from db for new
     permissions since adding a new permission also requires application restart
     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 = request.path_qs
     if message:
         h.flash(h.literal(message), category='warning')
     log.debug('Redirecting to login page, origin: %s', p)
     raise HTTPFound(location=url('login_home', came_from=p))
 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):
         controller = fargs[0]
         user = controller.authuser
         loc = "%s:%s" % (controller.__class__.__name__, func.__name__)
         log.debug('Checking access for user %s @ %s', user, loc)
         if not AuthUser.check_ip_allowed(user, controller.ip_addr):
             return redirect_to_login(_('IP %s not allowed') % controller.ip_addr)
         # check if we used an API key and it's a valid one
         api_key = request.GET.get('api_key')
         if api_key is not None:
             # explicit controller is enabled or API is in our whitelist
             if self.api_access or allowed_api_access(loc, api_key=api_key):
                 if api_key in user.api_keys:
                     log.info('user %s authenticated with API key ****%s @ %s',
                              user, api_key[-4:], loc)
                     return func(*fargs, **fkwargs)
                 else:
                     log.warning('API key ****%s is NOT valid', api_key[-4:])
                     return redirect_to_login(_('Invalid API key'))
             else:
                 # controller does not allow API access
                 log.warning('API access to %s is not allowed', loc)
                 raise HTTPForbidden()
         # Only allow the following HTTP request methods. (We sometimes use POST
         # requests with a '_method' set to 'PUT' or 'DELETE'; but that is only
         # used for the route lookup, and does not affect request.method.)
         if request.method not in ['GET', 'HEAD', 'POST', 'PUT']:
             raise HTTPMethodNotAllowed()
         # Make sure CSRF token never appears in the URL. If so, invalidate it.
         if secure_form.token_key in request.GET:
             log.error('CSRF key leak detected')
             session.pop(secure_form.token_key, None)
             session.save()
             from kallithea.lib import helpers as h
             h.flash(_("CSRF token leak has been detected - all form tokens have been expired"),
                     category='error')
         # CSRF protection: Whenever a request has ambient authority (whether
         # through a session cookie or its origin IP address), it must include
         # the correct token, unless the HTTP method is GET or HEAD (and thus
         # guaranteed to be side effect free. In practice, the only situation
         # where we allow side effects without ambient authority is when the
         # authority comes from an API key; and that is handled above.
         if request.method not in ['GET', 'HEAD']:
             token = request.POST.get(secure_form.token_key)
             if not token or token != secure_form.authentication_token():
                 log.error('CSRF check failed')
                 raise HTTPForbidden()
         # WebOb already ignores request payload parameters for anything other
         # than POST/PUT, but double-check since other Kallithea code relies on
         # this assumption.
         if request.method not in ['POST', 'PUT'] and request.POST:
             log.error('%r request with payload parameters; WebOb should have stopped this', request.method)
             raise HTTPBadRequest()
         # regular user authentication
         if user.is_authenticated or user.is_default_user:
             log.info('user %s authenticated with regular auth @ %s', user, loc)
             return func(*fargs, **fkwargs)
         else:
             log.warning('user %s NOT authenticated with regular auth @ %s', user, loc)
             return redirect_to_login()
 class NotAnonymous(object):
     """
     Must be logged in to execute this function else
     redirect to login page"""
     def __call__(self, func):
         return decorator(self.__wrapper, func)
     def __wrapper(self, func, *fargs, **fkwargs):
         cls = fargs[0]
         self.user = cls.authuser
         log.debug('Checking if user is not anonymous @%s', cls)
         if self.user.is_default_user:
             return redirect_to_login(_('You need to be a registered user to '
                     'perform this action'))
         else:
             return func(*fargs, **fkwargs)
 class PermsDecorator(object):
     """Base class for controller decorators"""
     def __init__(self, *required_perms):
         self.required_perms = set(required_perms)
         self.user_perms = None
     def __call__(self, func):
         return decorator(self.__wrapper, func)
     def __wrapper(self, func, *fargs, **fkwargs):
         cls = fargs[0]
         self.user = cls.authuser
         self.user_perms = self.user.permissions
         log.debug('checking %s permissions %s for %s %s',
           self.__class__.__name__, self.required_perms, cls, self.user)
         if self.check_permissions():
             log.debug('Permission granted for %s %s', cls, self.user)

kallithea/lib/base.py

➞

Show inline comments

 # -*- 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.lib.base
 ~~~~~~~~~~~~~~~~~~
 The base Controller API
 Provides the BaseController class for subclassing. And usage in different
 controllers
 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: Oct 06, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import datetime
 import logging
 import time
 import traceback
 import webob.exc
 import paste.httpexceptions
 import paste.auth.basic
 import paste.httpheaders
 from pylons import config, tmpl_context as c, request, session, url
 from pylons.controllers import WSGIController
 from pylons.templating import render_mako as render  # don't remove this import
 from pylons.i18n.translation import _
 from kallithea import __version__, BACKENDS
 from kallithea.lib.utils2 import str2bool, safe_unicode, AttributeDict,\
     safe_str, safe_int
 from kallithea.lib import auth_modules
 from kallithea.lib.auth import AuthUser, HasPermissionAnyMiddleware
 from kallithea.lib.utils import get_repo_slug
 from kallithea.lib.exceptions import UserCreationError
 from kallithea.lib.vcs.exceptions import RepositoryError, EmptyRepositoryError, ChangesetDoesNotExistError
 from kallithea.model import meta
 from kallithea.model.db import Repository, Ui, User, Setting
 from kallithea.model.notification import NotificationModel
 from kallithea.model.scm import ScmModel
 from kallithea.model.pull_request import PullRequestModel
 log = logging.getLogger(__name__)
 def _filter_proxy(ip):
     """
     HEADERS can have multiple ips inside the left-most being the original
     client, and each successive proxy that passed the request adding the IP
     address where it received the request from.
     :param ip:
     """
     if ',' in ip:
         _ips = ip.split(',')
         _first_ip = _ips[0].strip()
         log.debug('Got multiple IPs %s, using %s', ','.join(_ips), _first_ip)
         return _first_ip
     return ip
 def _get_ip_addr(environ):
     proxy_key = 'HTTP_X_REAL_IP'
     proxy_key2 = 'HTTP_X_FORWARDED_FOR'
     def_key = 'REMOTE_ADDR'
     ip = environ.get(proxy_key)
     if ip:
         return _filter_proxy(ip)
     ip = environ.get(proxy_key2)
     if ip:
         return _filter_proxy(ip)
     ip = environ.get(def_key, '0.0.0.0')
     return _filter_proxy(ip)
 def _get_access_path(environ):
     path = environ.get('PATH_INFO')
     org_req = environ.get('pylons.original_request')
     if org_req:
         path = org_req.environ.get('PATH_INFO')
     return path
 def log_in_user(user, remember, is_external_auth):
     """
     Log a `User` in and update session and cookies. If `remember` is True,
     the session cookie is set to expire in a year; otherwise, it expires at
     the end of the browser session.
     Returns populated `AuthUser` object.
     """
     user.update_lastlogin()
     meta.Session().commit()
     auth_user = AuthUser(dbuser=user,
                          is_external_auth=is_external_auth)
     if not auth_user.is_default_user:
         auth_user.is_authenticated = True
     # It should not be possible to explicitly log in as the default user.
     assert not auth_user.is_default_user
     auth_user.is_authenticated = True
     # Start new session to prevent session fixation attacks.
     session.invalidate()
     session['authuser'] = cookie = auth_user.to_cookie()
     # If they want to be remembered, update the cookie.
     # NOTE: Assumes that beaker defaults to browser session cookie.
     if remember:
         t = datetime.datetime.now() + datetime.timedelta(days=365)
         session._set_cookie_expires(t)
     session.save()
     log.info('user %s is now authenticated and stored in '
              'session, session attrs %s', user.username, cookie)
     # dumps session attrs back to cookie
     session._update_cookie_out()
     return auth_user
 class BasicAuth(paste.auth.basic.AuthBasicAuthenticator):
     def __init__(self, realm, authfunc, auth_http_code=None):
         self.realm = realm
         self.authfunc = authfunc
         self._rc_auth_http_code = auth_http_code
     def build_authentication(self):
         head = paste.httpheaders.WWW_AUTHENTICATE.tuples('Basic realm="%s"' % self.realm)
         if self._rc_auth_http_code and self._rc_auth_http_code == '403':
             # return 403 if alternative http return code is specified in
             # Kallithea config
             return paste.httpexceptions.HTTPForbidden(headers=head)
         return paste.httpexceptions.HTTPUnauthorized(headers=head)
     def authenticate(self, environ):
         authorization = paste.httpheaders.AUTHORIZATION(environ)
         if not authorization:
             return self.build_authentication()
         (authmeth, auth) = authorization.split(' ', 1)
         if 'basic' != authmeth.lower():
             return self.build_authentication()
         auth = auth.strip().decode('base64')
         _parts = auth.split(':', 1)
         if len(_parts) == 2:
             username, password = _parts
             if self.authfunc(username, password, environ) is not None:
                 return username
         return self.build_authentication()
     __call__ = authenticate
 class BaseVCSController(object):
     def __init__(self, application, config):
         self.application = application
         self.config = config
         # base path of repo locations
         self.basepath = self.config['base_path']
         # authenticate this VCS request using the authentication modules
         self.authenticate = BasicAuth('', auth_modules.authenticate,
                                       config.get('auth_ret_code'))
         self.ip_addr = '0.0.0.0'
     def _handle_request(self, environ, start_response):
         raise NotImplementedError()
     def _get_by_id(self, repo_name):
         """
         Gets a special pattern _<ID> from clone url and tries to replace it
         with a repository_name for support of _<ID> permanent URLs
         :param repo_name:
         """
         data = repo_name.split('/')
         if len(data) >= 2:
             from kallithea.lib.utils import get_repo_by_id
             by_id_match = get_repo_by_id(repo_name)
             if by_id_match:
                 data[1] = by_id_match
         return '/'.join(data)
     def _invalidate_cache(self, repo_name):
         """
         Sets cache for this repository for invalidation on next access
         :param repo_name: full repo name, also a cache key
         """
         ScmModel().mark_for_invalidation(repo_name)
     def _check_permission(self, action, user, repo_name, ip_addr=None):
         """
         Checks permissions using action (push/pull) user and repository
         name
         :param action: push or pull action
         :param user: `User` instance
         :param repo_name: repository name
         """
         # check IP
         ip_allowed = AuthUser.check_ip_allowed(user, ip_addr)
         if ip_allowed:
             log.info('Access for IP:%s allowed', ip_addr)
         else:
             return False
         if action == 'push':
             if not HasPermissionAnyMiddleware('repository.write',
                                               'repository.admin')(user,
                                                                   repo_name):
                 return False
         else:
             #any other action need at least read permission
             if not HasPermissionAnyMiddleware('repository.read',
                                               'repository.write',
                                               'repository.admin')(user,
                                                                   repo_name):
                 return False
         return True
     def _get_ip_addr(self, environ):
         return _get_ip_addr(environ)
     def _check_ssl(self, environ):
         """
         Checks the SSL check flag and returns False if SSL is not present
         and required True otherwise
         """
         #check if we have SSL required  ! if not it's a bad request !
         if str2bool(Ui.get_by_key('push_ssl').ui_value):
             org_proto = environ.get('wsgi._org_proto', environ['wsgi.url_scheme'])
             if org_proto != 'https':
                 log.debug('proto is %s and SSL is required BAD REQUEST !',
                           org_proto)
                 return False
         return True
     def _check_locking_state(self, environ, action, repo, user_id):
         """
         Checks locking on this repository, if locking is enabled and lock is
         present returns a tuple of make_lock, locked, locked_by.
         make_lock can have 3 states None (do nothing) True, make lock
         False release lock, This value is later propagated to hooks, which
         do the locking. Think about this as signals passed to hooks what to do.
         """
         locked = False  # defines that locked error should be thrown to user
         make_lock = None
         repo = Repository.get_by_repo_name(repo)
         user = User.get(user_id)
         # this is kind of hacky, but due to how mercurial handles client-server
         # server see all operation on changeset; bookmarks, phases and
         # obsolescence marker in different transaction, we don't want to check
         # locking on those
         obsolete_call = environ['QUERY_STRING'] in ['cmd=listkeys',]
         locked_by = repo.locked
         if repo and repo.enable_locking and not obsolete_call:
             if action == 'push':
                 #check if it's already locked !, if it is compare users
                 user_id, _date = repo.locked
                 if user.user_id == user_id:
                     log.debug('Got push from user %s, now unlocking', user)
                     # unlock if we have push from user who locked
                     make_lock = False
                 else:
                     # we're not the same user who locked, ban with 423 !
                     locked = True
             if action == 'pull':
                 if repo.locked[0] and repo.locked[1]:
                     locked = True
                 else:
                     log.debug('Setting lock on repo %s by %s', repo, user)
                     make_lock = True
         else:
             log.debug('Repository %s do not have locking enabled', repo)
         log.debug('FINAL locking values make_lock:%s,locked:%s,locked_by:%s',
                   make_lock, locked, locked_by)
         return make_lock, locked, locked_by
     def __call__(self, environ, start_response):
         start = time.time()
         try:
             return self._handle_request(environ, start_response)
         finally:
             log = logging.getLogger('kallithea.' + self.__class__.__name__)
             log.debug('Request time: %.3fs', time.time() - start)
             meta.Session.remove()
 class BaseController(WSGIController):
     def __before__(self):
         """
         __before__ is called before controller methods and after __call__
         """
         c.kallithea_version = __version__
         rc_config = Setting.get_app_settings()
         # Visual options
         c.visual = AttributeDict({})
         ## DB stored
         c.visual.show_public_icon = str2bool(rc_config.get('show_public_icon'))
         c.visual.show_private_icon = str2bool(rc_config.get('show_private_icon'))
         c.visual.stylify_metatags = str2bool(rc_config.get('stylify_metatags'))
         c.visual.dashboard_items = safe_int(rc_config.get('dashboard_items', 100))
         c.visual.admin_grid_items = safe_int(rc_config.get('admin_grid_items', 100))
         c.visual.repository_fields = str2bool(rc_config.get('repository_fields'))
         c.visual.show_version = str2bool(rc_config.get('show_version'))
         c.visual.use_gravatar = str2bool(rc_config.get('use_gravatar'))
         c.visual.gravatar_url = rc_config.get('gravatar_url')
         c.ga_code = rc_config.get('ga_code')
         # TODO: replace undocumented backwards compatibility hack with db upgrade and rename ga_code
         if c.ga_code and '<' not in c.ga_code:
             c.ga_code = '''<script type="text/javascript">
                 var _gaq = _gaq || [];
                 _gaq.push(['_setAccount', '%s']);
                 _gaq.push(['_trackPageview']);
                 (function() {
                     var ga = document.createElement('script'); ga.type = 'text/javascript'; ga.async = true;
                     ga.src = ('https:' == document.location.protocol ? 'https://ssl' : 'http://www') + '.google-analytics.com/ga.js';
                     var s = document.getElementsByTagName('script')[0]; s.parentNode.insertBefore(ga, s);
                     })();
             </script>''' % c.ga_code
         c.site_name = rc_config.get('title')
         c.clone_uri_tmpl = rc_config.get('clone_uri_tmpl')
         ## INI stored
         c.visual.allow_repo_location_change = str2bool(config.get('allow_repo_location_change', True))
         c.visual.allow_custom_hooks_settings = str2bool(config.get('allow_custom_hooks_settings', True))
         c.instance_id = config.get('instance_id')
         c.issues_url = config.get('bugtracker', url('issues_url'))
         # END CONFIG VARS
         c.repo_name = get_repo_slug(request)  # can be empty
         c.backends = BACKENDS.keys()
         c.unread_notifications = NotificationModel()\
                         .get_unread_cnt_for_user(c.authuser.user_id)
         self.cut_off_limit = safe_int(config.get('cut_off_limit'))
         c.my_pr_count = PullRequestModel().get_pullrequest_cnt_for_user(c.authuser.user_id)
         self.sa = meta.Session
         self.scm_model = ScmModel(self.sa)
     @staticmethod
     def _determine_auth_user(api_key, session_authuser):
         """
         Create an `AuthUser` object given the API key (if any) and the
         value of the authuser session cookie.
         """
         # Authenticate by API key
         if api_key:
             # when using API_KEY we are sure user exists.
             return AuthUser(dbuser=User.get_by_api_key(api_key),
                             is_external_auth=True)
         # Authenticate by session cookie
         # In ancient login sessions, 'authuser' may not be a dict.
         # In that case, the user will have to log in again.
         if isinstance(session_authuser, dict):
         # v0.3 and earlier included an 'is_authenticated' key; if present,
         # this must be True.
         if isinstance(session_authuser, dict) and session_authuser.get('is_authenticated', True):
             try:
                 return AuthUser.from_cookie(session_authuser)
             except UserCreationError as e:
                 # container auth or other auth functions that create users on
                 # the fly can throw UserCreationError to signal issues with
                 # user creation. Explanation should be provided in the
                 # exception object.
                 from kallithea.lib import helpers as h
                 h.flash(e, 'error', logf=log.error)
         # Authenticate by auth_container plugin (if enabled)
         if any(
             auth_modules.importplugin(name).is_container_auth
             for name in Setting.get_auth_plugins()
         ):
             try:
                 user_info = auth_modules.authenticate('', '', request.environ)
             except UserCreationError as e:
                 from kallithea.lib import helpers as h
                 h.flash(e, 'error', logf=log.error)
             else:
                 if user_info is not None:
                     username = user_info['username']
                     user = User.get_by_username(username, case_insensitive=True)
                     return log_in_user(user, remember=False,
                                        is_external_auth=True)
         # User is anonymous
         return AuthUser()
     def __call__(self, environ, start_response):
         """Invoke the Controller"""
         # WSGIController.__call__ dispatches to the Controller method
         # the request is routed to. This routing information is
         # available in environ['pylons.routes_dict']
         try:
             self.ip_addr = _get_ip_addr(environ)
             # make sure that we update permissions each time we call controller
             #set globals for auth user
             self.authuser = c.authuser = request.user = self._determine_auth_user(
                 request.GET.get('api_key'),
                 session.get('authuser'),
+            )
             log.info('IP: %s User: %s accessed %s',
                 self.ip_addr, self.authuser,
                 safe_unicode(_get_access_path(environ)),
+            )
             return WSGIController.__call__(self, environ, start_response)
         finally:
             meta.Session.remove()
 class BaseRepoController(BaseController):
     """
     Base class for controllers responsible for loading all needed data for
     repository loaded items are
     c.db_repo_scm_instance: instance of scm repository
     c.db_repo: instance of db
     c.repository_followers: number of followers
     c.repository_forks: number of forks
     c.repository_following: weather the current user is following the current repo
     """
     def __before__(self):
         super(BaseRepoController, self).__before__()
         if c.repo_name:  # extracted from routes
             _dbr = Repository.get_by_repo_name(c.repo_name)
             if not _dbr:
                 return
             log.debug('Found repository in database %s with state `%s`',
                       safe_unicode(_dbr), safe_unicode(_dbr.repo_state))
             route = getattr(request.environ.get('routes.route'), 'name', '')
             # allow to delete repos that are somehow damages in filesystem
             if route in ['delete_repo']:
                 return
             if _dbr.repo_state in [Repository.STATE_PENDING]:
                 if route in ['repo_creating_home']:
                     return
                 check_url = url('repo_creating_home', repo_name=c.repo_name)
                 raise webob.exc.HTTPFound(location=check_url)
             dbr = c.db_repo = _dbr
             c.db_repo_scm_instance = c.db_repo.scm_instance
             if c.db_repo_scm_instance is None:
                 log.error('%s this repository is present in database but it '
                           'cannot be created as an scm instance', c.repo_name)
                 from kallithea.lib import helpers as h
                 h.flash(h.literal(_('Repository not found in the filesystem')),
                         category='error')
                 raise paste.httpexceptions.HTTPNotFound()
             # some globals counter for menu
             c.repository_followers = self.scm_model.get_followers(dbr)
             c.repository_forks = self.scm_model.get_forks(dbr)
             c.repository_pull_requests = self.scm_model.get_pull_requests(dbr)
             c.repository_following = self.scm_model.is_following_repo(
                                     c.repo_name, self.authuser.user_id)
     @staticmethod
     def _get_ref_rev(repo, ref_type, ref_name, returnempty=False):
         """
         Safe way to get changeset. If error occurs show error.
         """
         from kallithea.lib import helpers as h
         try:
             return repo.scm_instance.get_ref_revision(ref_type, ref_name)
         except EmptyRepositoryError as e:
             if returnempty:
                 return repo.scm_instance.EMPTY_CHANGESET
             h.flash(h.literal(_('There are no changesets yet')),
                     category='error')
             raise webob.exc.HTTPNotFound()
         except ChangesetDoesNotExistError as e:
             h.flash(h.literal(_('Changeset not found')),
                     category='error')
             raise webob.exc.HTTPNotFound()
         except RepositoryError as e:
             log.error(traceback.format_exc())
             h.flash(safe_str(e), category='error')
             raise webob.exc.HTTPBadRequest()
 class WSGIResultCloseCallback(object):
     """Wrap a WSGI result and let close call close after calling the
     close method on the result.
     """
     def __init__(self, result, close):
         self._result = result
         self._close = close
     def __iter__(self):
         return iter(self._result)
     def close(self):
         if hasattr(self._result, 'close'):
             self._result.close()
         self._close()

kallithea/tests/__init__.py

➞

Show inline comments

@@ @@ -31,212 +31,211 @@ optional FLAGS: @@
     KALLITHEA_WHOOSH_TEST_DISABLE=1 - skip whoosh index building and tests
     KALLITHEA_NO_TMP_PATH=1 - disable new temp path for tests, used mostly for test_vcs_operations
 """
 import os
 import re
 import time
 import logging
 import datetime
 import hashlib
 import tempfile
 from os.path import join as jn
 from tempfile import _RandomNameSequence
 import pylons
 import pylons.test
 from pylons import config, url
 from pylons.i18n.translation import _get_translator
 from pylons.util import ContextObj
 from routes.util import URLGenerator
 from webtest import TestApp
 from nose.plugins.skip import SkipTest
 from kallithea.lib.compat import unittest
 from kallithea import is_windows
 from kallithea.model.db import Notification, User, UserNotification
 from kallithea.model.meta import Session
 from kallithea.tests.parameterized import parameterized
 from kallithea.lib.utils2 import safe_str
 os.environ['TZ'] = 'UTC'
 if not is_windows:
     time.tzset()
 log = logging.getLogger(__name__)
 __all__ = [
     'parameterized', 'environ', 'url', 'TestController',
     'SkipTest', 'ldap_lib_installed', 'pam_lib_installed', 'BaseTestCase', 'init_stack',
     'TESTS_TMP_PATH', 'HG_REPO', 'GIT_REPO', 'NEW_HG_REPO', 'NEW_GIT_REPO',
     'HG_FORK', 'GIT_FORK', 'TEST_USER_ADMIN_LOGIN', 'TEST_USER_ADMIN_PASS',
     'TEST_USER_ADMIN_EMAIL', 'TEST_USER_REGULAR_LOGIN', 'TEST_USER_REGULAR_PASS',
     'TEST_USER_REGULAR_EMAIL', 'TEST_USER_REGULAR2_LOGIN',
     'TEST_USER_REGULAR2_PASS', 'TEST_USER_REGULAR2_EMAIL', 'TEST_HG_REPO',
     'TEST_HG_REPO_CLONE', 'TEST_HG_REPO_PULL', 'TEST_GIT_REPO',
     'TEST_GIT_REPO_CLONE', 'TEST_GIT_REPO_PULL', 'HG_REMOTE_REPO',
     'GIT_REMOTE_REPO', 'SCM_TESTS',
+]
 # Invoke websetup with the current config file
 # SetupCommand('setup-app').run([config_file])
 environ = {}
 #SOME GLOBALS FOR TESTS
 TESTS_TMP_PATH = jn('/', 'tmp', 'rc_test_%s' % _RandomNameSequence().next())
 TEST_USER_ADMIN_LOGIN = 'test_admin'
 TEST_USER_ADMIN_PASS = 'test12'
 TEST_USER_ADMIN_EMAIL = 'test_admin@example.com'
 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 = 'vcs_test_hg'
 GIT_REPO = 'vcs_test_git'
 NEW_HG_REPO = 'vcs_test_hg_new'
 NEW_GIT_REPO = 'vcs_test_git_new'
 HG_FORK = 'vcs_test_hg_fork'
 GIT_FORK = 'vcs_test_git_fork'
 ## VCS
 SCM_TESTS = ['hg', 'git']
 uniq_suffix = str(int(time.mktime(datetime.datetime.now().timetuple())))
 GIT_REMOTE_REPO = 'git://github.com/codeinn/vcs.git'
 TEST_GIT_REPO = jn(TESTS_TMP_PATH, GIT_REPO)
 TEST_GIT_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcsgitclone%s' % uniq_suffix)
 TEST_GIT_REPO_PULL = jn(TESTS_TMP_PATH, 'vcsgitpull%s' % uniq_suffix)
 HG_REMOTE_REPO = 'http://bitbucket.org/marcinkuzminski/vcs'
 TEST_HG_REPO = jn(TESTS_TMP_PATH, HG_REPO)
 TEST_HG_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcshgclone%s' % uniq_suffix)
 TEST_HG_REPO_PULL = jn(TESTS_TMP_PATH, 'vcshgpull%s' % uniq_suffix)
 TEST_DIR = tempfile.gettempdir()
 TEST_REPO_PREFIX = 'vcs-test'
 # cached repos if any !
 # comment out to get some other repos from bb or github
 GIT_REMOTE_REPO = jn(TESTS_TMP_PATH, GIT_REPO)
 HG_REMOTE_REPO = jn(TESTS_TMP_PATH, HG_REPO)
 #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
 import logging
 class NullHandler(logging.Handler):
     def emit(self, record):
         pass
 def init_stack(config=None):
     if not config:
         config = pylons.test.pylonsapp.config
     url._push_object(URLGenerator(config['routes.map'], environ))
     pylons.app_globals._push_object(config['pylons.app_globals'])
     pylons.config._push_object(config)
     pylons.tmpl_context._push_object(ContextObj())
     # Initialize a translator for tests that utilize i18n
     translator = _get_translator(pylons.config.get('lang'))
     pylons.translator._push_object(translator)
     h = NullHandler()
     logging.getLogger("kallithea").addHandler(h)
 class BaseTestCase(unittest.TestCase):
     def __init__(self, *args, **kwargs):
         self.wsgiapp = pylons.test.pylonsapp
         init_stack(self.wsgiapp.config)
         unittest.TestCase.__init__(self, *args, **kwargs)
     def remove_all_notifications(self):
         Notification.query().delete()
         # Because query().delete() does not (by default) trigger cascades.
         # http://docs.sqlalchemy.org/en/rel_0_7/orm/collections.html#passive-deletes
         UserNotification.query().delete()
         Session().commit()
 class TestController(BaseTestCase):
     def __init__(self, *args, **kwargs):
         BaseTestCase.__init__(self, *args, **kwargs)
         self.app = TestApp(self.wsgiapp)
         self.maxDiff = None
         self.index_location = config['app_conf']['index_dir']
     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})
         if 'Invalid username or password' in response.body:
             self.fail('could not login using %s %s' % (username, password))
         self.assertEqual(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
         self.assertEqual(user, expected_username)
         self.assertEqual(cookie.get('is_authenticated'), True)
     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:
             self.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
         self.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)

0 comments (0 inline, 0 general)