kallithea Changeset - f103b1a2383b

Changeset - f103b1a2383b

Parent rev.

Child rev.

[Not reviewed]

default

0 5 0

Søren Løvborg - 10 years ago 2015-07-14 14:00:17
kwi@kwi.dk

BaseController: hide "Log out" link for external login sessions

If user is authorized by external means (API key or container auth),
Kallithea is not actually able to log the user out and should not show
the "Log out" link.

5 files changed with 29 insertions and 7 deletions:

kallithea/controllers/login.py

kallithea/lib/auth.py

kallithea/lib/base.py

kallithea/templates/base/base.html

kallithea/tests/functional/test_admin_auth_settings.py

0 comments (0 inline, 0 general)

kallithea/controllers/login.py

➞

Show inline comments

@@ @@ -23,193 +23,194 @@ Original author and date, and relevant c @@
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import formencode
 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.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, 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 _validate_came_from(self, came_from):
         """Return True if came_from is valid and can and should be used"""
         if not came_from:
             return False
         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))
             return False
         if server_parsed.netloc != parsed.netloc:
             log.error('Suspicious NETLOC detected %s for url %s server url '
                       'is: %s' % (parsed.netloc, parsed, server_parsed))
             return False
         return True
     def _redirect_to_origin(self, origin):
         '''redirect to the original page, preserving any get arguments given'''
         request.GET.pop('came_from', None)
         raise HTTPFound(location=url(origin, **request.GET))
     def index(self):
         c.came_from = safe_str(request.GET.get('came_from', ''))
         if not self._validate_came_from(c.came_from):
             c.came_from = url('home')
         not_default = self.authuser.username != User.DEFAULT_USER
         ip_allowed = AuthUser.check_ip_allowed(self.authuser, self.ip_addr)
         # redirect if already logged in
         if self.authuser.is_authenticated and not_default and ip_allowed:
             return self._redirect_to_origin(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, 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')
             else:
                 log_in_user(user, c.form_result['remember'])
                 log_in_user(user, c.form_result['remember'],
                     is_external_auth=False)
                 return self._redirect_to_origin(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()
                 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",
                     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')
         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 = PasswordResetForm()()
             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)
                 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",
                     force_defaults=False)
         return render('/password_reset.html')
     def password_reset_confirmation(self):

kallithea/lib/auth.py

➞

Show inline comments

@@ @@ -383,360 +383,364 @@ def _cached_perms_data(user_id, user_is_ @@
         if multiple_counter[g_k] > 1:
             p = _choose_perm(p, cur_perm)
         permissions[GK][g_k] = p
     # user explicit permissions for repository groups
     user_repo_groups_perms = Permission.get_default_group_perms(uid)
     for perm in user_repo_groups_perms:
         rg_k = perm.UserRepoGroupToPerm.group.group_name
         p = perm.Permission.permission_name
         cur_perm = permissions[GK][rg_k]
         if not explicit:
             p = _choose_perm(p, cur_perm)
         permissions[GK][rg_k] = p
     #======================================================================
     # !! PERMISSIONS FOR USER GROUPS !!
     #======================================================================
     # user group for user group permissions
     user_group_user_groups_perms = \
      Session().query(UserGroupUserGroupToPerm, Permission, UserGroup)\
      .join((UserGroup, UserGroupUserGroupToPerm.target_user_group_id
             == UserGroup.users_group_id))\
      .join((Permission, UserGroupUserGroupToPerm.permission_id
             == Permission.permission_id))\
      .join((UserGroupMember, UserGroupUserGroupToPerm.user_group_id
             == UserGroupMember.users_group_id))\
      .filter(UserGroupMember.user_id == uid)\
      .join((UserGroup, UserGroupMember.users_group_id ==
             UserGroup.users_group_id), aliased=True, from_joinpoint=True)\
      .filter(UserGroup.users_group_active == True)\
      .all()
     multiple_counter = collections.defaultdict(int)
     for perm in user_group_user_groups_perms:
         g_k = perm.UserGroupUserGroupToPerm.target_user_group.users_group_name
         multiple_counter[g_k] += 1
         p = perm.Permission.permission_name
         cur_perm = permissions[UK][g_k]
         if multiple_counter[g_k] > 1:
             p = _choose_perm(p, cur_perm)
         permissions[UK][g_k] = p
     #user explicit permission for user groups
     user_user_groups_perms = Permission.get_default_user_group_perms(uid)
     for perm in user_user_groups_perms:
         u_k = perm.UserUserGroupToPerm.user_group.users_group_name
         p = perm.Permission.permission_name
         cur_perm = permissions[UK][u_k]
         if not explicit:
             p = _choose_perm(p, cur_perm)
         permissions[UK][u_k] = p
     return permissions
 def allowed_api_access(controller_name, whitelist=None, api_key=None):
     """
     Check if given controller_name is in whitelist API access
     """
     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 user ID, username, API key or cookie dict, it looks
     up the matching database `User` 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, except when falling back
     to the default anonymous user (if enabled). 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.
     """
     def __init__(self, user_id=None, api_key=None, username=None):
     def __init__(self, user_id=None, api_key=None, username=None,
             is_external_auth=False):
         self.user_id = user_id
         self._api_key = api_key
         self.api_key = None
         self.username = username
         self.name = ''
         self.lastname = ''
         self.email = ''
         self.is_authenticated = False
         self.admin = False
         self.inherit_default_permissions = False
         self.is_external_auth = is_external_auth
         self.propagate_data()
         self._instance = None
     @LazyProperty
     def permissions(self):
         return self.__get_perms(user=self, cache=False)
     @property
     def api_keys(self):
         return self.get_api_keys()
     def propagate_data(self):
         user_model = UserModel()
         self.anonymous_user = User.get_default_user(cache=True)
         is_user_loaded = False
         # lookup by userid
         if self.user_id is not None and self.user_id != self.anonymous_user.user_id:
             log.debug('Auth User lookup by USER ID %s' % self.user_id)
             is_user_loaded = user_model.fill_data(self, user_id=self.user_id)
         # try go get user by API key
         elif self._api_key and self._api_key != self.anonymous_user.api_key:
             log.debug('Auth User lookup by API key %s' % self._api_key)
             is_user_loaded = user_model.fill_data(self, api_key=self._api_key)
         # lookup by username
         elif self.username:
             log.debug('Auth User lookup by USER NAME %s' % self.username)
             is_user_loaded = user_model.fill_data(self, username=self.username)
         else:
             log.debug('No data in %s that could been used to log in' % self)
         if not is_user_loaded:
             # if we cannot authenticate user try anonymous
             if self.anonymous_user.active:
                 user_model.fill_data(self, user_id=self.anonymous_user.user_id)
                 # then we set this user is logged in
                 self.is_authenticated = True
             else:
                 self.user_id = None
                 self.username = None
                 self.is_authenticated = False
         if not self.username:
             self.username = 'None'
         log.debug('Auth User is now %s' % self)
     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)
     def set_authenticated(self, authenticated=True):
         if self.user_id != self.anonymous_user.user_id:
             self.is_authenticated = authenticated
     def to_cookie(self):
         """ Serializes this login session to a cookie `dict`. """
         return {
             'user_id': self.user_id,
             'username': self.username,
             'is_authenticated': self.is_authenticated,
             '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'),
             username=cookie.get('username'),
             is_external_auth=cookie.get('is_external_auth', False),
+        )
         if not au.is_authenticated and au.user_id is not None:
             # user is not authenticated and not empty
             au.set_authenticated(cookie.get('is_authenticated'))
         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 = 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, **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):
         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)

kallithea/lib/base.py

➞

Show inline comments

@@ @@ -11,204 +11,205 @@ @@
+#
 # 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.controllers.util import redirect
 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):
+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(user_id=user.user_id)
     auth_user = AuthUser(user_id=user.user_id,
                          is_external_auth=is_external_auth)
     auth_user.set_authenticated()
     # 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
     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):
                 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 authfunc
         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):
@@ @@ -291,224 +292,225 @@ class BaseVCSController(object): @@
                 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 IP address of the request, the
         API key (if any), and the authuser from the session.
         """
         # Authenticate by API key
         if api_key:
             # when using API_KEY we are sure user exists.
             return AuthUser(api_key=api_key)
+            return AuthUser(api_key=api_key, is_external_auth=True)
         # Authenticate by session cookie
         cookie = session.get('authuser')
         # In ancient login sessions, 'authuser' may not be a dict.
         # In that case, the user will have to log in again.
         if isinstance(cookie, dict):
             try:
                 return AuthUser.from_cookie(cookie)
             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:
                 auth_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 auth_info:
                     username = auth_info['username']
                     user = User.get_by_username(username, case_insensitive=True)
                     return log_in_user(user, remember=False)
                     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)
                 return redirect(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')

kallithea/templates/base/base.html

➞

Show inline comments

@@ @@ -255,193 +255,196 @@ @@
         </a>
     </li>
     % if h.HasPermissionAll('hg.admin')('access admin main page'):
       <li ${is_current('admin')}>
         <a class="menu_link childs" title="${_('Admin')}" href="${h.url('admin_home')}">
           <i class="icon-gear"></i> ${_('Admin')}
         </a>
         ${admin_menu()}
       </li>
     % elif c.authuser.repositories_admin or c.authuser.repository_groups_admin or c.authuser.user_groups_admin:
     <li ${is_current('admin')}>
         <a class="menu_link childs" title="${_('Admin')}">
           <i class="icon-gear"></i> ${_('Admin')}
         </a>
         ${admin_menu_simple(c.authuser.repositories_admin,
                             c.authuser.repository_groups_admin,
                             c.authuser.user_groups_admin or h.HasPermissionAny('hg.usergroup.create.true')())}
     </li>
     % endif
     <li ${is_current('my_pullrequests')}>
       <a class="menu_link" title="${_('My Pull Requests')}" href="${h.url('my_pullrequests')}">
         <i class="icon-git-pull-request"></i> ${_('My Pull Requests')}
         %if c.my_pr_count != 0:
           <span class="menu_link_notifications">${c.my_pr_count}</span>
         %endif
       </a>
     </li>
     ## USER MENU
     <li>
       <a class="menu_link childs" id="quick_login_link">
           <span class="icon">
             ${h.gravatar(c.authuser.email, size=20)}
           </span>
           %if c.authuser.username != 'default':
             <span class="menu_link_user">${c.authuser.username}</span>
             %if c.unread_notifications != 0:
               <span class="menu_link_notifications">${c.unread_notifications}</span>
             %endif
           %else:
               <span>${_('Not Logged In')}</span>
           %endif
       </a>
       <div class="user-menu">
         <div id="quick_login">
           %if c.authuser.username == 'default' or c.authuser.user_id is None:
             <h4>${_('Login to Your Account')}</h4>
             ${h.form(h.url('login_home',came_from=h.url.current()))}
             <div class="form">
                 <div class="fields">
                     <div class="field">
                         <div class="label">
                             <label for="username">${_('Username')}:</label>
                         </div>
                         <div class="input">
                             ${h.text('username',class_='focus')}
                         </div>
                     </div>
                     <div class="field">
                         <div class="label">
                             <label for="password">${_('Password')}:</label>
                         </div>
                         <div class="input">
                             ${h.password('password',class_='focus')}
                         </div>
                     </div>
                     <div class="buttons">
                         <div class="password_forgoten">${h.link_to(_('Forgot password ?'),h.url('reset_password'))}</div>
                         <div class="register">
                         %if h.HasPermissionAny('hg.admin', 'hg.register.auto_activate', 'hg.register.manual_activate')():
                          ${h.link_to(_("Don't have an account ?"),h.url('register'))}
                         %endif
                         </div>
                         <div class="submit">
                             ${h.submit('sign_in',_('Log In'),class_="btn btn-mini")}
                         </div>
                     </div>
                 </div>
             </div>
             ${h.end_form()}
           %else:
             <div class="links_left">
                 <div class="big_gravatar">
                   ${h.gravatar(c.authuser.email, size=48)}
                 </div>
                 <div class="full_name">${c.authuser.full_name_or_username}</div>
                 <div class="email">${c.authuser.email}</div>
             </div>
             <div class="links_right">
             <ol class="links">
               <li><a href="${h.url('notifications')}">${_('Notifications')}: ${c.unread_notifications}</a></li>
               <li>${h.link_to(_(u'My Account'),h.url('my_account'))}</li>
               <li class="logout">${h.link_to(_(u'Log Out'),h.url('logout_home'))}</li>
               %if not c.authuser.is_external_auth:
                 ## Cannot log out if using external (container) authentication.
                 <li class="logout">${h.link_to(_(u'Log Out'), h.url('logout_home'))}</li>
               %endif
             </ol>
             </div>
           %endif
         </div>
       </div>
     </li>
     <script type="text/javascript">
         var visual_show_public_icon = "${c.visual.show_public_icon}" == "True";
         var cache = {}
         /*format the look of items in the list*/
         var format = function(state){
             if (!state.id){
               return state.text; // optgroup
+            }
             var obj_dict = state.obj;
             var tmpl = '';
             if(obj_dict && state.type == 'repo'){
                 tmpl += '<span class="repo-icons">';
                 if(obj_dict['repo_type'] === 'hg'){
                     tmpl += '<span class="repotag">hg</span> ';
+                }
                 else if(obj_dict['repo_type'] === 'git'){
                     tmpl += '<span class="repotag">git</span> ';
+                }
                 if(obj_dict['private']){
                     tmpl += '<i class="icon-keyhole-circled"></i> ';
+                }
                 else if(visual_show_public_icon){
                     tmpl += '<i class="icon-globe"></i> ';
+                }
                 tmpl += '</span>';
+            }
             if(obj_dict && state.type == 'group'){
                     tmpl += '<i class="icon-folder"></i> ';
+            }
             tmpl += state.text;
             return tmpl;
+        }
         $("#repo_switcher").select2({
             placeholder: '<i class="icon-database"></i> ${_('Repositories')}',
             dropdownAutoWidth: true,
             formatResult: format,
             formatSelection: format,
             formatNoMatches: function(term){
                 return "${_('No matches found')}";
             },
             containerCssClass: "repo-switcher",
             dropdownCssClass: "repo-switcher-dropdown",
             escapeMarkup: function(m){
                 // don't escape our custom placeholder
                 if(m.substr(0,29) == '<i class="icon-database"></i>'){
                     return m;
+                }
                 return Select2.util.escapeMarkup(m);
             },
             query: function(query){
               var key = 'cache';
               var cached = cache[key] ;
               if(cached) {
                 var data = {results: []};
                 //filter results
                 $.each(cached.results, function(){
                     var section = this.text;
                     var children = [];
                     $.each(this.children, function(){
                         if(query.term.length == 0 || this.text.toUpperCase().indexOf(query.term.toUpperCase()) >= 0 ){
                             children.push({'id': this.id, 'text': this.text, 'type': this.type, 'obj': this.obj});
+                        }
                     });
                     if(children.length !== 0){
                         data.results.push({'text': section, 'children': children});
+                    }
                 });
                 query.callback(data);
               }else{
                   $.ajax({
                     url: "${h.url('repo_switcher_data')}",
                     data: {},
                     dataType: 'json',
                     type: 'GET',
                     success: function(data) {
                       cache[key] = data;
                       query.callback({results: data.results});
+                    }
                   });
+              }
+            }
         });
         $("#repo_switcher").on('select2-selecting', function(e){
             e.preventDefault();

kallithea/tests/functional/test_admin_auth_settings.py

➞

Show inline comments

@@ @@ -82,96 +82,108 @@ class TestAuthSettingsController(TestCon @@
         self.log_user()
         if not ldap_lib_installed:
             raise SkipTest('skipping due to missing ldap lib')
         params = self._enable_plugins('kallithea.lib.auth_modules.auth_internal,kallithea.lib.auth_modules.auth_ldap')
         params.update({'auth_ldap_host': 'Host',
                        'auth_ldap_port': '123',
                        'auth_ldap_tls_kind': 'PLAIN',
                        'auth_ldap_tls_reqcert': 'NEVER',
                        'auth_ldap_dn_user': '',
                        'auth_ldap_dn_pass': '',
                        'auth_ldap_base_dn': '',
                        'auth_ldap_filter': '',
                        'auth_ldap_search_scope': 'BASE',
                        'auth_ldap_attr_login': '',  # <----- missing required input
                        'auth_ldap_attr_firstname': '',
                        'auth_ldap_attr_lastname': '',
                        'auth_ldap_attr_email': ''})
         test_url = url(controller='admin/auth_settings',
                        action='auth_settings')
         response = self.app.post(url=test_url, params=params)
         response.mustcontain("""<span class="error-message">The LDAP Login"""
                              """ attribute of the CN must be specified""")
     def test_ldap_login(self):
         pass
     def test_ldap_login_incorrect(self):
         pass
     def _container_auth_setup(self, **settings):
         self.log_user()
         params = self._enable_plugins('kallithea.lib.auth_modules.auth_internal,kallithea.lib.auth_modules.auth_container')
         params.update(settings)
         test_url = url(controller='admin/auth_settings',
                        action='auth_settings')
         response = self.app.post(url=test_url, params=params)
         response = response.follow()
         response.click('Log Out') # end admin login session
     def _container_auth_verify_login(self, resulting_username, **get_kwargs):
         response = self.app.get(
             url=url(controller='admin/my_account', action='my_account'),
             **get_kwargs
+        )
         response.mustcontain('My Account %s' % resulting_username)
     def test_container_auth_login_header(self):
         self._container_auth_setup(
             auth_container_header='THE_USER_NAME',
             auth_container_fallback_header='',
             auth_container_clean_username='False',
+        )
         self._container_auth_verify_login(
             extra_environ={'THE_USER_NAME': 'john@example.org'},
             resulting_username='john@example.org',
+        )
     def test_container_auth_login_fallback_header(self):
         self._container_auth_setup(
             auth_container_header='THE_USER_NAME',
             auth_container_fallback_header='HTTP_X_YZZY',
             auth_container_clean_username='False',
+        )
         self._container_auth_verify_login(
             headers={'X-Yzzy': r'foo\bar'},
             resulting_username=r'foo\bar',
+        )
     def test_container_auth_clean_username_at(self):
         self._container_auth_setup(
             auth_container_header='REMOTE_USER',
             auth_container_fallback_header='',
             auth_container_clean_username='True',
+        )
         self._container_auth_verify_login(
             extra_environ={'REMOTE_USER': 'john@example.org'},
             resulting_username='john',
+        )
     def test_container_auth_clean_username_backslash(self):
         self._container_auth_setup(
             auth_container_header='REMOTE_USER',
             auth_container_fallback_header='',
             auth_container_clean_username='True',
+        )
         self._container_auth_verify_login(
             extra_environ={'REMOTE_USER': r'example\jane'},
             resulting_username=r'jane',
+        )
     def test_container_auth_no_logout(self):
         self._container_auth_setup(
             auth_container_header='REMOTE_USER',
             auth_container_fallback_header='',
             auth_container_clean_username='True',
+        )
         response = self.app.get(
             url=url(controller='admin/my_account', action='my_account'),
             extra_environ={'REMOTE_USER': 'john'},
+        )
         self.assertNotIn('Log Out', response.normal_body)

0 comments (0 inline, 0 general)