"""

import os

import pprint

import random

import sys

from kallithea.lib import ext_json

from kallithea.lib.utils2 import ascii_bytes

CONFIG_NAME = '.config/kallithea'

        }

    if not method:

        raise Exception('please specify method name !')

    apihost = apihost.rstrip('/')

    id_ = random.randrange(1, 9999)

                      data=ascii_bytes(ext_json.dumps(_build_data(id_))),

                      headers={'content-type': 'text/plain'})

    raw_json = ret.read()

    json_data = ext_json.loads(raw_json)

    id_ret = json_data['id']

    if id_ret == id_:

        return json_data

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

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

"""

from __future__ import print_function

import uuid

from configparser import ConfigParser

import ldap

from kallithea.lib import ext_json

        """

        uid = str(uuid.uuid1())

        data = self.get_api_data(uid, method, args)

        data = ascii_bytes(ext_json.dumps(data))

        headers = {'content-type': 'text/plain'}

        response = ext_json.load(response)

        if uid != response["id"]:

            raise InvalidResponseIDError("UUID does not match.")

        if response["error"] is not None:

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

"""

import base64

import logging

from kallithea.lib import auth_modules, ext_json

from kallithea.lib.compat import hybrid_property

from kallithea.lib.utils2 import ascii_bytes, ascii_str, safe_bytes

    def set_credentials(self, user, passwd):

        self.user = user

        self.passwd = passwd

        self._make_opener()

    def _make_opener(self):

        mgr.add_password(None, self._uri, self.user, self.passwd)

    def _request(self, url, body=None, headers=None,

                 method=None, noformat=False,

                 empty_response_ok=False):

        _headers = {"Content-type": "application/json",

                    "Accept": "application/json"}

        if self.user and self.passwd:

            authstring = ascii_str(base64.b64encode(safe_bytes("%s:%s" % (self.user, self.passwd))))

            _headers["Authorization"] = "Basic %s" % authstring

        if headers:

            _headers.update(headers)

        log.debug("Sent to crowd at %s:\nHeaders: %s\nBody:\n%s", url, _headers, body)

        if method:

            req.get_method = lambda: method

        global msg

        msg = ""

        try:

        return rval

    def user_auth(self, username, password):

        """Authenticate a user against crowd. Returns brief information about

        the user."""

        url = ("%s/rest/usermanagement/%s/authentication?username=%s"

        body = ascii_bytes(ext_json.dumps({"value": password}))

        return self._request(url, body)

    def user_groups(self, username):

        """Retrieve a list of groups to which this user belongs."""

        url = ("%s/rest/usermanagement/%s/user/group/nested?username=%s"

        return self._request(url)

class KallitheaAuthPlugin(auth_modules.KallitheaExternalAuthPlugin):

    def __init__(self):

        self._protocol_values = ["http", "https"]

"""

import logging

import os

import mercurial.hgweb

from kallithea.lib.base import BaseVCSController, get_path_info

from kallithea.lib.utils import make_ui

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

                        hgarg = get_header_hgarg(environ)

                        if not hgarg.startswith('cmds='):

                            action = 'push' # paranoid and safe

                            break

                        action = 'pull'

                        for cmd_arg in hgarg[5:].split(';'):

                            op = cmd_mapping.get(cmd, 'push')

                            if op != 'pull':

                                assert op == 'push'

                                action = 'push'

                                break

                    else:

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

import json

class RecaptchaResponse(object):

    def __init__(self, is_valid, error_code=None):

        self.is_valid = is_valid

        self.error_code = error_code

    def encode_if_necessary(s):

        if isinstance(s, unicode):

            return s.encode('utf-8')

        return s

        'secret': encode_if_necessary(private_key),

        'remoteip': encode_if_necessary(remoteip),

        'response': encode_if_necessary(g_recaptcha_response),

    }).encode('ascii')

        url="https://www.google.com/recaptcha/api/siteverify",

        data=params,

        headers={

            "Content-type": "application/x-www-form-urlencoded",

            "User-agent": "reCAPTCHA Python"

        }

    )

    return_values = json.loads(httpresp.read())

    httpresp.close()

    if not (isinstance(return_values, dict)):

        return RecaptchaResponse(is_valid=False, error_code='incorrect-captcha-sol')

    elif (("success" in return_values) and ((return_values["success"] is True) or (return_values["success"] == "true"))):

import datetime

import json

import os

import pwd

import re

import time

import urlobject

from tg.i18n import ugettext as _

from tg.i18n import ungettext

from webhelpers2.text import collapse, remove_formatting, strip_tags

    return ''.join(uri)

def get_clone_url(clone_uri_tmpl, prefix_url, repo_name, repo_id, username=None):

    parsed_url = urlobject.URLObject(prefix_url)

    try:

        system_user = pwd.getpwuid(os.getuid()).pw_name

    except Exception: # TODO: support all systems - especially Windows

        system_user = 'kallithea' # hardcoded default value ...

    args = {

        'scheme': parsed_url.scheme,

        'netloc': parsed_url.netloc + prefix,  # like "hostname:port/prefix" (with optional ":port" and "/prefix")

        'prefix': prefix, # undocumented, empty or starting with /

        'repo': repo_name,

        'repoid': str(repo_id),

        'system_user': safe_unicode(system_user),

        'hostname': parsed_url.hostname,

import errno

import logging

import os

import re

import time

from collections import OrderedDict

import mercurial.url  # import httpbasicauthhandler, httpdigestauthhandler

import mercurial.util  # import url as hg_url

from dulwich.config import ConfigFile

from dulwich.objects import Tag

        url_obj.passwd = b'*****'

        cleaned_uri = str(url_obj)

        if authinfo:

            # create a password manager

            passmgr.add_password(*authinfo)

            handlers.extend((mercurial.url.httpbasicauthhandler(passmgr),

                             mercurial.url.httpdigestauthhandler(passmgr)))

        o.addheaders = [('User-Agent', 'git/1.7.8.0')]  # fake some git

            "%s?%s" % (

                test_uri,

            ))

        try:

            resp = o.open(req)

            if resp.code != 200:

                raise Exception('Return Code is not 200')

        except Exception as e:

            # means it cannot be cloned

        # now detect if it's proper git repo

        gitdata = resp.read()

        if 'service=git-upload-pack' not in gitdata:

                "url [%s] does not look like an git" % cleaned_uri)

        return True

    def _get_repo(self, create, src_url=None, update_after_clone=False,

                  bare=False):

"""

import datetime

import logging

import os

import time

from collections import OrderedDict

import mercurial.commands

import mercurial.error

import mercurial.exchange

import mercurial.hg

        test_uri, authinfo = url_obj.authinfo()

        url_obj.passwd = b'*****'

        cleaned_uri = str(url_obj)

        if authinfo:

            # create a password manager

            passmgr.add_password(*authinfo)

            handlers.extend((mercurial.url.httpbasicauthhandler(passmgr),

                             mercurial.url.httpdigestauthhandler(passmgr)))

        o.addheaders = [('Content-Type', 'application/mercurial-0.1'),

                        ('Accept', 'application/mercurial-0.1')]

            "%s?%s" % (

                test_uri,

                    'cmd': 'between',

                    'pairs': "%s-%s" % ('0' * 40, '0' * 40),

                })

            ))

        try:

            resp = o.open(req)

            if resp.code != 200:

                raise Exception('Return Code is not 200')

        except Exception as e:

            # means it cannot be cloned

        if not url_prefix: # skip svn+http://... (and git+... too)

            # now check if it's a proper hg repo

            try:

                mercurial.httppeer.instance(repoui or mercurial.ui.ui(), url, False).lookup(b'tip')

            except Exception as e:

                    "url [%s] does not look like an hg repo org_exc: %s"

                    % (cleaned_uri, e))

        return True

    def _get_repo(self, create, src_url=None, update_after_clone=False):

        Returns normalized url. If schema is not given, would fall

        to filesystem

        (``file:///``) schema.

        """

        url = safe_str(url)

        if url != 'default' and '://' not in url:

        return url

    def get_changeset(self, revision=None):

        """

        Returns ``MercurialChangeset`` object representing repository's

        changeset at the given ``revision``.

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

import os

import mock

import pytest

from kallithea.lib import vcs

from kallithea.lib.utils2 import safe_str, safe_unicode

                                                _session_csrf_secret_token=self.session_csrf_secret_token()))

        ## run the check page that triggers the flash message

        response = self.app.get(base.url('repo_check_home', repo_name=repo_name))

        assert response.json == {u'result': True}

        self.checkSessionFlash(response,

                               u'Created repository <a href="/%s">%s</a>'

        # test if the repo was created in the database

        new_repo = Session().query(Repository) \

            .filter(Repository.repo_name == repo_name_unicode).one()

        assert new_repo.repo_name == repo_name_unicode

        assert new_repo.description == description_unicode

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

from kallithea.lib.utils2 import safe_str, safe_unicode

from kallithea.model.db import Repository, User

from kallithea.model.meta import Session

from kallithea.model.repo import RepoModel

from kallithea.model.user import UserModel

            '_session_csrf_secret_token': self.session_csrf_secret_token()}

        self.app.post(base.url(controller='forks', action='fork_create',

                          repo_name=repo_name), creation_args)

        response = self.app.get(base.url(controller='forks', action='forks',

                                    repo_name=repo_name))

        response.mustcontain(

        )

        fork_repo = Repository.get_by_repo_name(safe_unicode(fork_name))

        assert fork_repo

        # fork the fork

        fork_name_2 = safe_str(self.REPO_FORK + u'-blåbærgrød')

            '_session_csrf_secret_token': self.session_csrf_secret_token()}

        self.app.post(base.url(controller='forks', action='fork_create',

                          repo_name=fork_name), creation_args)

        response = self.app.get(base.url(controller='forks', action='forks',

                                    repo_name=fork_name))

        response.mustcontain(

        )

        # remove these forks

        response = self.app.post(base.url('delete_repo', repo_name=fork_name_2),

            params={'_session_csrf_secret_token': self.session_csrf_secret_token()})

        response = self.app.post(base.url('delete_repo', repo_name=fork_name),

import json

import os

import re

import tempfile

import time

from subprocess import PIPE, Popen

from tempfile import _RandomNameSequence

import pytest

from kallithea import CONFIG

            "api_key": usr.api_key,

            "method": 'create_repo',

            "args": dict(repo_name=repo_name,

                         owner=base.TEST_USER_ADMIN_LOGIN,

                         repo_type=vt.repo_type),

        }

            'http://%s:%s/_admin/api' % webserver.server_address,

            data=ascii_bytes(json.dumps(params)),

            headers={'content-type': 'application/json'})

        result = json.loads(response.read())

        # Expect something like:

        # {u'result': {u'msg': u'Created new repository `new_XXX`', u'task': None, u'success': True}, u'id': 7, u'error': None}

        assert result[u'result'][u'success']

        # Create local clone of the empty server repo

import http.cookiejar

import os

import sys

import tempfile

import time

from os.path import dirname

from kallithea.lib import vcs

from kallithea.lib.compat import OrderedSet

from kallithea.lib.vcs.exceptions import RepositoryError

    'CPython',

    'kallithea',

]

cj = http.cookiejar.FileCookieJar(os.path.join(tempfile.gettempdir(), 'rc_test_cookie.txt'))

o.addheaders = [

    ('User-agent', 'kallithea-crawler'),

    ('Accept-Language', 'en - us, en;q = 0.5')

]

def _get_repo(proj):

    if isinstance(proj, str):

        repo = vcs.get_repo(os.path.join(PROJECT_PATH, proj))

        proj = proj

    total_time = 0

    for i in range(1, pages):

        page = '/'.join((proj, 'changelog',))

        s = time.time()

        f = o.open(full_uri)

        assert f.url == full_uri, 'URL:%s does not match %s' % (f.url, full_uri)

        size = len(f.read())

import datetime

import os

import sys

import mock

import pytest

from kallithea.lib.vcs.backends.git import GitChangeset, GitRepository

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

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

        with pytest.raises(RepositoryError):

            GitRepository(wrong_repo_path)

    def test_git_cmd_injection(self):

        repo_inject_path = TEST_GIT_REPO + '; echo "Cake";'

            # Should fail because URL will contain the parts after ; too

            GitRepository(get_new_dir('injection-repo'), src_url=repo_inject_path, update_after_clone=True, create=True)

        with pytest.raises(RepositoryError):

            # Should fail on direct clone call, which as of this writing does not happen outside of class

            clone_fail_repo = GitRepository(get_new_dir('injection-repo'), create=True)