@LoginRequired()

    def __before__(self):

        super(SettingsController, self).__before__()

    def _get_hg_ui_settings(self):

        ret = Ui.query().all()

        if not ret:

            raise Exception('Could not get application ui settings !')

        settings = {}

        for each in ret:

            k = each.ui_key

            v = each.ui_value

            if k == '/':

                k = 'root_path'

            if k == 'push_ssl':

                v = str2bool(v)

            if k.find('.') != -1:

                k = k.replace('.', '_')

            if each.ui_section in ['hooks', 'extensions']:

                v = each.ui_active

            settings[each.ui_section + '_' + k] = v

        return settings

    @HasPermissionAllDecorator('hg.admin')

    def settings_vcs(self):

        """GET /admin/settings: All items in the collection"""

        # url('admin_settings')

        c.active = 'vcs'

        if request.POST:

            application_form = ApplicationUiSettingsForm()()

            try:

                form_result = application_form.to_python(dict(request.POST))

            except formencode.Invalid as errors:

                return htmlfill.render(

                     render('admin/settings/settings.html'),

                     defaults=errors.value,

                     errors=errors.error_dict or {},

                     prefix_error=False,

                     encoding="UTF-8",

                     force_defaults=False)

            try:

                sett.ui_value = form_result['web_push_ssl']

                if c.visual.allow_repo_location_change:

                    sett.ui_value = form_result['paths_root_path']

                #HOOKS

                sett.ui_active = form_result['hooks_changegroup_update']

                sett.ui_active = form_result['hooks_changegroup_repo_size']

                sett.ui_active = form_result['hooks_changegroup_push_logger']

                sett.ui_active = form_result['hooks_outgoing_pull_logger']

                ## EXTENSIONS

                sett = Ui.get_or_create('extensions', 'largefiles')

                sett.ui_active = form_result['extensions_largefiles']

                sett = Ui.get_or_create('extensions', 'hgsubversion')

                sett.ui_active = form_result['extensions_hgsubversion']

                if sett.ui_active:

                    try:

                        import hgsubversion  # pragma: no cover

                    except ImportError:

                        raise HgsubversionImportError

#                sett = Ui.get_or_create('extensions', 'hggit')

#                sett.ui_active = form_result['extensions_hggit']

                Session().commit()

                h.flash(_('Updated VCS settings'), category='success')

            except HgsubversionImportError:

                log.error(traceback.format_exc())

                h.flash(_('Unable to activate hgsubversion support. '

                          'The "hgsubversion" library is missing'),

                        category='error')

            except Exception:

                log.error(traceback.format_exc())

                h.flash(_('Error occurred while updating '

                          'application settings'), category='error')

        defaults = Setting.get_app_settings()

        defaults.update(self._get_hg_ui_settings())

        return htmlfill.render(

            render('admin/settings/settings.html'),

            defaults=defaults,

            encoding="UTF-8",

            force_defaults=False)

    @HasPermissionAllDecorator('hg.admin')

    def settings_mapping(self):

        """GET /admin/settings/mapping: All items in the collection"""

        # url('admin_settings_mapping')

        c.active = 'mapping'

        if request.POST:

            rm_obsolete = request.POST.get('destroy', False)

forks controller for Kallithea

This file was forked by the Kallithea project in July 2014.

Original author and date, and relevant copyright and licensing information is below:

:created_on: Apr 23, 2011

:author: marcink

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

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

"""

import logging

import formencode

import traceback

from formencode import htmlfill

from pylons import tmpl_context as c, request, url

from pylons.i18n.translation import _

from webob.exc import HTTPFound

import kallithea.lib.helpers as h

from kallithea.lib.helpers import Page

from kallithea.lib.auth import LoginRequired, HasRepoPermissionAnyDecorator, \

    NotAnonymous, HasRepoPermissionAny, HasPermissionAnyDecorator, HasPermissionAny

from kallithea.lib.base import BaseRepoController, render

from kallithea.model.db import Repository, UserFollowing, User, Ui

from kallithea.model.repo import RepoModel

from kallithea.model.forms import RepoForkForm

from kallithea.model.scm import ScmModel, AvailableRepoGroupChoices

from kallithea.lib.utils2 import safe_int

log = logging.getLogger(__name__)

class ForksController(BaseRepoController):

    def __before__(self):

        super(ForksController, self).__before__()

    def __load_defaults(self):

        repo_group_perms = ['group.admin']

        if HasPermissionAny('hg.create.write_on_repogroup.true')():

            repo_group_perms.append('group.write')

        c.repo_groups = AvailableRepoGroupChoices(['hg.create.repository'], repo_group_perms)

        c.landing_revs_choices, c.landing_revs = ScmModel().get_repo_landing_revs()

    def __load_data(self, repo_name=None):

        """

        Load defaults settings for edit, and update

        :param repo_name:

        """

        self.__load_defaults()

        c.repo_info = db_repo = Repository.get_by_repo_name(repo_name)

        repo = db_repo.scm_instance

        if c.repo_info is None:

            h.not_mapped_error(repo_name)

            raise HTTPFound(location=url('repos'))

        c.default_user_id = User.get_default_user().user_id

        c.in_public_journal = UserFollowing.query() \

            .filter(UserFollowing.user_id == c.default_user_id) \

            .filter(UserFollowing.follows_repository == c.repo_info).scalar()

        if c.repo_info.stats:

            last_rev = c.repo_info.stats.stat_on_revision+1

        else:

            last_rev = 0

        c.stats_revision = last_rev

        c.repo_last_rev = repo.count() if repo.revisions else 0

        if last_rev == 0 or c.repo_last_rev == 0:

            c.stats_percentage = 0

        else:

            c.stats_percentage = '%.2f' % ((float((last_rev)) /

                                            c.repo_last_rev) * 100)

        defaults = RepoModel()._get_defaults(repo_name)

        # alter the description to indicate a fork

        defaults['description'] = ('fork of repository: %s \n%s'

                                   % (defaults['repo_name'],

                                      defaults['description']))

        # add suffix to fork

        defaults['repo_name'] = '%s-fork' % defaults['repo_name']

        return defaults

    @LoginRequired()

    @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',

                                   'repository.admin')

            if not HasRepoPermissionAny(

                'repository.read', 'repository.write', 'repository.admin'

            )(r.repo_name, 'get forks check'):

                continue

            d.append(r)

        c.forks_pager = Page(d, page=p, items_per_page=20)

        if request.environ.get('HTTP_X_PARTIAL_XHR'):

            return render('/forks/forks_data.html')

        return render('/forks/forks.html')

    @LoginRequired()

    @NotAnonymous()

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

    @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',

                                   'repository.admin')

    def fork(self, repo_name):

        c.repo_info = Repository.get_by_repo_name(repo_name)

        if not c.repo_info:

            h.not_mapped_error(repo_name)

            raise HTTPFound(location=url('home'))

        defaults = self.__load_data(repo_name)

        return htmlfill.render(

            render('forks/fork.html'),

            defaults=defaults,

            encoding="UTF-8",

            force_defaults=False)

    @LoginRequired()

    @NotAnonymous()

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

    @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',

                                   'repository.admin')

    def fork_create(self, repo_name):

        self.__load_defaults()

        c.repo_info = Repository.get_by_repo_name(repo_name)

        _form = RepoForkForm(old_data={'repo_type': c.repo_info.repo_type},

                             repo_groups=c.repo_groups,

                             landing_revs=c.landing_revs_choices)()

        form_result = {}

        task_id = None

        try:

            form_result = _form.to_python(dict(request.POST))

            # an approximation that is better than nothing

                form_result['update_after_clone'] = False

            # create fork is done sometimes async on celery, db transaction

            # management is handled there.

            task = RepoModel().create_fork(form_result, self.authuser.user_id)

            from celery.result import BaseAsyncResult

            if isinstance(task, BaseAsyncResult):

                task_id = task.task_id

        except formencode.Invalid as errors:

            return htmlfill.render(

                render('forks/fork.html'),

                defaults=errors.value,

                errors=errors.error_dict or {},

                prefix_error=False,

                encoding="UTF-8",

                force_defaults=False)

        except Exception:

            log.error(traceback.format_exc())

            h.flash(_('An error occurred during repository forking %s') %

                    repo_name, category='error')

        raise HTTPFound(location=h.url('repo_creating_home',

                              repo_name=form_result['repo_name_full'],

                              task_id=task_id))

        """

        Sets cache for this repository for invalidation on next access

        :param repo_name: full repo name, also a cache key

        """

        ScmModel().mark_for_invalidation(repo_name)

    def _check_permission(self, action, user, repo_name, ip_addr=None):

        """

        Checks permissions using action (push/pull) user and repository

        name

        :param action: push or pull action

        :param user: `User` instance

        :param repo_name: repository name

        """

        # check IP

        ip_allowed = AuthUser.check_ip_allowed(user, ip_addr)

        if ip_allowed:

            log.info('Access for IP:%s allowed', ip_addr)

        else:

            return False

        if action == 'push':

            if not HasPermissionAnyMiddleware('repository.write',

                                              'repository.admin')(user,

                                                                  repo_name):

                return False

        else:

            #any other action need at least read permission

            if not HasPermissionAnyMiddleware('repository.read',

                                              'repository.write',

                                              'repository.admin')(user,

                                                                  repo_name):

                return False

        return True

    def _get_ip_addr(self, environ):

        return _get_ip_addr(environ)

    def _check_ssl(self, environ):

        """

        Checks the SSL check flag and returns False if SSL is not present

        and required True otherwise

        """

        #check if we have SSL required  ! if not it's a bad request !

            org_proto = environ.get('wsgi._org_proto', environ['wsgi.url_scheme'])

            if org_proto != 'https':

                log.debug('proto is %s and SSL is required BAD REQUEST !',

                          org_proto)

                return False

        return True

    def _check_locking_state(self, environ, action, repo, user_id):

        """

        Checks locking on this repository, if locking is enabled and lock is

        present returns a tuple of make_lock, locked, locked_by.

        make_lock can have 3 states None (do nothing) True, make lock

        False release lock, This value is later propagated to hooks, which

        do the locking. Think about this as signals passed to hooks what to do.

        """

        locked = False  # defines that locked error should be thrown to user

        make_lock = None

        repo = Repository.get_by_repo_name(repo)

        user = User.get(user_id)

        # this is kind of hacky, but due to how mercurial handles client-server

        # server see all operation on changeset; bookmarks, phases and

        # obsolescence marker in different transaction, we don't want to check

        # locking on those

        obsolete_call = environ['QUERY_STRING'] in ['cmd=listkeys',]

        locked_by = repo.locked

        if repo and repo.enable_locking and not obsolete_call:

            if action == 'push':

                #check if it's already locked !, if it is compare users

                user_id, _date = repo.locked

                if user.user_id == user_id:

                    log.debug('Got push from user %s, now unlocking', user)

                    # unlock if we have push from user who locked

                    make_lock = False

                else:

                    # we're not the same user who locked, ban with 423 !

                    locked = True

            if action == 'pull':

                if repo.locked[0] and repo.locked[1]:

                    locked = True

                else:

                    log.debug('Setting lock on repo %s by %s', repo, user)

                    make_lock = True

        else:

            log.debug('Repository %s do not have locking enabled', repo)

        log.debug('FINAL locking values make_lock:%s,locked:%s,locked_by:%s',

            if strip_prefix:

                key = remove_prefix(key, prefix='default_')

            fd.update({key: row.app_settings_value})

        return fd

    @classmethod

    def get_server_info(cls):

        import pkg_resources

        import platform

        import kallithea

        from kallithea.lib.utils import check_git_version

        mods = [(p.project_name, p.version) for p in pkg_resources.working_set]

        info = {

            'modules': sorted(mods, key=lambda k: k[0].lower()),

            'py_version': platform.python_version(),

            'platform': safe_unicode(platform.platform()),

            'kallithea_version': kallithea.__version__,

            'git_version': safe_unicode(check_git_version()),

            'git_path': kallithea.CONFIG.get('git_path')

        }

        return info

class Ui(Base, BaseModel):

    __tablename__ = DB_PREFIX + 'ui'

    __table_args__ = (

        # FIXME: ui_key as key is wrong and should be removed when the corresponding

        # Ui.get_by_key has been replaced by the composite key

        UniqueConstraint('ui_key'),

        UniqueConstraint('ui_section', 'ui_key'),

        _table_args_default_dict,

    )

    HOOK_UPDATE = 'changegroup.update'

    HOOK_REPO_SIZE = 'changegroup.repo_size'

    HOOK_PUSH = 'changegroup.push_logger'

    HOOK_PRE_PUSH = 'prechangegroup.pre_push'

    HOOK_PULL = 'outgoing.pull_logger'

    HOOK_PRE_PULL = 'preoutgoing.pre_pull'

    ui_id = Column(Integer(), unique=True, primary_key=True)

    ui_section = Column(String(255), nullable=False)

    ui_key = Column(String(255), nullable=False)

    ui_value = Column(String(255), nullable=True) # FIXME: not nullable?

    ui_active = Column(Boolean(), nullable=False, default=True)

    @classmethod

        """ Return specified Ui object, or None if not found. """

    @classmethod

    def get_or_create(cls, section, key):

        """ Return specified Ui object, creating it if necessary. """

        if setting is None:

            setting = cls(ui_section=section, ui_key=key)

            Session().add(setting)

        return setting

    @classmethod

    def get_builtin_hooks(cls):

        q = cls.query()

        q = q.filter(cls.ui_key.in_([cls.HOOK_UPDATE, cls.HOOK_REPO_SIZE,

                                     cls.HOOK_PUSH, cls.HOOK_PRE_PUSH,

                                     cls.HOOK_PULL, cls.HOOK_PRE_PULL]))

        return q.all()

    @classmethod

    def get_custom_hooks(cls):

        q = cls.query()

        q = q.filter(~cls.ui_key.in_([cls.HOOK_UPDATE, cls.HOOK_REPO_SIZE,

                                      cls.HOOK_PUSH, cls.HOOK_PRE_PUSH,

                                      cls.HOOK_PULL, cls.HOOK_PRE_PULL]))

        q = q.filter(cls.ui_section == 'hooks')

        return q.all()

    @classmethod

    def get_repos_location(cls):

    @classmethod

    def create_or_update_hook(cls, key, val):

        new_ui = cls.get_or_create('hooks', key)

        new_ui.ui_active = True

        new_ui.ui_value = val

    def __repr__(self):

        return '<%s[%s]%s=>%s]>' % (self.__class__.__name__, self.ui_section,

                                    self.ui_key, self.ui_value)

class User(Base, BaseModel):

    __tablename__ = 'users'

    __table_args__ = (

        Index('u_username_idx', 'username'),

        Index('u_email_idx', 'email'),

        _table_args_default_dict,

    )

    DEFAULT_USER = 'default'

    DEFAULT_GRAVATAR_URL = 'https://secure.gravatar.com/avatar/{md5email}?d=identicon&s={size}'

    user_id = Column(Integer(), unique=True, primary_key=True)

    username = Column(String(255), nullable=False, unique=True)

    password = Column(String(255), nullable=False)

    active = Column(Boolean(), nullable=False, default=True)

    admin = Column(Boolean(), nullable=False, default=False)

    name = Column("firstname", Unicode(255), nullable=False)

    lastname = Column(Unicode(255), nullable=False)

    _email = Column("email", String(255), nullable=True, unique=True) # FIXME: not nullable?

    last_login = Column(DateTime(timezone=False), nullable=True)

    extern_type = Column(String(255), nullable=True) # FIXME: not nullable?

    extern_name = Column(String(255), nullable=True) # FIXME: not nullable?

    api_key = Column(String(255), nullable=False)

    inherit_default_permissions = Column(Boolean(), nullable=False, default=True)

    created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)

    _user_data = Column("user_data", LargeBinary(), nullable=True)  # JSON data # FIXME: not nullable?

    user_log = relationship('UserLog')

    user_perms = relationship('UserToPerm', primaryjoin="User.user_id==UserToPerm.user_id", cascade='all')

    repositories = relationship('Repository')

    repo_groups = relationship('RepoGroup')

    user_groups = relationship('UserGroup')

    user_followers = relationship('UserFollowing', primaryjoin='UserFollowing.follows_user_id==User.user_id', cascade='all')

    followings = relationship('UserFollowing', primaryjoin='UserFollowing.user_id==User.user_id', cascade='all')

"""

kallithea.model.repo_group

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

repo group 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: Jan 25, 2011

:author: marcink

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

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

"""

import os

import logging

import traceback

import shutil

import datetime

from kallithea.lib.utils2 import LazyProperty

from kallithea.model import BaseModel

from kallithea.model.db import RepoGroup, Ui, UserRepoGroupToPerm, \

    User, Permission, UserGroupRepoGroupToPerm, UserGroup, Repository

log = logging.getLogger(__name__)

class RepoGroupModel(BaseModel):

    cls = RepoGroup

    def _get_user_group(self, users_group):

        return self._get_instance(UserGroup, users_group,

                                  callback=UserGroup.get_by_group_name)

    def _get_repo_group(self, repo_group):

        return self._get_instance(RepoGroup, repo_group,

                                  callback=RepoGroup.get_by_group_name)

    @LazyProperty

    def repos_path(self):

        """

        Gets the repositories root path from database

        """

        return q.ui_value

    def _create_default_perms(self, new_group):

        # create default permission

        default_perm = 'group.read'

        def_user = User.get_default_user()

        for p in def_user.user_perms:

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

                default_perm = p.permission.permission_name

                break

        repo_group_to_perm = UserRepoGroupToPerm()

        repo_group_to_perm.permission = Permission.get_by_key(default_perm)

        repo_group_to_perm.group = new_group

        repo_group_to_perm.user_id = def_user.user_id

        return repo_group_to_perm

    def _create_group(self, group_name):

        """

        makes repository group on filesystem

        :param repo_name:

        :param parent_id:

        """

        create_path = os.path.join(self.repos_path, group_name)

        log.debug('creating new group in %s', create_path)

        if os.path.isdir(create_path):

            raise Exception('That directory already exists !')

        os.makedirs(create_path)

        log.debug('Created group in %s', create_path)

    def _rename_group(self, old, new):

        """

        Renames a group on filesystem

        :param group_name:

        """

        if old == new:

            log.debug('skipping group rename')

            return

        log.debug('renaming repository group from %s to %s', old, new)

class TestAdminSettingsController(TestController):

    def test_index_main(self):

        self.log_user()

        response = self.app.get(url('admin_settings'))

    def test_index_mapping(self):

        self.log_user()

        response = self.app.get(url('admin_settings_mapping'))

    def test_index_global(self):

        self.log_user()

        response = self.app.get(url('admin_settings_global'))

    def test_index_visual(self):

        self.log_user()

        response = self.app.get(url('admin_settings_visual'))

    def test_index_email(self):

        self.log_user()

        response = self.app.get(url('admin_settings_email'))

    def test_index_hooks(self):

        self.log_user()

        response = self.app.get(url('admin_settings_hooks'))

    def test_create_custom_hook(self):

        self.log_user()

        response = self.app.post(url('admin_settings_hooks'),

                                params=dict(new_hook_ui_key='test_hooks_1',

                                            new_hook_ui_value='cd /tmp',

                                            _authentication_token=self.authentication_token()))

        response = response.follow()

        response.mustcontain('test_hooks_1')

        response.mustcontain('cd /tmp')

    def test_create_custom_hook_delete(self):

        self.log_user()

        response = self.app.post(url('admin_settings_hooks'),

                                params=dict(new_hook_ui_key='test_hooks_2',

                                            new_hook_ui_value='cd /tmp2',

                                            _authentication_token=self.authentication_token()))

        response = response.follow()

        response.mustcontain('test_hooks_2')

        response.mustcontain('cd /tmp2')

        ## delete

        self.app.post(url('admin_settings_hooks'),

                        params=dict(hook_id=hook_id, _authentication_token=self.authentication_token()))

        response = self.app.get(url('admin_settings_hooks'))

        response.mustcontain(no=['test_hooks_2'])

        response.mustcontain(no=['cd /tmp2'])

    def test_index_search(self):

        self.log_user()

        response = self.app.get(url('admin_settings_search'))

    def test_index_system(self):

        self.log_user()

        response = self.app.get(url('admin_settings_system'))

    def test_ga_code_active(self):

        self.log_user()

        old_title = 'Kallithea'

        old_realm = 'Kallithea authentication'

        new_ga_code = 'ga-test-123456789'

        response = self.app.post(url('admin_settings_global'),

                        params=dict(title=old_title,

                                 realm=old_realm,

                                 ga_code=new_ga_code,

                                 captcha_private_key='',

                                 captcha_public_key='',

                                 _authentication_token=self.authentication_token(),

                                 ))

        self.checkSessionFlash(response, 'Updated application settings')

        self.assertEqual(Setting

                         .get_app_settings()['ga_code'], new_ga_code)

        response = response.follow()

        response.mustcontain("""_gaq.push(['_setAccount', '%s']);""" % new_ga_code)

    def test_ga_code_inactive(self):

        self.log_user()

        old_title = 'Kallithea'

        old_realm = 'Kallithea authentication'

        new_ga_code = ''

        response = self.app.post(url('admin_settings_global'),

                        params=dict(title=old_title,

                                 realm=old_realm,

                                 ga_code=new_ga_code,

                                 captcha_private_key='',

                                 captcha_public_key='',