import pprint

import random

import sys

from kallithea.lib import ext_json

from kallithea.lib.utils2 import ascii_bytes

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

from __future__ import print_function

import uuid

from configparser import ConfigParser

        data = ascii_bytes(ext_json.dumps(data))

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

        response = ext_json.load(response)

        if uid != response["id"]:

import base64

import logging

from kallithea.lib import auth_modules, ext_json

from kallithea.lib.compat import hybrid_property

        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,

        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

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

import logging

import os

import mercurial.hgweb

                            break

                        action = 'pull'

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

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

                            if op != 'pull':

                                assert op == 'push'

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

import json

class RecaptchaResponse(object):

            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={

        }

    )

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

    httpresp.close()

import pwd

import re

import time

import urlobject

from tg.i18n import ugettext as _

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,

import os

import re

import time

from collections import OrderedDict

import mercurial.url  # import httpbasicauthhandler, httpdigestauthhandler

        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:

                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

import logging

import os

import time

from collections import OrderedDict

import mercurial.commands

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

                })

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

        """

        url = safe_str(url)

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

        return url

    def get_changeset(self, revision=None):

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

import os

import mock

import pytest

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

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

from kallithea.lib.utils2 import safe_str, safe_unicode

from kallithea.model.db import Repository, User

        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

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

                                    repo_name=fork_name))

        response.mustcontain(

        )

        # remove these forks

import re

import tempfile

import time

from subprocess import PIPE, Popen

from tempfile import _RandomNameSequence

                         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}

import sys

import tempfile

import time

from os.path import dirname

from kallithea.lib import vcs

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

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

        s = time.time()

        f = o.open(full_uri)

import datetime

import os

import sys

import mock

import pytest

    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)