from kallithea.config.routing import url

from kallithea.lib.utils2 import str2bool, safe_unicode, AttributeDict, \

    safe_str, safe_int

from kallithea.lib import auth_modules

from kallithea.lib.auth import AuthUser, HasPermissionAnyMiddleware

from kallithea.lib.compat import json

from kallithea.lib.utils import get_repo_slug

from kallithea.lib.exceptions import UserCreationError

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

from kallithea.model import meta

from kallithea.model.scm import ScmModel

log = logging.getLogger(__name__)

def render(template_path):

    return render_template({'url': url}, 'mako', template_path)

def _filter_proxy(ip):

    """

    HEADERS can have multiple ips inside the left-most being the original

    if ip:

        return _filter_proxy(ip)

    ip = environ.get(proxy_key2)

    if ip:

        return _filter_proxy(ip)

    ip = environ.get(def_key, '0.0.0.0')

    return _filter_proxy(ip)

def _get_access_path(environ):

    org_req = environ.get('tg.original_request')

def log_in_user(user, remember, is_external_auth, ip_addr):

    """

    Log a `User` in and update session and cookies. If `remember` is True,

    the session cookie is set to expire in a year; otherwise, it expires at

    the end of the browser session.

    Returns populated `AuthUser` object.

    """

    # It should not be possible to explicitly log in as the default user.

    assert not user.is_default_user, user

        self.basepath = self.config['base_path']

        # authenticate this VCS request using the authentication modules

        self.authenticate = BasicAuth('', auth_modules.authenticate,

                                      config.get('auth_ret_code'))

    @classmethod

    def parse_request(cls, environ):

        """If request is parsed as a request for this VCS, return a namespace with the parsed request.

        If the request is unknown, return None.

        """

        raise NotImplementedError()

        """Authenticate and authorize user.

        Since we're dealing with a VCS client and not a browser, we only

        support HTTP basic authentication, either directly via raw header

        inspection, or by using container authentication to delegate the

        authentication to the web server.

        Returns (user, None) on successful authentication and authorization.

        Returns (None, wsgi_app) to send the wsgi_app response to the client.

        """

        # Use anonymous access if allowed for action on repo.

        default_user = User.get_default_user(cache=True)

It's implemented with basic auth function

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

:author: marcink

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

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

"""

import re

import logging

import traceback

from kallithea.model.db import Ui, Repository

from kallithea.lib.utils2 import safe_str, safe_unicode, get_server_url, \

    _set_extras

from kallithea.lib.base import BaseVCSController

from kallithea.lib.utils import make_ui, is_valid_repo

from kallithea.lib.hooks import log_pull_action

from kallithea.lib.middleware.pygrack import make_wsgi_app

log = logging.getLogger(__name__)

            query_string = environ['QUERY_STRING']

            if cmd == 'info/refs' and query_string.startswith('service='):

                service = query_string.split('=', 1)[1]

                action = cmd_mapping.get(service)

            else:

                service = None

                action = cmd_mapping.get(cmd)

        return parsed_request

    def _handle_request(self, parsed_request, environ, start_response):

        # skip passing error to error controller

        environ['pylons.status_code_redirect'] = True

        # quick check if repo exists...

        if not is_valid_repo(parsed_request.repo_name, self.basepath, self.scm_alias):

            raise HTTPNotFound()

        if parsed_request.action is None:

            # Note: the client doesn't get the helpful error message

            raise HTTPBadRequest('Unable to detect pull/push action for %r! Are you using a nonstandard command or client?' % parsed_request.repo_name)

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

        # CHECK PERMISSIONS

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

        if response_app is not None:

            return response_app(environ, start_response)

        # extras are injected into Mercurial UI object and later available

        # in hooks executed by Kallithea

        from kallithea import CONFIG

        extras = {

            'ip': ip_addr,

            'username': user.username,

            'action': parsed_request.action,

            'repository': parsed_request.repo_name,

            'scm': self.scm_alias,

            'config': CONFIG['__file__'],

        }

        log.debug('HOOKS extras is %s', extras)

        try:

            log.info('%s action on %s repo "%s" by "%s" from %s',

                     parsed_request.action, self.scm_alias, parsed_request.repo_name, safe_str(user.username), ip_addr)

            app = self._make_app(parsed_request)

            return app(environ, start_response)

        except Exception:

            log.error(traceback.format_exc())

            raise HTTPInternalServerError()

    def _make_app(self, parsed_request):

        """

:author: marcink

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

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

"""

import os

import logging

import traceback

import urllib

from kallithea.lib.utils2 import safe_str, safe_unicode, get_server_url, \

    _set_extras

from kallithea.lib.base import BaseVCSController

from kallithea.lib.utils import make_ui, is_valid_repo

from kallithea.lib.vcs.utils.hgcompat import RepoError, hgweb_mod

log = logging.getLogger(__name__)

def get_header_hgarg(environ):

    """Decode the special Mercurial encoding of big requests over multiple headers.

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

                            if op != 'pull':

                                assert op == 'push'

                                action = 'push'

                                break

                    else:

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

                    break # only process one cmd

        return parsed_request

    def _handle_request(self, parsed_request, environ, start_response):

        # skip passing error to error controller

        environ['pylons.status_code_redirect'] = True

        # quick check if repo exists...

        if not is_valid_repo(parsed_request.repo_name, self.basepath, self.scm_alias):

            raise HTTPNotFound()

        if parsed_request.action is None:

            # Note: the client doesn't get the helpful error message

            raise HTTPBadRequest('Unable to detect pull/push action for %r! Are you using a nonstandard command or client?' % parsed_request.repo_name)

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

        # CHECK PERMISSIONS

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

        if response_app is not None:

            return response_app(environ, start_response)

        # extras are injected into Mercurial UI object and later available

        # in hooks executed by Kallithea

        from kallithea import CONFIG

        extras = {

            'ip': ip_addr,

            'username': user.username,

            'action': parsed_request.action,

            'repository': parsed_request.repo_name,

            'scm': self.scm_alias,

            'config': CONFIG['__file__'],

        }

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

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

        log.debug('HOOKS extras is %s', extras)

        try:

            log.info('%s action on %s repo "%s" by "%s" from %s',

                     parsed_request.action, self.scm_alias, parsed_request.repo_name, safe_str(user.username), ip_addr)

            app = self._make_app(parsed_request)

            return app(environ, start_response)

        except Exception:

            log.error(traceback.format_exc())

            raise HTTPInternalServerError()

    def _make_app(self, parsed_request):

        """

def is_valid_repo(repo_name, base_path, scm=None):

    """

    Returns True if given path is a valid repository False otherwise.

    If scm param is given also compare if given scm is the same as expected

    from scm parameter

    :param repo_name:

    :param base_path:

    :param scm:

    :return True: if given path is a valid repository

    """

    full_path = os.path.join(safe_str(base_path), safe_str(repo_name))

    try:

        scm_ = get_scm(full_path)

        if scm:

            return scm_[0] == scm

        return True

    except VCSError:

        return False

def is_valid_repo_group(repo_group_name, base_path, skip_path_check=False):