def generate_api_key():

    """

    Generates a random (presumably unique) API key.

    This value is used in URLs and "Bearer" HTTP Authorization headers,

    which in practice means it should only contain URL-safe characters

    (RFC 3986):

        unreserved = ALPHA / DIGIT / "-" / "." / "_" / "~"

    """

    # Hexadecimal certainly qualifies as URL-safe.

    return ascii_str(binascii.hexlify(os.urandom(20)))

def safe_int(val, default=None):

    """

    Returns int() of val if val is not convertable to int use default

    instead

    :param val:

    :param default:

    """

    try:

        val = int(val)

    except (ValueError, TypeError):

        val = default

    return val

def remove_suffix(s, suffix):

    if s.endswith(suffix):

        s = s[:-1 * len(suffix)]

    return s

def remove_prefix(s, prefix):

    if s.startswith(prefix):

        s = s[len(prefix):]

    return s

def uri_filter(uri):

    """

    Removes user:password from given url string

    :param uri:

    :rtype: str

    :returns: filtered list of strings

    """

    if not uri:

        return []

    proto = ''

    for pat in ('https://', 'http://', 'git://'):

        if uri.startswith(pat):

            uri = uri[len(pat):]

            proto = pat

            break

    # remove passwords and username

    uri = uri[uri.find('@') + 1:]

    # get the port

    cred_pos = uri.find(':')

    if cred_pos == -1:

        host, port = uri, None

    else:

        host, port = uri[:cred_pos], uri[cred_pos + 1:]

    return [_f for _f in [proto, host, port] if _f]

def credentials_filter(uri):

    """

    Returns a url with removed credentials

    :param uri:

    """

    uri = uri_filter(uri)

    # check if we have port

    if len(uri) > 2 and uri[2]:

        uri[2] = ':' + uri[2]

    return ''.join(uri)

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

    parsed_url = urlobject.URLObject(prefix_url)

    prefix = urllib.parse.unquote(parsed_url.path.rstrip('/'))

    try:

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

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

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

    args = {

        'scheme': parsed_url.scheme,

        'user': urllib.parse.quote(username or ''),

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

        'hostname': parsed_url.hostname,

    }

    url = re.sub('{([^{}]+)}', lambda m: args.get(m.group(1), m.group(0)), clone_uri_tmpl)

    # remove leading @ sign if it's present. Case of empty user

    url_obj = urlobject.URLObject(url)

    if not url_obj.username:

        url_obj = url_obj.with_username(None)

    return str(url_obj)

def short_ref_name(ref_type, ref_name):

    """Return short description of PR ref - revs will be truncated"""

    if ref_type == 'rev':

        return ref_name[:12]

    return ref_name

def link_to_ref(repo_name, ref_type, ref_name, rev=None):

    """

    Return full markup for a PR ref to changeset_home for a changeset.

    If ref_type is 'branch', it will link to changelog.

    ref_name is shortened if ref_type is 'rev'.

    if rev is specified, show it too, explicitly linking to that revision.

    """

    txt = short_ref_name(ref_type, ref_name)

    if ref_type == 'branch':

        u = webutils.url('changelog_home', repo_name=repo_name, branch=ref_name)

    else:

        u = webutils.url('changeset_home', repo_name=repo_name, revision=ref_name)

    l = webutils.link_to(repo_name + '#' + txt, u)

    if rev and ref_type != 'rev':

        l = webutils.literal('%s (%s)' % (l, webutils.link_to(rev[:12], webutils.url('changeset_home', repo_name=repo_name, revision=rev))))

    return l

def get_changeset_safe(repo, rev):

    """

    Safe version of get_changeset if this changeset doesn't exists for a

    repo it returns a Dummy one instead

    :param repo:

    :param rev:

    """

    if not isinstance(repo, BaseRepository):

        raise Exception('You must pass an Repository '

                        'object as first argument got %s' % type(repo))

    try:

        cs = repo.get_changeset(rev)

    except (RepositoryError, LookupError):

        cs = EmptyChangeset(requested_revision=rev)

    return cs

def datetime_to_time(dt):

    if dt:

        return time.mktime(dt.timetuple())

def time_to_datetime(tm):

    if tm:

        if isinstance(tm, str):

            try:

                tm = float(tm)

            except ValueError:

                return

        return datetime.datetime.fromtimestamp(tm)

class AttributeDict(dict):

    def __getattr__(self, attr):

        return self.get(attr, None)

    __setattr__ = dict.__setitem__

    __delattr__ = dict.__delitem__

def obfuscate_url_pw(engine):

    try:

        _url = sa_url.make_url(engine or '')

    except ArgumentError:

        return engine

    if _url.password:

        _url.password = 'XXXXX'

    return str(_url)

def get_hook_environment():

    """

    Get hook context by deserializing the global KALLITHEA_EXTRAS environment

    variable.

    Called early in Git out-of-process hooks to get .ini config path so the

    basic environment can be configured properly. Also used in all hooks to get

    information about the action that triggered it.

    """

    try:

        kallithea_extras = os.environ['KALLITHEA_EXTRAS']

    except KeyError:

        raise HookEnvironmentError("Environment variable KALLITHEA_EXTRAS not found")

    extras = json.loads(kallithea_extras)

    for k in ['username', 'repository', 'scm', 'action', 'ip', 'config']:

        try:

            extras[k]

        except KeyError:

            raise HookEnvironmentError('Missing key %s in KALLITHEA_EXTRAS %s' % (k, extras))

    return AttributeDict(extras)

def set_hook_environment(username, ip_addr, repo_name, repo_alias, action=None):

    """Prepare global context for running hooks by serializing data in the

    global KALLITHEA_EXTRAS environment variable.

    Most importantly, this allow Git hooks to do proper logging and updating of

    caches after pushes.

    Must always be called before anything with hooks are invoked.

    """

    extras = {

        'ip': ip_addr, # used in action_logger

        'username': username,

        'action': action or 'push_local', # used in process_pushed_raw_ids action_logger

        'repository': repo_name,

        'scm': repo_alias,

        'config': kallithea.CONFIG['__file__'], # used by git hook to read config

    }

    os.environ['KALLITHEA_EXTRAS'] = json.dumps(extras)

def get_current_authuser():

    """

    Gets kallithea user from threadlocal tmpl_context variable if it's

    defined, else returns None.

    """

    try:

        return getattr(tmpl_context, 'authuser', None)

    except TypeError:  # No object (name: context) has been registered for this thread

        return None

def urlreadable(s, _cleanstringsub=re.compile('[^-a-zA-Z0-9./]+').sub):

    return _cleanstringsub('_', s).rstrip('_')

def recursive_replace(str_, replace=' '):

    """

    Recursive replace of given sign to just one instance

    :param str_: given string

    :param replace: char to find and replace multiple instances

    Examples::

    >>> recursive_replace("Mighty---Mighty-Bo--sstones",'-')

    'Mighty-Mighty-Bo-sstones'

    """

    if str_.find(replace * 2) == -1:

        return str_

    else:

        str_ = str_.replace(replace * 2, replace)

        return recursive_replace(str_, replace)

def repo_name_slug(value):

    """

    Return slug of name of repository

    This function is called on each creation/modification

    of repository to prevent bad names in repo

    """

    slug = remove_formatting(value)

    slug = strip_tags(slug)

    for c in r"""`?=[]\;'"<>,/~!@#$%^&*()+{}|: """:

        slug = slug.replace(c, '-')

    slug = recursive_replace(slug, '-')

    slug = collapse(slug, '-')

    return slug

def ask_ok(prompt, retries=4, complaint='Yes or no please!'):

    while True:

        ok = input(prompt)

        if ok in ('y', 'ye', 'yes'):

            return True

        if ok in ('n', 'no', 'nop', 'nope'):

            return False

        retries = retries - 1

        if retries < 0:

            raise IOError

        print(complaint)

class PasswordGenerator(object):

    """

    This is a simple class for generating password from different sets of

    characters

    usage::

        passwd_gen = PasswordGenerator()

        #print 8-letter password containing only big and small letters

            of alphabet

        passwd_gen.gen_password(8, passwd_gen.ALPHABETS_BIG_SMALL)

    """

    ALPHABETS_NUM = r'''1234567890'''

    ALPHABETS_SMALL = r'''qwertyuiopasdfghjklzxcvbnm'''

    ALPHABETS_BIG = r'''QWERTYUIOPASDFGHJKLZXCVBNM'''

    ALPHABETS_SPECIAL = r'''`-=[]\;',./~!@#$%^&*()_+{}|:"<>?'''

    ALPHABETS_FULL = ALPHABETS_BIG + ALPHABETS_SMALL \

        + ALPHABETS_NUM + ALPHABETS_SPECIAL

    ALPHABETS_ALPHANUM = ALPHABETS_BIG + ALPHABETS_SMALL + ALPHABETS_NUM

    ALPHABETS_BIG_SMALL = ALPHABETS_BIG + ALPHABETS_SMALL

    ALPHABETS_ALPHANUM_BIG = ALPHABETS_BIG + ALPHABETS_NUM

    ALPHABETS_ALPHANUM_SMALL = ALPHABETS_SMALL + ALPHABETS_NUM

    def gen_password(self, length, alphabet=ALPHABETS_FULL):

        assert len(alphabet) <= 256, alphabet

        l = []

        while len(l) < length:

            i = ord(os.urandom(1))

            if i < len(alphabet):

                l.append(alphabet[i])

        return ''.join(l)

def get_crypt_password(password):

    """

    Cryptographic function used for bcrypt password hashing.

    :param password: password to hash

    """

    return ascii_str(bcrypt.hashpw(safe_bytes(password), bcrypt.gensalt(10)))

def check_password(password, hashed):

    """

    Checks password match the hashed value using bcrypt.

    Remains backwards compatible and accept plain sha256 hashes which used to

    be used on Windows.

    :param password: password

    :param hashed: password in hashed form

    """

    # sha256 hashes will always be 64 hex chars

    # bcrypt hashes will always contain $ (and be shorter)

    if len(hashed) == 64 and all(x in string.hexdigits for x in hashed):

        return hashlib.sha256(password).hexdigest() == hashed

    try:

        return bcrypt.checkpw(safe_bytes(password), ascii_bytes(hashed))

    except ValueError as e:

        # bcrypt will throw ValueError 'Invalid hashed_password salt' on all password errors

        log.error('error from bcrypt checking password: %s', e)

        return False

    log.error('check_password failed - no method found for hash length %s', len(hashed))

    return False

git_req_ver = StrictVersion('1.7.4')

def check_git_version():

    """

    Checks what version of git is installed on the system, and raise a system exit

    if it's too old for Kallithea to work properly.

    """

    if 'git' not in kallithea.BACKENDS:

        return None

    if not settings.GIT_EXECUTABLE_PATH:

        log.warning('No git executable configured - check "git_path" in the ini file.')

        return None

    try:

        stdout, stderr = GitRepository._run_git_command(['--version'])

    except RepositoryError as e:

        dest_dir = _get_tmp_dir()

        stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'init', dest_dir)

        clone_url = vt.repo_url_param(webserver, vt.repo_name)

        stdout, stderr = Command(dest_dir).execute(vt.repo_type, 'pull', clone_url)

        meta.Session.close()  # make sure SA fetches all new log entries (apparently only needed for MariaDB/MySQL ...)

        if vt.repo_type == 'git':

            assert 'FETCH_HEAD' in stderr

        elif vt.repo_type == 'hg':

            assert 'new changesets' in stdout

        action_parts = [ul.action for ul in db.UserLog.query().order_by(db.UserLog.user_log_id)]

        assert action_parts == ['pull']

        # Test handling of URLs with extra '/' around repo_name

        stdout, stderr = Command(dest_dir).execute(vt.repo_type, 'pull', clone_url.replace('/' + vt.repo_name, '/./%s/' % vt.repo_name), ignoreReturnCode=True)

        if issubclass(vt, HttpVcsTest):

            if vt.repo_type == 'git':

                # NOTE: when pulling from http://hostname/./vcs_test_git/ , the git client will normalize that and issue an HTTP request to /vcs_test_git/info/refs

                assert 'Already up to date.' in stdout

            else:

                assert vt.repo_type == 'hg'

                assert "abort: HTTP Error 404: Not Found" in stderr

        else:

            assert issubclass(vt, SshVcsTest)

            if vt.repo_type == 'git':

                assert "abort: Access to './%s' denied" % vt.repo_name in stderr

            else:

                assert "abort: Access to './%s' denied" % vt.repo_name in stdout + stderr

        stdout, stderr = Command(dest_dir).execute(vt.repo_type, 'pull', clone_url.replace('/' + vt.repo_name, '/%s/' % vt.repo_name), ignoreReturnCode=True)

        if vt.repo_type == 'git':

            assert 'Already up to date.' in stdout

        else:

            assert vt.repo_type == 'hg'

            assert "no changes found" in stdout

        assert "denied" not in stderr

        assert "denied" not in stdout

        assert "404" not in stdout

    @parametrize_vcs_test

    def test_push_invalidates_cache(self, webserver, testfork, vt):

        pre_cached_tip = [repo.get_api_data()['last_changeset']['short_id'] for repo in db.Repository.query().filter(db.Repository.repo_name == testfork[vt.repo_type])]

        dest_dir = _get_tmp_dir()

        clone_url = vt.repo_url_param(webserver, testfork[vt.repo_type])

        stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)

        stdout, stderr = _add_files_and_push(webserver, vt, dest_dir, files_no=1, clone_url=clone_url)

        if vt.repo_type == 'git':

            _check_proper_git_push(stdout, stderr)

        meta.Session.close()  # expire session to make sure SA fetches new Repository instances after last_changeset has been updated by server side hook in another process

        post_cached_tip = [repo.get_api_data()['last_changeset']['short_id'] for repo in db.Repository.query().filter(db.Repository.repo_name == testfork[vt.repo_type])]

        assert pre_cached_tip != post_cached_tip

    @parametrize_vcs_test_http

    def test_push_wrong_credentials(self, webserver, vt):

        dest_dir = _get_tmp_dir()

        clone_url = vt.repo_url_param(webserver, vt.repo_name)

        stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)

        clone_url = webserver.repo_url(vt.repo_name, username='bad', password='name')

        stdout, stderr = _add_files_and_push(webserver, vt, dest_dir,

                                             clone_url=clone_url, ignoreReturnCode=True)

        if vt.repo_type == 'git':

            assert 'fatal: Authentication failed' in stderr

        elif vt.repo_type == 'hg':

            assert 'abort: authorization failed' in stderr

    @parametrize_vcs_test

    def test_push_with_readonly_credentials(self, webserver, vt):

        db.UserLog.query().delete()

        meta.Session().commit()

        dest_dir = _get_tmp_dir()

        clone_url = vt.repo_url_param(webserver, vt.repo_name, username=base.TEST_USER_REGULAR_LOGIN, password=base.TEST_USER_REGULAR_PASS)

        stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)

        stdout, stderr = _add_files_and_push(webserver, vt, dest_dir, ignoreReturnCode=True, clone_url=clone_url)

        if vt.repo_type == 'git':

            assert 'The requested URL returned error: 403' in stderr or 'abort: Push access to %r denied' % str(vt.repo_name) in stderr

        elif vt.repo_type == 'hg':

            assert 'abort: HTTP Error 403: Forbidden' in stderr or 'abort: push failed on remote' in stderr and 'remote: Push access to %r denied' % str(vt.repo_name) in stdout

        meta.Session.close()  # make sure SA fetches all new log entries (apparently only needed for MariaDB/MySQL ...)

        action_parts = [ul.action.split(':', 1) for ul in db.UserLog.query().order_by(db.UserLog.user_log_id)]

        assert [(t[0], (t[1].count(',') + 1) if len(t) == 2 else 0) for t in action_parts] == \

            [('pull', 0)]

    @parametrize_vcs_test

    def test_push_back_to_wrong_url(self, webserver, vt):

        dest_dir = _get_tmp_dir()

        clone_url = vt.repo_url_param(webserver, vt.repo_name)

        stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)

        stdout, stderr = _add_files_and_push(

            webserver, vt, dest_dir, clone_url='http://%s:%s/tmp' % (

                webserver.server_address[0], webserver.server_address[1]),

            ignoreReturnCode=True)

        if vt.repo_type == 'git':

            assert 'not found' in stderr

        elif vt.repo_type == 'hg':

            assert 'HTTP Error 404: Not Found' in stderr

    @parametrize_vcs_test

    def test_ip_restriction(self, webserver, vt):

        user_model = UserModel()

        try:

            # Add IP constraint that excludes the test context:

            user_model.add_extra_ip(base.TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')

            meta.Session().commit()

            # IP permissions are cached, need to wait for the cache in the server process to expire

            time.sleep(1.5)

            clone_url = vt.repo_url_param(webserver, vt.repo_name)

            stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

            if vt.repo_type == 'git':

                # The message apparently changed in Git 1.8.3, so match it loosely.

                assert re.search(r'\b403\b', stderr) or 'abort: User test_admin from 127.0.0.127 cannot be authorized' in stderr

            elif vt.repo_type == 'hg':

                assert 'abort: HTTP Error 403: Forbidden' in stderr or 'remote: abort: User test_admin from 127.0.0.127 cannot be authorized' in stdout + stderr

        finally:

            # release IP restrictions

            for ip in db.UserIpMap.query():

                db.UserIpMap.delete(ip.ip_id)

            meta.Session().commit()

            # IP permissions are cached, need to wait for the cache in the server process to expire

            time.sleep(1.5)

        clone_url = vt.repo_url_param(webserver, vt.repo_name)

        stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir())

        if vt.repo_type == 'git':

            assert 'Cloning into' in stdout + stderr

            assert stderr == '' or stdout == ''

        elif vt.repo_type == 'hg':

            assert 'requesting all changes' in stdout

            assert 'adding changesets' in stdout

            assert 'adding manifests' in stdout

            assert 'adding file changes' in stdout

            assert stderr == ''

    @parametrize_vcs_test_hg # git hooks doesn't work like hg hooks

    def test_custom_hooks_preoutgoing(self, testhook_cleanup, webserver, testfork, vt):

        # set prechangegroup to failing hook (returns True)

        db.Ui.create_or_update_hook('preoutgoing.testhook', 'python:kallithea.tests.fixture.failing_test_hook')

        meta.Session().commit()

        # clone repo

        clone_url = vt.repo_url_param(webserver, testfork[vt.repo_type], username=base.TEST_USER_ADMIN_LOGIN, password=base.TEST_USER_ADMIN_PASS)

        dest_dir = _get_tmp_dir()

        stdout, stderr = Command(base.TESTS_TMP_PATH) \

            .execute(vt.repo_type, 'clone', clone_url, dest_dir, ignoreReturnCode=True)

        if vt.repo_type == 'hg':

            assert 'preoutgoing.testhook hook failed' in stdout + stderr

        elif vt.repo_type == 'git':

            assert 'error: 406' in stderr

    @parametrize_vcs_test_hg # git hooks doesn't work like hg hooks

    def test_custom_hooks_prechangegroup(self, testhook_cleanup, webserver, testfork, vt):

        # set prechangegroup to failing hook (returns exit code 1)

        db.Ui.create_or_update_hook('prechangegroup.testhook', 'python:kallithea.tests.fixture.failing_test_hook')

        meta.Session().commit()

        # clone repo

        clone_url = vt.repo_url_param(webserver, testfork[vt.repo_type], username=base.TEST_USER_ADMIN_LOGIN, password=base.TEST_USER_ADMIN_PASS)

        dest_dir = _get_tmp_dir()

        stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)

        stdout, stderr = _add_files_and_push(webserver, vt, dest_dir, clone_url,

                                             ignoreReturnCode=True)

        assert 'failing_test_hook failed' in stdout + stderr

        assert 'Traceback' not in stdout + stderr

        assert 'prechangegroup.testhook hook failed' in stdout + stderr

        # there are still outgoing changesets

        stdout, stderr = _check_outgoing(vt.repo_type, dest_dir, clone_url)

        assert stdout != ''

        # set prechangegroup hook to exception throwing method

        db.Ui.create_or_update_hook('prechangegroup.testhook', 'python:kallithea.tests.fixture.exception_test_hook')

        meta.Session().commit()

        # re-try to push

        stdout, stderr = Command(dest_dir).execute('%s push' % vt.repo_type, clone_url, ignoreReturnCode=True)

        if vt is HgHttpVcsTest:

            # like with 'hg serve...' 'HTTP Error 500: INTERNAL SERVER ERROR' should be returned

            assert 'HTTP Error 500: INTERNAL SERVER ERROR' in stderr

        elif vt is HgSshVcsTest:

            assert 'remote: Exception: exception_test_hook threw an exception' in stdout

        # there are still outgoing changesets

        stdout, stderr = _check_outgoing(vt.repo_type, dest_dir, clone_url)

        assert stdout != ''

        # set prechangegroup hook to method that returns False

        db.Ui.create_or_update_hook('prechangegroup.testhook', 'python:kallithea.tests.fixture.passing_test_hook')

        meta.Session().commit()

        # re-try to push

        stdout, stderr = Command(dest_dir).execute('%s push' % vt.repo_type, clone_url, ignoreReturnCode=True)

        assert 'passing_test_hook succeeded' in stdout + stderr

        assert 'Traceback' not in stdout + stderr

        assert 'prechangegroup.testhook hook failed' not in stdout + stderr

        # no more outgoing changesets

        stdout, stderr = _check_outgoing(vt.repo_type, dest_dir, clone_url)

        assert stdout == ''

        assert stderr == ''

    def test_add_submodule_git(self, webserver, testfork):

        dest_dir = _get_tmp_dir()

        clone_url = GitHttpVcsTest.repo_url_param(webserver, base.GIT_REPO)

        fork_url = GitHttpVcsTest.repo_url_param(webserver, testfork['git'])

        # add submodule

        stdout, stderr = Command(base.TESTS_TMP_PATH).execute('git clone', fork_url, dest_dir)

        stdout, stderr = Command(dest_dir).execute('git submodule add', clone_url, 'testsubmodule')

        stdout, stderr = Command(dest_dir).execute('git commit -am "added testsubmodule pointing to', clone_url, '"', EMAIL=base.TEST_USER_ADMIN_EMAIL)

        stdout, stderr = Command(dest_dir).execute('git push', fork_url, 'master')

        # check for testsubmodule link in files page

        self.log_user()

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

                                    repo_name=testfork['git'],

                                    revision='tip',

                                    f_path='/'))

        # check _repo_files_url that will be used to reload as AJAX

        response.mustcontain('var _repo_files_url = ("/%s/files/");' % testfork['git'])

        response.mustcontain('<a class="submodule-dir" href="%s" target="_blank"><i class="icon-file-submodule"></i><span>testsubmodule @ ' % clone_url)

        # check that following a submodule link actually works - and redirects

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

                                    repo_name=testfork['git'],

                                    revision='tip',

                                    f_path='/testsubmodule'),

                                status=302)

        assert response.location == clone_url