kallithea.lib.hooks

Hooks run by Kallithea

import time

from kallithea.lib import webutils

from kallithea.lib.exceptions import UserCreationError

from kallithea.lib.utils import action_logger, make_ui

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(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 = webutils.format_byte_size(size_scm)

    size_root_f = webutils.format_byte_size(size_root)

    size_total_f = webutils.format_byte_size(size_root + size_scm)

    return size_scm_f, size_root_f, size_total_f

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

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

def log_pull_action(*args, **kwargs):

    """Logs user last pull action

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

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

    """

    ex = get_hook_environment()

    action = 'pull'

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

    # extension hook call

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

    if callable(callback):

        kw = {}

        kw.update(ex)

        callback(**kw)

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

    Called as Mercurial hook changegroup.kallithea_log_push_action .

    process_pushed_raw_ids(revs)

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.kallithea_log_push_action 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

    callback = getattr(kallithea.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'

    """

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

    if callable(callback):

        kw = {}

        kw.update(repository_dict)

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

        kw.update(kwargs)

        callback(**kw)

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

    # pre create hooks

    callback = getattr(kallithea.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',

    """

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

    if callable(callback):

        callback(created_by=created_by, **user_dict)

def log_create_pullrequest(pullrequest_dict, created_by, **kwargs):

    """

    Post create pull request hook.

    :param pullrequest_dict: dict dump of pull request object

    """

    callback = getattr(kallithea.EXTENSIONS, 'CREATE_PULLREQUEST_HOOK', None)

    if callable(callback):

        return callback(created_by=created_by, **pullrequest_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'

    """

    callback = getattr(kallithea.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)

        callback(**kw)

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',

    """

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

    if callable(callback):

        callback(deleted_by=deleted_by, **user_dict)

def _hook_environment(repo_path):

    import kallithea.config.application

    baseui = make_ui()

    return baseui, repo

def handle_git_pre_receive(repo_path, git_stdin_lines):

def handle_git_post_receive(repo_path, git_stdin_lines):

        baseui, repo = _hook_environment(repo_path)

    process_pushed_raw_ids(git_revs)

                # Run hooks like Mercurial outgoing.kallithea_log_pull_action does

Hooks run by Kallithea

import os

import sys

import mercurial.scmutil

import paste.deploy

from kallithea.lib import webutils

from kallithea.lib.utils import action_logger, make_ui

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

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

from kallithea.model import db

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(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 = webutils.format_byte_size(size_scm)

    size_root_f = webutils.format_byte_size(size_root)

    size_total_f = webutils.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.

    Called as Mercurial hook changegroup.repo_size after push.

    """

    size_hg_f, size_root_f, size_total_f = _get_scm_size('.hg', safe_str(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(*args, **kwargs):

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

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

    """

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

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

    caches! The function should perhaps be renamed.

    Called as Mercurial hook changegroup.kallithea_log_push_action .

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

    """

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

    process_pushed_raw_ids(revs)

    Called from Mercurial changegroup.kallithea_log_push_action calling hook log_push_action,

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 kallithea.config.application

    extras = get_hook_environment()

    path_to_ini_file = extras['config']

    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.application.make_app(config.global_conf, **config.local_conf)

    # fix if it's not a bare repo

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

        repo_path = repo_path[:-5]

    repo = db.Repository.get_by_full_path(repo_path)

    if not repo:

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

    baseui = make_ui()

    return baseui, repo

def handle_git_pre_receive(repo_path, git_stdin_lines):

    """Called from Git pre-receive hook.

    The returned value is used as hook exit code and must be 0.

    """

    # Currently unused. TODO: remove?

    return 0

def handle_git_post_receive(repo_path, git_stdin_lines):

    """Called from Git post-receive hook.

    The returned value is used as hook exit code and must be 0.

    """

    try:

        baseui, repo = _hook_environment(repo_path)

    except HookEnvironmentError as e:

        sys.stderr.write("Skipping Kallithea Git post-receive hook %r.\nGit was apparently not invoked by Kallithea: %s\n" % (sys.argv[0], e))

        return 0

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

    Return value 1 will make the hook fail and reject the push.

    """

    ex = get_hook_environment()

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

    return 1

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

        baseui.setconfig(b'hooks', ascii_bytes(db.Ui.HOOK_REPO_SIZE), b'python:kallithea.lib.hooks.repo_size')

            baseui.setconfig(b'hooks', b'pretxnopen._ssh_reject', b'python:kallithea.lib.hooks.rejectpush')

            baseui.setconfig(b'hooks', b'prepushkey._ssh_reject', b'python:kallithea.lib.hooks.rejectpush')

import kallithea.lib.hooks

    sys.exit(kallithea.lib.hooks.handle_git_post_receive(repo_path, git_stdin_lines))

import kallithea.lib.hooks

    sys.exit(kallithea.lib.hooks.handle_git_pre_receive(repo_path, git_stdin_lines))