following list, modules may only depend on modules below them:

``tests``

  Just get the job done - anything goes.

``bin/`` & ``config/`` & ``alembic/``

  The main entry points, defined in ``setup.py``. Note: The TurboGears template

  use ``config`` for the high WSGI application - this is not for low level

  configuration.

``controllers/``

  The top level web application, with TurboGears using the ``root`` controller

  as entry point, and ``routing`` dispatching to other controllers.

``templates/**.html``

  The "view", rendering to HTML. Invoked by controllers which can pass them

  anything from lower layers - especially ``helpers`` available as ``h`` will

  cut through all layers, and ``c`` gives access to global variables.

``lib/helpers.py``

  High level helpers, exposing everything to templates as ``h``. It depends on

  everything and has a huge dependency chain, so it should not be used for

  anything else. TODO.

  The base class of controllers, with lots of model knowledge.

``lib/auth.py``

  All things related to authentication. TODO.

``lib/utils.py``

  High level utils with lots of model knowledge. TODO.

``lib/hooks.py``

  Hooks into "everything" to give centralized logging to database, cache

  invalidation, and extension handling. TODO.

``model/``

  Convenience business logic wrappers around database models.

``model/db.py``

  Defines the database schema and provides some additional logic.

``model/scm.py``

  All things related to anything. TODO.

SQLAlchemy

  Database session and transaction in thread-local variables.

#

# 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.config.middleware.simplegit

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

SimpleGit middleware for handling Git protocol requests (push/clone etc.)

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 logging

import re

from kallithea.config.middleware.pygrack import make_wsgi_app

from kallithea.lib import hooks

log = logging.getLogger(__name__)

GIT_PROTO_PAT = re.compile(r'^/(.+)/(info/refs|git-upload-pack|git-receive-pack)$')

cmd_mapping = {

    'git-receive-pack': 'push',

    'git-upload-pack': 'pull',

}

    scm_alias = 'git'

    @classmethod

    def parse_request(cls, environ):

        m = GIT_PROTO_PAT.match(path_info)

        if m is None:

            return None

        class parsed_request(object):

            # See https://git-scm.com/book/en/v2/Git-Internals-Transfer-Protocols#_the_smart_protocol

            repo_name = m.group(1).rstrip('/')

            cmd = m.group(2)

            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 _make_app(self, parsed_request):

        """

        Return a pygrack wsgi application.

        """

        pygrack_app = make_wsgi_app(parsed_request.repo_name, self.basepath)

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""

kallithea.config.middleware.simplehg

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

SimpleHg middleware for handling Mercurial protocol requests (push/clone etc.).

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 logging

import os

import urllib.parse

import mercurial.hgweb

from kallithea.lib.utils import make_ui

from kallithea.lib.utils2 import safe_bytes

log = logging.getLogger(__name__)

def get_header_hgarg(environ):

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

    >>> get_header_hgarg({})

    ''

    >>> get_header_hgarg({'HTTP_X_HGARG_0': ' ', 'HTTP_X_HGARG_1': 'a','HTTP_X_HGARG_2': '','HTTP_X_HGARG_3': 'b+c %20'})

    'ab+c %20'

    """

    chunks = []

    i = 1

    while True:

        v = environ.get('HTTP_X_HGARG_%d' % i)

        if v is None:

            break

        chunks.append(v)

        i += 1

    return ''.join(chunks)

    'changegroupsubset': 'pull',

    'changesetdata': 'pull',

    'clonebundles': 'pull',

    'debugwireargs': 'pull',

    'filedata': 'pull',

    'getbundle': 'pull',

    'getlfile': 'pull',

    'heads': 'pull',

    'hello': 'pull',

    'known': 'pull',

    'lheads': 'pull',

    'listkeys': 'pull',

    'lookup': 'pull',

    'manifestdata': 'pull',

    'narrow_widen': 'pull',

    'protocaps': 'pull',

    'statlfile': 'pull',

    'stream_out': 'pull',

    'pushkey': 'push',

    'putlfile': 'push',

    'unbundle': 'push',

    }

    scm_alias = 'hg'

    @classmethod

    def parse_request(cls, environ):

        http_accept = environ.get('HTTP_ACCEPT', '')

        if not http_accept.startswith('application/mercurial'):

            return None

        if not path_info.startswith('/'): # it must!

            return None

        class parsed_request(object):

            repo_name = path_info[1:].rstrip('/')

            query_string = environ['QUERY_STRING']

            action = None

            for qry in query_string.split('&'):

                parts = qry.split('=', 1)

                if len(parts) == 2 and parts[0] == 'cmd':

                    cmd = parts[1]

                    if cmd == 'batch':

                        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(';'):

                            cmd, _args = urllib.parse.unquote_plus(cmd_arg).split(' ', 1)

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

                            if op != 'pull':

                                assert op == 'push'

# 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.config.middleware.wrapper

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

Wrap app to measure request and response time ... all the way to the response

WSGI iterator has been closed.

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: May 23, 2013

:author: marcink

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

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

"""

import logging

import time

log = logging.getLogger(__name__)

class Meter:

    def __init__(self, start_response):

        self._start_response = start_response

        self._start = time.time()

        self.status = None

        self._size = 0

    def duration(self):

        return time.time() - self._start

    def start_response(self, status, response_headers, exc_info=None):

        self.status = status

        write = self._start_response(status, response_headers, exc_info)

        def metered_write(s):

            self.measure(s)

            write(s)

        return metered_write

        self._description = description

    def __iter__(self):

        return self

    def __next__(self):

        chunk = self._next()

        self._meter.measure(chunk)

        return chunk

    def close(self):

        self._result_close()

        log.info("%s responded %r after %.3fs with %s bytes", self._description, self._meter.status, self._meter.duration(), self._meter.size())

class RequestWrapper(object):

    def __init__(self, app, config):

        self.application = app

        self.config = config

    def __call__(self, environ, start_response):

        meter = Meter(start_response)

        description = "Request from %s for %s" % (

        )

        log.info("%s received", description)

        try:

            result = self.application(environ, meter.start_response)

        finally:

            log.info("%s responding %r after %.3fs", description, meter.status, meter.duration())

        return ResultIter(result, meter, description)

kallithea.controllers.admin.admin

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

Controller for Admin panel of 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 7, 2010

:author: marcink

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

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

"""

import logging

from sqlalchemy.orm import joinedload

from sqlalchemy.sql.expression import and_, func, or_

from tg import request

from tg import tmpl_context as c

from whoosh import query

from whoosh.qparser.dateparse import DateParserPlugin

from whoosh.qparser.default import QueryParser

from kallithea.lib.auth import HasPermissionAnyDecorator, LoginRequired

from kallithea.lib.indexers import JOURNAL_SCHEMA

from kallithea.lib.page import Page

from kallithea.lib.utils2 import remove_prefix, remove_suffix, safe_int

from kallithea.model import db

log = logging.getLogger(__name__)

def _journal_filter(user_log, search_term):

    """

    Filters sqlalchemy user_log based on search_term with whoosh Query language

    http://packages.python.org/Whoosh/querylang.html

    :param user_log:

    :param search_term:

    """

    log.debug('Initial search term: %r', search_term)

    qry = None

    if search_term:

        qp = QueryParser('repository', schema=JOURNAL_SCHEMA)

        qp.add_plugin(DateParserPlugin())

        qry = qp.parse(search_term)

        log.debug('Filtering using parsed query %r', qry)

            return and_(field >= val[0], field <= val[1])

        return func.lower(field) == func.lower(val)

    if isinstance(qry, (query.And, query.Term, query.Prefix, query.Wildcard,

                        query.DateRange)):

        if not isinstance(qry, query.And):

            qry = [qry]

        for term in qry:

            field = term.fieldname

            val = (term.text if not isinstance(term, query.DateRange)

                   else [term.startdate, term.enddate])

            user_log = user_log.filter(get_filterion(field, val, term))

    elif isinstance(qry, query.Or):

        filters = []

        for term in qry:

            field = term.fieldname

            val = (term.text if not isinstance(term, query.DateRange)

                   else [term.startdate, term.enddate])

            filters.append(get_filterion(field, val, term))

        user_log = user_log.filter(or_(*filters))

    return user_log

    @LoginRequired(allow_default_user=True)

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

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

    @HasPermissionAnyDecorator('hg.admin')

    def index(self):

        users_log = db.UserLog.query() \

                .options(joinedload(db.UserLog.user)) \

                .options(joinedload(db.UserLog.repository))

        # FILTERING

        c.search_term = request.GET.get('filter')

        users_log = _journal_filter(users_log, c.search_term)

        users_log = users_log.order_by(db.UserLog.action_date.desc())

        p = safe_int(request.GET.get('page'), 1)

        c.users_log = Page(users_log, page=p, items_per_page=10,

                           filter=c.search_term)

        if request.environ.get('HTTP_X_PARTIAL_XHR'):

#

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

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

pluggable authentication 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: Nov 26, 2010

:author: akesterson

"""

import logging

import traceback

import formencode.htmlfill

from tg import request

from tg import tmpl_context as c

from tg.i18n import ugettext as _

from webob.exc import HTTPFound

from kallithea.lib import auth_modules, webutils

from kallithea.lib.auth import HasPermissionAnyDecorator, LoginRequired

from kallithea.lib.webutils import url

from kallithea.model import db, meta

from kallithea.model.forms import AuthSettingsForm

log = logging.getLogger(__name__)

    @LoginRequired()

    @HasPermissionAnyDecorator('hg.admin')

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

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

    def __load_defaults(self):

        c.available_plugins = [

            'kallithea.lib.auth_modules.auth_internal',

            'kallithea.lib.auth_modules.auth_container',

            'kallithea.lib.auth_modules.auth_ldap',

            'kallithea.lib.auth_modules.auth_crowd',

            'kallithea.lib.auth_modules.auth_pam'

        ]

        self.enabled_plugins = auth_modules.get_auth_plugins()

        c.enabled_plugin_names = [plugin.__class__.__module__ for plugin in self.enabled_plugins]

    def __render(self, defaults, errors):

        c.defaults = {}

        c.plugin_settings = {}

        c.plugin_shortnames = {}

        for plugin in self.enabled_plugins:

            module = plugin.__class__.__module__

            c.plugin_shortnames[module] = plugin.name

            c.plugin_settings[module] = plugin.plugin_settings()

            for v in c.plugin_settings[module]:

                fullname = "auth_%s_%s" % (plugin.name, v["name"])

                if "default" in v:

                    c.defaults[fullname] = v["default"]

                # Current values will be the default on the form, if there are any

                setting = db.Setting.get_by_name(fullname)

                if setting is not None:

                    c.defaults[fullname] = setting.app_settings_value

        if defaults:

            c.defaults.update(defaults)

        # we want to show , separated list of enabled plugins

        c.defaults['auth_plugins'] = ','.join(c.enabled_plugin_names)

        log.debug('defaults: %s', defaults)

        return formencode.htmlfill.render(

            defaults=c.defaults,

            errors=errors,

            prefix_error=False,

            encoding="UTF-8",

            force_defaults=False)

    def index(self):

        self.__load_defaults()

        return self.__render(defaults=None, errors=None)

    def auth_settings(self):

        """POST create and store auth settings"""

        self.__load_defaults()

        log.debug("POST Result: %s", dict(request.POST))

        # First, parse only the plugin list (not the plugin settings).

        _auth_plugins_validator = AuthSettingsForm([]).fields['auth_plugins']

        try:

            new_enabled_plugins = _auth_plugins_validator.to_python(request.POST.get('auth_plugins'))

        except formencode.Invalid:

            # User provided an invalid plugin list. Just fall back to

            # the list of currently enabled plugins. (We'll re-validate

            # and show an error message to the user, below.)

            pass

# 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.lib import webutils

from kallithea.lib.auth import HasPermissionAnyDecorator, LoginRequired

from kallithea.lib.webutils import url

from kallithea.model import db, meta

from kallithea.model.forms import DefaultsForm

log = logging.getLogger(__name__)

    @LoginRequired()

    @HasPermissionAnyDecorator('hg.admin')

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

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

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

        defaults = db.Setting.get_default_repo_settings()

        return htmlfill.render(

            defaults=defaults,

            encoding="UTF-8",

            force_defaults=False

        )

    def update(self, id):

        _form = DefaultsForm()()

        try:

            form_result = _form.to_python(dict(request.POST))

            for k, v in form_result.items():

                setting = db.Setting.create_or_update(k, v)

            meta.Session().commit()

            webutils.flash(_('Default settings updated successfully'),

                    category='success')

        except formencode.Invalid as errors:

            defaults = errors.value

            return htmlfill.render(

                defaults=defaults,

                errors=errors.error_dict or {},

                prefix_error=False,

                encoding="UTF-8",

                force_defaults=False)

        except Exception:

            log.error(traceback.format_exc())

            webutils.flash(_('Error occurred during update of defaults'),

                    category='error')

        raise HTTPFound(location=url('defaults'))

"""

kallithea.controllers.admin.gists

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

gist 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: May 9, 2013

:author: marcink

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

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

"""

import logging

import traceback

import formencode.htmlfill

from sqlalchemy.sql.expression import or_

from tg import request, response

from tg import tmpl_context as c

from tg.i18n import ugettext as _

from webob.exc import HTTPForbidden, HTTPFound, HTTPNotFound

from kallithea.lib import auth, webutils

from kallithea.lib.auth import LoginRequired

from kallithea.lib.page import Page

from kallithea.lib.utils2 import safe_int, safe_str, time_to_datetime

from kallithea.lib.vcs.exceptions import NodeNotChangedError, VCSError

from kallithea.lib.webutils import url

from kallithea.model import db, meta

from kallithea.model.forms import GistForm

from kallithea.model.gist import GistModel

log = logging.getLogger(__name__)

    def __load_defaults(self, extra_values=None):

        c.lifetime_values = [

            (str(-1), _('Forever')),

            (str(5), _('5 minutes')),

            (str(60), _('1 hour')),

            (str(60 * 24), _('1 day')),

            (str(60 * 24 * 30), _('1 month')),

        ]

        if extra_values:

            c.lifetime_values.append(extra_values)

        c.lifetime_options = [(c.lifetime_values, _("Lifetime"))]

    @LoginRequired(allow_default_user=True)

    def index(self):

        not_default_user = not request.authuser.is_default_user

        c.show_private = request.GET.get('private') and not_default_user

        c.show_public = request.GET.get('public') and not_default_user

        url_params = {}

        if c.show_public:

            url_params['public'] = 1

        elif c.show_private:

            url_params['private'] = 1

        # MY private

        if c.show_private and not c.show_public:

            gists = gists.filter(db.Gist.gist_type == db.Gist.GIST_PRIVATE) \

                             .filter(db.Gist.owner_id == request.authuser.user_id)

        # MY public

        elif c.show_public and not c.show_private:

            gists = gists.filter(db.Gist.gist_type == db.Gist.GIST_PUBLIC) \

                             .filter(db.Gist.owner_id == request.authuser.user_id)

        # MY public+private

        elif c.show_private and c.show_public:

            gists = gists.filter(or_(db.Gist.gist_type == db.Gist.GIST_PUBLIC,

                                     db.Gist.gist_type == db.Gist.GIST_PRIVATE)) \

                             .filter(db.Gist.owner_id == request.authuser.user_id)

        # default show ALL public gists

        if not c.show_public and not c.show_private:

            gists = gists.filter(db.Gist.gist_type == db.Gist.GIST_PUBLIC)

        c.gists = gists

        p = safe_int(request.GET.get('page'), 1)

        c.gists_pager = Page(c.gists, page=p, items_per_page=10,

                             **url_params)

    @LoginRequired()

    def create(self):

        self.__load_defaults()

        gist_form = GistForm([x[0] for x in c.lifetime_values])()

        try:

            form_result = gist_form.to_python(dict(request.POST))

            # TODO: multiple files support, from the form

            filename = form_result['filename'] or db.Gist.DEFAULT_FILENAME

            nodes = {

                filename: {

                    'content': form_result['content'],

                    'lexer': form_result['mimetype']  # None is autodetect

                }

            }

            _public = form_result['public']

            gist_type = db.Gist.GIST_PUBLIC if _public else db.Gist.GIST_PRIVATE

            gist = GistModel().create(

                description=form_result['description'],

                owner=request.authuser.user_id,

                ip_addr=request.ip_addr,

                gist_mapping=nodes,

                gist_type=gist_type,

                lifetime=form_result['lifetime']

            )

            meta.Session().commit()

            new_gist_id = gist.gist_access_id

        except formencode.Invalid as errors:

            defaults = errors.value

            return formencode.htmlfill.render(

                defaults=defaults,

                errors=errors.error_dict or {},

                prefix_error=False,

                encoding="UTF-8",

                force_defaults=False)

        except Exception as e:

            log.error(traceback.format_exc())

            webutils.flash(_('Error occurred during gist creation'), category='error')

            raise HTTPFound(location=url('new_gist'))

        raise HTTPFound(location=url('gist', gist_id=new_gist_id))

    @LoginRequired()

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

        self.__load_defaults()

    @LoginRequired()

    def delete(self, gist_id):

        gist = GistModel().get_gist(gist_id)

        owner = gist.owner_id == request.authuser.user_id

        if auth.HasPermissionAny('hg.admin')() or owner:

            GistModel().delete(gist)

            meta.Session().commit()

            webutils.flash(_('Deleted gist %s') % gist.gist_access_id, category='success')

        else:

            raise HTTPForbidden()

        raise HTTPFound(location=url('gists'))

    @LoginRequired(allow_default_user=True)

    def show(self, gist_id, revision='tip', format='html', f_path=None):

        c.gist = db.Gist.get_or_404(gist_id)

        if c.gist.is_expired:

            log.error('Gist expired at %s',

                      time_to_datetime(c.gist.gist_expires))

            raise HTTPNotFound()

        try:

            c.file_changeset, c.files = GistModel().get_gist_files(gist_id,

                                                            revision=revision)

        except VCSError:

            log.error(traceback.format_exc())

            raise HTTPNotFound()

        if format == 'raw':

            content = '\n\n'.join(

                safe_str(f.content)

                for f in c.files if (f_path is None or f.path == f_path)

            )

            response.content_type = 'text/plain'

            return content

    @LoginRequired()

    def edit(self, gist_id, format='html'):

        c.gist = db.Gist.get_or_404(gist_id)

        if c.gist.is_expired:

            log.error('Gist expired at %s',

                      time_to_datetime(c.gist.gist_expires))

            raise HTTPNotFound()

        try:

            c.file_changeset, c.files = GistModel().get_gist_files(gist_id)

        except VCSError:

            log.error(traceback.format_exc())

            raise HTTPNotFound()

        self.__load_defaults(extra_values=('0', _('Unmodified')))

        if request.POST:

            rpost = request.POST

            nodes = {}

            for org_filename, filename, mimetype, content in zip(

                                                    rpost.getall('org_files'),