# -*- 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 logging

import os

import shlex

import sys

import click

import kallithea

import kallithea.bin.kallithea_cli_base as cli_base

from kallithea.lib.utils2 import str2bool

from kallithea.lib.vcs.backends.git.ssh import GitSshHandler

from kallithea.lib.vcs.backends.hg.ssh import MercurialSshHandler

from kallithea.model.ssh_key import SshKeyModel

log = logging.getLogger(__name__)

@cli_base.register_command(config_file_initialize_app=True, hidden=True)

@click.argument('user-id', type=click.INT, required=True)

@click.argument('key-id', type=click.INT, required=True)

def ssh_serve(user_id, key_id):

    """Serve SSH repository protocol access.

    The trusted command that is invoked from .ssh/authorized_keys to serve SSH

    protocol access. The access will be granted as the specified user ID, and

# 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.controllers.admin.defaults

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

default settings 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 27, 2010

:author: marcink

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

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

"""

import logging

import traceback

import formencode

from formencode import htmlfill

from tg import request

from tg.i18n import ugettext as _

from webob.exc import HTTPFound

from kallithea.config.routing import url

from kallithea.lib import helpers as h

from kallithea.lib.auth import HasPermissionAnyDecorator, LoginRequired

from kallithea.lib.base import BaseController, render

from kallithea.model.db import Setting

from kallithea.model.forms import DefaultsForm

from kallithea.model.meta import Session

log = logging.getLogger(__name__)

class DefaultsController(BaseController):

    @LoginRequired()

    @HasPermissionAnyDecorator('hg.admin')

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

        super(DefaultsController, self)._before(*args, **kwargs)

    def index(self, format='html'):

        defaults = Setting.get_default_repo_settings()

kallithea.controllers.admin.repo_groups

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

Repository groups 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: Mar 23, 2010

:author: marcink

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

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

"""

import logging

import traceback

import formencode

from formencode import htmlfill

from tg import app_globals, request

from tg import tmpl_context as c

from tg.i18n import ugettext as _

from tg.i18n import ungettext

from webob.exc import HTTPForbidden, HTTPFound, HTTPInternalServerError, HTTPNotFound

from kallithea.config.routing import url

from kallithea.lib import helpers as h

from kallithea.lib.auth import HasPermissionAny, HasRepoGroupPermissionLevel, HasRepoGroupPermissionLevelDecorator, LoginRequired

from kallithea.lib.base import BaseController, render

from kallithea.lib.utils2 import safe_int, safe_unicode

from kallithea.model.db import RepoGroup, Repository

from kallithea.model.forms import RepoGroupForm, RepoGroupPermsForm

from kallithea.model.meta import Session

from kallithea.model.repo import RepoModel

from kallithea.model.repo_group import RepoGroupModel

from kallithea.model.scm import AvailableRepoGroupChoices, RepoGroupList

log = logging.getLogger(__name__)

class RepoGroupsController(BaseController):

    @LoginRequired(allow_default_user=True)

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

        super(RepoGroupsController, self)._before(*args, **kwargs)

    def __load_defaults(self, extras=(), exclude=()):

        """extras is used for keeping current parent ignoring permissions

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

User Groups crud controller

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 logging

import traceback

import formencode

from formencode import htmlfill

from sqlalchemy.orm import joinedload

from sqlalchemy.sql.expression import func

from tg import app_globals, request

from tg import tmpl_context as c

from tg.i18n import ugettext as _

from webob.exc import HTTPFound, HTTPInternalServerError

from kallithea.config.routing import url

from kallithea.lib import helpers as h

from kallithea.lib.auth import HasPermissionAnyDecorator, HasUserGroupPermissionLevelDecorator, LoginRequired

from kallithea.lib.base import BaseController, render

from kallithea.lib.exceptions import RepoGroupAssignmentError, UserGroupsAssignedException

from kallithea.lib.utils import action_logger

from kallithea.lib.utils2 import safe_int, safe_unicode

from kallithea.model.db import User, UserGroup, UserGroupRepoGroupToPerm, UserGroupRepoToPerm, UserGroupToPerm

from kallithea.model.forms import CustomDefaultPermissionsForm, UserGroupForm, UserGroupPermsForm

from kallithea.model.meta import Session

from kallithea.model.scm import UserGroupList

from kallithea.model.user_group import UserGroupModel

log = logging.getLogger(__name__)

class UserGroupsController(BaseController):

    """REST Controller styled on the Atom Publishing Protocol"""

    @LoginRequired(allow_default_user=True)

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

        super(UserGroupsController, self)._before(*args, **kwargs)

kallithea.controllers.admin.users

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

Users crud controller

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 4, 2010

:author: marcink

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

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

"""

import logging

import traceback

import formencode

from formencode import htmlfill

from sqlalchemy.sql.expression import func

from tg import app_globals, request

from tg import tmpl_context as c

from tg.i18n import ugettext as _

from webob.exc import HTTPFound, HTTPNotFound

from kallithea.config.routing import url

from kallithea.lib import auth_modules

from kallithea.lib import helpers as h

from kallithea.lib.auth import AuthUser, HasPermissionAnyDecorator, LoginRequired

from kallithea.lib.base import BaseController, IfSshEnabled, render

from kallithea.lib.exceptions import DefaultUserException, UserCreationError, UserOwnsReposException

from kallithea.lib.utils import action_logger

from kallithea.lib.utils2 import datetime_to_time, generate_api_key, safe_int

from kallithea.model.api_key import ApiKeyModel

from kallithea.model.db import User, UserEmailMap, UserIpMap, UserToPerm

from kallithea.model.forms import CustomDefaultPermissionsForm, UserForm

from kallithea.model.meta import Session

from kallithea.model.ssh_key import SshKeyModel, SshKeyModelException

from kallithea.model.user import UserModel

log = logging.getLogger(__name__)

class UsersController(BaseController):

    """REST Controller styled on the Atom Publishing Protocol"""

    @LoginRequired()

    @HasPermissionAnyDecorator('hg.admin')

changelog 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 21, 2010

:author: marcink

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

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

"""

import logging

import traceback

from tg import request, session

from tg import tmpl_context as c

from tg.i18n import ugettext as _

from webob.exc import HTTPBadRequest, HTTPFound, HTTPNotFound

import kallithea.lib.helpers as h

from kallithea.config.routing import url

from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired

from kallithea.lib.base import BaseRepoController, render

from kallithea.lib.graphmod import graph_data

from kallithea.lib.page import Page

from kallithea.lib.vcs.exceptions import ChangesetDoesNotExistError, ChangesetError, EmptyRepositoryError, NodeDoesNotExistError, RepositoryError

log = logging.getLogger(__name__)

class ChangelogController(BaseRepoController):

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

        super(ChangelogController, self)._before(*args, **kwargs)

        c.affected_files_cut_off = 60

    @staticmethod

    def __get_cs(rev, repo):

        """

        Safe way to get changeset. If error occur fail with error message.

        :param rev: revision to fetch

        :param repo: repo instance

        """

        try:

            return c.db_repo_scm_instance.get_changeset(rev)

        except EmptyRepositoryError as e:

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

"""

import calendar

import itertools

import logging

import traceback

from datetime import date, timedelta

from time import mktime

from beaker.cache import cache_region

from tg import request

from tg import tmpl_context as c

from tg.i18n import ugettext as _

from webob.exc import HTTPBadRequest

import kallithea.lib.helpers as h

from kallithea.config.conf import ALL_EXTS, ALL_READMES, LANGUAGES_EXTENSIONS_MAP

from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired

from kallithea.lib.base import BaseRepoController, jsonify, render

from kallithea.lib.celerylib.tasks import get_commits_stats

from kallithea.lib.compat import json

from kallithea.lib.markup_renderer import MarkupRenderer

from kallithea.lib.page import Page

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

from kallithea.lib.vcs.exceptions import ChangesetError, EmptyRepositoryError, NodeDoesNotExistError

from kallithea.lib.vcs.nodes import FileNode

from kallithea.model.db import Statistics

log = logging.getLogger(__name__)

README_FILES = [''.join([x[0][0], x[1][0]]) for x in

                    sorted(list(itertools.product(ALL_READMES, ALL_EXTS)),

                           key=lambda y:y[0][1] + y[1][1])]

class SummaryController(BaseRepoController):

    def __get_readme_data(self, db_repo):

        repo_name = db_repo.repo_name

        log.debug('Looking for README file')

        @cache_region('long_term', '_get_readme_from_cache')

        def _get_readme_from_cache(*_cache_keys):  # parameters are not really used - only as caching key

            readme_data = None

            readme_file = None

            try:

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

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

Kallithea authentication plugin for built in internal auth

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

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

:created_on: Created on Nov 17, 2012

:author: marcink

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

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

"""

import logging

from kallithea.lib import auth_modules

from kallithea.lib.compat import formatted_json, hybrid_property

log = logging.getLogger(__name__)

class KallitheaAuthPlugin(auth_modules.KallitheaAuthPluginBase):

    def __init__(self):

        pass

    @hybrid_property

    def name(self):

        # Also found as kallithea.lib.model.db.User.DEFAULT_AUTH_TYPE

        return 'internal'

    def settings(self):

        return []

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

        """

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

        know that user exists in database.

        """

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

                                                        accepts_empty=False)

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

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

Python backward compatibility functions and common libs

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: Oct 7, 2011

:author: marcink

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

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

"""

import functools

import os

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

# Hybrid property/method

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

from sqlalchemy.ext.hybrid import hybrid_property

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

# OrderedSet

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

from sqlalchemy.util import OrderedSet

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

# json

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

from kallithea.lib.ext_json import json

# alias for formatted json

formatted_json = functools.partial(json.dumps, indent=4, sort_keys=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/>.

import logging

import os

from kallithea.lib.hooks import log_pull_action

from kallithea.lib.utils import make_ui

from kallithea.lib.vcs.backends.ssh import BaseSshHandler

log = logging.getLogger(__name__)

class GitSshHandler(BaseSshHandler):

    vcs_type = 'git'

    @classmethod

    def make(cls, ssh_command_parts):

        r"""

        >>> import shlex

        >>> GitSshHandler.make(shlex.split("git-upload-pack '/foo bar'")).repo_name

        u'foo bar'

        >>> GitSshHandler.make(shlex.split("git-upload-pack '/foo bar'")).verb

        'git-upload-pack'

        >>> GitSshHandler.make(shlex.split(" git-upload-pack /blåbærgrød ")).repo_name # might not be necessary to support no quoting ... but we can

        u'bl\xe5b\xe6rgr\xf8d'

        >>> GitSshHandler.make(shlex.split('''git-upload-pack "/foo'bar"''')).repo_name

        u"foo'bar"

        >>> GitSshHandler.make(shlex.split("git-receive-pack '/foo'")).repo_name

        u'foo'

        >>> GitSshHandler.make(shlex.split("git-receive-pack '/foo'")).verb

# -*- 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 logging

from mercurial import hg

from mercurial.wireprotoserver import sshserver

from kallithea.lib.utils import make_ui

from kallithea.lib.vcs.backends.ssh import BaseSshHandler

log = logging.getLogger(__name__)

class MercurialSshHandler(BaseSshHandler):

    vcs_type = 'hg'

    @classmethod

    def make(cls, ssh_command_parts):

        r"""

        >>> import shlex

        >>> MercurialSshHandler.make(shlex.split('hg -R "foo bar" serve --stdio')).repo_name

        u'foo bar'

        >>> MercurialSshHandler.make(shlex.split(' hg -R blåbærgrød serve --stdio ')).repo_name

        u'bl\xe5b\xe6rgr\xf8d'

        >>> MercurialSshHandler.make(shlex.split('''hg -R 'foo"bar' serve --stdio''')).repo_name

        u'foo"bar'

        >>> MercurialSshHandler.make(shlex.split('/bin/hg -R "foo" serve --stdio'))

        >>> MercurialSshHandler.make(shlex.split('''hg -R "foo"bar" serve --stdio''')) # ssh-serve will report: Error parsing SSH command "...": invalid syntax

        Traceback (most recent call last):

        ValueError: No closing quotation

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

"""

vcs.backends.ssh

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

SSH backend for all available SCMs

"""

import datetime

import logging

import sys

from kallithea.lib.auth import AuthUser, HasPermissionAnyMiddleware

from kallithea.model.db import Repository, User, UserSshKeys

from kallithea.model.meta import Session

log = logging.getLogger(__name__)

class BaseSshHandler(object):

    # Protocol for setting properties:

    # Set by sub class:

    #   vcs_type: 'hg' or 'git'

    # Set by make() / __init__():

    #   repo_name: requested repo name - only validated by serve()

    # Set by serve() - must not be accessed before:

    #   db_repo: repository db object

    #   authuser: user that has been authenticated - like request.authuser ... which isn't used here

    #   allow_push: false for read-only access to the repo

    # Set defaults, in case .exit should be called early

    vcs_type = None

    repo_name = None

    @staticmethod

    def make(ssh_command):

            .filter(Repository.repo_name == repo_name)

        if cache:

            repo = repo.options(FromCache("sql_cache_short",

                                          "get_repo_%s" % repo_name))

        return repo.scalar()

    def get_all_user_repos(self, user):

        """

        Gets all repositories that user have at least read access

        :param user:

        """

        from kallithea.lib.auth import AuthUser

        user = User.guess_instance(user)

        repos = AuthUser(dbuser=user).permissions['repositories']

        access_check = lambda r: r[1] in ['repository.read',

                                          'repository.write',

                                          'repository.admin']

        repos = [x[0] for x in filter(access_check, repos.items())]

        return Repository.query().filter(Repository.repo_name.in_(repos))

    @classmethod

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

        from tg import tmpl_context as c, request, app_globals

        from tg.i18n import ugettext as _

        _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 tmpl_context as c

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

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

                           short_name=short_name)

import os

import mock

import pytest

from kallithea.lib.vcs.backends.hg import MercurialChangeset, MercurialRepository

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

from kallithea.lib.vcs.nodes import NodeKind, NodeState

from kallithea.tests.vcs.conf import TEST_HG_REPO, TEST_HG_REPO_CLONE, TEST_HG_REPO_PULL, TESTS_TMP_PATH

class TestMercurialRepository(object):

    def __check_for_existing_repo(self):

        if os.path.exists(TEST_HG_REPO_CLONE):

            pytest.fail('Cannot test mercurial clone repo as location %s already '

                      'exists. You should manually remove it first.'

                      % TEST_HG_REPO_CLONE)

    def setup_method(self):

        self.repo = MercurialRepository(safe_str(TEST_HG_REPO))

    def test_wrong_repo_path(self):

        wrong_repo_path = os.path.join(TESTS_TMP_PATH, 'errorrepo')

        with pytest.raises(RepositoryError):

            MercurialRepository(wrong_repo_path)

    def test_unicode_path_repo(self):

        with pytest.raises(VCSError):

            MercurialRepository(u'iShouldFail')

    def test_repo_clone(self):

# encoding: utf-8

"""

Tests so called "in memory changesets" commit API of vcs.

"""

import datetime

import pytest

from kallithea.lib.vcs.exceptions import (

    EmptyRepositoryError, NodeAlreadyAddedError, NodeAlreadyChangedError, NodeAlreadyExistsError, NodeAlreadyRemovedError, NodeDoesNotExistError, NodeNotChangedError)

from kallithea.lib.vcs.nodes import DirNode, FileNode

from kallithea.lib.vcs.utils import safe_unicode

from kallithea.tests.vcs.base import _BackendTestMixin

class InMemoryChangesetTestMixin(_BackendTestMixin):

    @classmethod

    def _get_commits(cls):

        # Note: this is slightly different than the regular _get_commits methods

        # as we don't actually return any commits. The creation of commits is

        # handled in the tests themselves.

        cls.nodes = [

            FileNode('foobar', content='Foo & bar'),

            FileNode('foobar2', content='Foo & bar, doubled!'),

            FileNode('foo bar with spaces', content=''),

            FileNode('foo/bar/baz', content='Inside'),

            FileNode('foo/bar/file.bin', content='\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00;\x00\x03\x00\xfe\xff\t\x00\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x1a\x00\x00\x00\x00\x00\x00\x00\x00\x10\x00\x00\x18\x00\x00\x00\x01\x00\x00\x00\xfe\xff\xff\xff\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff'),

        ]

        commits = []

        return commits

import os

import shutil

import pytest

from kallithea.lib.vcs import VCSError, get_backend, get_repo

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

from kallithea.tests.vcs.conf import TEST_GIT_REPO, TEST_HG_REPO, TESTS_TMP_PATH

class TestVCS(object):

    """

    Tests for main module's methods.

    """

    def test_get_backend(self):

        hg = get_backend('hg')

        assert hg == MercurialRepository

    def test_alias_detect_hg(self):

        alias = 'hg'

        path = TEST_HG_REPO

        backend = get_backend(alias)

        repo = backend(safe_str(path))

        assert 'hg' == repo.alias

    def test_alias_detect_git(self):

        alias = 'git'

        path = TEST_GIT_REPO

        backend = get_backend(alias)

        repo = backend(safe_str(path))