imports: always import the whole kallithea module to use top level kallithea variables

This is slightly more lazy and might avoid some depeendency issues.

docs/conf.py

@@ -1,229 +1,229 b''
 # -*- coding: utf-8 -*-
 #
 # Kallithea documentation build configuration file, created by
 # sphinx-quickstart on Sun Oct 10 16:46:37 2010.
 #
 # This file is execfile()d with the current directory set to its containing dir.
 #
 # Note that not all possible configuration values are present in this
 # autogenerated file.
 #
 # All configuration values have a default; values that are commented out
 # serve to show the default.
 
 import os
 import sys
 
-from kallithea import __version__
+import kallithea
 
 
 # If extensions (or modules to document with autodoc) are in another directory,
 # add these directories to sys.path here. If the directory is relative to the
 # documentation root, use os.path.abspath to make it absolute, like shown here.
 sys.path.insert(0, os.path.abspath('..'))
 
 # -- General configuration -----------------------------------------------------
 
 # If your documentation needs a minimal Sphinx version, state it here.
 #needs_sphinx = '1.0'
 
 # Add any Sphinx extension module names here, as strings. They can be extensions
 # coming with Sphinx (named 'sphinx.ext.*') or your custom ones.
 extensions = ['sphinx.ext.autodoc', 'sphinx.ext.doctest',
               'sphinx.ext.intersphinx', 'sphinx.ext.todo',
               'sphinx.ext.viewcode']
 
 # Add any paths that contain templates here, relative to this directory.
 templates_path = ['_templates']
 
 # The suffix of source filenames.
 source_suffix = '.rst'
 
 # The encoding of source files.
 #source_encoding = 'utf-8-sig'
 
 # The master toctree document.
 master_doc = 'index'
 
 # General information about the project.
 project = 'Kallithea'
 copyright = '2010-2020 by various authors, licensed as GPLv3.'
 
 # The version info for the project you're documenting, acts as replacement for
 # |version| and |release|, also used in various other places throughout the
 # built documents.
 #
 # The short X.Y version.
 root = os.path.dirname(os.path.dirname(__file__))
 sys.path.append(root)
-version = __version__
+version = kallithea.__version__
 # The full version, including alpha/beta/rc tags.
-release = __version__
+release = kallithea.__version__
 
 # The language for content autogenerated by Sphinx. Refer to documentation
 # for a list of supported languages.
 #language = None
 
 # There are two options for replacing |today|: either, you set today to some
 # non-false value, then it is used:
 #today = ''
 # Else, today_fmt is used as the format for a strftime call.
 #today_fmt = '%B %d, %Y'
 
 # List of patterns, relative to source directory, that match files and
 # directories to ignore when looking for source files.
 exclude_patterns = ['_build']
 
 # The reST default role (used for this markup: `text`) to use for all documents.
 #default_role = None
 
 # If true, '()' will be appended to :func: etc. cross-reference text.
 #add_function_parentheses = True
 
 # If true, the current module name will be prepended to all description
 # unit titles (such as .. function::).
 #add_module_names = True
 
 # If true, sectionauthor and moduleauthor directives will be shown in the
 # output. They are ignored by default.
 #show_authors = False
 
 # The name of the Pygments (syntax highlighting) style to use.
 pygments_style = 'sphinx'
 highlight_language = 'none'
 
 # A list of ignored prefixes for module index sorting.
 #modindex_common_prefix = []
 
 
 # -- Options for HTML output ---------------------------------------------------
 
 # The theme to use for HTML and HTML Help pages.  See the documentation for
 # a list of builtin themes.
 html_theme = 'nature'
 
 # Theme options are theme-specific and customize the look and feel of a theme
 # further.  For a list of options available for each theme, see the
 # documentation.
 #html_theme_options = {}
 
 # Add any paths that contain custom themes here, relative to this directory.
 html_theme_path = ['theme']
 
 # The name for this set of Sphinx documents.  If None, it defaults to
 # "<project> v<release> documentation".
 #html_title = None
 
 # A shorter title for the navigation bar.  Default is the same as html_title.
 #html_short_title = None
 
 # The name of an image file (relative to this directory) to place at the top
 # of the sidebar.
 #html_logo = None
 
 # The name of an image file (within the static path) to use as favicon of the
 # docs.  This file should be a Windows icon file (.ico) being 16x16 or 32x32
 # pixels large.
 #html_favicon = None
 
 # Add any paths that contain custom static files (such as style sheets) here,
 # relative to this directory. They are copied after the builtin static files,
 # so a file named "default.css" will overwrite the builtin "default.css".
 #html_static_path = ['_static']
 
 # If not '', a 'Last updated on:' timestamp is inserted at every page bottom,
 # using the given strftime format.
 #html_last_updated_fmt = '%b %d, %Y'
 
 # If true, SmartyPants will be used to convert quotes and dashes to
 # typographically correct entities.
 #html_use_smartypants = True
 
 # Custom sidebar templates, maps document names to template names.
 #html_sidebars = {}
 
 # Additional templates that should be rendered to pages, maps page names to
 # template names.
 #html_additional_pages = {}
 
 # If false, no module index is generated.
 #html_domain_indices = True
 
 # If false, no index is generated.
 #html_use_index = True
 
 # If true, the index is split into individual pages for each letter.
 #html_split_index = False
 
 # If true, links to the reST sources are added to the pages.
 #html_show_sourcelink = True
 
 # If true, "Created using Sphinx" is shown in the HTML footer. Default is True.
 #html_show_sphinx = True
 
 # If true, "(C) Copyright ..." is shown in the HTML footer. Default is True.
 #html_show_copyright = True
 
 # If true, an OpenSearch description file will be output, and all pages will
 # contain a <link> tag referring to it.  The value of this option must be the
 # base URL from which the finished HTML is served.
 #html_use_opensearch = ''
 
 # This is the file name suffix for HTML files (e.g. ".xhtml").
 #html_file_suffix = None
 
 # Output file base name for HTML help builder.
 htmlhelp_basename = 'Kallithea-docs'
 
 
 # -- Options for LaTeX output --------------------------------------------------
 
 # The paper size ('letter' or 'a4').
 #latex_paper_size = 'letter'
 
 # The font size ('10pt', '11pt' or '12pt').
 #latex_font_size = '10pt'
 
 # Grouping the document tree into LaTeX files. List of tuples
 # (source start file, target name, title, author, documentclass [howto/manual]).
 latex_documents = [
   ('index', 'Kallithea.tex', 'Kallithea Documentation',
    'Kallithea Developers', 'manual'),
 ]
 
 # The name of an image file (relative to this directory) to place at the top of
 # the title page.
 #latex_logo = None
 
 # For "manual" documents, if this is true, then toplevel headings are parts,
 # not chapters.
 #latex_use_parts = False
 
 # If true, show page references after internal links.
 #latex_show_pagerefs = False
 
 # If true, show URL addresses after external links.
 #latex_show_urls = False
 
 # Additional stuff for the LaTeX preamble.
 #latex_preamble = ''
 
 # Documents to append as an appendix to all manuals.
 #latex_appendices = []
 
 # If false, no module index is generated.
 #latex_domain_indices = True
 
 
 # -- Options for manual page output --------------------------------------------
 
 # One entry per manual page. List of tuples
 # (source start file, name, description, authors, manual section).
 man_pages = [
     ('index', 'kallithea', 'Kallithea Documentation',
      ['Kallithea Developers'], 1)
 ]
 
 
 # Example configuration for intersphinx: refer to the Python standard library.
 intersphinx_mapping = {'http://docs.python.org/': None}

kallithea/lib/base.py

@@ -1,649 +1,649 b''
 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # 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.lib.base
 ~~~~~~~~~~~~~~~~~~
 
 The base Controller API
 Provides the BaseController class for subclassing. And usage in different
 controllers
 
 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: Oct 06, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 
 import base64
 import datetime
 import logging
 import traceback
 import warnings
 
 import decorator
 import paste.auth.basic
 import paste.httpexceptions
 import paste.httpheaders
 import webob.exc
 from tg import TGController, config, render_template, request, response, session
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 
-from kallithea import BACKENDS, __version__
+import kallithea
 from kallithea.config.routing import url
 from kallithea.lib import auth_modules, ext_json
 from kallithea.lib.auth import AuthUser, HasPermissionAnyMiddleware
 from kallithea.lib.exceptions import UserCreationError
 from kallithea.lib.utils import get_repo_slug, is_valid_repo
 from kallithea.lib.utils2 import AttributeDict, asbool, ascii_bytes, safe_int, safe_str, set_hook_environment
 from kallithea.lib.vcs.exceptions import ChangesetDoesNotExistError, EmptyRepositoryError, RepositoryError
 from kallithea.model import meta
 from kallithea.model.db import PullRequest, Repository, Setting, User
 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
     client, and each successive proxy that passed the request adding the IP
     address where it received the request from.
 
     :param ip:
     """
     if ',' in ip:
         _ips = ip.split(',')
         _first_ip = _ips[0].strip()
         log.debug('Got multiple IPs %s, using %s', ','.join(_ips), _first_ip)
         return _first_ip
     return ip
 
 
 def _get_ip_addr(environ):
     proxy_key = 'HTTP_X_REAL_IP'
     proxy_key2 = 'HTTP_X_FORWARDED_FOR'
     def_key = 'REMOTE_ADDR'
 
     ip = environ.get(proxy_key)
     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_path_info(environ):
     """Return PATH_INFO from environ ... using tg.original_request if available.
 
     In Python 3 WSGI, PATH_INFO is a unicode str, but kind of contains encoded
     bytes. The code points are guaranteed to only use the lower 8 bit bits, and
     encoding the string with the 1:1 encoding latin1 will give the
     corresponding byte string ... which then can be decoded to proper unicode.
     """
     org_req = environ.get('tg.original_request')
     if org_req is not None:
         environ = org_req.environ
     return safe_str(environ['PATH_INFO'].encode('latin1'))
 
 
 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
 
     auth_user = AuthUser.make(dbuser=user, is_external_auth=is_external_auth, ip_addr=ip_addr)
     if auth_user is None:
         return None
 
     user.update_lastlogin()
     meta.Session().commit()
 
     # Start new session to prevent session fixation attacks.
     session.invalidate()
     session['authuser'] = cookie = auth_user.to_cookie()
 
     # If they want to be remembered, update the cookie.
     # NOTE: Assumes that beaker defaults to browser session cookie.
     if remember:
         t = datetime.datetime.now() + datetime.timedelta(days=365)
         session._set_cookie_expires(t)
 
     session.save()
 
     log.info('user %s is now authenticated and stored in '
              'session, session attrs %s', user.username, cookie)
 
     # dumps session attrs back to cookie
     session._update_cookie_out()
 
     return auth_user
 
 
 class BasicAuth(paste.auth.basic.AuthBasicAuthenticator):
 
     def __init__(self, realm, authfunc, auth_http_code=None):
         self.realm = realm
         self.authfunc = authfunc
         self._rc_auth_http_code = auth_http_code
 
     def build_authentication(self, environ):
         head = paste.httpheaders.WWW_AUTHENTICATE.tuples('Basic realm="%s"' % self.realm)
         # Consume the whole body before sending a response
         try:
             request_body_size = int(environ.get('CONTENT_LENGTH', 0))
         except (ValueError):
             request_body_size = 0
         environ['wsgi.input'].read(request_body_size)
         if self._rc_auth_http_code and self._rc_auth_http_code == '403':
             # return 403 if alternative http return code is specified in
             # Kallithea config
             return paste.httpexceptions.HTTPForbidden(headers=head)
         return paste.httpexceptions.HTTPUnauthorized(headers=head)
 
     def authenticate(self, environ):
         authorization = paste.httpheaders.AUTHORIZATION(environ)
         if not authorization:
             return self.build_authentication(environ)
         (authmeth, auth) = authorization.split(' ', 1)
         if 'basic' != authmeth.lower():
             return self.build_authentication(environ)
         auth = safe_str(base64.b64decode(auth.strip()))
         _parts = auth.split(':', 1)
         if len(_parts) == 2:
             username, password = _parts
             if self.authfunc(username, password, environ) is not None:
                 return username
         return self.build_authentication(environ)
 
     __call__ = authenticate
 
 
 class BaseVCSController(object):
     """Base controller for handling Mercurial/Git protocol requests
     (coming from a VCS client, and not a browser).
     """
 
     scm_alias = None # 'hg' / 'git'
 
     def __init__(self, application, config):
         self.application = application
         self.config = config
         # base path of repo locations
         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()
 
     def _authorize(self, environ, action, repo_name, ip_addr):
         """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()
         default_authuser = AuthUser.make(dbuser=default_user, ip_addr=ip_addr)
         if default_authuser is None:
             log.debug('No anonymous access at all') # move on to proper user auth
         else:
             if self._check_permission(action, default_authuser, repo_name):
                 return default_authuser, None
             log.debug('Not authorized to access this repository as anonymous user')
 
         username = None
         #==============================================================
         # DEFAULT PERM FAILED OR ANONYMOUS ACCESS IS DISABLED SO WE
         # NEED TO AUTHENTICATE AND ASK FOR AUTH USER PERMISSIONS
         #==============================================================
 
         # try to auth based on environ, container auth methods
         log.debug('Running PRE-AUTH for container based authentication')
         pre_auth = auth_modules.authenticate('', '', environ)
         if pre_auth is not None and pre_auth.get('username'):
             username = pre_auth['username']
         log.debug('PRE-AUTH got %s as username', username)
 
         # If not authenticated by the container, running basic auth
         if not username:
             self.authenticate.realm = self.config['realm']
             result = self.authenticate(environ)
             if isinstance(result, str):
                 paste.httpheaders.AUTH_TYPE.update(environ, 'basic')
                 paste.httpheaders.REMOTE_USER.update(environ, result)
                 username = result
             else:
                 return None, result.wsgi_application
 
         #==============================================================
         # CHECK PERMISSIONS FOR THIS REQUEST USING GIVEN USERNAME
         #==============================================================
         try:
             user = User.get_by_username_or_email(username)
         except Exception:
             log.error(traceback.format_exc())
             return None, webob.exc.HTTPInternalServerError()
 
         authuser = AuthUser.make(dbuser=user, ip_addr=ip_addr)
         if authuser is None:
             return None, webob.exc.HTTPForbidden()
         if not self._check_permission(action, authuser, repo_name):
             return None, webob.exc.HTTPForbidden()
 
         return user, None
 
     def _handle_request(self, environ, start_response):
         raise NotImplementedError()
 
     def _check_permission(self, action, authuser, repo_name):
         """
         :param action: 'push' or 'pull'
         :param user: `AuthUser` instance
         :param repo_name: repository name
         """
         if action == 'push':
             if not HasPermissionAnyMiddleware('repository.write',
                                               'repository.admin')(authuser,
                                                                   repo_name):
                 return False
 
         elif action == 'pull':
             #any other action need at least read permission
             if not HasPermissionAnyMiddleware('repository.read',
                                               'repository.write',
                                               'repository.admin')(authuser,
                                                                   repo_name):
                 return False
 
         else:
             assert False, action
 
         return True
 
     def _get_ip_addr(self, environ):
         return _get_ip_addr(environ)
 
     def __call__(self, environ, start_response):
         try:
             # try parsing a request for this VCS - if it fails, call the wrapped app
             parsed_request = self.parse_request(environ)
             if parsed_request is None:
                 return self.application(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 webob.exc.HTTPNotFound()
 
             if parsed_request.action is None:
                 # Note: the client doesn't get the helpful error message
                 raise webob.exc.HTTPBadRequest('Unable to detect pull/push action for %r! Are you using a nonstandard command or client?' % parsed_request.repo_name)
 
             #======================================================================
             # CHECK PERMISSIONS
             #======================================================================
             ip_addr = self._get_ip_addr(environ)
             user, response_app = self._authorize(environ, parsed_request.action, parsed_request.repo_name, ip_addr)
             if response_app is not None:
                 return response_app(environ, start_response)
 
             #======================================================================
             # REQUEST HANDLING
             #======================================================================
             set_hook_environment(user.username, ip_addr,
                 parsed_request.repo_name, self.scm_alias, parsed_request.action)
 
             try:
                 log.info('%s action on %s repo "%s" by "%s" from %s',
                          parsed_request.action, self.scm_alias, parsed_request.repo_name, user.username, ip_addr)
                 app = self._make_app(parsed_request)
                 return app(environ, start_response)
             except Exception:
                 log.error(traceback.format_exc())
                 raise webob.exc.HTTPInternalServerError()
 
         except webob.exc.HTTPException as e:
             return e(environ, start_response)
 
 
 class BaseController(TGController):
 
     def _before(self, *args, **kwargs):
         """
         _before is called before controller methods and after __call__
         """
         if request.needs_csrf_check:
             # CSRF protection: Whenever a request has ambient authority (whether
             # through a session cookie or its origin IP address), it must include
             # the correct token, unless the HTTP method is GET or HEAD (and thus
             # guaranteed to be side effect free. In practice, the only situation
             # where we allow side effects without ambient authority is when the
             # authority comes from an API key; and that is handled above.
             from kallithea.lib import helpers as h
             token = request.POST.get(h.session_csrf_secret_name)
             if not token or token != h.session_csrf_secret_token():
                 log.error('CSRF check failed')
                 raise webob.exc.HTTPForbidden()
 
-        c.kallithea_version = __version__
+        c.kallithea_version = kallithea.__version__
         rc_config = Setting.get_app_settings()
 
         # Visual options
         c.visual = AttributeDict({})
 
         ## DB stored
         c.visual.show_public_icon = asbool(rc_config.get('show_public_icon'))
         c.visual.show_private_icon = asbool(rc_config.get('show_private_icon'))
         c.visual.stylify_metalabels = asbool(rc_config.get('stylify_metalabels'))
         c.visual.page_size = safe_int(rc_config.get('dashboard_items', 100))
         c.visual.admin_grid_items = safe_int(rc_config.get('admin_grid_items', 100))
         c.visual.repository_fields = asbool(rc_config.get('repository_fields'))
         c.visual.show_version = asbool(rc_config.get('show_version'))
         c.visual.use_gravatar = asbool(rc_config.get('use_gravatar'))
         c.visual.gravatar_url = rc_config.get('gravatar_url')
 
         c.ga_code = rc_config.get('ga_code')
         # TODO: replace undocumented backwards compatibility hack with db upgrade and rename ga_code
         if c.ga_code and '<' not in c.ga_code:
             c.ga_code = '''<script type="text/javascript">
                 var _gaq = _gaq || [];
                 _gaq.push(['_setAccount', '%s']);
                 _gaq.push(['_trackPageview']);
 
                 (function() {
                     var ga = document.createElement('script'); ga.type = 'text/javascript'; ga.async = true;
                     ga.src = ('https:' == document.location.protocol ? 'https://ssl' : 'http://www') + '.google-analytics.com/ga.js';
                     var s = document.getElementsByTagName('script')[0]; s.parentNode.insertBefore(ga, s);
                     })();
             </script>''' % c.ga_code
         c.site_name = rc_config.get('title')
         c.clone_uri_tmpl = rc_config.get('clone_uri_tmpl') or Repository.DEFAULT_CLONE_URI
         c.clone_ssh_tmpl = rc_config.get('clone_ssh_tmpl') or Repository.DEFAULT_CLONE_SSH
 
         ## INI stored
         c.visual.allow_repo_location_change = asbool(config.get('allow_repo_location_change', True))
         c.visual.allow_custom_hooks_settings = asbool(config.get('allow_custom_hooks_settings', True))
         c.ssh_enabled = asbool(config.get('ssh_enabled', False))
 
         c.instance_id = config.get('instance_id')
         c.issues_url = config.get('bugtracker', url('issues_url'))
         # END CONFIG VARS
 
         c.repo_name = get_repo_slug(request)  # can be empty
-        c.backends = list(BACKENDS)
+        c.backends = list(kallithea.BACKENDS)
 
         self.cut_off_limit = safe_int(config.get('cut_off_limit'))
 
         c.my_pr_count = PullRequest.query(reviewer_id=request.authuser.user_id, include_closed=False).count()
 
         self.scm_model = ScmModel()
 
     @staticmethod
     def _determine_auth_user(session_authuser, ip_addr):
         """
         Create an `AuthUser` object given the API key/bearer token
         (if any) and the value of the authuser session cookie.
         Returns None if no valid user is found (like not active or no access for IP).
         """
 
         # Authenticate by session cookie
         # In ancient login sessions, 'authuser' may not be a dict.
         # In that case, the user will have to log in again.
         # v0.3 and earlier included an 'is_authenticated' key; if present,
         # this must be True.
         if isinstance(session_authuser, dict) and session_authuser.get('is_authenticated', True):
             return AuthUser.from_cookie(session_authuser, ip_addr=ip_addr)
 
         # Authenticate by auth_container plugin (if enabled)
         if any(
             plugin.is_container_auth
             for plugin in auth_modules.get_auth_plugins()
         ):
             try:
                 user_info = auth_modules.authenticate('', '', request.environ)
             except UserCreationError as e:
                 from kallithea.lib import helpers as h
                 h.flash(e, 'error', logf=log.error)
             else:
                 if user_info is not None:
                     username = user_info['username']
                     user = User.get_by_username(username, case_insensitive=True)
                     return log_in_user(user, remember=False, is_external_auth=True, ip_addr=ip_addr)
 
         # User is default user (if active) or anonymous
         default_user = User.get_default_user()
         authuser = AuthUser.make(dbuser=default_user, ip_addr=ip_addr)
         if authuser is None: # fall back to anonymous
             authuser = AuthUser(dbuser=default_user) # TODO: somehow use .make?
         return authuser
 
     @staticmethod
     def _basic_security_checks():
         """Perform basic security/sanity checks before processing the request."""
 
         # Only allow the following HTTP request methods.
         if request.method not in ['GET', 'HEAD', 'POST']:
             raise webob.exc.HTTPMethodNotAllowed()
 
         # Also verify the _method override - no longer allowed.
         if request.params.get('_method') is None:
             pass # no override, no problem
         else:
             raise webob.exc.HTTPMethodNotAllowed()
 
         # Make sure CSRF token never appears in the URL. If so, invalidate it.
         from kallithea.lib import helpers as h
         if h.session_csrf_secret_name in request.GET:
             log.error('CSRF key leak detected')
             session.pop(h.session_csrf_secret_name, None)
             session.save()
             h.flash(_('CSRF token leak has been detected - all form tokens have been expired'),
                     category='error')
 
         # WebOb already ignores request payload parameters for anything other
         # than POST/PUT, but double-check since other Kallithea code relies on
         # this assumption.
         if request.method not in ['POST', 'PUT'] and request.POST:
             log.error('%r request with payload parameters; WebOb should have stopped this', request.method)
             raise webob.exc.HTTPBadRequest()
 
     def __call__(self, environ, context):
         try:
             ip_addr = _get_ip_addr(environ)
             self._basic_security_checks()
 
             api_key = request.GET.get('api_key')
             try:
                 # Request.authorization may raise ValueError on invalid input
                 type, params = request.authorization
             except (ValueError, TypeError):
                 pass
             else:
                 if type.lower() == 'bearer':
                     api_key = params # bearer token is an api key too
 
             if api_key is None:
                 authuser = self._determine_auth_user(
                     session.get('authuser'),
                     ip_addr=ip_addr,
                 )
                 needs_csrf_check = request.method not in ['GET', 'HEAD']
 
             else:
                 dbuser = User.get_by_api_key(api_key)
                 if dbuser is None:
                     log.info('No db user found for authentication with API key ****%s from %s',
                              api_key[-4:], ip_addr)
                 authuser = AuthUser.make(dbuser=dbuser, is_external_auth=True, ip_addr=ip_addr)
                 needs_csrf_check = False # API key provides CSRF protection
 
             if authuser is None:
                 log.info('No valid user found')
                 raise webob.exc.HTTPForbidden()
 
             # set globals for auth user
             request.authuser = authuser
             request.ip_addr = ip_addr
             request.needs_csrf_check = needs_csrf_check
 
             log.info('IP: %s User: %s Request: %s',
                 request.ip_addr, request.authuser,
                 get_path_info(environ),
             )
             return super(BaseController, self).__call__(environ, context)
         except webob.exc.HTTPException as e:
             return e
 
 
 class BaseRepoController(BaseController):
     """
     Base class for controllers responsible for loading all needed data for
     repository loaded items are
 
     c.db_repo_scm_instance: instance of scm repository
     c.db_repo: instance of db
     c.repository_followers: number of followers
     c.repository_forks: number of forks
     c.repository_following: weather the current user is following the current repo
     """
 
     def _before(self, *args, **kwargs):
         super(BaseRepoController, self)._before(*args, **kwargs)
         if c.repo_name:  # extracted from request by base-base BaseController._before
             _dbr = Repository.get_by_repo_name(c.repo_name)
             if not _dbr:
                 return
 
             log.debug('Found repository in database %s with state `%s`',
                       _dbr, _dbr.repo_state)
             route = getattr(request.environ.get('routes.route'), 'name', '')
 
             # allow to delete repos that are somehow damages in filesystem
             if route in ['delete_repo']:
                 return
 
             if _dbr.repo_state in [Repository.STATE_PENDING]:
                 if route in ['repo_creating_home']:
                     return
                 check_url = url('repo_creating_home', repo_name=c.repo_name)
                 raise webob.exc.HTTPFound(location=check_url)
 
             dbr = c.db_repo = _dbr
             c.db_repo_scm_instance = c.db_repo.scm_instance
             if c.db_repo_scm_instance is None:
                 log.error('%s this repository is present in database but it '
                           'cannot be created as an scm instance', c.repo_name)
                 from kallithea.lib import helpers as h
                 h.flash(_('Repository not found in the filesystem'),
                         category='error')
                 raise webob.exc.HTTPNotFound()
 
             # some globals counter for menu
             c.repository_followers = self.scm_model.get_followers(dbr)
             c.repository_forks = self.scm_model.get_forks(dbr)
             c.repository_pull_requests = self.scm_model.get_pull_requests(dbr)
             c.repository_following = self.scm_model.is_following_repo(
                                     c.repo_name, request.authuser.user_id)
 
     @staticmethod
     def _get_ref_rev(repo, ref_type, ref_name, returnempty=False):
         """
         Safe way to get changeset. If error occurs show error.
         """
         from kallithea.lib import helpers as h
         try:
             return repo.scm_instance.get_ref_revision(ref_type, ref_name)
         except EmptyRepositoryError as e:
             if returnempty:
                 return repo.scm_instance.EMPTY_CHANGESET
             h.flash(_('There are no changesets yet'), category='error')
             raise webob.exc.HTTPNotFound()
         except ChangesetDoesNotExistError as e:
             h.flash(_('Changeset for %s %s not found in %s') %
                               (ref_type, ref_name, repo.repo_name),
                     category='error')
             raise webob.exc.HTTPNotFound()
         except RepositoryError as e:
             log.error(traceback.format_exc())
             h.flash(e, category='error')
             raise webob.exc.HTTPBadRequest()
 
 
 @decorator.decorator
 def jsonify(func, *args, **kwargs):
     """Action decorator that formats output for JSON
 
     Given a function that will return content, this decorator will turn
     the result into JSON, with a content-type of 'application/json' and
     output it.
     """
     response.headers['Content-Type'] = 'application/json; charset=utf-8'
     data = func(*args, **kwargs)
     if isinstance(data, (list, tuple)):
         # A JSON list response is syntactically valid JavaScript and can be
         # loaded and executed as JavaScript by a malicious third-party site
         # using <script>, which can lead to cross-site data leaks.
         # JSON responses should therefore be scalars or objects (i.e. Python
         # dicts), because a JSON object is a syntax error if intepreted as JS.
         msg = "JSON responses with Array envelopes are susceptible to " \
               "cross-site data leak attacks, see " \
               "https://web.archive.org/web/20120519231904/http://wiki.pylonshq.com/display/pylonsfaq/Warnings"
         warnings.warn(msg, Warning, 2)
         log.warning(msg)
     log.debug("Returning JSON wrapped action output")
     return ascii_bytes(ext_json.dumps(data))
 
 @decorator.decorator
 def IfSshEnabled(func, *args, **kwargs):
     """Decorator for functions that can only be called if SSH access is enabled.
 
     If SSH access is disabled in the configuration file, HTTPNotFound is raised.
     """
     if not c.ssh_enabled:
         from kallithea.lib import helpers as h
         h.flash(_("SSH access is disabled."), category='warning')
         raise webob.exc.HTTPNotFound()
     return func(*args, **kwargs)

kallithea/lib/hooks.py

@@ -1,413 +1,406 b''
 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # 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.lib.hooks
 ~~~~~~~~~~~~~~~~~~~
 
 Hooks run by 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: Aug 6, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 
 import os
 import sys
 import time
 
 import mercurial.scmutil
 
+import kallithea
 from kallithea.lib import helpers as h
 from kallithea.lib.exceptions import UserCreationError
 from kallithea.lib.utils import action_logger, make_ui
 from kallithea.lib.utils2 import HookEnvironmentError, ascii_str, get_hook_environment, safe_bytes, safe_str
 from kallithea.lib.vcs.backends.base import EmptyChangeset
 from kallithea.model.db import Repository, User
 
 
 def _get_scm_size(alias, root_path):
     if not alias.startswith('.'):
         alias += '.'
 
     size_scm, size_root = 0, 0
     for path, dirs, files in os.walk(root_path):
         if path.find(alias) != -1:
             for f in files:
                 try:
                     size_scm += os.path.getsize(os.path.join(path, f))
                 except OSError:
                     pass
         else:
             for f in files:
                 try:
                     size_root += os.path.getsize(os.path.join(path, f))
                 except OSError:
                     pass
 
     size_scm_f = h.format_byte_size(size_scm)
     size_root_f = h.format_byte_size(size_root)
     size_total_f = h.format_byte_size(size_root + size_scm)
 
     return size_scm_f, size_root_f, size_total_f
 
 
 def repo_size(ui, repo, hooktype=None, **kwargs):
     """Show size of Mercurial repository.
 
     Called as Mercurial hook changegroup.repo_size after push.
     """
     size_hg_f, size_root_f, size_total_f = _get_scm_size('.hg', safe_str(repo.root))
 
     last_cs = repo[len(repo) - 1]
 
     msg = ('Repository size .hg: %s Checkout: %s Total: %s\n'
            'Last revision is now r%s:%s\n') % (
         size_hg_f, size_root_f, size_total_f, last_cs.rev(), ascii_str(last_cs.hex())[:12]
     )
     ui.status(safe_bytes(msg))
 
 
 def log_pull_action(ui, repo, **kwargs):
     """Logs user last pull action
 
     Called as Mercurial hook outgoing.pull_logger or from Kallithea before invoking Git.
 
     Does *not* use the action from the hook environment but is always 'pull'.
     """
     ex = get_hook_environment()
 
     user = User.get_by_username(ex.username)
     action = 'pull'
     action_logger(user, action, ex.repository, ex.ip, commit=True)
     # extension hook call
-    from kallithea import EXTENSIONS
-    callback = getattr(EXTENSIONS, 'PULL_HOOK', None)
+    callback = getattr(kallithea.EXTENSIONS, 'PULL_HOOK', None)
     if callable(callback):
         kw = {}
         kw.update(ex)
         callback(**kw)
 
 
 def log_push_action(ui, repo, node, node_last, **kwargs):
     """
     Register that changes have been added to the repo - log the action *and* invalidate caches.
     Note: This hook is not only logging, but also the side effect invalidating
     caches! The function should perhaps be renamed.
 
     Called as Mercurial hook changegroup.kallithea_log_push_action .
 
     The pushed changesets is given by the revset 'node:node_last'.
     """
     revs = [ascii_str(repo[r].hex()) for r in mercurial.scmutil.revrange(repo, [b'%s:%s' % (node, node_last)])]
     process_pushed_raw_ids(revs)
 
 
 def process_pushed_raw_ids(revs):
     """
     Register that changes have been added to the repo - log the action *and* invalidate caches.
 
     Called from Mercurial changegroup.kallithea_log_push_action calling hook log_push_action,
     or from the Git post-receive hook calling handle_git_post_receive ...
     or from scm _handle_push.
     """
     ex = get_hook_environment()
 
     action = '%s:%s' % (ex.action, ','.join(revs))
     action_logger(ex.username, action, ex.repository, ex.ip, commit=True)
 
     from kallithea.model.scm import ScmModel
     ScmModel().mark_for_invalidation(ex.repository)
 
     # extension hook call
-    from kallithea import EXTENSIONS
-    callback = getattr(EXTENSIONS, 'PUSH_HOOK', None)
+    callback = getattr(kallithea.EXTENSIONS, 'PUSH_HOOK', None)
     if callable(callback):
         kw = {'pushed_revs': revs}
         kw.update(ex)
         callback(**kw)
 
 
 def log_create_repository(repository_dict, created_by, **kwargs):
     """
     Post create repository Hook.
 
     :param repository: dict dump of repository object
     :param created_by: username who created repository
 
     available keys of repository_dict:
 
      'repo_type',
      'description',
      'private',
      'created_on',
      'enable_downloads',
      'repo_id',
      'owner_id',
      'enable_statistics',
      'clone_uri',
      'fork_id',
      'group_id',
      'repo_name'
 
     """
-    from kallithea import EXTENSIONS
-    callback = getattr(EXTENSIONS, 'CREATE_REPO_HOOK', None)
+    callback = getattr(kallithea.EXTENSIONS, 'CREATE_REPO_HOOK', None)
     if callable(callback):
         kw = {}
         kw.update(repository_dict)
         kw.update({'created_by': created_by})
         kw.update(kwargs)
         callback(**kw)
 
 
 def check_allowed_create_user(user_dict, created_by, **kwargs):
     # pre create hooks
-    from kallithea import EXTENSIONS
-    callback = getattr(EXTENSIONS, 'PRE_CREATE_USER_HOOK', None)
+    callback = getattr(kallithea.EXTENSIONS, 'PRE_CREATE_USER_HOOK', None)
     if callable(callback):
         allowed, reason = callback(created_by=created_by, **user_dict)
         if not allowed:
             raise UserCreationError(reason)
 
 
 def log_create_user(user_dict, created_by, **kwargs):
     """
     Post create user Hook.
 
     :param user_dict: dict dump of user object
 
     available keys for user_dict:
 
      'username',
      'full_name_or_username',
      'full_contact',
      'user_id',
      'name',
      'firstname',
      'short_contact',
      'admin',
      'lastname',
      'ip_addresses',
      'ldap_dn',
      'email',
      'api_key',
      'last_login',
      'full_name',
      'active',
      'password',
      'emails',
 
     """
-    from kallithea import EXTENSIONS
-    callback = getattr(EXTENSIONS, 'CREATE_USER_HOOK', None)
+    callback = getattr(kallithea.EXTENSIONS, 'CREATE_USER_HOOK', None)
     if callable(callback):
         callback(created_by=created_by, **user_dict)
 
 
 def log_create_pullrequest(pullrequest_dict, created_by, **kwargs):
     """
     Post create pull request hook.
 
     :param pullrequest_dict: dict dump of pull request object
     """
-    from kallithea import EXTENSIONS
-    callback = getattr(EXTENSIONS, 'CREATE_PULLREQUEST_HOOK', None)
+    callback = getattr(kallithea.EXTENSIONS, 'CREATE_PULLREQUEST_HOOK', None)
     if callable(callback):
         return callback(created_by=created_by, **pullrequest_dict)
 
     return 0
 
 def log_delete_repository(repository_dict, deleted_by, **kwargs):
     """
     Post delete repository Hook.
 
     :param repository: dict dump of repository object
     :param deleted_by: username who deleted the repository
 
     available keys of repository_dict:
 
      'repo_type',
      'description',
      'private',
      'created_on',
      'enable_downloads',
      'repo_id',
      'owner_id',
      'enable_statistics',
      'clone_uri',
      'fork_id',
      'group_id',
      'repo_name'
 
     """
-    from kallithea import EXTENSIONS
-    callback = getattr(EXTENSIONS, 'DELETE_REPO_HOOK', None)
+    callback = getattr(kallithea.EXTENSIONS, 'DELETE_REPO_HOOK', None)
     if callable(callback):
         kw = {}
         kw.update(repository_dict)
         kw.update({'deleted_by': deleted_by,
                    'deleted_on': time.time()})
         kw.update(kwargs)
         callback(**kw)
 
 
 def log_delete_user(user_dict, deleted_by, **kwargs):
     """
     Post delete user Hook.
 
     :param user_dict: dict dump of user object
 
     available keys for user_dict:
 
      'username',
      'full_name_or_username',
      'full_contact',
      'user_id',
      'name',
      'firstname',
      'short_contact',
      'admin',
      'lastname',
      'ip_addresses',
      'ldap_dn',
      'email',
      'api_key',
      'last_login',
      'full_name',
      'active',
      'password',
      'emails',
 
     """
-    from kallithea import EXTENSIONS
-    callback = getattr(EXTENSIONS, 'DELETE_USER_HOOK', None)
+    callback = getattr(kallithea.EXTENSIONS, 'DELETE_USER_HOOK', None)
     if callable(callback):
         callback(deleted_by=deleted_by, **user_dict)
 
 
 def _hook_environment(repo_path):
     """
     Create a light-weight environment for stand-alone scripts and return an UI and the
     db repository.
 
     Git hooks are executed as subprocess of Git while Kallithea is waiting, and
     they thus need enough info to be able to create an app environment and
     connect to the database.
     """
     import paste.deploy
 
     import kallithea.config.application
 
     extras = get_hook_environment()
 
     path_to_ini_file = extras['config']
     config = paste.deploy.appconfig('config:' + path_to_ini_file)
     #logging.config.fileConfig(ini_file_path) # Note: we are in a different process - don't use configured logging
     kallithea.config.application.make_app(config.global_conf, **config.local_conf)
 
     # fix if it's not a bare repo
     if repo_path.endswith(os.sep + '.git'):
         repo_path = repo_path[:-5]
 
     repo = Repository.get_by_full_path(repo_path)
     if not repo:
         raise OSError('Repository %s not found in database' % repo_path)
 
     baseui = make_ui()
     return baseui, repo
 
 
 def handle_git_pre_receive(repo_path, git_stdin_lines):
     """Called from Git pre-receive hook.
     The returned value is used as hook exit code and must be 0.
     """
     # Currently unused. TODO: remove?
     return 0
 
 
 def handle_git_post_receive(repo_path, git_stdin_lines):
     """Called from Git post-receive hook.
     The returned value is used as hook exit code and must be 0.
     """
     try:
         baseui, repo = _hook_environment(repo_path)
     except HookEnvironmentError as e:
         sys.stderr.write("Skipping Kallithea Git post-recieve hook %r.\nGit was apparently not invoked by Kallithea: %s\n" % (sys.argv[0], e))
         return 0
 
     # the post push hook should never use the cached instance
     scm_repo = repo.scm_instance_no_cache()
 
     rev_data = []
     for l in git_stdin_lines:
         old_rev, new_rev, ref = l.strip().split(' ')
         _ref_data = ref.split('/')
         if _ref_data[1] in ['tags', 'heads']:
             rev_data.append({'old_rev': old_rev,
                              'new_rev': new_rev,
                              'ref': ref,
                              'type': _ref_data[1],
                              'name': '/'.join(_ref_data[2:])})
 
     git_revs = []
     for push_ref in rev_data:
         _type = push_ref['type']
         if _type == 'heads':
             if push_ref['old_rev'] == EmptyChangeset().raw_id:
                 # update the symbolic ref if we push new repo
                 if scm_repo.is_empty():
                     scm_repo._repo.refs.set_symbolic_ref(
                         b'HEAD',
                         b'refs/heads/%s' % safe_bytes(push_ref['name']))
 
                 # build exclude list without the ref
                 cmd = ['for-each-ref', '--format=%(refname)', 'refs/heads/*']
                 stdout = scm_repo.run_git_command(cmd)
                 ref = push_ref['ref']
                 heads = [head for head in stdout.splitlines() if head != ref]
                 # now list the git revs while excluding from the list
                 cmd = ['log', push_ref['new_rev'], '--reverse', '--pretty=format:%H']
                 cmd.append('--not')
                 cmd.extend(heads) # empty list is ok
                 stdout = scm_repo.run_git_command(cmd)
                 git_revs += stdout.splitlines()
 
             elif push_ref['new_rev'] == EmptyChangeset().raw_id:
                 # delete branch case
                 git_revs += ['delete_branch=>%s' % push_ref['name']]
             else:
                 cmd = ['log', '%(old_rev)s..%(new_rev)s' % push_ref,
                        '--reverse', '--pretty=format:%H']
                 stdout = scm_repo.run_git_command(cmd)
                 git_revs += stdout.splitlines()
 
         elif _type == 'tags':
             git_revs += ['tag=>%s' % push_ref['name']]
 
     process_pushed_raw_ids(git_revs)
 
     return 0
 
 
 # Almost exactly like Mercurial contrib/hg-ssh:
 def rejectpush(ui, **kwargs):
     """Mercurial hook to be installed as pretxnopen and prepushkey for read-only repos.
     Return value 1 will make the hook fail and reject the push.
     """
     ex = get_hook_environment()
     ui.warn(safe_bytes("Push access to %r denied\n" % ex.repository))
     return 1

kallithea/lib/utils2.py

@@ -1,519 +1,519 b''
 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # 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.lib.utils2
 ~~~~~~~~~~~~~~~~~~~~
 
 Some simple helper functions.
 Note: all these functions should be independent of Kallithea classes, i.e.
 models, controllers, etc.  to prevent import cycles.
 
 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: Jan 5, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 
 import binascii
 import datetime
 import json
 import os
 import re
 import time
 import urllib.parse
 
 import urlobject
 from tg.i18n import ugettext as _
 from tg.i18n import ungettext
 from tg.support.converters import asbool, aslist
 from webhelpers2.text import collapse, remove_formatting, strip_tags
 
+import kallithea
 from kallithea.lib.vcs.utils import ascii_bytes, ascii_str, safe_bytes, safe_str  # re-export
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 
 
 try:
     import pwd
 except ImportError:
     pass
 
 
 # mute pyflakes "imported but unused"
 assert asbool
 assert aslist
 assert ascii_bytes
 assert ascii_str
 assert safe_bytes
 assert safe_str
 assert LazyProperty
 
 
 def convert_line_endings(line, mode):
     """
     Converts a given line  "line end" according to given mode
 
     Available modes are::
         0 - Unix
         1 - Mac
         2 - DOS
 
     :param line: given line to convert
     :param mode: mode to convert to
     :rtype: str
     :return: converted line according to mode
     """
     if mode == 0:
         line = line.replace('\r\n', '\n')
         line = line.replace('\r', '\n')
     elif mode == 1:
         line = line.replace('\r\n', '\r')
         line = line.replace('\n', '\r')
     elif mode == 2:
         line = re.sub("\r(?!\n)|(?<!\r)\n", "\r\n", line)
     return line
 
 
 def detect_mode(line, default):
     """
     Detects line break for given line, if line break couldn't be found
     given default value is returned
 
     :param line: str line
     :param default: default
     :rtype: int
     :return: value of line end on of 0 - Unix, 1 - Mac, 2 - DOS
     """
     if line.endswith('\r\n'):
         return 2
     elif line.endswith('\n'):
         return 0
     elif line.endswith('\r'):
         return 1
     else:
         return default
 
 
 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 age(prevdate, show_short_version=False, now=None):
     """
     turns a datetime into an age string.
     If show_short_version is True, then it will generate a not so accurate but shorter string,
     example: 2days ago, instead of 2 days and 23 hours ago.
 
     :param prevdate: datetime object
     :param show_short_version: if it should approximate the date and return a shorter string
     :rtype: str
     :returns: str words describing age
     """
     now = now or datetime.datetime.now()
     order = ['year', 'month', 'day', 'hour', 'minute', 'second']
     deltas = {}
     future = False
 
     if prevdate > now:
         now, prevdate = prevdate, now
         future = True
     if future:
         prevdate = prevdate.replace(microsecond=0)
     # Get date parts deltas
     from dateutil import relativedelta
     for part in order:
         d = relativedelta.relativedelta(now, prevdate)
         deltas[part] = getattr(d, part + 's')
 
     # Fix negative offsets (there is 1 second between 10:59:59 and 11:00:00,
     # not 1 hour, -59 minutes and -59 seconds)
     for num, length in [(5, 60), (4, 60), (3, 24)]:  # seconds, minutes, hours
         part = order[num]
         carry_part = order[num - 1]
 
         if deltas[part] < 0:
             deltas[part] += length
             deltas[carry_part] -= 1
 
     # Same thing for days except that the increment depends on the (variable)
     # number of days in the month
     month_lengths = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
     if deltas['day'] < 0:
         if prevdate.month == 2 and (prevdate.year % 4 == 0 and
             (prevdate.year % 100 != 0 or prevdate.year % 400 == 0)
         ):
             deltas['day'] += 29
         else:
             deltas['day'] += month_lengths[prevdate.month - 1]
 
         deltas['month'] -= 1
 
     if deltas['month'] < 0:
         deltas['month'] += 12
         deltas['year'] -= 1
 
     # In short version, we want nicer handling of ages of more than a year
     if show_short_version:
         if deltas['year'] == 1:
             # ages between 1 and 2 years: show as months
             deltas['month'] += 12
             deltas['year'] = 0
         if deltas['year'] >= 2:
             # ages 2+ years: round
             if deltas['month'] > 6:
                 deltas['year'] += 1
                 deltas['month'] = 0
 
     # Format the result
     fmt_funcs = {
         'year': lambda d: ungettext('%d year', '%d years', d) % d,
         'month': lambda d: ungettext('%d month', '%d months', d) % d,
         'day': lambda d: ungettext('%d day', '%d days', d) % d,
         'hour': lambda d: ungettext('%d hour', '%d hours', d) % d,
         'minute': lambda d: ungettext('%d minute', '%d minutes', d) % d,
         'second': lambda d: ungettext('%d second', '%d seconds', d) % d,
     }
 
     for i, part in enumerate(order):
         value = deltas[part]
         if value == 0:
             continue
 
         if i < 5:
             sub_part = order[i + 1]
             sub_value = deltas[sub_part]
         else:
             sub_value = 0
 
         if sub_value == 0 or show_short_version:
             if future:
                 return _('in %s') % fmt_funcs[part](value)
             else:
                 return _('%s ago') % fmt_funcs[part](value)
         if future:
             return _('in %s and %s') % (fmt_funcs[part](value),
                 fmt_funcs[sub_part](sub_value))
         else:
             return _('%s and %s ago') % (fmt_funcs[part](value),
                 fmt_funcs[sub_part](sub_value))
 
     return _('just now')
 
 
 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 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:
     """
     from kallithea.lib.vcs.backends.base import BaseRepository, EmptyChangeset
     from kallithea.lib.vcs.exceptions import RepositoryError
     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)
 
 
 # Must match regexp in kallithea/public/js/base.js MentionsAutoComplete()
 # Check char before @ - it must not look like we are in an email addresses.
 # Matching is greedy so we don't have to look beyond the end.
 MENTIONS_REGEX = re.compile(r'(?:^|(?<=[^a-zA-Z0-9]))@([a-zA-Z0-9][-_.a-zA-Z0-9]*[a-zA-Z0-9])')
 
 
 def extract_mentioned_usernames(text):
     r"""
     Returns list of (possible) usernames @mentioned in given text.
 
     >>> extract_mentioned_usernames('@1-2.a_X,@1234 not@not @ddd@not @n @ee @ff @gg, @gg;@hh @n\n@zz,')
     ['1-2.a_X', '1234', 'ddd', 'ee', 'ff', 'gg', 'gg', 'hh', 'zz']
     """
     return MENTIONS_REGEX.findall(text)
 
 
 def extract_mentioned_users(text):
     """ Returns set of actual database Users @mentioned in given text. """
     from kallithea.model.db import User
     result = set()
     for name in extract_mentioned_usernames(text):
         user = User.get_by_username(name, case_insensitive=True)
         if user is not None and not user.is_default_user:
             result.add(user)
     return result
 
 
 class AttributeDict(dict):
     def __getattr__(self, attr):
         return self.get(attr, None)
     __setattr__ = dict.__setitem__
     __delattr__ = dict.__delitem__
 
 
 def obfuscate_url_pw(engine):
     from sqlalchemy.engine import url as sa_url
     from sqlalchemy.exc import ArgumentError
     try:
         _url = sa_url.make_url(engine or '')
     except ArgumentError:
         return engine
     if _url.password:
         _url.password = 'XXXXX'
     return str(_url)
 
 
 class HookEnvironmentError(Exception): pass
 
 
 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.
     """
-    from kallithea import CONFIG
     extras = {
         'ip': ip_addr, # used in log_push/pull_action action_logger
         'username': username,
         'action': action or 'push_local', # used in log_push_action_raw_ids action_logger
         'repository': repo_name,
         'scm': repo_alias, # used to pick hack in log_push_action_raw_ids
-        'config': CONFIG['__file__'], # used by git hook to read config
+        '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.
     """
     from tg import tmpl_context
     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)

kallithea/model/forms.py

@@ -1,564 +1,564 b''
 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # 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/>.
 """
 these are form validation classes
 http://formencode.org/module-formencode.validators.html
 for list of all available validators
 
 we can create our own validators
 
 The table below outlines the options which can be used in a schema in addition to the validators themselves
 pre_validators          []     These validators will be applied before the schema
 chained_validators      []     These validators will be applied after the schema
 allow_extra_fields      False     If True, then it is not an error when keys that aren't associated with a validator are present
 filter_extra_fields     False     If True, then keys that aren't associated with a validator are removed
 if_key_missing          NoDefault If this is given, then any keys that aren't available but are expected will be replaced with this value (and then validated). This does not override a present .if_missing attribute on validators. NoDefault is a special FormEncode class to mean that no default values has been specified and therefore missing keys shouldn't take a default value.
 ignore_key_missing      False     If True, then missing keys will be missing in the result, if the validator doesn't have .if_missing on it already
 
 
 <name> = formencode.validators.<name of validator>
 <name> must equal form name
 list=[1,2,3,4,5]
 for SELECT use formencode.All(OneOf(list), Int())
 
 """
 import logging
 
 import formencode
 from formencode import All
 from tg.i18n import ugettext as _
 
-from kallithea import BACKENDS
+import kallithea
 from kallithea.model import validators as v
 
 
 log = logging.getLogger(__name__)
 
 
 def LoginForm():
     class _LoginForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         username = v.UnicodeString(
             strip=True,
             min=1,
             not_empty=True,
             messages={
                'empty': _('Please enter a login'),
                'tooShort': _('Enter a value %(min)i characters long or more')}
         )
 
         password = v.UnicodeString(
             strip=False,
             min=3,
             not_empty=True,
             messages={
                 'empty': _('Please enter a password'),
                 'tooShort': _('Enter %(min)i characters or more')}
         )
 
         remember = v.StringBoolean(if_missing=False)
 
         chained_validators = [v.ValidAuth()]
     return _LoginForm
 
 
 def PasswordChangeForm(username):
     class _PasswordChangeForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
 
         current_password = v.ValidOldPassword(username)(not_empty=True)
         new_password = All(v.ValidPassword(), v.UnicodeString(strip=False, min=6))
         new_password_confirmation = All(v.ValidPassword(), v.UnicodeString(strip=False, min=6))
 
         chained_validators = [v.ValidPasswordsMatch('new_password',
                                                     'new_password_confirmation')]
     return _PasswordChangeForm
 
 
 def UserForm(edit=False, old_data=None):
     old_data = old_data or {}
 
     class _UserForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         username = All(v.UnicodeString(strip=True, min=1, not_empty=True),
                        v.ValidUsername(edit, old_data))
         if edit:
             new_password = All(
                 v.ValidPassword(),
                 v.UnicodeString(strip=False, min=6, not_empty=False)
             )
             password_confirmation = All(
                 v.ValidPassword(),
                 v.UnicodeString(strip=False, min=6, not_empty=False),
             )
             admin = v.StringBoolean(if_missing=False)
             chained_validators = [v.ValidPasswordsMatch('new_password',
                                                         'password_confirmation')]
         else:
             password = All(
                 v.ValidPassword(),
                 v.UnicodeString(strip=False, min=6, not_empty=True)
             )
             password_confirmation = All(
                 v.ValidPassword(),
                 v.UnicodeString(strip=False, min=6, not_empty=False)
             )
             chained_validators = [v.ValidPasswordsMatch('password',
                                                         'password_confirmation')]
 
         active = v.StringBoolean(if_missing=False)
         firstname = v.UnicodeString(strip=True, min=1, not_empty=False)
         lastname = v.UnicodeString(strip=True, min=1, not_empty=False)
         email = All(v.Email(not_empty=True), v.UniqSystemEmail(old_data))
         extern_name = v.UnicodeString(strip=True, if_missing=None)
         extern_type = v.UnicodeString(strip=True, if_missing=None)
     return _UserForm
 
 
 def UserGroupForm(edit=False, old_data=None, available_members=None):
     old_data = old_data or {}
     available_members = available_members or []
 
     class _UserGroupForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
 
         users_group_name = All(
             v.UnicodeString(strip=True, min=1, not_empty=True),
             v.ValidUserGroup(edit, old_data)
         )
         user_group_description = v.UnicodeString(strip=True, min=1,
                                                  not_empty=False)
 
         users_group_active = v.StringBoolean(if_missing=False)
 
         if edit:
             users_group_members = v.OneOf(
                 available_members, hideList=False, testValueList=True,
                 if_missing=None, not_empty=False
             )
 
     return _UserGroupForm
 
 
 def RepoGroupForm(edit=False, old_data=None, repo_groups=None,
                    can_create_in_root=False):
     old_data = old_data or {}
     repo_groups = repo_groups or []
     repo_group_ids = [rg[0] for rg in repo_groups]
 
     class _RepoGroupForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
 
         group_name = All(v.UnicodeString(strip=True, min=1, not_empty=True),
                          v.SlugifyName(),
                          v.ValidRegex(msg=_('Name must not contain only digits'))(r'(?!^\d+$)^.+$'))
         group_description = v.UnicodeString(strip=True, min=1,
                                             not_empty=False)
         group_copy_permissions = v.StringBoolean(if_missing=False)
 
         if edit:
             # FIXME: do a special check that we cannot move a group to one of
             # its children
             pass
 
         parent_group_id = All(v.CanCreateGroup(can_create_in_root),
                               v.OneOf(repo_group_ids, hideList=False,
                                       testValueList=True,
                                       if_missing=None, not_empty=True),
                               v.Int(min=-1, not_empty=True))
         chained_validators = [v.ValidRepoGroup(edit, old_data)]
 
     return _RepoGroupForm
 
 
 def RegisterForm(edit=False, old_data=None):
     class _RegisterForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         username = All(
             v.ValidUsername(edit, old_data),
             v.UnicodeString(strip=True, min=1, not_empty=True)
         )
         password = All(
             v.ValidPassword(),
             v.UnicodeString(strip=False, min=6, not_empty=True)
         )
         password_confirmation = All(
             v.ValidPassword(),
             v.UnicodeString(strip=False, min=6, not_empty=True)
         )
         active = v.StringBoolean(if_missing=False)
         firstname = v.UnicodeString(strip=True, min=1, not_empty=False)
         lastname = v.UnicodeString(strip=True, min=1, not_empty=False)
         email = All(v.Email(not_empty=True), v.UniqSystemEmail(old_data))
 
         chained_validators = [v.ValidPasswordsMatch('password',
                                                     'password_confirmation')]
 
     return _RegisterForm
 
 
 def PasswordResetRequestForm():
     class _PasswordResetRequestForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         email = v.Email(not_empty=True)
     return _PasswordResetRequestForm
 
 
 def PasswordResetConfirmationForm():
     class _PasswordResetConfirmationForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
 
         email = v.UnicodeString(strip=True, not_empty=True)
         timestamp = v.Number(strip=True, not_empty=True)
         token = v.UnicodeString(strip=True, not_empty=True)
         password = All(v.ValidPassword(), v.UnicodeString(strip=False, min=6))
         password_confirm = All(v.ValidPassword(), v.UnicodeString(strip=False, min=6))
 
         chained_validators = [v.ValidPasswordsMatch('password',
                                                     'password_confirm')]
     return _PasswordResetConfirmationForm
 
 
-def RepoForm(edit=False, old_data=None, supported_backends=BACKENDS,
+def RepoForm(edit=False, old_data=None, supported_backends=kallithea.BACKENDS,
              repo_groups=None, landing_revs=None):
     old_data = old_data or {}
     repo_groups = repo_groups or []
     landing_revs = landing_revs or []
     repo_group_ids = [rg[0] for rg in repo_groups]
 
     class _RepoForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         repo_name = All(v.UnicodeString(strip=True, min=1, not_empty=True),
                         v.SlugifyName())
         repo_group = All(v.CanWriteGroup(old_data),
                          v.OneOf(repo_group_ids, hideList=True),
                          v.Int(min=-1, not_empty=True))
         repo_type = v.OneOf(supported_backends, required=False,
                             if_missing=old_data.get('repo_type'))
         repo_description = v.UnicodeString(strip=True, min=1, not_empty=False)
         repo_private = v.StringBoolean(if_missing=False)
         repo_landing_rev = v.OneOf(landing_revs, hideList=True)
         repo_copy_permissions = v.StringBoolean(if_missing=False)
         clone_uri = All(v.UnicodeString(strip=True, min=1, not_empty=False))
 
         repo_enable_statistics = v.StringBoolean(if_missing=False)
         repo_enable_downloads = v.StringBoolean(if_missing=False)
 
         if edit:
             owner = All(v.UnicodeString(not_empty=True), v.ValidRepoUser())
             # Not a real field - just for reference for validation:
             # clone_uri_hidden = v.UnicodeString(if_missing='')
 
         chained_validators = [v.ValidCloneUri(),
                               v.ValidRepoName(edit, old_data)]
     return _RepoForm
 
 
 def RepoPermsForm():
     class _RepoPermsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         chained_validators = [v.ValidPerms(type_='repo')]
     return _RepoPermsForm
 
 
 def RepoGroupPermsForm(valid_recursive_choices):
     class _RepoGroupPermsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         recursive = v.OneOf(valid_recursive_choices)
         chained_validators = [v.ValidPerms(type_='repo_group')]
     return _RepoGroupPermsForm
 
 
 def UserGroupPermsForm():
     class _UserPermsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         chained_validators = [v.ValidPerms(type_='user_group')]
     return _UserPermsForm
 
 
 def RepoFieldForm():
     class _RepoFieldForm(formencode.Schema):
         filter_extra_fields = True
         allow_extra_fields = True
 
         new_field_key = All(v.FieldKey(),
                             v.UnicodeString(strip=True, min=3, not_empty=True))
         new_field_value = v.UnicodeString(not_empty=False, if_missing='')
         new_field_type = v.OneOf(['str', 'unicode', 'list', 'tuple'],
                                  if_missing='str')
         new_field_label = v.UnicodeString(not_empty=False)
         new_field_desc = v.UnicodeString(not_empty=False)
 
     return _RepoFieldForm
 
 
-def RepoForkForm(edit=False, old_data=None, supported_backends=BACKENDS,
+def RepoForkForm(edit=False, old_data=None, supported_backends=kallithea.BACKENDS,
                  repo_groups=None, landing_revs=None):
     old_data = old_data or {}
     repo_groups = repo_groups or []
     landing_revs = landing_revs or []
     repo_group_ids = [rg[0] for rg in repo_groups]
 
     class _RepoForkForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         repo_name = All(v.UnicodeString(strip=True, min=1, not_empty=True),
                         v.SlugifyName())
         repo_group = All(v.CanWriteGroup(),
                          v.OneOf(repo_group_ids, hideList=True),
                          v.Int(min=-1, not_empty=True))
         repo_type = All(v.ValidForkType(old_data), v.OneOf(supported_backends))
         description = v.UnicodeString(strip=True, min=1, not_empty=True)
         private = v.StringBoolean(if_missing=False)
         copy_permissions = v.StringBoolean(if_missing=False)
         update_after_clone = v.StringBoolean(if_missing=False)
         fork_parent_id = v.UnicodeString()
         chained_validators = [v.ValidForkName(edit, old_data)]
         landing_rev = v.OneOf(landing_revs, hideList=True)
 
     return _RepoForkForm
 
 
 def ApplicationSettingsForm():
     class _ApplicationSettingsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         title = v.UnicodeString(strip=True, not_empty=False)
         realm = v.UnicodeString(strip=True, min=1, not_empty=True)
         ga_code = v.UnicodeString(strip=True, min=1, not_empty=False)
         captcha_public_key = v.UnicodeString(strip=True, min=1, not_empty=False)
         captcha_private_key = v.UnicodeString(strip=True, min=1, not_empty=False)
 
     return _ApplicationSettingsForm
 
 
 def ApplicationVisualisationForm():
     class _ApplicationVisualisationForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         show_public_icon = v.StringBoolean(if_missing=False)
         show_private_icon = v.StringBoolean(if_missing=False)
         stylify_metalabels = v.StringBoolean(if_missing=False)
 
         repository_fields = v.StringBoolean(if_missing=False)
         lightweight_journal = v.StringBoolean(if_missing=False)
         dashboard_items = v.Int(min=5, not_empty=True)
         admin_grid_items = v.Int(min=5, not_empty=True)
         show_version = v.StringBoolean(if_missing=False)
         use_gravatar = v.StringBoolean(if_missing=False)
         gravatar_url = v.UnicodeString(min=3)
         clone_uri_tmpl = v.UnicodeString(min=3)
         clone_ssh_tmpl = v.UnicodeString()
 
     return _ApplicationVisualisationForm
 
 
 def ApplicationUiSettingsForm():
     class _ApplicationUiSettingsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         paths_root_path = All(
             v.ValidPath(),
             v.UnicodeString(strip=True, min=1, not_empty=True)
         )
         hooks_changegroup_update = v.StringBoolean(if_missing=False)
         hooks_changegroup_repo_size = v.StringBoolean(if_missing=False)
 
         extensions_largefiles = v.StringBoolean(if_missing=False)
         extensions_hggit = v.StringBoolean(if_missing=False)
 
     return _ApplicationUiSettingsForm
 
 
 def DefaultPermissionsForm(repo_perms_choices, group_perms_choices,
                            user_group_perms_choices, create_choices,
                            user_group_create_choices, fork_choices,
                            register_choices, extern_activate_choices):
     class _DefaultPermissionsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         overwrite_default_repo = v.StringBoolean(if_missing=False)
         overwrite_default_group = v.StringBoolean(if_missing=False)
         overwrite_default_user_group = v.StringBoolean(if_missing=False)
         anonymous = v.StringBoolean(if_missing=False)
         default_repo_perm = v.OneOf(repo_perms_choices)
         default_group_perm = v.OneOf(group_perms_choices)
         default_user_group_perm = v.OneOf(user_group_perms_choices)
 
         default_repo_create = v.OneOf(create_choices)
         default_user_group_create = v.OneOf(user_group_create_choices)
         default_fork = v.OneOf(fork_choices)
 
         default_register = v.OneOf(register_choices)
         default_extern_activate = v.OneOf(extern_activate_choices)
     return _DefaultPermissionsForm
 
 
 def CustomDefaultPermissionsForm():
     class _CustomDefaultPermissionsForm(formencode.Schema):
         filter_extra_fields = True
         allow_extra_fields = True
 
         create_repo_perm = v.StringBoolean(if_missing=False)
         create_user_group_perm = v.StringBoolean(if_missing=False)
         #create_repo_group_perm Impl. later
 
         fork_repo_perm = v.StringBoolean(if_missing=False)
 
     return _CustomDefaultPermissionsForm
 
 
-def DefaultsForm(edit=False, old_data=None, supported_backends=BACKENDS):
+def DefaultsForm(edit=False, old_data=None, supported_backends=kallithea.BACKENDS):
     class _DefaultsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         default_repo_type = v.OneOf(supported_backends)
         default_repo_private = v.StringBoolean(if_missing=False)
         default_repo_enable_statistics = v.StringBoolean(if_missing=False)
         default_repo_enable_downloads = v.StringBoolean(if_missing=False)
 
     return _DefaultsForm
 
 
 def AuthSettingsForm(current_active_modules):
     class _AuthSettingsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         auth_plugins = All(v.ValidAuthPlugins(),
                            v.UniqueListFromString()(not_empty=True))
 
         def __init__(self, *args, **kwargs):
             # The auth plugins tell us what form validators they use
             if current_active_modules:
                 import kallithea.lib.auth_modules
                 from kallithea.lib.auth_modules import LazyFormencode
                 for module in current_active_modules:
                     plugin = kallithea.lib.auth_modules.loadplugin(module)
                     plugin_name = plugin.name
                     for sv in plugin.plugin_settings():
                         newk = "auth_%s_%s" % (plugin_name, sv["name"])
                         # can be a LazyFormencode object from plugin settings
                         validator = sv["validator"]
                         if isinstance(validator, LazyFormencode):
                             validator = validator()
                         # init all lazy validators from formencode.All
                         if isinstance(validator, All):
                             init_validators = []
                             for validator in validator.validators:
                                 if isinstance(validator, LazyFormencode):
                                     validator = validator()
                                 init_validators.append(validator)
                             validator.validators = init_validators
 
                         self.add_field(newk, validator)
             formencode.Schema.__init__(self, *args, **kwargs)
 
     return _AuthSettingsForm
 
 
 def LdapSettingsForm(tls_reqcert_choices, search_scope_choices,
                      tls_kind_choices):
     class _LdapSettingsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         #pre_validators = [LdapLibValidator]
         ldap_active = v.StringBoolean(if_missing=False)
         ldap_host = v.UnicodeString(strip=True,)
         ldap_port = v.Number(strip=True,)
         ldap_tls_kind = v.OneOf(tls_kind_choices)
         ldap_tls_reqcert = v.OneOf(tls_reqcert_choices)
         ldap_dn_user = v.UnicodeString(strip=True,)
         ldap_dn_pass = v.UnicodeString(strip=True,)
         ldap_base_dn = v.UnicodeString(strip=True,)
         ldap_filter = v.UnicodeString(strip=True,)
         ldap_search_scope = v.OneOf(search_scope_choices)
         ldap_attr_login = v.AttrLoginValidator()(not_empty=True)
         ldap_attr_firstname = v.UnicodeString(strip=True,)
         ldap_attr_lastname = v.UnicodeString(strip=True,)
         ldap_attr_email = v.UnicodeString(strip=True,)
 
     return _LdapSettingsForm
 
 
 def UserExtraEmailForm():
     class _UserExtraEmailForm(formencode.Schema):
         email = All(v.UniqSystemEmail(), v.Email(not_empty=True))
     return _UserExtraEmailForm
 
 
 def UserExtraIpForm():
     class _UserExtraIpForm(formencode.Schema):
         ip = v.ValidIp()(not_empty=True)
     return _UserExtraIpForm
 
 
 def PullRequestForm(repo_id):
     class _PullRequestForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
 
         org_repo = v.UnicodeString(strip=True, required=True)
         org_ref = v.UnicodeString(strip=True, required=True)
         other_repo = v.UnicodeString(strip=True, required=True)
         other_ref = v.UnicodeString(strip=True, required=True)
 
         pullrequest_title = v.UnicodeString(strip=True, required=True)
         pullrequest_desc = v.UnicodeString(strip=True, required=False)
 
     return _PullRequestForm
 
 
 def PullRequestPostForm():
     class _PullRequestPostForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
 
         pullrequest_title = v.UnicodeString(strip=True, required=True)
         pullrequest_desc = v.UnicodeString(strip=True, required=False)
         org_review_members = v.Set()
         review_members = v.Set()
         updaterev = v.UnicodeString(strip=True, required=False, if_missing=None)
         owner = All(v.UnicodeString(strip=True, required=True),
                     v.ValidRepoUser())
 
     return _PullRequestPostForm
 
 
 def GistForm(lifetime_options):
     class _GistForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
 
         filename = All(v.BasePath()(),
                        v.UnicodeString(strip=True, required=False))
         description = v.UnicodeString(required=False, if_missing='')
         lifetime = v.OneOf(lifetime_options)
         mimetype = v.UnicodeString(required=False, if_missing=None)
         content = v.UnicodeString(required=True, not_empty=True)
         public = v.UnicodeString(required=False, if_missing='')
         private = v.UnicodeString(required=False, if_missing='')
 
     return _GistForm

kallithea/model/scm.py

@@ -1,762 +1,761 b''
 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # 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.model.scm
 ~~~~~~~~~~~~~~~~~~~
 
 Scm model 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 9, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 
 import logging
 import os
 import posixpath
 import re
 import sys
 import traceback
 
 import pkg_resources
 from tg.i18n import ugettext as _
 
 import kallithea
-from kallithea import BACKENDS
 from kallithea.lib.auth import HasPermissionAny, HasRepoGroupPermissionLevel, HasRepoPermissionLevel, HasUserGroupPermissionLevel
 from kallithea.lib.exceptions import IMCCommitError, NonRelativePathError
 from kallithea.lib.hooks import process_pushed_raw_ids
 from kallithea.lib.utils import action_logger, get_filesystem_repos, make_ui
 from kallithea.lib.utils2 import safe_bytes, set_hook_environment
 from kallithea.lib.vcs import get_backend
 from kallithea.lib.vcs.backends.base import EmptyChangeset
 from kallithea.lib.vcs.exceptions import RepositoryError
 from kallithea.lib.vcs.nodes import FileNode
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 from kallithea.model.db import PullRequest, RepoGroup, Repository, Session, Ui, User, UserFollowing, UserLog
 
 
 log = logging.getLogger(__name__)
 
 
 class UserTemp(object):
     def __init__(self, user_id):
         self.user_id = user_id
 
     def __repr__(self):
         return "<%s('id:%s')>" % (self.__class__.__name__, self.user_id)
 
 
 class RepoTemp(object):
     def __init__(self, repo_id):
         self.repo_id = repo_id
 
     def __repr__(self):
         return "<%s('id:%s')>" % (self.__class__.__name__, self.repo_id)
 
 
 class _PermCheckIterator(object):
     def __init__(self, obj_list, obj_attr, perm_set, perm_checker, extra_kwargs=None):
         """
         Creates iterator from given list of objects, additionally
         checking permission for them from perm_set var
 
         :param obj_list: list of db objects
         :param obj_attr: attribute of object to pass into perm_checker
         :param perm_set: list of permissions to check
         :param perm_checker: callable to check permissions against
         """
         self.obj_list = obj_list
         self.obj_attr = obj_attr
         self.perm_set = perm_set
         self.perm_checker = perm_checker
         self.extra_kwargs = extra_kwargs or {}
 
     def __len__(self):
         return len(self.obj_list)
 
     def __repr__(self):
         return '<%s (%s)>' % (self.__class__.__name__, self.__len__())
 
     def __iter__(self):
         for db_obj in self.obj_list:
             # check permission at this level
             name = getattr(db_obj, self.obj_attr, None)
             if not self.perm_checker(*self.perm_set)(
                     name, self.__class__.__name__, **self.extra_kwargs):
                 continue
 
             yield db_obj
 
 
 class RepoList(_PermCheckIterator):
 
     def __init__(self, db_repo_list, perm_level, extra_kwargs=None):
         super(RepoList, self).__init__(obj_list=db_repo_list,
                     obj_attr='repo_name', perm_set=[perm_level],
                     perm_checker=HasRepoPermissionLevel,
                     extra_kwargs=extra_kwargs)
 
 
 class RepoGroupList(_PermCheckIterator):
 
     def __init__(self, db_repo_group_list, perm_level, extra_kwargs=None):
         super(RepoGroupList, self).__init__(obj_list=db_repo_group_list,
                     obj_attr='group_name', perm_set=[perm_level],
                     perm_checker=HasRepoGroupPermissionLevel,
                     extra_kwargs=extra_kwargs)
 
 
 class UserGroupList(_PermCheckIterator):
 
     def __init__(self, db_user_group_list, perm_level, extra_kwargs=None):
         super(UserGroupList, self).__init__(obj_list=db_user_group_list,
                     obj_attr='users_group_name', perm_set=[perm_level],
                     perm_checker=HasUserGroupPermissionLevel,
                     extra_kwargs=extra_kwargs)
 
 
 class ScmModel(object):
     """
     Generic Scm Model
     """
 
     def __get_repo(self, instance):
         cls = Repository
         if isinstance(instance, cls):
             return instance
         elif isinstance(instance, int):
             return cls.get(instance)
         elif isinstance(instance, str):
             if instance.isdigit():
                 return cls.get(int(instance))
             return cls.get_by_repo_name(instance)
         raise Exception('given object must be int, basestr or Instance'
                         ' of %s got %s' % (type(cls), type(instance)))
 
     @LazyProperty
     def repos_path(self):
         """
         Gets the repositories root path from database
         """
 
         q = Ui.query().filter(Ui.ui_key == '/').one()
 
         return q.ui_value
 
     def repo_scan(self, repos_path=None):
         """
         Listing of repositories in given path. This path should not be a
         repository itself. Return a dictionary of repository objects mapping to
         vcs instances.
 
         :param repos_path: path to directory containing repositories
         """
 
         if repos_path is None:
             repos_path = self.repos_path
 
         log.info('scanning for repositories in %s', repos_path)
 
         baseui = make_ui()
         repos = {}
 
         for name, path in get_filesystem_repos(repos_path):
             # name need to be decomposed and put back together using the /
             # since this is internal storage separator for kallithea
             name = Repository.normalize_repo_name(name)
 
             try:
                 if name in repos:
                     raise RepositoryError('Duplicate repository name %s '
                                           'found in %s' % (name, path))
                 else:
 
                     klass = get_backend(path[0])
 
-                    if path[0] == 'hg' and path[0] in BACKENDS:
+                    if path[0] == 'hg' and path[0] in kallithea.BACKENDS:
                         repos[name] = klass(path[1], baseui=baseui)
 
-                    if path[0] == 'git' and path[0] in BACKENDS:
+                    if path[0] == 'git' and path[0] in kallithea.BACKENDS:
                         repos[name] = klass(path[1])
             except OSError:
                 continue
         log.debug('found %s paths with repositories', len(repos))
         return repos
 
     def get_repos(self, repos):
         """Return the repos the user has access to"""
         return RepoList(repos, perm_level='read')
 
     def get_repo_groups(self, groups=None):
         """Return the repo groups the user has access to
         If no groups are specified, use top level groups.
         """
         if groups is None:
             groups = RepoGroup.query() \
                 .filter(RepoGroup.parent_group_id == None).all()
         return RepoGroupList(groups, perm_level='read')
 
     def mark_for_invalidation(self, repo_name):
         """
         Mark caches of this repo invalid in the database.
 
         :param repo_name: the repo for which caches should be marked invalid
         """
         log.debug("Marking %s as invalidated and update cache", repo_name)
         repo = Repository.get_by_repo_name(repo_name)
         if repo is not None:
             repo.set_invalidate()
             repo.update_changeset_cache()
 
     def toggle_following_repo(self, follow_repo_id, user_id):
 
         f = UserFollowing.query() \
             .filter(UserFollowing.follows_repository_id == follow_repo_id) \
             .filter(UserFollowing.user_id == user_id).scalar()
 
         if f is not None:
             try:
                 Session().delete(f)
                 action_logger(UserTemp(user_id),
                               'stopped_following_repo',
                               RepoTemp(follow_repo_id))
                 return
             except Exception:
                 log.error(traceback.format_exc())
                 raise
 
         try:
             f = UserFollowing()
             f.user_id = user_id
             f.follows_repository_id = follow_repo_id
             Session().add(f)
 
             action_logger(UserTemp(user_id),
                           'started_following_repo',
                           RepoTemp(follow_repo_id))
         except Exception:
             log.error(traceback.format_exc())
             raise
 
     def toggle_following_user(self, follow_user_id, user_id):
         f = UserFollowing.query() \
             .filter(UserFollowing.follows_user_id == follow_user_id) \
             .filter(UserFollowing.user_id == user_id).scalar()
 
         if f is not None:
             try:
                 Session().delete(f)
                 return
             except Exception:
                 log.error(traceback.format_exc())
                 raise
 
         try:
             f = UserFollowing()
             f.user_id = user_id
             f.follows_user_id = follow_user_id
             Session().add(f)
         except Exception:
             log.error(traceback.format_exc())
             raise
 
     def is_following_repo(self, repo_name, user_id):
         r = Repository.query() \
             .filter(Repository.repo_name == repo_name).scalar()
 
         f = UserFollowing.query() \
             .filter(UserFollowing.follows_repository == r) \
             .filter(UserFollowing.user_id == user_id).scalar()
 
         return f is not None
 
     def is_following_user(self, username, user_id):
         u = User.get_by_username(username)
 
         f = UserFollowing.query() \
             .filter(UserFollowing.follows_user == u) \
             .filter(UserFollowing.user_id == user_id).scalar()
 
         return f is not None
 
     def get_followers(self, repo):
         repo = Repository.guess_instance(repo)
 
         return UserFollowing.query() \
                 .filter(UserFollowing.follows_repository == repo).count()
 
     def get_forks(self, repo):
         repo = Repository.guess_instance(repo)
         return Repository.query() \
                 .filter(Repository.fork == repo).count()
 
     def get_pull_requests(self, repo):
         repo = Repository.guess_instance(repo)
         return PullRequest.query() \
                 .filter(PullRequest.other_repo == repo) \
                 .filter(PullRequest.status != PullRequest.STATUS_CLOSED).count()
 
     def mark_as_fork(self, repo, fork, user):
         repo = self.__get_repo(repo)
         fork = self.__get_repo(fork)
         if fork and repo.repo_id == fork.repo_id:
             raise Exception("Cannot set repository as fork of itself")
 
         if fork and repo.repo_type != fork.repo_type:
             raise RepositoryError("Cannot set repository as fork of repository with other type")
 
         repo.fork = fork
         return repo
 
     def _handle_push(self, repo, username, ip_addr, action, repo_name, revisions):
         """
         Handle that the repository has changed.
         Adds an action log entry with the new revisions, and the head revision
         cache and in-memory caches are invalidated/updated.
 
         :param username: username who pushes
         :param action: push/push_local/push_remote
         :param repo_name: name of repo
         :param revisions: list of revisions that we pushed
         """
         set_hook_environment(username, ip_addr, repo_name, repo_alias=repo.alias, action=action)
         process_pushed_raw_ids(revisions) # also calls mark_for_invalidation
 
     def _get_IMC_module(self, scm_type):
         """
         Returns InMemoryCommit class based on scm_type
 
         :param scm_type:
         """
         if scm_type == 'hg':
             from kallithea.lib.vcs.backends.hg import MercurialInMemoryChangeset
             return MercurialInMemoryChangeset
 
         if scm_type == 'git':
             from kallithea.lib.vcs.backends.git import GitInMemoryChangeset
             return GitInMemoryChangeset
 
         raise Exception('Invalid scm_type, must be one of hg,git got %s'
                         % (scm_type,))
 
     def pull_changes(self, repo, username, ip_addr, clone_uri=None):
         """
         Pull from "clone URL" or fork origin.
         """
         dbrepo = self.__get_repo(repo)
         if clone_uri is None:
             clone_uri = dbrepo.clone_uri or dbrepo.fork and dbrepo.fork.repo_full_path
         if not clone_uri:
             raise Exception("This repository doesn't have a clone uri")
 
         repo = dbrepo.scm_instance
         repo_name = dbrepo.repo_name
         try:
             if repo.alias == 'git':
                 repo.fetch(clone_uri)
                 # git doesn't really have something like post-fetch action
                 # we fake that now.
                 # TODO: extract fetched revisions ... somehow ...
                 self._handle_push(repo,
                                   username=username,
                                   ip_addr=ip_addr,
                                   action='push_remote',
                                   repo_name=repo_name,
                                   revisions=[])
             else:
                 set_hook_environment(username, ip_addr, dbrepo.repo_name,
                                            repo.alias, action='push_remote')
                 repo.pull(clone_uri)
         except Exception:
             log.error(traceback.format_exc())
             raise
 
     def commit_change(self, repo, repo_name, cs, user, ip_addr, author, message,
                       content, f_path):
         """
         Commit a change to a single file
 
         :param repo: a db_repo.scm_instance
         """
         user = User.guess_instance(user)
         IMC = self._get_IMC_module(repo.alias)
         imc = IMC(repo)
         imc.change(FileNode(f_path, content, mode=cs.get_file_mode(f_path)))
         try:
             tip = imc.commit(message=message, author=author,
                              parents=[cs], branch=cs.branch)
         except Exception as e:
             log.error(traceback.format_exc())
             # clear caches - we also want a fresh object if commit fails
             self.mark_for_invalidation(repo_name)
             raise IMCCommitError(str(e))
         self._handle_push(repo,
                           username=user.username,
                           ip_addr=ip_addr,
                           action='push_local',
                           repo_name=repo_name,
                           revisions=[tip.raw_id])
         return tip
 
     def _sanitize_path(self, f_path):
         if f_path.startswith('/') or f_path.startswith('.') or '../' in f_path:
             raise NonRelativePathError('%s is not an relative path' % f_path)
         if f_path:
             f_path = posixpath.normpath(f_path)
         return f_path
 
     def get_nodes(self, repo_name, revision, root_path='/', flat=True):
         """
         Recursively walk root dir and return a set of all paths found.
 
         :param repo_name: name of repository
         :param revision: revision for which to list nodes
         :param root_path: root path to list
         :param flat: return as a list, if False returns a dict with description
 
         """
         _files = list()
         _dirs = list()
         try:
             _repo = self.__get_repo(repo_name)
             changeset = _repo.scm_instance.get_changeset(revision)
             root_path = root_path.lstrip('/')
             for topnode, dirs, files in changeset.walk(root_path):
                 for f in files:
                     _files.append(f.path if flat else {"name": f.path,
                                                        "type": "file"})
                 for d in dirs:
                     _dirs.append(d.path if flat else {"name": d.path,
                                                       "type": "dir"})
         except RepositoryError:
             log.debug(traceback.format_exc())
             raise
 
         return _dirs, _files
 
     def create_nodes(self, user, ip_addr, repo, message, nodes, parent_cs=None,
                      author=None, trigger_push_hook=True):
         """
         Commits specified nodes to repo.
 
         :param user: Kallithea User object or user_id, the committer
         :param repo: Kallithea Repository object
         :param message: commit message
         :param nodes: mapping {filename:{'content':content},...}
         :param parent_cs: parent changeset, can be empty than it's initial commit
         :param author: author of commit, cna be different that committer only for git
         :param trigger_push_hook: trigger push hooks
 
         :returns: new committed changeset
         """
 
         user = User.guess_instance(user)
         scm_instance = repo.scm_instance_no_cache()
 
         processed_nodes = []
         for f_path in nodes:
             content = nodes[f_path]['content']
             f_path = self._sanitize_path(f_path)
             if not isinstance(content, str) and not isinstance(content, bytes):
                 content = content.read()
             processed_nodes.append((f_path, content))
 
         message = message
         committer = user.full_contact
         if not author:
             author = committer
 
         IMC = self._get_IMC_module(scm_instance.alias)
         imc = IMC(scm_instance)
 
         if not parent_cs:
             parent_cs = EmptyChangeset(alias=scm_instance.alias)
 
         if isinstance(parent_cs, EmptyChangeset):
             # EmptyChangeset means we we're editing empty repository
             parents = None
         else:
             parents = [parent_cs]
         # add multiple nodes
         for path, content in processed_nodes:
             imc.add(FileNode(path, content=content))
 
         tip = imc.commit(message=message,
                          author=author,
                          parents=parents,
                          branch=parent_cs.branch)
 
         if trigger_push_hook:
             self._handle_push(scm_instance,
                               username=user.username,
                               ip_addr=ip_addr,
                               action='push_local',
                               repo_name=repo.repo_name,
                               revisions=[tip.raw_id])
         else:
             self.mark_for_invalidation(repo.repo_name)
         return tip
 
     def update_nodes(self, user, ip_addr, repo, message, nodes, parent_cs=None,
                      author=None, trigger_push_hook=True):
         """
         Commits specified nodes to repo. Again.
         """
         user = User.guess_instance(user)
         scm_instance = repo.scm_instance_no_cache()
 
         message = message
         committer = user.full_contact
         if not author:
             author = committer
 
         imc_class = self._get_IMC_module(scm_instance.alias)
         imc = imc_class(scm_instance)
 
         if not parent_cs:
             parent_cs = EmptyChangeset(alias=scm_instance.alias)
 
         if isinstance(parent_cs, EmptyChangeset):
             # EmptyChangeset means we we're editing empty repository
             parents = None
         else:
             parents = [parent_cs]
 
         # add multiple nodes
         for _filename, data in nodes.items():
             # new filename, can be renamed from the old one
             filename = self._sanitize_path(data['filename'])
             old_filename = self._sanitize_path(_filename)
             content = data['content']
 
             filenode = FileNode(old_filename, content=content)
             op = data['op']
             if op == 'add':
                 imc.add(filenode)
             elif op == 'del':
                 imc.remove(filenode)
             elif op == 'mod':
                 if filename != old_filename:
                     # TODO: handle renames, needs vcs lib changes
                     imc.remove(filenode)
                     imc.add(FileNode(filename, content=content))
                 else:
                     imc.change(filenode)
 
         # commit changes
         tip = imc.commit(message=message,
                          author=author,
                          parents=parents,
                          branch=parent_cs.branch)
 
         if trigger_push_hook:
             self._handle_push(scm_instance,
                               username=user.username,
                               ip_addr=ip_addr,
                               action='push_local',
                               repo_name=repo.repo_name,
                               revisions=[tip.raw_id])
         else:
             self.mark_for_invalidation(repo.repo_name)
 
     def delete_nodes(self, user, ip_addr, repo, message, nodes, parent_cs=None,
                      author=None, trigger_push_hook=True):
         """
         Deletes specified nodes from repo.
 
         :param user: Kallithea User object or user_id, the committer
         :param repo: Kallithea Repository object
         :param message: commit message
         :param nodes: mapping {filename:{'content':content},...}
         :param parent_cs: parent changeset, can be empty than it's initial commit
         :param author: author of commit, cna be different that committer only for git
         :param trigger_push_hook: trigger push hooks
 
         :returns: new committed changeset after deletion
         """
 
         user = User.guess_instance(user)
         scm_instance = repo.scm_instance_no_cache()
 
         processed_nodes = []
         for f_path in nodes:
             f_path = self._sanitize_path(f_path)
             # content can be empty but for compatibility it allows same dicts
             # structure as add_nodes
             content = nodes[f_path].get('content')
             processed_nodes.append((f_path, content))
 
         message = message
         committer = user.full_contact
         if not author:
             author = committer
 
         IMC = self._get_IMC_module(scm_instance.alias)
         imc = IMC(scm_instance)
 
         if not parent_cs:
             parent_cs = EmptyChangeset(alias=scm_instance.alias)
 
         if isinstance(parent_cs, EmptyChangeset):
             # EmptyChangeset means we we're editing empty repository
             parents = None
         else:
             parents = [parent_cs]
         # add multiple nodes
         for path, content in processed_nodes:
             imc.remove(FileNode(path, content=content))
 
         tip = imc.commit(message=message,
                          author=author,
                          parents=parents,
                          branch=parent_cs.branch)
 
         if trigger_push_hook:
             self._handle_push(scm_instance,
                               username=user.username,
                               ip_addr=ip_addr,
                               action='push_local',
                               repo_name=repo.repo_name,
                               revisions=[tip.raw_id])
         else:
             self.mark_for_invalidation(repo.repo_name)
         return tip
 
     def get_unread_journal(self):
         return UserLog.query().count()
 
     def get_repo_landing_revs(self, repo=None):
         """
         Generates select option with tags branches and bookmarks (for hg only)
         grouped by type
 
         :param repo:
         """
 
         hist_l = []
         choices = []
         hist_l.append(('rev:tip', _('latest tip')))
         choices.append('rev:tip')
         if repo is None:
             return choices, hist_l
 
         repo = self.__get_repo(repo)
         repo = repo.scm_instance
 
         branches_group = ([('branch:%s' % k, k) for k, v in
                            repo.branches.items()], _("Branches"))
         hist_l.append(branches_group)
         choices.extend([x[0] for x in branches_group[0]])
 
         if repo.alias == 'hg':
             bookmarks_group = ([('book:%s' % k, k) for k, v in
                                 repo.bookmarks.items()], _("Bookmarks"))
             hist_l.append(bookmarks_group)
             choices.extend([x[0] for x in bookmarks_group[0]])
 
         tags_group = ([('tag:%s' % k, k) for k, v in
                        repo.tags.items()], _("Tags"))
         hist_l.append(tags_group)
         choices.extend([x[0] for x in tags_group[0]])
 
         return choices, hist_l
 
     def _get_git_hook_interpreter(self):
         """Return a suitable interpreter for Git hooks.
 
         Return a suitable string to be written in the POSIX #! shebang line for
         Git hook scripts so they invoke Kallithea code with the right Python
         interpreter and in the right environment.
         """
         # Note: sys.executable might not point at a usable Python interpreter. For
         # example, when using uwsgi, it will point at the uwsgi program itself.
         # FIXME This may not work on Windows and may need a shell wrapper script.
         return (kallithea.CONFIG.get('git_hook_interpreter')
                 or sys.executable
                 or '/usr/bin/env python3')
 
     def install_git_hooks(self, repo, force=False):
         """
         Creates a kallithea hook inside a git repository
 
         :param repo: Instance of VCS repo
         :param force: Overwrite existing non-Kallithea hooks
         """
 
         hooks_path = os.path.join(repo.path, 'hooks')
         if not repo.bare:
             hooks_path = os.path.join(repo.path, '.git', 'hooks')
         if not os.path.isdir(hooks_path):
             os.makedirs(hooks_path)
 
         tmpl_post = b"#!%s\n" % safe_bytes(self._get_git_hook_interpreter())
         tmpl_post += pkg_resources.resource_string(
             'kallithea', os.path.join('config', 'post_receive_tmpl.py')
         )
         tmpl_pre = b"#!%s\n" % safe_bytes(self._get_git_hook_interpreter())
         tmpl_pre += pkg_resources.resource_string(
             'kallithea', os.path.join('config', 'pre_receive_tmpl.py')
         )
 
         for h_type, tmpl in [('pre', tmpl_pre), ('post', tmpl_post)]:
             hook_file = os.path.join(hooks_path, '%s-receive' % h_type)
             other_hook = False
             log.debug('Installing git hook in repo %s', repo)
             if os.path.exists(hook_file):
                 # let's take a look at this hook, maybe it's kallithea ?
                 log.debug('hook exists, checking if it is from kallithea')
                 with open(hook_file, 'rb') as f:
                     data = f.read()
                     matches = re.search(br'^KALLITHEA_HOOK_VER\s*=\s*(.*)$', data, flags=re.MULTILINE)
                     if matches:
                         ver = matches.groups()[0]
                         log.debug('Found Kallithea hook - it has KALLITHEA_HOOK_VER %r', ver)
                     else:
                         log.debug('Found non-Kallithea hook at %s', hook_file)
                         other_hook = True
 
             if other_hook and not force:
                 log.warning('skipping overwriting hook file %s', hook_file)
             else:
                 log.debug('writing %s hook file !', h_type)
                 try:
                     with open(hook_file, 'wb') as f:
                         tmpl = tmpl.replace(b'_TMPL_', safe_bytes(kallithea.__version__))
                         f.write(tmpl)
                     os.chmod(hook_file, 0o755)
                 except IOError as e:
                     log.error('error writing hook %s: %s', hook_file, e)
 
 
 def AvailableRepoGroupChoices(repo_group_perm_level, extras=()):
     """Return group_id,string tuples with choices for all the repo groups where
     the user has the necessary permissions.
 
     Top level is -1.
     """
     groups = RepoGroup.query().all()
     if HasPermissionAny('hg.admin')('available repo groups'):
         groups.append(None)
     else:
         groups = list(RepoGroupList(groups, perm_level=repo_group_perm_level))
         if HasPermissionAny('hg.create.repository')('available repo groups'):
             groups.append(None)
         for extra in extras:
             if not any(rg == extra for rg in groups):
                 groups.append(extra)
     return RepoGroup.groups_choices(groups=groups)