# Additional mappings that are not present in the pygments lexers
# used for building stats
# format is {'ext':['Names']} eg. {'py':['Python']} note: there can be
# more than one name for extension
# NOTE: that this will overide any mappings in LANGUAGES_EXTENSIONS_MAP
# build by pygments
EXTRA_MAPPINGS = {}
#==============================================================================
# WHOOSH INDEX EXTENSIONS
# if INDEX_EXTENSIONS is [] it'll use pygments lexers extensions by default.
# To set your own just add to this list extensions to index with content
INDEX_EXTENSIONS = []
# additional extensions for indexing besides the default from pygments
# those get's added to INDEX_EXTENSIONS
EXTRA_INDEX_EXTENSIONS = []
# POST CREATE REPOSITORY HOOK
# this function will be executed after each repository is created
def _crhook(*args, **kwargs):
"""
Post create repository HOOK
kwargs available:
:param repo_name:
:param repo_type:
:param description:
:param private:
:param created_on:
:param enable_downloads:
:param repo_id:
:param user_id:
:param enable_statistics:
:param clone_uri:
:param fork_id:
:param group_id:
:param created_by:
return 0
CREATE_REPO_HOOK = _crhook
# POST DELETE REPOSITORY HOOK
# this function will be executed after each repository deletion
def _dlhook(*args, **kwargs):
:param deleted_by:
:param deleted_on:
DELETE_REPO_HOOK = _dlhook
# POST PUSH HOOK
# this function will be executed after each push it's runned after the build-in
# hook that rhodecode uses for logging pushes
# this function will be executed after each push it's executed after the
# build-in hook that RhodeCode uses for logging pushes
def _pushhook(*args, **kwargs):
Post push hook
:param server_url: url of instance that triggered this hook
:param config: path to .ini config used
:param scm: type of VS 'git' or 'hg'
:param username: name of user who pushed
:param ip: ip of who pushed
:param action: pull
:param action: push
:param repository: repository name
:param pushed_revs: generator of pushed revisions
:param pushed_revs: list of pushed revisions
PUSH_HOOK = _pushhook
# POST PULL HOOK
# build-in hook that RhodeCode uses for logging pulls
def _pullhook(*args, **kwargs):
Post pull hook
kwargs available::
:param username: name of user who pulled
:param ip: ip of who pulled
PULL_HOOK = _pullhook
# -*- coding: utf-8 -*-
rhodecode.lib.middleware.simplegit
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
SimpleGit middleware for handling git protocol request (push/clone etc.)
It's implemented with basic auth function
:created_on: Apr 28, 2010
:author: marcink
:copyright: (C) 2010-2012 Marcin Kuzminski <marcin@python-works.com>
:license: GPLv3, see COPYING for more details.
# 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/>.
import os
import re
import logging
import traceback
from dulwich import server as dulserver
from dulwich.web import LimitedInputFilter, GunzipFilter
from rhodecode.lib.exceptions import HTTPLockedRC
from rhodecode.lib.hooks import pre_pull
class SimpleGitUploadPackHandler(dulserver.UploadPackHandler):
def handle(self):
write = lambda x: self.proto.write_sideband(1, x)
graph_walker = dulserver.ProtocolGraphWalker(self,
self.repo.object_store,
self.repo.get_peeled)
objects_iter = self.repo.fetch_objects(
graph_walker.determine_wants, graph_walker, self.progress,
get_tagged=self.get_tagged)
# Did the process short-circuit (e.g. in a stateless RPC call)? Note
# that the client still expects a 0-object pack in most cases.
if objects_iter is None:
return
self.progress("counting objects: %d, done.\n" % len(objects_iter))
dulserver.write_pack_objects(dulserver.ProtocolFile(None, write),
objects_iter)
messages = []
messages.append('thank you for using rhodecode')
for msg in messages:
self.progress(msg + "\n")
# we are done
self.proto.write("0000")
dulserver.DEFAULT_HANDLERS = {
#git-ls-remote, git-clone, git-fetch and git-pull
'git-upload-pack': SimpleGitUploadPackHandler,
#git-push
'git-receive-pack': dulserver.ReceivePackHandler,
}
# not used for now until dulwich get's fixed
#from dulwich.repo import Repo
#from dulwich.web import make_wsgi_chain
from paste.httpheaders import REMOTE_USER, AUTH_TYPE
from webob.exc import HTTPNotFound, HTTPForbidden, HTTPInternalServerError, \
HTTPBadRequest, HTTPNotAcceptable
from rhodecode.lib.utils2 import safe_str, fix_PATH
from rhodecode.lib.utils2 import safe_str, fix_PATH, get_server_url
from rhodecode.lib.base import BaseVCSController
from rhodecode.lib.auth import get_container_username
from rhodecode.lib.utils import is_valid_repo, make_ui
from rhodecode.lib.compat import json
from rhodecode.model.db import User, RhodeCodeUi
log = logging.getLogger(__name__)
GIT_PROTO_PAT = re.compile(r'^/(.+)/(info/refs|git-upload-pack|git-receive-pack)')
def is_git(environ):
path_info = environ['PATH_INFO']
isgit_path = GIT_PROTO_PAT.match(path_info)
log.debug('pathinfo: %s detected as GIT %s' % (
path_info, isgit_path != None)
)
return isgit_path
class SimpleGit(BaseVCSController):
def _handle_request(self, environ, start_response):
if not is_git(environ):
return self.application(environ, start_response)
if not self._check_ssl(environ, start_response):
return HTTPNotAcceptable('SSL REQUIRED !')(environ, start_response)
ipaddr = self._get_ip_addr(environ)
username = None
self._git_first_op = False
# skip passing error to error controller
environ['pylons.status_code_redirect'] = True
#======================================================================
# EXTRACT REPOSITORY NAME FROM ENV
try:
repo_name = self.__get_repository(environ)
log.debug('Extracted repo name is %s' % repo_name)
except:
return HTTPInternalServerError()(environ, start_response)
# quick check if that dir exists...
if is_valid_repo(repo_name, self.basepath, 'git') is False:
return HTTPNotFound()(environ, start_response)
# GET ACTION PULL or PUSH
action = self.__get_action(environ)
# CHECK ANONYMOUS PERMISSION
if action in ['pull', 'push']:
anonymous_user = self.__get_user('default')
username = anonymous_user.username
anonymous_perm = self._check_permission(action, anonymous_user,
repo_name)
if anonymous_perm is not True or anonymous_user.active is False:
if anonymous_perm is not True:
log.debug('Not enough credentials to access this '
'repository as anonymous user')
if anonymous_user.active is False:
log.debug('Anonymous access is disabled, running '
'authentication')
#==============================================================
# DEFAULT PERM FAILED OR ANONYMOUS ACCESS IS DISABLED SO WE
# NEED TO AUTHENTICATE AND ASK FOR AUTH USER PERMISSIONS
# Attempting to retrieve username from the container
username = get_container_username(environ, self.config)
# If not authenticated by the container, running basic auth
if not username:
self.authenticate.realm = \
safe_str(self.config['rhodecode_realm'])
result = self.authenticate(environ)
if isinstance(result, str):
AUTH_TYPE.update(environ, 'basic')
REMOTE_USER.update(environ, result)
username = result
else:
return result.wsgi_application(environ, start_response)
# CHECK PERMISSIONS FOR THIS REQUEST USING GIVEN USERNAME
user = self.__get_user(username)
if user is None or not user.active:
return HTTPForbidden()(environ, start_response)
username = user.username
log.error(traceback.format_exc())
#check permissions for this repository
perm = self._check_permission(action, user, repo_name)
if perm is not True:
# extras are injected into UI object and later available
# in hooks executed by rhodecode
from rhodecode import CONFIG
server_url = get_server_url(environ)
extras = {
'ip': ipaddr,
'username': username,
'action': action,
'repository': repo_name,
'scm': 'git',
'config': CONFIG['__file__'],
'server_url': server_url,
'make_lock': None,
'locked_by': [None, None]
#===================================================================
# GIT REQUEST HANDLING
repo_path = os.path.join(safe_str(self.basepath), safe_str(repo_name))
log.debug('Repository path is %s' % repo_path)
# CHECK LOCKING only if it's not ANONYMOUS USER
if username != User.DEFAULT_USER:
log.debug('Checking locking on repository')
(make_lock,
locked,
locked_by) = self._check_locking_state(
environ=environ, action=action,
repo=repo_name, user_id=user.user_id
# store the make_lock for later evaluation in hooks
extras.update({'make_lock': make_lock,
'locked_by': locked_by})
# set the environ variables for this request
os.environ['RC_SCM_DATA'] = json.dumps(extras)
fix_PATH()
log.debug('HOOKS extras is %s' % extras)
baseui = make_ui('db')
self.__inject_extras(repo_path, baseui, extras)
# invalidate cache on push
if action == 'push':
self._invalidate_cache(repo_name)
self._handle_githooks(repo_name, action, baseui, environ)
log.info('%s action on GIT repo "%s"' % (action, repo_name))
app = self.__make_app(repo_name, repo_path, extras)
return app(environ, start_response)
except HTTPLockedRC, e:
log.debug('Repositry LOCKED ret code 423!')
return e(environ, start_response)
except Exception:
def __make_app(self, repo_name, repo_path, extras):
Make an wsgi application using dulserver
:param repo_name: name of the repository
:param repo_path: full path to the repository
from rhodecode.lib.middleware.pygrack import make_wsgi_app
app = make_wsgi_app(
repo_root=safe_str(self.basepath),
repo_name=repo_name,
extras=extras,
app = GunzipFilter(LimitedInputFilter(app))
return app
def __get_repository(self, environ):
Get's repository name out of PATH_INFO header
:param environ: environ where PATH_INFO is stored
environ['PATH_INFO'] = self._get_by_id(environ['PATH_INFO'])
repo_name = GIT_PROTO_PAT.match(environ['PATH_INFO']).group(1)
raise
return repo_name
def __get_user(self, username):
return User.get_by_username(username)
def __get_action(self, environ):
Maps git request commands into a pull or push command.
:param environ:
service = environ['QUERY_STRING'].split('=')
if len(service) > 1:
service_cmd = service[1]
mapping = {
'git-receive-pack': 'push',
'git-upload-pack': 'pull',
op = mapping[service_cmd]
self._git_stored_op = op
rhodecode.lib.middleware.simplehg
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
SimpleHG middleware for handling mercurial protocol request
(push/clone etc.). It's implemented with basic auth function
from mercurial.error import RepoError
from mercurial.hgweb import hgweb_mod
from rhodecode.lib.utils import make_ui, is_valid_repo, ui_sections
from rhodecode.model.db import User
def is_mercurial(environ):
Returns True if request's target is mercurial server - header
``HTTP_ACCEPT`` of such request would start with ``application/mercurial``.
http_accept = environ.get('HTTP_ACCEPT')
if http_accept and http_accept.startswith('application/mercurial'):
ishg_path = True
ishg_path = False
log.debug('pathinfo: %s detected as HG %s' % (
path_info, ishg_path)
return ishg_path
class SimpleHg(BaseVCSController):
if not is_mercurial(environ):
repo_name = environ['REPO_NAME'] = self.__get_repository(environ)
if is_valid_repo(repo_name, self.basepath, 'hg') is False:
# extras are injected into mercurial UI object and later available
# in hg hooks executed by rhodecode
'scm': 'hg',
# MERCURIAL REQUEST HANDLING
log.info('%s action on HG repo "%s"' % (action, repo_name))
app = self.__make_app(repo_path, baseui, extras)
except RepoError, e:
if str(e).find('not found') != -1:
def __make_app(self, repo_name, baseui, extras):
Make an wsgi application using hgweb, and inject generated baseui
instance, additionally inject some extras into ui object
return hgweb_mod.hgweb(repo_name, name=repo_name, baseui=baseui)
repo_name = '/'.join(environ['PATH_INFO'].split('/')[1:])
if repo_name.endswith('/'):
repo_name = repo_name.rstrip('/')
Maps mercurial request commands into a clone,pull or push command.
This should always return a valid command string
mapping = {'changegroup': 'pull',
'changegroupsubset': 'pull',
'stream_out': 'pull',
'listkeys': 'pull',
'unbundle': 'push',
'pushkey': 'push', }
for qry in environ['QUERY_STRING'].split('&'):
if qry.startswith('cmd'):
cmd = qry.split('=')[-1]
if cmd in mapping:
return mapping[cmd]
return 'pull'
raise Exception('Unable to detect pull/push action !!'
'Are you using non standard command or client ?')
rhodecode.lib.utils
~~~~~~~~~~~~~~~~~~~
Some simple helper functions
:created_on: Jan 5, 2011
:copyright: (C) 2011-2012 Marcin Kuzminski <marcin@python-works.com>
import time
import datetime
import webob
from pylons.i18n.translation import _, ungettext
from rhodecode.lib.vcs.utils.lazy import LazyProperty
def __get_lem():
Get language extension map based on what's inside pygments lexers
from pygments import lexers
from string import lower
from collections import defaultdict
d = defaultdict(lambda: [])
def __clean(s):
s = s.lstrip('*')
s = s.lstrip('.')
if s.find('[') != -1:
exts = []
start, stop = s.find('['), s.find(']')
for suffix in s[start + 1:stop]:
exts.append(s[:s.find('[')] + suffix)
return map(lower, exts)
return map(lower, [s])
for lx, t in sorted(lexers.LEXERS.items()):
m = map(__clean, t[-2])
if m:
m = reduce(lambda x, y: x + y, m)
for ext in m:
desc = lx.replace('Lexer', '')
d[ext].append(desc)
return dict(d)
def str2bool(_str):
returs True/False value from given string, it tries to translate the
string into boolean
:param _str: string value to translate into boolean
:rtype: boolean
:returns: boolean from given string
if _str is None:
return False
if _str in (True, False):
return _str
_str = str(_str).strip().lower()
return _str in ('t', 'true', 'y', 'yes', 'on', '1')
def convert_line_endings(line, mode):
Converts a given line "line end" accordingly 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
from string import replace
if mode == 0:
line = replace(line, '\r\n', '\n')
line = replace(line, '\r', '\n')
elif mode == 1:
line = replace(line, '\r\n', '\r')
line = replace(line, '\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'):
@@ -423,96 +425,101 @@ def uri_filter(uri):
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_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 rhodecode.lib.vcs.backends.base import BaseRepository
from rhodecode.lib.vcs.exceptions import RepositoryError
from rhodecode.lib.vcs.backends.base import EmptyChangeset
if not isinstance(repo, BaseRepository):
raise Exception('You must pass an Repository '
'object as first argument got %s', type(repo))
cs = repo.get_changeset(rev)
except RepositoryError:
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, basestring):
tm = float(tm)
except ValueError:
return datetime.datetime.fromtimestamp(tm)
MENTIONS_REGEX = r'(?:^@|\s@)([a-zA-Z0-9]{1}[a-zA-Z0-9\-\_\.]+)(?:\s{1})'
def extract_mentioned_users(s):
Returns unique usernames from given string s that have @mention
:param s: string to get mentions
usrs = set()
for username in re.findall(MENTIONS_REGEX, s):
usrs.add(username)
return sorted(list(usrs), key=lambda k: k.lower())
class AttributeDict(dict):
def __getattr__(self, attr):
return self.get(attr, None)
__setattr__ = dict.__setitem__
__delattr__ = dict.__delitem__
def fix_PATH(os_=None):
Get current active python path, and append it to PATH variable to fix issues
of subprocess calls and different python versions
import sys
if os_ is None:
os = os_
cur_path = os.path.split(sys.executable)[0]
if not os.environ['PATH'].startswith(cur_path):
os.environ['PATH'] = '%s:%s' % (cur_path, os.environ['PATH'])
def obfuscate_url_pw(engine):
from sqlalchemy.engine import url
url = url.make_url(engine)
if url.password:
url.password = 'XXXXX'
return str(url)
def get_server_url(environ):
req = webob.Request(environ)
return req.host_url + req.script_name
Status change: