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

import functools

import io

import logging.config

import os

import re

import sys

import click

import paste.deploy

import kallithea

# kallithea_cli is usually invoked through the 'kallithea-cli' wrapper script

# that is installed by setuptools, as specified in setup.py console_scripts

# entry_points. The script will be using the right virtualenv (if any), and for

# Unix, it will contain #! pointing at the right python executable. The script

# also makes sure sys.argv[0] points back at the script path, and that is what

# can be used to invoke 'kallithea-cli' later.

kallithea_cli_path = sys.argv[0]

def read_config(ini_file_name, strip_section_prefix):

    """Read ini_file_name content, and for all sections like '[X:Y]' where X is

    strip_section_prefix, replace the section name with '[Y]'."""

    def repl(m):

        if m.group(1) == strip_section_prefix:

            return '[%s]' % m.group(2)

        return m.group(0)

    with open(ini_file_name) as f:

        return re.sub(r'^\[([^:]+):(.*)]', repl, f.read().decode(), flags=re.MULTILINE)

# This placeholder is the main entry point for the kallithea-cli command

@click.group(context_settings=dict(help_option_names=['-h', '--help']))

def cli():

    """Various commands to manage a Kallithea instance."""

def register_command(config_file=False, config_file_initialize_app=False, hidden=False):

    """Register a kallithea-cli subcommand.

    If one of the config_file flags are true, a config file must be specified

    with -c and it is read and logging is configured. The configuration is

    available in the kallithea.CONFIG dict.

    If config_file_initialize_app is true, Kallithea, TurboGears global state

    (including tg.config), and database access will also be fully initialized.

    """

    cli_command = cli.command(hidden=hidden)

    if config_file or config_file_initialize_app:

        def annotator(annotated):

            @click.option('--config_file', '-c', help="Path to .ini file with app configuration.",

                type=click.Path(dir_okay=False, exists=True, readable=True), required=True)

            @functools.wraps(annotated) # reuse meta data from the wrapped function so click can see other options

            def runtime_wrapper(config_file, *args, **kwargs):

                path_to_ini_file = os.path.realpath(config_file)

                kallithea.CONFIG = paste.deploy.appconfig('config:' + path_to_ini_file)

                config_string = read_config(path_to_ini_file, strip_section_prefix=annotated.__name__)

                logging.config.fileConfig(io.StringIO(config_string))

                if config_file_initialize_app:

                    kallithea.config.middleware.make_app_without_logging(kallithea.CONFIG.global_conf, **kallithea.CONFIG.local_conf)

                return annotated(*args, **kwargs)

            return cli_command(runtime_wrapper)

        return annotator

    return cli_command

# -*- coding: utf-8 -*-

# This program is free software: you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""

kallithea.lib.hooks

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

Hooks run by 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: Aug 6, 2010

:author: marcink

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

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

"""

import os

import time

import mercurial.scmutil

from kallithea.lib import helpers as h

from kallithea.lib.exceptions import UserCreationError

from kallithea.lib.utils import action_logger, make_ui

from kallithea.lib.utils2 import ascii_str, get_hook_environment, safe_bytes, safe_str, safe_unicode

from kallithea.lib.vcs.backends.base import EmptyChangeset

from kallithea.model.db import Repository, User

def _get_scm_size(alias, root_path):

    if not alias.startswith('.'):

        alias += '.'

    size_scm, size_root = 0, 0

    for path, dirs, files in os.walk(safe_str(root_path)):

        if path.find(alias) != -1:

            for f in files:

                try:

                    size_scm += os.path.getsize(os.path.join(path, f))

                except OSError:

                    pass

        else:

            for f in files:

                try:

                    size_root += os.path.getsize(os.path.join(path, f))

                except OSError:

                    pass

    size_scm_f = h.format_byte_size(size_scm)

    size_root_f = h.format_byte_size(size_root)

    size_total_f = h.format_byte_size(size_root + size_scm)

    return size_scm_f, size_root_f, size_total_f

def repo_size(ui, repo, hooktype=None, **kwargs):

    """Show size of Mercurial repository, to be called after push."""

    size_hg_f, size_root_f, size_total_f = _get_scm_size('.hg', repo.root)

    last_cs = repo[len(repo) - 1]

    msg = ('Repository size .hg: %s Checkout: %s Total: %s\n'

           'Last revision is now r%s:%s\n') % (

        size_hg_f, size_root_f, size_total_f, last_cs.rev(), ascii_str(last_cs.hex())[:12]

    )

    ui.status(safe_bytes(msg))

def log_pull_action(ui, repo, **kwargs):

    """Logs user last pull action

    Called as Mercurial hook outgoing.pull_logger or from Kallithea before invoking Git.

    Does *not* use the action from the hook environment but is always 'pull'.

    """

    ex = get_hook_environment()

    user = User.get_by_username(ex.username)

    action = 'pull'

    action_logger(user, action, ex.repository, ex.ip, commit=True)

    # extension hook call

    from kallithea import EXTENSIONS

    callback = getattr(EXTENSIONS, 'PULL_HOOK', None)

    if callable(callback):

        kw = {}

        kw.update(ex)

        callback(**kw)

    return 0

def log_push_action(ui, repo, node, node_last, **kwargs):

    """

    Entry point for Mercurial hook changegroup.push_logger.

    The pushed changesets is given by the revset 'node:node_last'.

    Note: This hook is not only logging, but also the side effect invalidating

    cahes! The function should perhaps be renamed.

    """

    revs = [ascii_str(repo[r].hex()) for r in mercurial.scmutil.revrange(repo, [b'%s:%s' % (node, node_last)])]

    process_pushed_raw_ids(revs)

    return 0

def process_pushed_raw_ids(revs):

    """

    Register that changes have been added to the repo - log the action *and* invalidate caches.

    Called from  Mercurial changegroup.push_logger calling hook log_push_action,

    or from the Git post-receive hook calling handle_git_post_receive ...

    or from scm _handle_push.

    """

    ex = get_hook_environment()

    action = '%s:%s' % (ex.action, ','.join(revs))

    action_logger(ex.username, action, ex.repository, ex.ip, commit=True)

    from kallithea.model.scm import ScmModel

    ScmModel().mark_for_invalidation(ex.repository)

    # extension hook call

    from kallithea import EXTENSIONS

    callback = getattr(EXTENSIONS, 'PUSH_HOOK', None)

    if callable(callback):

        kw = {'pushed_revs': revs}

        kw.update(ex)

        callback(**kw)

def log_create_repository(repository_dict, created_by, **kwargs):

    """

    Post create repository Hook.

    :param repository: dict dump of repository object

    :param created_by: username who created repository

    available keys of repository_dict:

     'repo_type',

     'description',

     'private',

     'created_on',

     'enable_downloads',

     'repo_id',

     'owner_id',

     'enable_statistics',

     'clone_uri',

     'fork_id',

     'group_id',

     'repo_name'

    """

    from kallithea import EXTENSIONS

    callback = getattr(EXTENSIONS, 'CREATE_REPO_HOOK', None)

    if callable(callback):

        kw = {}

        kw.update(repository_dict)

        kw.update({'created_by': created_by})

        kw.update(kwargs)

        return callback(**kw)

    return 0

def check_allowed_create_user(user_dict, created_by, **kwargs):

    # pre create hooks

    from kallithea import EXTENSIONS

    callback = getattr(EXTENSIONS, 'PRE_CREATE_USER_HOOK', None)

    if callable(callback):

        allowed, reason = callback(created_by=created_by, **user_dict)

        if not allowed:

            raise UserCreationError(reason)

def log_create_user(user_dict, created_by, **kwargs):

    """

    Post create user Hook.

    :param user_dict: dict dump of user object

    available keys for user_dict:

     'username',

     'full_name_or_username',

     'full_contact',

     'user_id',

     'name',

     'firstname',

     'short_contact',

     'admin',

     'lastname',

     'ip_addresses',

     'ldap_dn',

     'email',

     'api_key',

     'last_login',

     'full_name',

     'active',

     'password',

     'emails',

    """

    from kallithea import EXTENSIONS

    callback = getattr(EXTENSIONS, 'CREATE_USER_HOOK', None)

    if callable(callback):

        return callback(created_by=created_by, **user_dict)

    return 0

def log_delete_repository(repository_dict, deleted_by, **kwargs):

    """

    Post delete repository Hook.

    :param repository: dict dump of repository object

    :param deleted_by: username who deleted the repository

    available keys of repository_dict:

     'repo_type',

     'description',

     'private',

     'created_on',

     'enable_downloads',

     'repo_id',

     'owner_id',

     'enable_statistics',

     'clone_uri',

     'fork_id',

     'group_id',

     'repo_name'

    """

    from kallithea import EXTENSIONS

    callback = getattr(EXTENSIONS, 'DELETE_REPO_HOOK', None)

    if callable(callback):

        kw = {}

        kw.update(repository_dict)

        kw.update({'deleted_by': deleted_by,

                   'deleted_on': time.time()})

        kw.update(kwargs)

        return callback(**kw)

    return 0

def log_delete_user(user_dict, deleted_by, **kwargs):

    """

    Post delete user Hook.

    :param user_dict: dict dump of user object

    available keys for user_dict:

     'username',

     'full_name_or_username',

     'full_contact',

     'user_id',

     'name',

     'firstname',

     'short_contact',

     'admin',

     'lastname',

     'ip_addresses',

     'ldap_dn',

     'email',

     'api_key',

     'last_login',

     'full_name',

     'active',

     'password',

     'emails',

    """

    from kallithea import EXTENSIONS

    callback = getattr(EXTENSIONS, 'DELETE_USER_HOOK', None)

    if callable(callback):

        return callback(deleted_by=deleted_by, **user_dict)

    return 0

def _hook_environment(repo_path):

    """

    Create a light-weight environment for stand-alone scripts and return an UI and the

    db repository.

    Git hooks are executed as subprocess of Git while Kallithea is waiting, and

    they thus need enough info to be able to create an app environment and

    connect to the database.

    """

    import paste.deploy

    import kallithea.config.middleware

    extras = get_hook_environment()

    path_to_ini_file = extras['config']

    kallithea.CONFIG = paste.deploy.appconfig('config:' + path_to_ini_file)

    #logging.config.fileConfig(ini_file_path) # Note: we are in a different process - don't use configured logging

    kallithea.config.middleware.make_app_without_logging(kallithea.CONFIG.global_conf, **kallithea.CONFIG.local_conf)

    repo_path = safe_unicode(repo_path)

    # fix if it's not a bare repo

    if repo_path.endswith(os.sep + '.git'):

        repo_path = repo_path[:-5]

    repo = Repository.get_by_full_path(repo_path)

    if not repo:

        raise OSError('Repository %s not found in database'

                      % (safe_str(repo_path)))

    baseui = make_ui()

    return baseui, repo

def handle_git_pre_receive(repo_path, git_stdin_lines):

    """Called from Git pre-receive hook"""

    # Currently unused. TODO: remove?

    return 0

def handle_git_post_receive(repo_path, git_stdin_lines):

    """Called from Git post-receive hook"""

    baseui, repo = _hook_environment(repo_path)

    # the post push hook should never use the cached instance

    scm_repo = repo.scm_instance_no_cache()

    rev_data = []

    for l in git_stdin_lines:

        old_rev, new_rev, ref = l.strip().split(' ')

        _ref_data = ref.split('/')

        if _ref_data[1] in ['tags', 'heads']:

            rev_data.append({'old_rev': old_rev,

                             'new_rev': new_rev,

                             'ref': ref,

                             'type': _ref_data[1],

                             'name': '/'.join(_ref_data[2:])})

    git_revs = []

    for push_ref in rev_data:

        _type = push_ref['type']

        if _type == 'heads':

            if push_ref['old_rev'] == EmptyChangeset().raw_id:

                # update the symbolic ref if we push new repo

                if scm_repo.is_empty():

                    scm_repo._repo.refs.set_symbolic_ref(

                        b'HEAD',

                        b'refs/heads/%s' % safe_bytes(push_ref['name']))

                # build exclude list without the ref

                cmd = ['for-each-ref', '--format=%(refname)', 'refs/heads/*']

                stdout = scm_repo.run_git_command(cmd)

                ref = push_ref['ref']

                heads = [head for head in stdout.splitlines() if head != ref]

                # now list the git revs while excluding from the list

                cmd = ['log', push_ref['new_rev'], '--reverse', '--pretty=format:%H']

                cmd.append('--not')

                cmd.extend(heads) # empty list is ok

                stdout = scm_repo.run_git_command(cmd)

                git_revs += stdout.splitlines()

            elif push_ref['new_rev'] == EmptyChangeset().raw_id:

                # delete branch case

                git_revs += ['delete_branch=>%s' % push_ref['name']]

            else:

                cmd = ['log', '%(old_rev)s..%(new_rev)s' % push_ref,

                       '--reverse', '--pretty=format:%H']

                stdout = scm_repo.run_git_command(cmd)

                git_revs += stdout.splitlines()

        elif _type == 'tags':

            git_revs += ['tag=>%s' % push_ref['name']]

    process_pushed_raw_ids(git_revs)

    return 0

# Almost exactly like Mercurial contrib/hg-ssh:

def rejectpush(ui, **kwargs):

    """Mercurial hook to be installed as pretxnopen and prepushkey for read-only repos"""

    ex = get_hook_environment()

    ui.warn(safe_bytes("Push access to %r denied\n" % safe_str(ex.repository)))

    return 1

# -*- coding: utf-8 -*-

# This program is free software: you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""

kallithea.lib.utils

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

Utilities library 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 18, 2010

:author: marcink

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

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

"""

import datetime

import logging

import os

import re

import sys

import traceback

from distutils.version import StrictVersion

import mercurial.config

import mercurial.ui

from tg.i18n import ugettext as _

import kallithea.config.conf

from kallithea.lib.exceptions import HgsubversionImportError

from kallithea.lib.utils2 import ascii_bytes, aslist, get_current_authuser, safe_bytes, safe_str, safe_unicode

from kallithea.lib.vcs.backends.git.repository import GitRepository

from kallithea.lib.vcs.backends.hg.repository import MercurialRepository

from kallithea.lib.vcs.conf import settings

from kallithea.lib.vcs.exceptions import RepositoryError, VCSError

from kallithea.lib.vcs.utils.fakemod import create_module

from kallithea.lib.vcs.utils.helpers import get_scm

from kallithea.model import meta

from kallithea.model.db import RepoGroup, Repository, Setting, Ui, User, UserGroup, UserLog

log = logging.getLogger(__name__)

REMOVED_REPO_PAT = re.compile(r'rm__\d{8}_\d{6}_\d{6}_.*')

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

# PERM DECORATOR HELPERS FOR EXTRACTING NAMES FOR PERM CHECKS

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

def get_repo_slug(request):

    _repo = request.environ['pylons.routes_dict'].get('repo_name')

    if _repo:

        _repo = _repo.rstrip('/')

    return _repo

def get_repo_group_slug(request):

    _group = request.environ['pylons.routes_dict'].get('group_name')

    if _group:

        _group = _group.rstrip('/')

    return _group

def get_user_group_slug(request):

    _group = request.environ['pylons.routes_dict'].get('id')

    _group = UserGroup.get(_group)

    if _group:

        return _group.users_group_name

    return None

def _get_permanent_id(s):

    """Helper for decoding stable URLs with repo ID. For a string like '_123'

    return 123.

    """

    by_id_match = re.match(r'^_(\d+)$', s)

    if by_id_match is None:

        return None

    return int(by_id_match.group(1))

def fix_repo_id_name(path):

    """

    Rewrite repo_name for _<ID> permanent URLs.

    Given a path, if the first path element is like _<ID>, return the path with

    this part expanded to the corresponding full repo name, else return the

    provided path.

    """

    first, rest = path, ''

    if '/' in path:

        first, rest_ = path.split('/', 1)

        rest = '/' + rest_

    repo_id = _get_permanent_id(first)

    if repo_id is not None:

        repo = Repository.get(repo_id)

        if repo is not None:

            return repo.repo_name + rest

    return path

def action_logger(user, action, repo, ipaddr='', commit=False):

    """

    Action logger for various actions made by users

    :param user: user that made this action, can be a unique username string or

        object containing user_id attribute

    :param action: action to log, should be on of predefined unique actions for

        easy translations

    :param repo: string name of repository or object containing repo_id,

        that action was made on

    :param ipaddr: optional IP address from what the action was made

    """

    # if we don't get explicit IP address try to get one from registered user

    # in tmpl context var

    if not ipaddr:

        ipaddr = getattr(get_current_authuser(), 'ip_addr', '')

    if getattr(user, 'user_id', None):

        user_obj = User.get(user.user_id)

    elif isinstance(user, basestring):

        user_obj = User.get_by_username(user)

    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 = u''

    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

    meta.Session().add(user_log)

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

             action, safe_unicode(repo), user_obj, ipaddr)

    if commit:

        meta.Session().commit()

def get_filesystem_repos(path):

    """

    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 = safe_str(path.rstrip(os.sep))

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

    def isdir(*n):

        return os.path.isdir(os.path.join(*n))

    for root, dirs, _files in os.walk(path):

        recurse_dirs = []

        for subdir in dirs:

            # skip removed repos

            if REMOVED_REPO_PAT.match(subdir):

                continue

            # skip .<something> dirs TODO: rly? then we should prevent creating them ...

            if subdir.startswith('.'):

                continue

            cur_path = os.path.join(root, subdir)

            if isdir(cur_path, '.git'):

                log.warning('ignoring non-bare Git repo: %s', cur_path)

                continue

            if (isdir(cur_path, '.hg') or

                isdir(cur_path, '.svn') or

                isdir(cur_path, 'objects') and (isdir(cur_path, 'refs') or

                                                os.path.isfile(os.path.join(cur_path, 'packed-refs')))):

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

                    log.warning('ignoring repo path without access: %s', cur_path)

                    continue

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

                    log.warning('repo path without write access: %s', cur_path)

                try:

                    scm_info = get_scm(cur_path)

                    assert cur_path.startswith(path)

                    repo_path = cur_path[len(path) + 1:]

                    yield repo_path, scm_info

                    continue # no recursion

                except VCSError:

                    # We should perhaps ignore such broken repos, but especially

                    # the bare git detection is unreliable so we dive into it

                    pass

            recurse_dirs.append(subdir)

        dirs[:] = recurse_dirs

def is_valid_repo_uri(repo_type, url, ui):

    """Check if the url seems like a valid remote repo location - raise an Exception if any problems"""

    if repo_type == 'hg':

        if url.startswith('http') or url.startswith('ssh'):

            # initially check if it's at least the proper URL

            # or does it pass basic auth

            MercurialRepository._check_url(url, ui)

        elif url.startswith('svn+http'):

            try:

                from hgsubversion.svnrepo import svnremoterepo

            except ImportError:

                raise HgsubversionImportError(_('Unable to activate hgsubversion support. '

                                                'The "hgsubversion" library is missing'))

            svnremoterepo(ui, url).svn.uuid

        elif url.startswith('git+http'):

            raise NotImplementedError()

        else:

            raise Exception('URI %s not allowed' % (url,))

    elif repo_type == 'git':

        if url.startswith('http') or url.startswith('git'):

            # initially check if it's at least the proper URL

            # or does it pass basic auth

            GitRepository._check_url(url)

        elif url.startswith('svn+http'):

            raise NotImplementedError()

        elif url.startswith('hg+http'):

            raise NotImplementedError()

        else:

            raise Exception('URI %s not allowed' % (url))

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

    """

    # TODO: paranoid security checks?

    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:

        return False

def is_valid_repo_group(repo_group_name, base_path, skip_path_check=False):

    """

    Returns True if given path is a repository group False otherwise

    :param repo_name:

    :param base_path:

    """

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

    # check if it's not a repo

    if is_valid_repo(repo_group_name, base_path):

        return False

    try:

        # we need to check bare git repos at higher level

        # since we might match branches/hooks/info/objects or possible

        # other things inside bare git repo

        get_scm(os.path.dirname(full_path))

        return False

    except VCSError:

        pass

    # check if it's a valid path

    if skip_path_check or os.path.isdir(full_path):

        return True

    return False

# propagated from mercurial documentation

ui_sections = ['alias', 'auth',

                'decode/encode', 'defaults',

                'diff', 'email',

                'extensions', 'format',

                'merge-patterns', 'merge-tools',

                'hooks', 'http_proxy',

                'smtp', 'patch',

                'paths', 'profiling',

                'server', 'trusted',

                'ui', 'web', ]

def make_ui(repo_path=None):

    """

    Create an Mercurial 'ui' object based on database Ui settings, possibly

    augmenting with content from a hgrc file.

    """

    baseui = mercurial.ui.ui()

    # clean the baseui object

    baseui._ocfg = mercurial.config.config()

    baseui._ucfg = mercurial.config.config()

    baseui._tcfg = mercurial.config.config()

    sa = meta.Session()

    for ui_ in sa.query(Ui).all():

        if ui_.ui_active:

            log.debug('config from db: [%s] %s=%r', ui_.ui_section,

                      ui_.ui_key, ui_.ui_value)

            baseui.setconfig(ascii_bytes(ui_.ui_section), ascii_bytes(ui_.ui_key),

                             b'' if ui_.ui_value is None else safe_bytes(ui_.ui_value))

    # force set push_ssl requirement to False, Kallithea handles that

    baseui.setconfig(b'web', b'push_ssl', False)

    baseui.setconfig(b'web', b'allow_push', b'*')

    # prevent interactive questions for ssh password / passphrase

    ssh = baseui.config(b'ui', b'ssh', default=b'ssh')

    baseui.setconfig(b'ui', b'ssh', b'%s -oBatchMode=yes -oIdentitiesOnly=yes' % ssh)

    # push / pull hooks

    baseui.setconfig(b'hooks', b'changegroup.kallithea_log_push_action', b'python:kallithea.lib.hooks.log_push_action')

    baseui.setconfig(b'hooks', b'outgoing.kallithea_log_pull_action', b'python:kallithea.lib.hooks.log_pull_action')

    if repo_path is not None:

        hgrc_path = os.path.join(repo_path, '.hg', 'hgrc')

        if os.path.isfile(hgrc_path):

            log.debug('reading hgrc from %s', hgrc_path)

            cfg = mercurial.config.config()

            cfg.read(safe_bytes(hgrc_path))

            for section in ui_sections:

                for k, v in cfg.items(section):

                    log.debug('config from file: [%s] %s=%s', section, k, v)

                    baseui.setconfig(ascii_bytes(section), ascii_bytes(k), safe_bytes(v))

        else:

            log.debug('hgrc file is not present at %s, skipping...', hgrc_path)

    return baseui

def set_app_settings(config):

    """

    Updates app config with new settings from database

    :param config:

    """

    hgsettings = Setting.get_app_settings()

    for k, v in hgsettings.items():

        config[k] = v

def set_vcs_config(config):

    """

    Patch VCS config with some Kallithea specific stuff