# 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
return op
# try to fallback to stored variable as we don't know if the last
# operation is pull/push
op = getattr(self, '_git_stored_op', 'pull')
def _handle_githooks(self, repo_name, action, baseui, environ):
Handles pull action, push is handled by post-receive hook
from rhodecode.lib.hooks import log_pull_action
if len(service) < 2:
from rhodecode.model.db import Repository
_repo = Repository.get_by_repo_name(repo_name)
_repo = _repo.scm_instance
_repo._repo.ui = baseui
_hooks = dict(baseui.configitems('hooks')) or {}
if action == 'pull':
# stupid git, emulate pre-pull hook !
pre_pull(ui=baseui, repo=_repo._repo)
if action == 'pull' and _hooks.get(RhodeCodeUi.HOOK_PULL):
log_pull_action(ui=baseui, repo=_repo._repo)
def __inject_extras(self, repo_path, baseui, extras={}):
Injects some extra params into baseui instance
:param baseui: baseui instance
:param extras: dict with extra params to put into baseui
# make our hgweb quiet so it doesn't print output
baseui.setconfig('ui', 'quiet', 'true')
#inject some additional parameters that will be available in ui
#for hooks
for k, v in extras.items():
baseui.setconfig('rhodecode_extras', k, v)
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 ?')
also overwrites global settings with those takes from local hgrc file
hgrc = os.path.join(repo_path, '.hg', 'hgrc')
repoui = make_ui('file', hgrc, False)
if repoui:
#overwrite our ui instance with the section from hgrc file
for section in ui_sections:
for k, v in repoui.configitems(section):
baseui.setconfig(section, k, v)
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'):
elif line.endswith('\r'):
return 1
return default
def generate_api_key(username, salt=None):
Generates unique API key for given username, if salt is not given
it'll be generated from some random string
:param username: username as string
:param salt: salt to hash generate KEY
:returns: sha1 hash from username+salt
from tempfile import _RandomNameSequence
import hashlib
if salt is None:
salt = _RandomNameSequence().next()
return hashlib.sha1(username + salt).hexdigest()
def safe_int(val, default=None):
Returns int() of val if val is not convertable to int use default
instead
:param val:
:param default:
val = int(val)
except ValueError:
val = default
return val
def safe_unicode(str_, from_encoding=None):
safe unicode function. Does few trick to turn str_ into unicode
In case of UnicodeDecode error we try to return it with encoding detected
by chardet library if it fails fallback to unicode with errors replaced
:param str_: string to decode
:rtype: unicode
:returns: unicode object
if isinstance(str_, unicode):
return str_
if not from_encoding:
import rhodecode
DEFAULT_ENCODING = rhodecode.CONFIG.get('default_encoding','utf8')
from_encoding = DEFAULT_ENCODING
return unicode(str_)
except UnicodeDecodeError:
pass
return unicode(str_, from_encoding)
import chardet
encoding = chardet.detect(str_)['encoding']
if encoding is None:
raise Exception()
return str_.decode(encoding)
except (ImportError, UnicodeDecodeError, Exception):
return unicode(str_, from_encoding, 'replace')
def safe_str(unicode_, to_encoding=None):
safe str function. Does few trick to turn unicode_ into string
In case of UnicodeEncodeError we try to return it with encoding detected
by chardet library if it fails fallback to string with errors replaced
:param unicode_: unicode to encode
:returns: str object
# if it's not basestr cast to str
if not isinstance(unicode_, basestring):
return str(unicode_)
@@ -327,192 +329,197 @@ def age(prevdate):
# 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
deltas['day'] += month_lengths[prevdate.month - 1]
deltas['month'] -= 1
if deltas['month'] < 0:
deltas['month'] += 12
deltas['year'] -= 1
# Format the result
fmt_funcs = {
'year': lambda d: ungettext(u'%d year', '%d years', d) % d,
'month': lambda d: ungettext(u'%d month', '%d months', d) % d,
'day': lambda d: ungettext(u'%d day', '%d days', d) % d,
'hour': lambda d: ungettext(u'%d hour', '%d hours', d) % d,
'minute': lambda d: ungettext(u'%d minute', '%d minutes', d) % d,
'second': lambda d: ungettext(u'%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]
sub_value = 0
if sub_value == 0:
if future:
return _(u'in %s') % fmt_funcs[part](value)
return _(u'%s ago') % fmt_funcs[part](value)
return _(u'in %s and %s') % (fmt_funcs[part](value),
fmt_funcs[sub_part](sub_value))
return _(u'%s and %s ago') % (fmt_funcs[part](value),
return _(u'just now')
def uri_filter(uri):
Removes user:password from given url string
:param uri:
:returns: filtered list of strings
if not uri:
return ''
proto = ''
for pat in ('https://', 'http://'):
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
host, port = uri[:cred_pos], uri[cred_pos + 1:]
return filter(None, [proto, host, port])
def credentials_filter(uri):
Returns a url with removed credentials
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)
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: