#==============================================================================

# CHECK DECORATORS

#==============================================================================

class LoginRequired(object):

    """

    Must be logged in to execute this function else

    redirect to login page

    :param api_access: if enabled this checks only for valid auth token

        and grants access based on valid token

    """

    def __init__(self, api_access=False):

        self.api_access = api_access

    def __call__(self, func):

        return decorator(self.__wrapper, func)

    def __wrapper(self, func, *fargs, **fkwargs):

        cls = fargs[0]

        user = cls.authuser

        loc = "%s:%s" % (cls.__class__.__name__, func.__name__)

        # check if our IP is allowed

        ip_access_valid = True

        if not user.ip_allowed:

            from kallithea.lib import helpers as h

            h.flash(h.literal(_('IP %s not allowed' % (user.ip_addr))),

                    category='warning')

            ip_access_valid = False

        # check if we used an APIKEY and it's a valid one

        # defined whitelist of controllers which API access will be enabled

        _api_key = request.GET.get('api_key', '')

        api_access_valid = allowed_api_access(loc, api_key=_api_key)

        # explicit controller is enabled or API is in our whitelist

        if self.api_access or api_access_valid:

            log.debug('Checking API KEY access for %s' % cls)

            if _api_key and _api_key in user.api_keys:

                api_access_valid = True

                log.debug('API KEY ****%s is VALID' % _api_key[-4:])

            else:

                api_access_valid = False

                if not _api_key:

                    log.debug("API KEY *NOT* present in request")

                else:

        log.debug('Checking if %s is authenticated @ %s' % (user.username, loc))

        reason = 'RegularAuth' if user.is_authenticated else 'APIAuth'

        if ip_access_valid and (user.is_authenticated or api_access_valid):

            log.info('user %s authenticating with:%s IS authenticated on func %s '

                     % (user, reason, loc)

            )

            return func(*fargs, **fkwargs)

        else:

                     'IP_ACCESS:%s API_ACCESS:%s'

                     % (user, reason, loc, ip_access_valid, api_access_valid)

            )

            p = url.current()

            log.debug('redirecting to login page with %s' % p)

            return redirect(url('login_home', came_from=p))

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)

        anonymous = self.user.username == User.DEFAULT_USER

        if anonymous:

            p = url.current()

            import kallithea.lib.helpers as h

            h.flash(_('You need to be a registered user to '

                      'perform this action'),

                    category='warning')

            return redirect(url('login_home', came_from=p))

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

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: Created on Nov 17, 2012

:author: marcink

:copyright: (c) 2013 RhodeCode GmbH, and others.

:license: GPLv3, see LICENSE.md for more details.

"""

import logging

from kallithea import EXTERN_TYPE_INTERNAL

from kallithea.lib import auth_modules

from kallithea.lib.compat import formatted_json, hybrid_property

from kallithea.model.db import User

log = logging.getLogger(__name__)

class KallitheaAuthPlugin(auth_modules.KallitheaAuthPluginBase):

    def __init__(self):

        pass

    @hybrid_property

    def name(self):

        return EXTERN_TYPE_INTERNAL

    def settings(self):

        return []

    def user_activation_state(self):

        def_user_perms = User.get_default_user().AuthUser.permissions['global']

        return 'hg.register.auto_activate' in def_user_perms

    def accepts(self, user, accepts_empty=True):

        """

        Custom accepts for this auth that doesn't accept empty users. We

        know that user exisits in database.

        """

        return super(KallitheaAuthPlugin, self).accepts(user,

                                                        accepts_empty=False)

    def auth(self, userobj, username, password, settings, **kwargs):

        if not userobj:

            log.debug('userobj was:%s skipping' % (userobj, ))

            return None

        if userobj.extern_type != self.name:

                     % (userobj, userobj.extern_type, self.name))

            return None

        user_attrs = {

            "username": userobj.username,

            "firstname": userobj.firstname,

            "lastname": userobj.lastname,

            "groups": [],

            "email": userobj.email,

            "admin": userobj.admin,

            "active": userobj.active,

            "active_from_extern": userobj.active,

            "extern_name": userobj.user_id,

            'extern_type': userobj.extern_type,

        }

        log.debug(formatted_json(user_attrs))

        if userobj.active:

            from kallithea.lib import auth

            password_match = auth.KallitheaCrypto.hash_check(password, userobj.password)

            if userobj.username == User.DEFAULT_USER and userobj.active:

                log.info('user %s authenticated correctly as anonymous user' %

                         username)

                return user_attrs

            elif userobj.username == username and password_match:

                log.info('user %s authenticated correctly' % user_attrs['username'])

                return user_attrs

            log.error("user %s had a bad password" % username)

            return None

        else:

            log.warning('user %s tried auth but is disabled' % username)

            return None

    def auth(self, userobj, username, password, settings, **kwargs):

        if username not in _auth_cache:

            # Need lock here, as PAM authentication is not thread safe

            _pam_lock.acquire()

            try:

                auth_result = pam.authenticate(username, password,

                                               settings["service"])

                # cache result only if we properly authenticated

                if auth_result:

                    _auth_cache[username] = time.time()

            finally:

                _pam_lock.release()

            if not auth_result:

                log.error("PAM was unable to authenticate user: %s" % (username,))

                return None

        else:

            log.debug("Using cached auth for user: %s" % (username,))

        # old attrs fetched from Kallithea database

        admin = getattr(userobj, 'admin', False)

        active = getattr(userobj, 'active', True)

        email = getattr(userobj, 'email', '') or "%s@%s" % (username, socket.gethostname())

        firstname = getattr(userobj, 'firstname', '')

        lastname = getattr(userobj, 'lastname', '')

        extern_type = getattr(userobj, 'extern_type', '')

        user_attrs = {

            'username': username,

            'firstname': firstname,

            'lastname': lastname,

            'groups': [g.gr_name for g in grp.getgrall() if username in g.gr_mem],

            'email': email,

            'admin': admin,

            'active': active,

            "active_from_extern": None,

            'extern_name': username,

            'extern_type': extern_type,

        }

        try:

            user_data = pwd.getpwnam(username)

            regex = settings["gecos"]

            match = re.search(regex, user_data.pw_gecos)

            if match:

                user_attrs["firstname"] = match.group('first_name')

                user_attrs["lastname"] = match.group('last_name')

        except Exception:

            pass

        log.debug("pamuser: \n%s" % formatted_json(user_attrs))

        log.info('user %s authenticated correctly' % user_attrs['username'])

        return user_attrs

            return

        u2p = UserToPerm.query()\

            .filter(UserToPerm.user == default_user).all()

        fixed = False

        if len(u2p) != len(Permission.DEFAULT_USER_PERMISSIONS):

            for p in u2p:

                Session().delete(p)

            fixed = True

            self.populate_default_permissions()

        return fixed

    def update_repo_info(self):

        RepoModel.update_repoinfo()

    def config_prompt(self, test_repo_path='', retries=3):

        defaults = self.cli_args

        _path = defaults.get('repos_location')

        if retries == 3:

            log.info('Setting up repositories config')

        if _path is not None:

            path = _path

        elif not self.tests and not test_repo_path:

            path = raw_input(

                 'Enter a valid absolute path to store repositories. '

                 'All repositories in that path will be added automatically:'

            )

        else:

            path = test_repo_path

        path_ok = True

        # check proper dir

        if not os.path.isdir(path):

            path_ok = False

            log.error('Given path %s is not a valid directory' % (path,))

        elif not os.path.isabs(path):

            path_ok = False

            log.error('Given path %s is not an absolute path' % (path,))

        # check if path is at least readable.

        if not os.access(path, os.R_OK):

            path_ok = False

            log.error('Given path %s is not readable' % (path,))

        # check write access, warn user about non writeable paths

        elif not os.access(path, os.W_OK) and path_ok:

            if not ask_ok('Given path %s is not writeable, do you want to '

                          'continue with read only mode ? [y/n]' % (path,)):

                log.error('Canceled by user')

                sys.exit(-1)

        if retries == 0:

            sys.exit('max retries reached')

        if not path_ok:

            retries -= 1

            return self.config_prompt(test_repo_path, retries)

        real_path = os.path.normpath(os.path.realpath(path))

        if real_path != os.path.normpath(path):

            if not ask_ok(('Path looks like a symlink, Kallithea will store '

                           'given path as %s ? [y/n]') % (real_path,)):

                log.error('Canceled by user')

                sys.exit(-1)

        return real_path

    def create_settings(self, path):

        self.create_ui_settings(path)

        ui_config = [

            ('web', 'push_ssl', 'false'),

            ('web', 'allow_archive', 'gz zip bz2'),

            ('web', 'allow_push', '*'),

            ('web', 'baseurl', '/'),

            ('paths', '/', path),

            #('phases', 'publish', 'false')

        ]

        for section, key, value in ui_config:

            ui_conf = Ui()

            setattr(ui_conf, 'ui_section', section)

            setattr(ui_conf, 'ui_key', key)

            setattr(ui_conf, 'ui_value', value)

            self.sa.add(ui_conf)

        settings = [

            ('realm', 'Kallithea', 'unicode'),

            ('title', '', 'unicode'),

            ('ga_code', '', 'unicode'),

            ('show_public_icon', True, 'bool'),

            ('show_private_icon', True, 'bool'),

            ('stylify_metatags', False, 'bool'),

            ('dashboard_items', 100, 'int'),

        else:

            raise Exception('You have to provide a user object or a username')

        if getattr(repo, 'repo_id', None):

            repo_obj = Repository.get(repo.repo_id)

            repo_name = repo_obj.repo_name

        elif isinstance(repo, basestring):

            repo_name = repo.lstrip('/')

            repo_obj = Repository.get_by_repo_name(repo_name)

        else:

            repo_obj = None

            repo_name = ''

        user_log = UserLog()

        user_log.user_id = user_obj.user_id

        user_log.username = user_obj.username

        user_log.action = safe_unicode(action)

        user_log.repository = repo_obj

        user_log.repository_name = repo_name

        user_log.action_date = datetime.datetime.now()

        user_log.user_ip = ipaddr

        sa.add(user_log)

        log.info('Logging action:%s on %s by user:%s ip:%s' %

                 (action, safe_unicode(repo), user_obj, ipaddr))

        if commit:

            sa.commit()

    except Exception:

        log.error(traceback.format_exc())

        raise

def get_filesystem_repos(path, recursive=False, skip_removed_repos=True):

    """

    Scans given path for repos and return (name,(type,path)) tuple

    :param path: path to scan for repositories

    :param recursive: recursive search and return names with subdirs in front

    """

    # remove ending slash for better results

    path = path.rstrip(os.sep)

    log.debug('now scanning in %s location recursive:%s...' % (path, recursive))

    def _get_repos(p):

        if not os.access(p, os.R_OK) or not os.access(p, os.X_OK):

            return

        if not os.access(p, os.W_OK):

        for dirpath in os.listdir(p):

            if os.path.isfile(os.path.join(p, dirpath)):

                continue

            cur_path = os.path.join(p, dirpath)

            # skip removed repos

            if skip_removed_repos and REMOVED_REPO_PAT.match(dirpath):

                continue

            #skip .<somethin> dirs

            if dirpath.startswith('.'):

                continue

            try:

                scm_info = get_scm(cur_path)

                yield scm_info[1].split(path, 1)[-1].lstrip(os.sep), scm_info

            except VCSError:

                if not recursive:

                    continue

                #check if this dir containts other repos for recursive scan

                rec_path = os.path.join(p, dirpath)

                if os.path.isdir(rec_path):

                    for inner_scm in _get_repos(rec_path):

                        yield inner_scm

    return _get_repos(path)

def is_valid_repo(repo_name, base_path, scm=None):

    """

    Returns True if given path is a valid repository False otherwise.

    If scm param is given also compare if given scm is the same as expected

    from scm parameter

    :param repo_name:

    :param base_path:

    :param scm:

    :return True: if given path is a valid repository

    """

    full_path = os.path.join(safe_str(base_path), safe_str(repo_name))

    try:

        scm_ = get_scm(full_path)

        if scm:

            return scm_[0] == scm

        return True

    except VCSError: