…

                                  {

                                    "name":     "<usergroup name>",

                                    "type" :    "user_group",

                                    "permission" : "usergroup.(read|write|admin)"

                                  },

                                  …

                                ]

                 "followers":   [<user_obj>, ...],

                 <if with_revision_names == True>

                 "tags": {

                            "<tagname>": "<raw_id>",

                            ...

                         },

                 "branches": {

                            "<branchname>": "<raw_id>",

                            ...

                         },

                 "bookmarks": {

                            "<bookmarkname>": "<raw_id>",

                            ...

                         },

            }

          }

          error :  null

        """

        repo = get_repo_or_error(repoid)

        if not HasPermissionAny('hg.admin')():

            if not HasRepoPermissionLevel('read')(repo.repo_name):

                raise JSONRPCError('repository `%s` does not exist' % (repoid,))

        members = []

        for user in repo.repo_to_perm:

            perm = user.permission.permission_name

            user = user.user

            user_data = {

                'name': user.username,

                'type': "user",

                'permission': perm

            }

            members.append(user_data)

        for user_group in repo.users_group_to_perm:

            perm = user_group.permission.permission_name

            user_group = user_group.users_group

            user_group_data = {

                'name': user_group.users_group_name,

                'type': "user_group",

                'permission': perm

            }

            members.append(user_group_data)

        followers = [

            uf.user.get_api_data()

            for uf in repo.followers

        ]

        data = repo.get_api_data(with_revision_names=with_revision_names,

                                 with_pullrequests=with_pullrequests)

        data['members'] = members

        data['followers'] = followers

        return data

    # permission check inside

    def get_repos(self):

        """

        Lists all existing repositories. This command can be executed only using

        api_key belonging to user with admin rights or regular user that have

        admin, write or read access to repository.

        OUTPUT::

            id : <id_given_in_input>

            result: [

                      {

                        "repo_id" :          "<repo_id>",

                        "repo_name" :        "<reponame>"

                        "repo_type" :        "<repo_type>",

                        "clone_uri" :        "<clone_uri>",

                        "private": :         "<bool>",

                        "created_on" :       "<datetimecreated>",

                        "description" :      "<description>",

                        "landing_rev":       "<landing_rev>",

                        "owner":             "<repo_owner>",

                        "fork_of":           "<name_of_fork_parent>",

                        "enable_downloads":  "<bool>",

                        "enable_statistics": "<bool>",

                      },

                      …

                    ]

            error:  null

        """

        if not HasPermissionAny('hg.admin')():

        else:

            repos = db.Repository.query()

        return [

            repo.get_api_data()

            for repo in repos

        ]

    # permission check inside

    def get_repo_nodes(self, repoid, revision, root_path,

                       ret_type='all'):

        """

        returns a list of nodes and it's children in a flat list for a given path

        at given revision. It's possible to specify ret_type to show only `files` or

        `dirs`.  This command can be executed only using api_key belonging to

        user with admin rights or regular user that have at least read access to repository.

        :param repoid: repository name or repository id

        :type repoid: str or int

        :param revision: revision for which listing should be done

        :type revision: str

        :param root_path: path from which start displaying

        :type root_path: str

        :param ret_type: return type 'all|files|dirs' nodes

        :type ret_type: Optional(str)

        OUTPUT::

            id : <id_given_in_input>

            result: [

                      {

                        "name" :        "<name>"

                        "type" :        "<type>",

                      },

                      …

                    ]

            error:  null

        """

        repo = get_repo_or_error(repoid)

        if not HasPermissionAny('hg.admin')():

            if not HasRepoPermissionLevel('read')(repo.repo_name):

                raise JSONRPCError('repository `%s` does not exist' % (repoid,))

        ret_type = ret_type

        _map = {}

        try:

            _d, _f = ScmModel().get_nodes(repo, revision, root_path,

                                          flat=False)

            _map = {

                'all': _d + _f,

                'files': _f,

                'dirs': _d,

            }

            return _map[ret_type]

        except KeyError:

            raise JSONRPCError('ret_type must be one of %s'

                               % (','.join(sorted(_map))))

        except Exception:

            log.error(traceback.format_exc())

            raise JSONRPCError(

                'failed to get repo: `%s` nodes' % repo.repo_name

            )

    @HasPermissionAnyDecorator('hg.admin', 'hg.create.repository')

    def create_repo(self, repo_name, owner=None,

                    repo_type=None, description='',

                    private=False, clone_uri=None,

                    landing_rev='rev:tip',

                    enable_statistics=None,

                    enable_downloads=None,

                    copy_permissions=False):

        """

        Creates a repository. The repository name contains the full path, but the

        parent repository group must exist. For example "foo/bar/baz" require the groups

        "foo" and "bar" (with "foo" as parent), and create "baz" repository with

        "bar" as group. This command can be executed only using api_key

        belonging to user with admin rights or regular user that have create

        repository permission. Regular users cannot specify owner parameter

        :param repo_name: repository name

        :type repo_name: str

        :param owner: user_id or username

        :type owner: Optional(str)

        :param repo_type: 'hg' or 'git'

        :type repo_type: Optional(str)

        :param description: repository description

        :type description: Optional(str)

        :param private:

        :type private: bool

        :param clone_uri:

        :type clone_uri: str

        :param landing_rev: <rev_type>:<rev>

        :type landing_rev: str

        :param enable_downloads:

    def has_user_group_permission_level(self, user_group_name, level, purpose=None):

        required_perms = {

            'read': ['usergroup.read', 'usergroup.write', 'usergroup.admin'],

            'write': ['usergroup.write', 'usergroup.admin'],

            'admin': ['usergroup.admin'],

        }[level]

        actual_perm = self.user_group_permissions.get(user_group_name)

        ok = actual_perm in required_perms

        log.debug('Checking if user %r can %r user group %r (%s): %s (has %r)',

            self.username, level, user_group_name, purpose, ok, actual_perm)

        return ok

    @property

    def api_keys(self):

        return self._get_api_keys()

    def _get_api_keys(self):

        api_keys = [self.api_key]

        for api_key in db.UserApiKeys.query() \

                .filter_by(user_id=self.user_id, is_expired=False):

            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.repository_permissions.items()

                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.repository_group_permissions.items()

                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.user_group_permissions.items()

                if x[1] == 'usergroup.admin']

    def __repr__(self):

        return "<%s %s: %r>" % (self.__class__.__name__, self.user_id, self.username)

    def to_cookie(self):

        """ Serializes this login session to a cookie `dict`. """

        return {

            'user_id': self.user_id,

            'is_external_auth': self.is_external_auth,

        }

    @staticmethod

    def from_cookie(cookie, ip_addr):

        """

        Deserializes an `AuthUser` from a cookie `dict` ... or return None.

        """

        return AuthUser.make(

            dbuser=UserModel().get(cookie.get('user_id')),

            is_external_auth=cookie.get('is_external_auth', False),

            ip_addr=ip_addr,

        )

    @classmethod

    def get_allowed_ips(cls, user_id):

        _set = set()

        default_ips = db.UserIpMap.query().filter(db.UserIpMap.user_id == kallithea.DEFAULT_USER_ID)

        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 = db.UserIpMap.query().filter(db.UserIpMap.user_id == 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'])

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

# CHECK DECORATORS

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

def _redirect_to_login(message=None):

    """Return an exception that must be raised. It will redirect to the login

    page which will redirect back to the current URL after authentication.

    The optional message will be shown in a flash message."""

    if message:

        webutils.flash(message, category='warning')

    p = request.path_qs

    log.debug('Redirecting to login page, origin: %s', p)

    return HTTPFound(location=url('login_home', came_from=p))

# Use as decorator

class LoginRequired(object):

    """Client must be logged in as a valid User, or we'll redirect to the login

    page.

    If the "default" user is enabled and allow_default_user is true, that is

    considered valid too.

    Also checks that IP address is allowed.

    """

    def __init__(self, allow_default_user=False):

        self.allow_default_user = allow_default_user

    def __call__(self, func):

        return decorator(self.__wrapper, func)

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

        controller = fargs[0]

        user = request.authuser

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

        log.debug('Checking access for user %s @ %s', user, loc)

        # regular user authentication

        if user.is_default_user:

            if self.allow_default_user:

                log.info('default user @ %s', loc)

                return func(*fargs, **fkwargs)

            log.info('default user is not accepted here @ %s', loc)

        elif user.is_anonymous: # default user is disabled and no proper authentication

            log.warning('user is anonymous and NOT authenticated with regular auth @ %s', loc)

        else: # regular authentication

            log.info('user %s authenticated with regular auth @ %s', user, loc)

            return func(*fargs, **fkwargs)

        raise _redirect_to_login()

# Use as decorator

class NotAnonymous(object):

    """Ensures that client is not logged in as the "default" user, and

    redirects to the login page otherwise. Must be used together with

    LoginRequired."""

    def __call__(self, func):

        return decorator(self.__wrapper, func)

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

        cls = fargs[0]

        user = request.authuser

        log.debug('Checking that user %s is not anonymous @%s', user.username, cls)

        if user.is_default_user:

            raise _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 with multiple permissions"""

    def __init__(self, *required_perms):

        self.required_perms = required_perms # usually very short - a list is thus fine

    def __call__(self, func):

        return decorator(self.__wrapper, func)

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

        cls = fargs[0]

        user = request.authuser

        log.debug('checking %s permissions %s for %s %s',

          self.__class__.__name__, self.required_perms, cls, user)

        if self.check_permissions(user):

            log.debug('Permission granted for %s %s', cls, user)

            return func(*fargs, **fkwargs)

        else:

            log.info('Permission denied for %s %s', cls, user)

# -*- 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/>.

"""

Authentication modules

"""

import importlib

import logging

import traceback

from inspect import isfunction

from kallithea.lib.auth import AuthUser

from kallithea.lib.compat import hybrid_property

from kallithea.model import db, meta, validators

from kallithea.model.user import UserModel

from kallithea.model.user_group import UserGroupModel

log = logging.getLogger(__name__)

class LazyFormencode(object):

    def __init__(self, formencode_obj, *args, **kwargs):

        self.formencode_obj = formencode_obj

        self.args = args

        self.kwargs = kwargs

    def __call__(self, *args, **kwargs):

        formencode_obj = self.formencode_obj

        if isfunction(formencode_obj):

            # case we wrap validators into functions

            formencode_obj = self.formencode_obj(*args, **kwargs)

        return formencode_obj(*self.args, **self.kwargs)

class KallitheaAuthPluginBase(object):

    auth_func_attrs = {

        "username": "unique username",

        "firstname": "first name",

        "lastname": "last name",

        "email": "email address",

        "groups": '["list", "of", "groups"]',

        "extern_name": "name in external source of record",

        "admin": 'True|False defines if user should be Kallithea admin',

    }

    @property

    def validators(self):

        """

        Exposes Kallithea validators modules

        """

        # this is a hack to overcome issues with pylons threadlocals and

        # translator object _() not being registered properly.

        class LazyCaller(object):

            def __init__(self, name):

                self.validator_name = name

            def __call__(self, *args, **kwargs):

                obj = getattr(validators, self.validator_name)

                #log.debug('Initializing lazy formencode object: %s', obj)

                return LazyFormencode(obj, *args, **kwargs)

        class ProxyGet(object):

            def __getattribute__(self, name):

                return LazyCaller(name)

        return ProxyGet()

    @hybrid_property

    def name(self):

        """

        Returns the name of this authentication plugin.

        :returns: string

        """

        raise NotImplementedError("Not implemented in base class")

    @hybrid_property

    def is_container_auth(self):

        """

        Returns bool if this module uses container auth.

        This property will trigger an automatic call to authenticate on

        a visit to the website or during a push/pull.

        :returns: bool

        """

        return False

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

        """

        Checks if this authentication module should accept a request for

        the current user.

        :param user: user object fetched using plugin's get_user() method.

        :param accepts_empty: if True accepts don't allow the user to be empty

        :returns: boolean

        """

        plugin_name = self.name

        if not user and not accepts_empty:

            log.debug('User is empty not allowed to authenticate')

            return False

        if user and user.extern_type and user.extern_type != plugin_name:

            log.debug('User %s should authenticate using %s this is %s, skipping',

                      user, user.extern_type, plugin_name)

            return False

        return True

# -*- 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.model.repo

~~~~~~~~~~~~~~~~~~~~

Repository model 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: Jun 5, 2010

:author: marcink

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

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

"""

import logging

import os

import shutil

import traceback

from datetime import datetime

import kallithea.lib.utils2

from kallithea.lib.auth import HasRepoPermissionLevel, HasUserGroupPermissionLevel

from kallithea.lib.exceptions import AttachedForksError

from kallithea.lib.hooks import log_delete_repository

from kallithea.lib.utils import is_valid_repo_uri, make_ui

from kallithea.lib.utils2 import LazyProperty, get_current_authuser, obfuscate_url_pw, remove_prefix

from kallithea.lib.vcs.backends import get_backend

from kallithea.model import db, meta, scm

log = logging.getLogger(__name__)

class RepoModel(object):

    def _create_default_perms(self, repository, private):

        # create default permission

        default = 'repository.read'

        def_user = db.User.get_default_user()

        for p in def_user.user_perms:

            if p.permission.permission_name.startswith('repository.'):

                default = p.permission.permission_name

                break

        default_perm = 'repository.none' if private else default

        repo_to_perm = db.UserRepoToPerm()

        repo_to_perm.permission = db.Permission.get_by_key(default_perm)

        repo_to_perm.repository = repository

        repo_to_perm.user_id = def_user.user_id

        meta.Session().add(repo_to_perm)

        return repo_to_perm

    @LazyProperty

    def repos_path(self):

        """

        Gets the repositories root path from database

        """

        q = db.Ui.query().filter(db.Ui.ui_key == '/').one()

        return q.ui_value

    def get(self, repo_id):

        repo = db.Repository.query() \

            .filter(db.Repository.repo_id == repo_id)

        return repo.scalar()

    def get_repo(self, repository):

        return db.Repository.guess_instance(repository)

    def get_by_repo_name(self, repo_name):

        repo = db.Repository.query() \

            .filter(db.Repository.repo_name == repo_name)

        return repo.scalar()

    @classmethod

    def _render_datatable(cls, tmpl, *args, **kwargs):

        from tg import app_globals, request

        from tg import tmpl_context as c

        from tg.i18n import ugettext as _

        import kallithea.lib.helpers as h

        _tmpl_lookup = app_globals.mako_lookup

        template = _tmpl_lookup.get_template('data_table/_dt_elements.html')

        tmpl = template.get_def(tmpl)

        kwargs.update(dict(_=_, h=h, c=c, request=request))

        return tmpl.render_unicode(*args, **kwargs)

    def get_repos_as_dict(self, repos_list, repo_groups_list=None,

                          admin=False,

                          short_name=False):

        """Return repository list for use by DataTable.

        repos_list: list of repositories - but will be filtered for read permission.

        repo_groups_list: added at top of list without permission check.

        admin: return data for action column.

        """

        _render = self._render_datatable

        from tg import request

        from tg import tmpl_context as c

        import kallithea.lib.helpers as h

        def repo_lnk(name, rtype, rstate, private, fork_of):

            return _render('repo_name', name, rtype, rstate, private, fork_of,

                           short_name=short_name)

        def following(repo_id, is_following):

            return _render('following', repo_id, is_following)

        def last_change(last_change):

            return _render("last_change", last_change)

        def rss_lnk(repo_name):

            return _render("rss", repo_name)

        def atom_lnk(repo_name):

            return _render("atom", repo_name)

        def last_rev(repo_name, cs_cache):

            return _render('revision', repo_name, cs_cache.get('revision'),

                           cs_cache.get('raw_id'), cs_cache.get('author'),

                           cs_cache.get('message'))

        def desc(desc):

            import kallithea.lib.helpers as h

            return h.urlify_text(desc, truncate=80, stylize=c.visual.stylify_metalabels)

        def state(repo_state):

            return _render("repo_state", repo_state)

        def repo_actions(repo_name):

            return _render('repo_actions', repo_name)

        def owner_actions(owner_id, username):

            return _render('user_name', owner_id, username)

        repos_data = []

        for gr in repo_groups_list or []:

            repos_data.append(dict(

                raw_name='\0' + gr.name, # sort before repositories

                just_name=gr.name,

                name=_render('group_name_html', group_name=gr.group_name, name=gr.name),

                desc=gr.group_description))

        for repo in repos_list:

            if not HasRepoPermissionLevel('read')(repo.repo_name, 'get_repos_as_dict check'):

                continue

            cs_cache = repo.changeset_cache

            row = {

                "raw_name": repo.repo_name,

                "just_name": repo.just_name,

                "name": repo_lnk(repo.repo_name, repo.repo_type,

                                 repo.repo_state, repo.private, repo.fork),

                "following": following(

                    repo.repo_id,

                    scm.ScmModel().is_following_repo(repo.repo_name, request.authuser.user_id),

                ),

                "last_change_iso": repo.last_db_change.isoformat(),

                "last_change": last_change(repo.last_db_change),

                "last_changeset": last_rev(repo.repo_name, cs_cache),

                "last_rev_raw": cs_cache.get('revision'),

                "desc": desc(repo.description),

                "owner": h.person(repo.owner),

                "state": state(repo.repo_state),

                "rss": rss_lnk(repo.repo_name),

                "atom": atom_lnk(repo.repo_name),

            }

            if admin:

        response = api_call(self, params)

        assert "v0.2.0" in response.json['result']['tags']

        assert 'pull_requests' in response.json['result']

    @base.parametrize('grant_perm', [

        ('repository.admin'),

        ('repository.write'),

        ('repository.read'),

    ])

    def test_api_get_repo_by_non_admin(self, grant_perm):

        RepoModel().grant_user_permission(repo=self.REPO,

                                          user=self.TEST_USER_LOGIN,

                                          perm=grant_perm)

        meta.Session().commit()

        id_, params = _build_data(self.apikey_regular, 'get_repo',

                                  repoid=self.REPO)

        response = api_call(self, params)

        repo = RepoModel().get_by_repo_name(self.REPO)

        assert len(repo.repo_to_perm) >= 2  # make sure we actually are testing something - probably the default 2 permissions, possibly more

        expected = repo.get_api_data()

        members = []

        for user in repo.repo_to_perm:

            perm = user.permission.permission_name

            user_obj = user.user

            user_data = {'name': user_obj.username, 'type': "user",

                         'permission': perm}

            members.append(user_data)

        for user_group in repo.users_group_to_perm:

            perm = user_group.permission.permission_name

            user_group_obj = user_group.users_group

            user_group_data = {'name': user_group_obj.users_group_name,

                               'type': "user_group", 'permission': perm}

            members.append(user_group_data)

        expected['members'] = members

        followers = []

        for user in repo.followers:

            followers.append(user.user.get_api_data())

        expected['followers'] = followers

        try:

            self._compare_ok(id_, expected, given=response.body)

        finally:

            RepoModel().revoke_user_permission(self.REPO, self.TEST_USER_LOGIN)

    def test_api_get_repo_by_non_admin_no_permission_to_repo(self):

        RepoModel().grant_user_permission(repo=self.REPO,

                                          user=db.User.DEFAULT_USER_NAME,

                                          perm='repository.none')

        try:

            RepoModel().grant_user_permission(repo=self.REPO,

                                              user=self.TEST_USER_LOGIN,

                                              perm='repository.none')

            id_, params = _build_data(self.apikey_regular, 'get_repo',

                                      repoid=self.REPO)

            response = api_call(self, params)

            expected = 'repository `%s` does not exist' % (self.REPO)

            self._compare_error(id_, expected, given=response.body)

        finally:

            RepoModel().grant_user_permission(repo=self.REPO,

                                              user=db.User.DEFAULT_USER_NAME,

                                              perm='repository.read')

    def test_api_get_repo_that_doesn_not_exist(self):

        id_, params = _build_data(self.apikey, 'get_repo',

                                  repoid='no-such-repo')

        response = api_call(self, params)

        ret = 'repository `%s` does not exist' % 'no-such-repo'

        expected = ret

        self._compare_error(id_, expected, given=response.body)

    def test_api_get_repos(self):

        id_, params = _build_data(self.apikey, 'get_repos')

        response = api_call(self, params)

        expected = jsonify([

            repo.get_api_data()

            for repo in db.Repository.query()

        ])

        self._compare_ok(id_, expected, given=response.body)

    def test_api_get_repos_non_admin(self):

        id_, params = _build_data(self.apikey_regular, 'get_repos')

        response = api_call(self, params)

        expected = jsonify([

            repo.get_api_data()

        ])

        self._compare_ok(id_, expected, given=response.body)

    @base.parametrize('name,ret_type', [

        ('all', 'all'),

        ('dirs', 'dirs'),

        ('files', 'files'),

    ])

    def test_api_get_repo_nodes(self, name, ret_type):

        rev = 'tip'

        path = '/'

        id_, params = _build_data(self.apikey, 'get_repo_nodes',

                                  repoid=self.REPO, revision=rev,

                                  root_path=path,

                                  ret_type=ret_type)

        response = api_call(self, params)

        # we don't the actual return types here since it's tested somewhere

        # else

        expected = response.json['result']

        self._compare_ok(id_, expected, given=response.body)

    def test_api_get_repo_nodes_bad_revisions(self):

        rev = 'i-dont-exist'

        path = '/'

        id_, params = _build_data(self.apikey, 'get_repo_nodes',

                                  repoid=self.REPO, revision=rev,

                                  root_path=path, )

        response = api_call(self, params)

        expected = 'failed to get repo: `%s` nodes' % self.REPO

        self._compare_error(id_, expected, given=response.body)

    def test_api_get_repo_nodes_bad_path(self):

        rev = 'tip'

        path = '/idontexits'

        id_, params = _build_data(self.apikey, 'get_repo_nodes',

                                  repoid=self.REPO, revision=rev,

                                  root_path=path, )

        response = api_call(self, params)