kallithea Changeset - f8971422795e

Changeset - f8971422795e

Parent rev.

Child rev.

[Not reviewed]

default

0 11 1

Mads Kiilerich (mads) - 5 years ago 2020-11-07 02:29:41
mads@kiilerich.com

Grafted from: a0cb3480068a

scripts: introduce source_format.py to fix up the module name in file headers

12 files changed with 42 insertions and 17 deletions:

kallithea/bin/vcs_hooks.py

kallithea/config/middleware/https_fixup.py

kallithea/config/middleware/permanent_repo_url.py

kallithea/config/middleware/pygrack.py

kallithea/config/middleware/simplegit.py

kallithea/config/middleware/simplehg.py

kallithea/config/middleware/wrapper.py

kallithea/lib/conf.py

kallithea/lib/webutils.py

kallithea/model/userlog.py

scripts/run-all-cleanup

scripts/source_format.py

0 comments (0 inline, 0 general)

kallithea/bin/vcs_hooks.py

➞

Show inline comments

 # -*- 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.bin.vcs_hooks
 ~~~~~~~~~~~~~~~~~~~
+~~~~~~~~~~~~~~~~~~~~~~~
 Entry points for Kallithea hooking into Mercurial and Git.
 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 mercurial.scmutil
 import paste.deploy
 import kallithea
 import kallithea.config.application
 from kallithea.lib import hooks, webutils
 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.lib.vcs.utils.helpers import get_scm_size
 from kallithea.model import db
 def repo_size(ui, repo, hooktype=None, **kwargs):
     """Show size of Mercurial repository.
     Called as Mercurial hook changegroup.repo_size after push.
     """
     size_hg, size_root = 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') % (
         webutils.format_byte_size(size_hg),
         webutils.format_byte_size(size_root),
         webutils.format_byte_size(size_hg + size_root),
         last_cs.rev(),
         ascii_str(last_cs.hex())[:12],
+    )
     ui.status(safe_bytes(msg))
 def pull_action(ui, repo, **kwargs):
     """Logs user pull action
     Called as Mercurial hook outgoing.kallithea_pull_action.
     """
     hooks.log_pull_action()
 def 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_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)])]
     hooks.process_pushed_raw_ids(revs)
 def _git_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.
     """
     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 = db.Repository.get_by_full_path(repo_path)
     if not repo:
         raise OSError('Repository %s not found in database' % repo_path)
     return repo
 def 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 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:
         repo = _git_hook_environment(repo_path)
     except HookEnvironmentError as e:
         sys.stderr.write("Skipping Kallithea Git post-receive 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']]
     hooks.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/config/middleware/https_fixup.py

➞

Show inline comments

 # -*- 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.middleware.https_fixup
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 kallithea.config.middleware.https_fixup
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 middleware to handle https correctly
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: May 23, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from kallithea.lib.utils2 import asbool
 class HttpsFixup(object):
     def __init__(self, app, config):
         self.application = app
         self.config = config
     def __call__(self, environ, start_response):
         self.__fixup(environ)
         debug = asbool(self.config.get('debug'))
         is_ssl = environ['wsgi.url_scheme'] == 'https'
         def custom_start_response(status, headers, exc_info=None):
             if is_ssl and asbool(self.config.get('use_htsts')) and not debug:
                 headers.append(('Strict-Transport-Security',
                                 'max-age=8640000; includeSubDomains'))
             return start_response(status, headers, exc_info)
         return self.application(environ, custom_start_response)
     def __fixup(self, environ):
         """
         Function to fixup the environ as needed. In order to use this
         middleware you should set this header inside your
         proxy ie. nginx, apache etc.
         """
         # DETECT PROTOCOL !
         if 'HTTP_X_URL_SCHEME' in environ:
             proto = environ.get('HTTP_X_URL_SCHEME')
         elif 'HTTP_X_FORWARDED_SCHEME' in environ:
             proto = environ.get('HTTP_X_FORWARDED_SCHEME')
         elif 'HTTP_X_FORWARDED_PROTO' in environ:
             proto = environ.get('HTTP_X_FORWARDED_PROTO')
         else:
             proto = 'http'
         org_proto = proto
         # if we have force, just override
         if asbool(self.config.get('force_https')):
             proto = 'https'
         environ['wsgi.url_scheme'] = proto
         environ['wsgi._org_proto'] = org_proto

kallithea/config/middleware/permanent_repo_url.py

➞

Show inline comments

 # -*- 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.middleware.permanent_repo_url
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 kallithea.config.middleware.permanent_repo_url
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 middleware to handle permanent repo URLs, replacing PATH_INFO '/_123/yada' with
 '/name/of/repo/yada' after looking 123 up in the database.
 """
 from kallithea.lib.utils import fix_repo_id_name
 from kallithea.lib.utils2 import safe_bytes, safe_str
 class PermanentRepoUrl(object):
     def __init__(self, app, config):
         self.application = app
         self.config = config
     def __call__(self, environ, start_response):
         # Extract path_info as get_path_info does, but do it explicitly because
         # we also have to do the reverse operation when patching it back in
         path_info = safe_str(environ['PATH_INFO'].encode('latin1'))
         if path_info.startswith('/'): # it must
             path_info = '/' + fix_repo_id_name(path_info[1:])
             environ['PATH_INFO'] = safe_bytes(path_info).decode('latin1')
         return self.application(environ, start_response)

kallithea/config/middleware/pygrack.py

➞

Show inline comments

 # -*- 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.middleware.pygrack
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 kallithea.config.middleware.pygrack
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Python implementation of git-http-backend's Smart HTTP protocol
 Based on original code from git_http_backend.py project.
 Copyright (c) 2010 Daniel Dotsenko <dotsa@hotmail.com>
 Copyright (c) 2012 Marcin Kuzminski <marcin@python-works.com>
 This file was forked by the Kallithea project in July 2014.
 """
 import logging
 import os
 import socket
 import traceback
 from dulwich.server import update_server_info
 from dulwich.web import GunzipFilter, LimitedInputFilter
 from webob import Request, Response, exc
 import kallithea
 from kallithea.lib.utils2 import ascii_bytes
 from kallithea.lib.vcs import get_repo, subprocessio
 log = logging.getLogger(__name__)
 class FileWrapper(object):
     def __init__(self, fd, content_length):
         self.fd = fd
         self.content_length = content_length
         self.remain = content_length
     def read(self, size):
         if size <= self.remain:
             try:
                 data = self.fd.read(size)
             except socket.error:
                 raise IOError(self)
             self.remain -= size
         elif self.remain:
             data = self.fd.read(self.remain)
             self.remain = 0
         else:
             data = None
         return data
     def __repr__(self):
         return '<FileWrapper %s len: %s, read: %s>' % (
             self.fd, self.content_length, self.content_length - self.remain
+        )
 class GitRepository(object):
     git_folder_signature = set(['config', 'head', 'info', 'objects', 'refs'])
     commands = ['git-upload-pack', 'git-receive-pack']
     def __init__(self, repo_name, content_path):
         files = set([f.lower() for f in os.listdir(content_path)])
         if not (self.git_folder_signature.intersection(files)
                 == self.git_folder_signature):
             raise OSError('%s missing git signature' % content_path)
         self.content_path = content_path
         self.valid_accepts = ['application/x-%s-result' %
                               c for c in self.commands]
         self.repo_name = repo_name
     def _get_fixedpath(self, path):
         """
         Small fix for repo_path
         :param path:
         """
         assert path.startswith('/' + self.repo_name + '/')
         return path[len(self.repo_name) + 2:].strip('/')
     def inforefs(self, req, environ):
         """
         WSGI Response producer for HTTP GET Git Smart
         HTTP /info/refs request.
         """
         git_command = req.GET.get('service')
         if git_command not in self.commands:
             log.debug('command %s not allowed', git_command)
             return exc.HTTPMethodNotAllowed()
         # From Documentation/technical/http-protocol.txt shipped with Git:
+        #
         # Clients MUST verify the first pkt-line is `# service=$servicename`.
         # Servers MUST set $servicename to be the request parameter value.
         # Servers SHOULD include an LF at the end of this line.
         # Clients MUST ignore an LF at the end of the line.
+        #
         #  smart_reply     =  PKT-LINE("# service=$servicename" LF)
         #                     ref_list
         #                     "0000"
         server_advert = '# service=%s\n' % git_command
         packet_len = hex(len(server_advert) + 4)[2:].rjust(4, '0').lower()
         _git_path = kallithea.CONFIG.get('git_path', 'git')
         cmd = [_git_path, git_command[4:],
                '--stateless-rpc', '--advertise-refs', self.content_path]
         log.debug('handling cmd %s', cmd)
         try:
             out = subprocessio.SubprocessIOChunker(cmd,
                 starting_values=[ascii_bytes(packet_len + server_advert + '0000')]
+            )
         except EnvironmentError as e:
             log.error(traceback.format_exc())
             raise exc.HTTPExpectationFailed()
         resp = Response()
         resp.content_type = 'application/x-%s-advertisement' % git_command
         resp.charset = None
         resp.app_iter = out
         return resp
     def backend(self, req, environ):
         """
         WSGI Response producer for HTTP POST Git Smart HTTP requests.
         Reads commands and data from HTTP POST's body.
         returns an iterator obj with contents of git command's
         response to stdout
         """
         _git_path = kallithea.CONFIG.get('git_path', 'git')
         git_command = self._get_fixedpath(req.path_info)
         if git_command not in self.commands:
             log.debug('command %s not allowed', git_command)
             return exc.HTTPMethodNotAllowed()
         if 'CONTENT_LENGTH' in environ:
             inputstream = FileWrapper(environ['wsgi.input'],
                                       req.content_length)
         else:
             inputstream = environ['wsgi.input']
         gitenv = dict(os.environ)
         # forget all configs
         gitenv['GIT_CONFIG_NOGLOBAL'] = '1'
         cmd = [_git_path, git_command[4:], '--stateless-rpc', self.content_path]
         log.debug('handling cmd %s', cmd)
         try:
             out = subprocessio.SubprocessIOChunker(
                 cmd,
                 inputstream=inputstream,
                 env=gitenv,
                 cwd=self.content_path,
+            )
         except EnvironmentError as e:
             log.error(traceback.format_exc())
             raise exc.HTTPExpectationFailed()
         if git_command in ['git-receive-pack']:
             # updating refs manually after each push.
             # Needed for pre-1.7.0.4 git clients using regular HTTP mode.
             repo = get_repo(self.content_path)
             if repo:
                 update_server_info(repo._repo)
         resp = Response()
         resp.content_type = 'application/x-%s-result' % git_command
         resp.charset = None
         resp.app_iter = out
         return resp
     def __call__(self, environ, start_response):
         req = Request(environ)
         _path = self._get_fixedpath(req.path_info)
         if _path.startswith('info/refs'):
             app = self.inforefs
         elif req.accept.acceptable_offers(self.valid_accepts):
             app = self.backend
         try:
             resp = app(req, environ)
         except exc.HTTPException as e:
             resp = e
             log.error(traceback.format_exc())
         except Exception as e:
             log.error(traceback.format_exc())
             resp = exc.HTTPInternalServerError()
         return resp(environ, start_response)
 class GitDirectory(object):
     def __init__(self, repo_root, repo_name):
         repo_location = os.path.join(repo_root, repo_name)
         if not os.path.isdir(repo_location):
             raise OSError(repo_location)
         self.content_path = repo_location
         self.repo_name = repo_name
         self.repo_location = repo_location
     def __call__(self, environ, start_response):
         content_path = self.content_path
         try:
             app = GitRepository(self.repo_name, content_path)
         except (AssertionError, OSError):
             content_path = os.path.join(content_path, '.git')
             if os.path.isdir(content_path):
                 app = GitRepository(self.repo_name, content_path)
             else:
                 return exc.HTTPNotFound()(environ, start_response)
         return app(environ, start_response)
 def make_wsgi_app(repo_name, repo_root):
     app = GitDirectory(repo_root, repo_name)
     return GunzipFilter(LimitedInputFilter(app))

kallithea/config/middleware/simplegit.py

➞

Show inline comments

 # -*- 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.middleware.simplegit
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 kallithea.config.middleware.simplegit
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 SimpleGit middleware for handling Git protocol requests (push/clone etc.)
 It's implemented with basic auth function
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 28, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import re
 from kallithea.config.middleware.pygrack import make_wsgi_app
 from kallithea.lib import hooks
 from kallithea.lib.base import BaseVCSController, get_path_info
 log = logging.getLogger(__name__)
 GIT_PROTO_PAT = re.compile(r'^/(.+)/(info/refs|git-upload-pack|git-receive-pack)$')
 cmd_mapping = {
     'git-receive-pack': 'push',
     'git-upload-pack': 'pull',
+}
 class SimpleGit(BaseVCSController):
     scm_alias = 'git'
     @classmethod
     def parse_request(cls, environ):
         path_info = get_path_info(environ)
         m = GIT_PROTO_PAT.match(path_info)
         if m is None:
             return None
         class parsed_request(object):
             # See https://git-scm.com/book/en/v2/Git-Internals-Transfer-Protocols#_the_smart_protocol
             repo_name = m.group(1).rstrip('/')
             cmd = m.group(2)
             query_string = environ['QUERY_STRING']
             if cmd == 'info/refs' and query_string.startswith('service='):
                 service = query_string.split('=', 1)[1]
                 action = cmd_mapping.get(service)
             else:
                 service = None
                 action = cmd_mapping.get(cmd)
         return parsed_request
     def _make_app(self, parsed_request):
         """
         Return a pygrack wsgi application.
         """
         pygrack_app = make_wsgi_app(parsed_request.repo_name, self.basepath)
         def wrapper_app(environ, start_response):
             if (parsed_request.cmd == 'info/refs' and
                 parsed_request.service == 'git-upload-pack'
             ):
                 # Run hooks like Mercurial outgoing.kallithea_pull_action does
                 hooks.log_pull_action()
             # Note: push hooks are handled by post-receive hook
             return pygrack_app(environ, start_response)
         return wrapper_app

kallithea/config/middleware/simplehg.py

➞

Show inline comments

 # -*- 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.middleware.simplehg
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 kallithea.config.middleware.simplehg
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 SimpleHg middleware for handling Mercurial protocol requests (push/clone etc.).
 It's implemented with basic auth function
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 28, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import os
 import urllib.parse
 import mercurial.hgweb
 from kallithea.lib.base import BaseVCSController, get_path_info
 from kallithea.lib.utils import make_ui
 from kallithea.lib.utils2 import safe_bytes
 log = logging.getLogger(__name__)
 def get_header_hgarg(environ):
     """Decode the special Mercurial encoding of big requests over multiple headers.
     >>> get_header_hgarg({})
     ''
     >>> get_header_hgarg({'HTTP_X_HGARG_0': ' ', 'HTTP_X_HGARG_1': 'a','HTTP_X_HGARG_2': '','HTTP_X_HGARG_3': 'b+c %20'})
     'ab+c %20'
     """
     chunks = []
     i = 1
     while True:
         v = environ.get('HTTP_X_HGARG_%d' % i)
         if v is None:
             break
         chunks.append(v)
         i += 1
     return ''.join(chunks)
 cmd_mapping = {
     # 'batch' is not in this list - it is handled explicitly
     'between': 'pull',
     'branches': 'pull',
     'branchmap': 'pull',
     'capabilities': 'pull',
     'changegroup': 'pull',
     'changegroupsubset': 'pull',
     'changesetdata': 'pull',
     'clonebundles': 'pull',
     'debugwireargs': 'pull',
     'filedata': 'pull',
     'getbundle': 'pull',
     'getlfile': 'pull',
     'heads': 'pull',
     'hello': 'pull',
     'known': 'pull',
     'lheads': 'pull',
     'listkeys': 'pull',
     'lookup': 'pull',
     'manifestdata': 'pull',
     'narrow_widen': 'pull',
     'protocaps': 'pull',
     'statlfile': 'pull',
     'stream_out': 'pull',
     'pushkey': 'push',
     'putlfile': 'push',
     'unbundle': 'push',
+    }
 class SimpleHg(BaseVCSController):
     scm_alias = 'hg'
     @classmethod
     def parse_request(cls, environ):
         http_accept = environ.get('HTTP_ACCEPT', '')
         if not http_accept.startswith('application/mercurial'):
             return None
         path_info = get_path_info(environ)
         if not path_info.startswith('/'): # it must!
             return None
         class parsed_request(object):
             repo_name = path_info[1:].rstrip('/')
             query_string = environ['QUERY_STRING']
             action = None
             for qry in query_string.split('&'):
                 parts = qry.split('=', 1)
                 if len(parts) == 2 and parts[0] == 'cmd':
                     cmd = parts[1]
                     if cmd == 'batch':
                         hgarg = get_header_hgarg(environ)
                         if not hgarg.startswith('cmds='):
                             action = 'push' # paranoid and safe
                             break
                         action = 'pull'
                         for cmd_arg in hgarg[5:].split(';'):
                             cmd, _args = urllib.parse.unquote_plus(cmd_arg).split(' ', 1)
                             op = cmd_mapping.get(cmd, 'push')
                             if op != 'pull':
                                 assert op == 'push'
                                 action = 'push'
                                 break
                     else:
                         action = cmd_mapping.get(cmd, 'push')
                     break # only process one cmd
         return parsed_request
     def _make_app(self, parsed_request):
         """
         Make an hgweb wsgi application.
         """
         repo_name = parsed_request.repo_name
         repo_path = os.path.join(self.basepath, repo_name)
         baseui = make_ui(repo_path=repo_path)
         hgweb_app = mercurial.hgweb.hgweb(safe_bytes(repo_path), name=safe_bytes(repo_name), baseui=baseui)
         def wrapper_app(environ, start_response):
             environ['REPO_NAME'] = repo_name # used by mercurial.hgweb.hgweb
             return hgweb_app(environ, start_response)
         return wrapper_app

kallithea/config/middleware/wrapper.py

➞

Show inline comments

 # -*- 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.middleware.wrapper
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 kallithea.config.middleware.wrapper
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Wrap app to measure request and response time ... all the way to the response
 WSGI iterator has been closed.
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: May 23, 2013
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import time
 from kallithea.lib.base import _get_ip_addr, get_path_info
 log = logging.getLogger(__name__)
 class Meter:
     def __init__(self, start_response):
         self._start_response = start_response
         self._start = time.time()
         self.status = None
         self._size = 0
     def duration(self):
         return time.time() - self._start
     def start_response(self, status, response_headers, exc_info=None):
         self.status = status
         write = self._start_response(status, response_headers, exc_info)
         def metered_write(s):
             self.measure(s)
             write(s)
         return metered_write
     def measure(self, chunk):
         self._size += len(chunk)
     def size(self):
         return self._size
 class ResultIter:
     def __init__(self, result, meter, description):
         self._result_close = getattr(result, 'close', None) or (lambda: None)
         self._next = iter(result).__next__
         self._meter = meter
         self._description = description
     def __iter__(self):
         return self
     def __next__(self):
         chunk = self._next()
         self._meter.measure(chunk)
         return chunk
     def close(self):
         self._result_close()
         log.info("%s responded %r after %.3fs with %s bytes", self._description, self._meter.status, self._meter.duration(), self._meter.size())
 class RequestWrapper(object):
     def __init__(self, app, config):
         self.application = app
         self.config = config
     def __call__(self, environ, start_response):
         meter = Meter(start_response)
         description = "Request from %s for %s" % (
             _get_ip_addr(environ),
             get_path_info(environ),
+        )
         log.info("%s received", description)
         try:
             result = self.application(environ, meter.start_response)
         finally:
             log.info("%s responding %r after %.3fs", description, meter.status, meter.duration())
         return ResultIter(result, meter, description)

kallithea/lib/conf.py

➞

Show inline comments

 # -*- 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.conf
-~~~~~~~~~~~~~~~~~~~~~
 ~~~~~~~~~~~~~~~~~~
 Various config settings 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: Mar 7, 2012
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from kallithea.lib import pygmentsutils
 # language map is also used by whoosh indexer, which for those specified
 # extensions will index it's content
 LANGUAGES_EXTENSIONS_MAP = pygmentsutils.get_extension_descriptions()
 # Whoosh index targets
 # Extensions we want to index content of using whoosh
 INDEX_EXTENSIONS = list(LANGUAGES_EXTENSIONS_MAP)
 # Filenames we want to index content of using whoosh
 INDEX_FILENAMES = pygmentsutils.get_index_filenames()
 # list of readme files to search in file tree and display in summary
 # attached weights defines the search  order lower is first
 ALL_READMES = [
     ('readme', 0), ('README', 0), ('Readme', 0),
     ('doc/readme', 1), ('doc/README', 1), ('doc/Readme', 1),
     ('Docs/readme', 2), ('Docs/README', 2), ('Docs/Readme', 2),
     ('DOCS/readme', 2), ('DOCS/README', 2), ('DOCS/Readme', 2),
     ('docs/readme', 2), ('docs/README', 2), ('docs/Readme', 2),
+]
 # extension together with weights to search lower is first
 RST_EXTS = [
     ('', 0), ('.rst', 1), ('.rest', 1),
     ('.RST', 2), ('.REST', 2),
     ('.txt', 3), ('.TXT', 3)
+]
 MARKDOWN_EXTS = [
     ('.md', 1), ('.MD', 1),
     ('.mkdn', 2), ('.MKDN', 2),
     ('.mdown', 3), ('.MDOWN', 3),
     ('.markdown', 4), ('.MARKDOWN', 4)
+]
 PLAIN_EXTS = [('.text', 2), ('.TEXT', 2)]
 ALL_EXTS = MARKDOWN_EXTS + RST_EXTS + PLAIN_EXTS

kallithea/lib/webutils.py

➞

Show inline comments

 # -*- 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.webutils
 ~~~~~~~~~~~~~~~~~~~~
+~~~~~~~~~~~~~~~~~~~~~~
 Helper functions that may rely on the current WSGI request, exposed in the TG2
 thread-local "global" variables. It should have few dependencies so it can be
 imported anywhere - just like the global variables can be used everywhere.
 """
 import logging
 import random
 from tg import request, session
 from webhelpers2.html import HTML, escape, literal
 from webhelpers2.html.tags import NotGiven, Option, Options, _input
 from webhelpers2.html.tags import _make_safe_id_component as safeid
 from webhelpers2.html.tags import checkbox, end_form
 from webhelpers2.html.tags import form as insecure_form
 from webhelpers2.html.tags import hidden, link_to, password, radio
 from webhelpers2.html.tags import select as webhelpers2_select
 from webhelpers2.html.tags import submit, text, textarea
 from webhelpers2.number import format_byte_size
 from webhelpers2.text import chop_at, truncate, wrap_paragraphs
 import kallithea
 log = logging.getLogger(__name__)
 # mute pyflakes "imported but unused"
 assert Option
 assert checkbox
 assert chop_at
 assert end_form
 assert escape
 assert format_byte_size
 assert link_to
 assert literal
 assert password
 assert radio
 assert safeid
 assert submit
 assert text
 assert textarea
 assert truncate
 assert wrap_paragraphs
+#
 # General Kallithea URL handling
+#
 class UrlGenerator(object):
     """Emulate pylons.url in providing a wrapper around routes.url
     This code was added during migration from Pylons to Turbogears2. Pylons
     already provided a wrapper like this, but Turbogears2 does not.
     When the routing of Kallithea is changed to use less Routes and more
     Turbogears2-style routing, this class may disappear or change.
     url() (the __call__ method) returns the URL based on a route name and
     arguments.
     url.current() returns the URL of the current page with arguments applied.
     Refer to documentation of Routes for details:
     https://routes.readthedocs.io/en/latest/generating.html#generation
     """
     def __call__(self, *args, **kwargs):
         return request.environ['routes.url'](*args, **kwargs)
     def current(self, *args, **kwargs):
         return request.environ['routes.url'].current(*args, **kwargs)
 url = UrlGenerator()
 def canonical_url(*args, **kargs):
     '''Like url(x, qualified=True), but returns url that not only is qualified
     but also canonical, as configured in canonical_url'''
     try:
         parts = kallithea.CONFIG.get('canonical_url', '').split('://', 1)
         kargs['host'] = parts[1]
         kargs['protocol'] = parts[0]
     except IndexError:
         kargs['qualified'] = True
     return url(*args, **kargs)
 def canonical_hostname():
     '''Return canonical hostname of system'''
     try:
         parts = kallithea.CONFIG.get('canonical_url', '').split('://', 1)
         return parts[1].split('/', 1)[0]
     except IndexError:
         parts = url('home', qualified=True).split('://', 1)
         return parts[1].split('/', 1)[0]
+#
 # Custom Webhelpers2 stuff
+#
 def html_escape(s):
     """Return string with all html escaped.
     This is also safe for javascript in html but not necessarily correct.
     """
     return (s
         .replace('&', '&amp;')
         .replace(">", "&gt;")
         .replace("<", "&lt;")
         .replace('"', "&quot;")
         .replace("'", "&apos;") # Note: this is HTML5 not HTML4 and might not work in mails
+        )
 def reset(name, value, id=NotGiven, **attrs):
     """Create a reset button, similar to webhelpers2.html.tags.submit ."""
     return _input("reset", name, value, id, attrs)
 def select(name, selected_values, options, id=NotGiven, **attrs):
     """Convenient wrapper of webhelpers2 to let it accept options as a tuple list"""
     if isinstance(options, list):
         option_list = options
         # Handle old value,label lists ... where value also can be value,label lists
         options = Options()
         for x in option_list:
             if isinstance(x, tuple) and len(x) == 2:
                 value, label = x
             elif isinstance(x, str):
                 value = label = x
             else:
                 log.error('invalid select option %r', x)
                 raise
             if isinstance(value, list):
                 og = options.add_optgroup(label)
                 for x in value:
                     if isinstance(x, tuple) and len(x) == 2:
                         group_value, group_label = x
                     elif isinstance(x, str):
                         group_value = group_label = x
                     else:
                         log.error('invalid select option %r', x)
                         raise
                     og.add_option(group_label, group_value)
             else:
                 options.add_option(label, value)
     return webhelpers2_select(name, selected_values, options, id=id, **attrs)
 session_csrf_secret_name = "_session_csrf_secret_token"
 def session_csrf_secret_token():
     """Return (and create) the current session's CSRF protection token."""
     if not session_csrf_secret_name in session:
         session[session_csrf_secret_name] = str(random.getrandbits(128))
         session.save()
     return session[session_csrf_secret_name]
 def form(url, method="post", **attrs):
     """Like webhelpers.html.tags.form , but automatically adding
     session_csrf_secret_token for POST. The secret is thus never leaked in GET
     URLs.
     """
     form = insecure_form(url, method, **attrs)
     if method.lower() == 'get':
         return form
     return form + HTML.div(hidden(session_csrf_secret_name, session_csrf_secret_token()), style="display: none;")
+#
 # Flash messages, stored in cookie
+#
 class _Message(object):
     """A message returned by ``pop_flash_messages()``.
     Converting the message to a string returns the message text. Instances
     also have the following attributes:
     * ``category``: the category specified when the message was created.
     * ``message``: the html-safe message text.
     """
     def __init__(self, category, message):
         self.category = category
         self.message = message
 def _session_flash_messages(append=None, clear=False):
     """Manage a message queue in tg.session: return the current message queue
     after appending the given message, and possibly clearing the queue."""
     key = 'flash'
     if key in session:
         flash_messages = session[key]
     else:
         if append is None:  # common fast path - also used for clearing empty queue
             return []  # don't bother saving
         flash_messages = []
         session[key] = flash_messages
     if append is not None and append not in flash_messages:
         flash_messages.append(append)
     if clear:
         session.pop(key, None)
     session.save()
     return flash_messages
 def flash(message, category, logf=None):
     """
     Show a message to the user _and_ log it through the specified function
     category: notice (default), warning, error, success
     logf: a custom log function - such as log.debug
     logf defaults to log.info, unless category equals 'success', in which
     case logf defaults to log.debug.
     """
     assert category in ('error', 'success', 'warning'), category
     if hasattr(message, '__html__'):
         # render to HTML for storing in cookie
         safe_message = str(message)
     else:
         # Apply str - the message might be an exception with __str__
         # Escape, so we can trust the result without further escaping, without any risk of injection
         safe_message = html_escape(str(message))
     if logf is None:
         logf = log.info
         if category == 'success':
             logf = log.debug
     logf('Flash %s: %s', category, safe_message)
     _session_flash_messages(append=(category, safe_message))
 def pop_flash_messages():
     """Return all accumulated messages and delete them from the session.
     The return value is a list of ``Message`` objects.
     """
     return [_Message(category, message) for category, message in _session_flash_messages(clear=True)]

kallithea/model/userlog.py

➞

Show inline comments

 # -*- 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.db
 ~~~~~~~~~~~~~~~~~~
 kallithea.model.userlog
 ~~~~~~~~~~~~~~~~~~~~~~~
 Database Models 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 08, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import datetime
 import logging
 from kallithea.lib.utils2 import get_current_authuser
 from kallithea.model import db, meta
 log = logging.getLogger(__name__)
 def action_logger(user, action, repo, ipaddr='', commit=False):
     """
     Action logger for various actions made by users
     :param user: user that made this action, can be a unique username string or
         object containing user_id attribute
     :param action: action to log, should be on of predefined unique actions for
         easy translations
     :param repo: string name of repository or object containing repo_id,
         that action was made on
     :param ipaddr: optional IP address from what the action was made
     """
     # if we don't get explicit IP address try to get one from registered user
     # in tmpl context var
     if not ipaddr:
         ipaddr = getattr(get_current_authuser(), 'ip_addr', '')
     if getattr(user, 'user_id', None):
         user_obj = db.User.get(user.user_id)
     elif isinstance(user, str):
         user_obj = db.User.get_by_username(user)
     else:
         raise Exception('You have to provide a user object or a username')
     if getattr(repo, 'repo_id', None):
         repo_obj = db.Repository.get(repo.repo_id)
         repo_name = repo_obj.repo_name
     elif isinstance(repo, str):
         repo_name = repo.lstrip('/')
         repo_obj = db.Repository.get_by_repo_name(repo_name)
     else:
         repo_obj = None
         repo_name = ''
     user_log = db.UserLog()
     user_log.user_id = user_obj.user_id
     user_log.username = user_obj.username
     user_log.action = action
     user_log.repository = repo_obj
     user_log.repository_name = repo_name
     user_log.action_date = datetime.datetime.now()
     user_log.user_ip = ipaddr
     meta.Session().add(user_log)
     log.info('Logging action:%s on %s by user:%s ip:%s',
              action, repo, user_obj, ipaddr)
     if commit:
         meta.Session().commit()

scripts/run-all-cleanup

➞

Show inline comments

 #!/bin/sh
 # Convenience script for running various idempotent source code cleanup scripts
 set -e
 set -x
 scripts/docs-headings.py
 scripts/generate-ini.py
 scripts/whitespacecleanup.sh
 hg files 'set:!binary()&grep("^#!.*python")' 'set:**.py' | xargs scripts/source_format.py
 hg files 'set:!binary()&grep("^#!.*python")' 'set:**.py' | xargs scripts/pyflakes
 echo "no blocking problems found by $0"

scripts/source_format.py

➞

Show inline comments

@@ new file 100755 @@
 #!/usr/bin/env python3
 # hg files 'set:!binary()&grep("^#!.*python")' 'set:**.py' | xargs scripts/source_format.py
 import re
 import sys
 filenames = sys.argv[1:]
 for fn in filenames:
     with open(fn) as f:
         org_content = f.read()
     mod_name = fn[:-3] if fn.endswith('.py') else fn
     mod_name = mod_name[:-9] if mod_name.endswith('/__init__') else mod_name
     mod_name = mod_name.replace('/', '.')
     def f(m):
         return '"""\n%s\n%s\n' % (mod_name, '~' * len(mod_name))
     new_content = re.sub(r'^"""\n(kallithea\..*\n)(~+\n)?', f, org_content, count=1, flags=re.MULTILINE)
     if new_content != org_content:
         with open(fn, 'w') as f:
             f.write(new_content)

0 comments (0 inline, 0 general)