# -*- coding: utf-8 -*-
"""
rhodecode.controllers.admin.admin
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Controller for Admin panel of Rhodecode
:created_on: Apr 7, 2010
:author: marcink
:copyright: (C) 2009-2010 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; version 2
# of the License or (at your opinion) any later version of the license.
#
# 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, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
# MA 02110-1301, USA.
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import logging
from pylons import request, tmpl_context as c
from rhodecode.lib.base import BaseController, render
from rhodecode.model.db import UserLog
from webhelpers.paginate import Page
from rhodecode.lib.auth import LoginRequired, HasPermissionAllDecorator
log = logging.getLogger(__name__)
class AdminController(BaseController):
@LoginRequired()
def __before__(self):
super(AdminController, self).__before__()
@HasPermissionAllDecorator('hg.admin')
def index(self):
users_log = self.sa.query(UserLog).order_by(UserLog.action_date.desc())
p = int(request.params.get('page', 1))
c.users_log = Page(users_log, page=p, items_per_page=10)
c.log_data = render('admin/admin_log.html')
if request.params.get('partial'):
return c.log_data
return render('admin/admin.html')
rhodecode.controllers.admin.ldap_settings
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
ldap controller for RhodeCode
:created_on: Nov 26, 2010
import formencode
import traceback
from formencode import htmlfill
from pylons import request, response, session, tmpl_context as c, url
from pylons.controllers.util import abort, redirect
from pylons.i18n.translation import _
from rhodecode.lib import helpers as h
from rhodecode.lib.auth_ldap import LdapImportError
from rhodecode.model.settings import SettingsModel
from rhodecode.model.forms import LdapSettingsForm
from sqlalchemy.exc import DatabaseError
class LdapSettingsController(BaseController):
c.admin_user = session.get('admin_user')
c.admin_username = session.get('admin_username')
super(LdapSettingsController, self).__before__()
defaults = SettingsModel().get_ldap_settings()
return htmlfill.render(
render('admin/ldap/ldap.html'),
defaults=defaults,
encoding="UTF-8",
force_defaults=True,)
def ldap_settings(self):
"""POST ldap create and store ldap settings"""
settings_model = SettingsModel()
post_data = dict(request.POST)
_form = LdapSettingsForm(post_data.get('ldap_active'))()
try:
form_result = _form.to_python(post_data)
for k, v in form_result.items():
if k.startswith('ldap_'):
setting = settings_model.get(k)
setting.app_settings_value = v
self.sa.add(setting)
self.sa.commit()
h.flash(_('Ldap settings updated successfully'),
category='success')
except (DatabaseError,):
raise
except LdapImportError:
h.flash(_('Unable to activate ldap. The "python-ldap" library '
'is missing.'), category='warning')
except formencode.Invalid, errors:
defaults=errors.value,
errors=errors.error_dict or {},
prefix_error=False,
encoding="UTF-8")
except Exception:
log.error(traceback.format_exc())
h.flash(_('error occurred during update of ldap settings'),
category='error')
return redirect(url('ldap_home'))
rhodecode.controllers.admin.permissions
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
permissions controller for Rhodecode
:created_on: Apr 27, 2010
from pylons import request, session, tmpl_context as c, url
from rhodecode.model.forms import LdapSettingsForm, DefaultPermissionsForm
from rhodecode.model.permission import PermissionModel
from rhodecode.model.user import UserModel
class PermissionsController(BaseController):
"""REST Controller styled on the Atom Publishing Protocol"""
# To properly map this controller, ensure your config/routing.py
# file has a resource setup:
# map.resource('permission', 'permissions')
super(PermissionsController, self).__before__()
self.perms_choices = [('repository.none', _('None'),),
('repository.read', _('Read'),),
('repository.write', _('Write'),),
('repository.admin', _('Admin'),)]
self.register_choices = [
('hg.register.none',
_('disabled')),
('hg.register.manual_activate',
_('allowed with manual account activation')),
('hg.register.auto_activate',
_('allowed with automatic account activation')), ]
self.create_choices = [('hg.create.none', _('Disabled')),
('hg.create.repository', _('Enabled'))]
def index(self, format='html'):
"""GET /permissions: All items in the collection"""
# url('permissions')
def create(self):
"""POST /permissions: Create a new item"""
def new(self, format='html'):
"""GET /permissions/new: Form to create a new item"""
# url('new_permission')
def update(self, id):
"""PUT /permissions/id: Update an existing item"""
# Forms posted to this method should contain a hidden field:
# <input type="hidden" name="_method" value="PUT" />
# Or using helpers:
# h.form(url('permission', id=ID),
# method='put')
# url('permission', id=ID)
permission_model = PermissionModel()
_form = DefaultPermissionsForm([x[0] for x in self.perms_choices],
[x[0] for x in self.register_choices],
[x[0] for x in self.create_choices])()
form_result = _form.to_python(dict(request.POST))
form_result.update({'perm_user_name':id})
permission_model.update(form_result)
h.flash(_('Default permissions updated successfully'),
c.perms_choices = self.perms_choices
c.register_choices = self.register_choices
c.create_choices = self.create_choices
defaults = errors.value
render('admin/permissions/permissions.html'),
h.flash(_('error occurred during update of permissions'),
return redirect(url('edit_permission', id=id))
def delete(self, id):
"""DELETE /permissions/id: Delete an existing item"""
# <input type="hidden" name="_method" value="DELETE" />
# method='delete')
def show(self, id, format='html'):
"""GET /permissions/id: Show a specific item"""
def edit(self, id, format='html'):
"""GET /permissions/id/edit: Form to edit an existing item"""
#url('edit_permission', id=ID)
if id == 'default':
default_user = UserModel().get_by_username('default')
defaults = {'_method':'put',
'anonymous':default_user.active}
for p in default_user.user_perms:
if p.permission.permission_name.startswith('repository.'):
defaults['default_perm'] = p.permission.permission_name
if p.permission.permission_name.startswith('hg.register.'):
defaults['default_register'] = p.permission.permission_name
if p.permission.permission_name.startswith('hg.create.'):
defaults['default_create'] = p.permission.permission_name
else:
return redirect(url('admin_home'))
rhodecode.controllers.admin.repos
Admin controller for RhodeCode
:copyright: (C) 2009-2011 Marcin Kuzminski <marcin@python-works.com>
from operator import itemgetter
from paste.httpexceptions import HTTPInternalServerError
from rhodecode.lib.auth import LoginRequired, HasPermissionAllDecorator, \
HasPermissionAnyDecorator
from rhodecode.lib.utils import invalidate_cache, action_logger
from rhodecode.model.db import User
from rhodecode.model.forms import RepoForm
from rhodecode.model.scm import ScmModel
from rhodecode.model.repo import RepoModel
class ReposController(BaseController):
# map.resource('repo', 'repos')
@HasPermissionAnyDecorator('hg.admin', 'hg.create.repository')
super(ReposController, self).__before__()
"""GET /repos: All items in the collection"""
# url('repos')
cached_repo_list = ScmModel().get_repos()
c.repos_list = sorted(cached_repo_list, key=itemgetter('name_sort'))
return render('admin/repos/repos.html')
"""POST /repos: Create a new item"""
repo_model = RepoModel()
_form = RepoForm()()
form_result = {}
repo_model.create(form_result, c.rhodecode_user)
h.flash(_('created repository %s') % form_result['repo_name'],
if request.POST.get('user_created'):
action_logger(self.rhodecode_user, 'user_created_repo',
form_result['repo_name'], '', self.sa)
action_logger(self.rhodecode_user, 'admin_created_repo',
c.new_repo = errors.value['repo_name']
r = render('admin/repos/repo_add_create_repository.html')
r = render('admin/repos/repo_add.html')
r,
msg = _('error occurred during creation of repository %s') \
% form_result.get('repo_name')
h.flash(msg, category='error')
return redirect(url('home'))
return redirect(url('repos'))
"""GET /repos/new: Form to create a new item"""
new_repo = request.GET.get('repo', '')
c.new_repo = h.repo_name_slug(new_repo)
return render('admin/repos/repo_add.html')
def update(self, repo_name):
"""PUT /repos/repo_name: Update an existing item"""
# h.form(url('repo', repo_name=ID),
# url('repo', repo_name=ID)
changed_name = repo_name
_form = RepoForm(edit=True, old_data={'repo_name':repo_name})()
repo_model.update(repo_name, form_result)
invalidate_cache('get_repo_cached_%s' % repo_name)
h.flash(_('Repository %s updated successfully' % repo_name),
changed_name = form_result['repo_name']
action_logger(self.rhodecode_user, 'admin_updated_repo',
changed_name, '', self.sa)
c.repo_info = repo_model.get_by_repo_name(repo_name)
if c.repo_info.stats:
last_rev = c.repo_info.stats.stat_on_revision
last_rev = 0
c.stats_revision = last_rev
r = ScmModel().get(repo_name)
c.repo_last_rev = r.revisions[-1] if r.revisions else 0
if last_rev == 0:
c.stats_percentage = 0
c.stats_percentage = '%.2f' % ((float((last_rev)) /
c.repo_last_rev) * 100)
c.users_array = repo_model.get_users_js()
errors.value.update({'user':c.repo_info.user.username})
render('admin/repos/repo_edit.html'),
h.flash(_('error occurred during update of repository %s') \
% repo_name, category='error')
return redirect(url('edit_repo', repo_name=changed_name))
def delete(self, repo_name):
"""DELETE /repos/repo_name: Delete an existing item"""
repo = repo_model.get_by_repo_name(repo_name)
if not repo:
h.flash(_('%s repository is not mapped to db perhaps'
' it was moved or renamed from the filesystem'
' please run the application again'
' in order to rescan repositories') % repo_name,
action_logger(self.rhodecode_user, 'admin_deleted_repo',
repo_name, '', self.sa)
repo_model.delete(repo)
h.flash(_('deleted repository %s') % repo_name, category='success')
except Exception, e:
h.flash(_('An error occurred during deletion of %s') % repo_name,
def delete_perm_user(self, repo_name):
DELETE an existing repository permission user
:param repo_name:
repo_model.delete_perm_user(request.POST, repo_name)
h.flash(_('An error occurred during deletion of repository user'),
raise HTTPInternalServerError()
def repo_stats(self, repo_name):
DELETE an existing repository statistics
repo_model.delete_stats(repo_name)
h.flash(_('An error occurred during deletion of repository stats'),
return redirect(url('edit_repo', repo_name=repo_name))
def repo_cache(self, repo_name):
INVALIDATE existing repository cache
ScmModel().mark_for_invalidation(repo_name)
h.flash(_('An error occurred during cache invalidation'),
def show(self, repo_name, format='html'):
"""GET /repos/repo_name: Show a specific item"""
def edit(self, repo_name, format='html'):
"""GET /repos/repo_name/edit: Form to edit an existing item"""
# url('edit_repo', repo_name=ID)
if c.repo_info is None:
' it was created or renamed from the filesystem'
if last_rev == 0 or c.repo_last_rev == 0:
defaults = c.repo_info.get_dict()
if c.repo_info.user:
defaults.update({'user':c.repo_info.user.username})
replacement_user = self.sa.query(User)\
.filter(User.admin == True).first().username
defaults.update({'user':replacement_user})
for p in c.repo_info.repo_to_perm:
defaults.update({'perm_%s' % p.user.username:
p.permission.permission_name})
force_defaults=False
)
rhodecode.controllers.admin.settings
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
settings controller for rhodecode admin
:created_on: Jul 14, 2010
from pylons import request, session, tmpl_context as c, url, app_globals as g, \
config
HasPermissionAnyDecorator, NotAnonymous
from rhodecode.lib.celerylib import tasks, run_task
from rhodecode.lib.utils import repo2db_mapper, invalidate_cache, \
set_rhodecode_config
from rhodecode.model.db import RhodeCodeUi, Repository
from rhodecode.model.forms import UserForm, ApplicationSettingsForm, \
ApplicationUiSettingsForm
from sqlalchemy import func
class SettingsController(BaseController):
# map.resource('setting', 'settings', controller='admin/settings',
# path_prefix='/admin', name_prefix='admin_')
super(SettingsController, self).__before__()
"""GET /admin/settings: All items in the collection"""
# url('admin_settings')
defaults = SettingsModel().get_app_settings()
defaults.update(self.get_hg_ui_settings())
render('admin/settings/settings.html'),
"""POST /admin/settings: Create a new item"""
"""GET /admin/settings/new: Form to create a new item"""
# url('admin_new_setting')
def update(self, setting_id):
"""PUT /admin/settings/setting_id: Update an existing item"""
# h.form(url('admin_setting', setting_id=ID),
# url('admin_setting', setting_id=ID)
if setting_id == 'mapping':
rm_obsolete = request.POST.get('destroy', False)
log.debug('Rescanning directories with destroy=%s', rm_obsolete)
initial = ScmModel().repo_scan(g.paths[0][1], g.baseui)
for repo_name in initial.keys():
repo2db_mapper(initial, rm_obsolete)
h.flash(_('Repositories successfully rescanned'), category='success')
if setting_id == 'whoosh':
repo_location = self.get_hg_ui_settings()['paths_root_path']
full_index = request.POST.get('full_index', False)
task = run_task(tasks.whoosh_index, repo_location, full_index)
h.flash(_('Whoosh reindex task scheduled'), category='success')
if setting_id == 'global':
application_form = ApplicationSettingsForm()()
form_result = application_form.to_python(dict(request.POST))
hgsettings1 = settings_model.get('title')
hgsettings1.app_settings_value = form_result['rhodecode_title']
hgsettings2 = settings_model.get('realm')
hgsettings2.app_settings_value = form_result['rhodecode_realm']
self.sa.add(hgsettings1)
self.sa.add(hgsettings2)
set_rhodecode_config(config)
h.flash(_('Updated application settings'),
except:
h.flash(_('error occurred during updating'
' application settings'), category='error')
self.sa.rollback()
if setting_id == 'mercurial':
application_form = ApplicationUiSettingsForm()()
hgsettings1 = self.sa.query(RhodeCodeUi)\
.filter(RhodeCodeUi.ui_key == 'push_ssl').one()
hgsettings1.ui_value = form_result['web_push_ssl']
hgsettings2 = self.sa.query(RhodeCodeUi)\
.filter(RhodeCodeUi.ui_key == '/').one()
hgsettings2.ui_value = form_result['paths_root_path']
#HOOKS
hgsettings3 = self.sa.query(RhodeCodeUi)\
.filter(RhodeCodeUi.ui_key == 'changegroup.update').one()
hgsettings3.ui_active = bool(form_result['hooks_changegroup_update'])
hgsettings4 = self.sa.query(RhodeCodeUi)\
.filter(RhodeCodeUi.ui_key == 'changegroup.repo_size').one()
hgsettings4.ui_active = bool(form_result['hooks_changegroup_repo_size'])
hgsettings5 = self.sa.query(RhodeCodeUi)\
.filter(RhodeCodeUi.ui_key == 'pretxnchangegroup.push_logger').one()
hgsettings5.ui_active = bool(form_result['hooks_pretxnchangegroup_push_logger'])
hgsettings6 = self.sa.query(RhodeCodeUi)\
.filter(RhodeCodeUi.ui_key == 'preoutgoing.pull_logger').one()
hgsettings6.ui_active = bool(form_result['hooks_preoutgoing_pull_logger'])
self.sa.add(hgsettings3)
self.sa.add(hgsettings4)
self.sa.add(hgsettings5)
self.sa.add(hgsettings6)
h.flash(_('Updated mercurial settings'),
h.flash(_('error occurred during updating application settings'),
return redirect(url('admin_settings'))
def delete(self, setting_id):
"""DELETE /admin/settings/setting_id: Delete an existing item"""
def show(self, setting_id, format='html'):
"""GET /admin/settings/setting_id: Show a specific item"""
def edit(self, setting_id, format='html'):
"""GET /admin/settings/setting_id/edit: Form to edit an existing item"""
# url('admin_edit_setting', setting_id=ID)
@NotAnonymous()
def my_account(self):
GET /_admin/my_account Displays info about my account
# url('admin_settings_my_account')
c.user = UserModel().get(c.rhodecode_user.user_id, cache=False)
all_repos = self.sa.query(Repository)\
.filter(Repository.user_id == c.user.user_id)\
.order_by(func.lower(Repository.repo_name))\
.all()
c.user_repos = ScmModel().get_repos(all_repos)
if c.user.username == 'default':
h.flash(_("You can't edit this user since it's"
" crucial for entire application"), category='warning')
return redirect(url('users'))
defaults = c.user.get_dict()
render('admin/users/user_edit_my_account.html'),
def my_account_update(self):
"""PUT /_admin/my_account_update: Update an existing item"""
# h.form(url('admin_settings_my_account_update'),
# url('admin_settings_my_account_update', id=ID)
user_model = UserModel()
uid = c.rhodecode_user.user_id
_form = UserForm(edit=True, old_data={'user_id':uid,
'email':c.rhodecode_user.email})()
user_model.update_my_account(uid, form_result)
h.flash(_('Your account was updated successfully'),
c.user = user_model.get(c.rhodecode_user.user_id, cache=False)
h.flash(_('error occurred during update of user %s') \
% form_result.get('username'), category='error')
return redirect(url('my_account'))
def create_repository(self):
"""GET /_admin/create_repository: Form to create a new item"""
return render('admin/repos/repo_add_create_repository.html')
def get_hg_ui_settings(self):
ret = self.sa.query(RhodeCodeUi).all()
if not ret:
raise Exception('Could not get application ui settings !')
settings = {}
for each in ret:
k = each.ui_key
v = each.ui_value
if k == '/':
k = 'root_path'
if k.find('.') != -1:
k = k.replace('.', '_')
if each.ui_section == 'hooks':
v = each.ui_active
settings[each.ui_section + '_' + k] = v
return settings
rhodecode.controllers.admin.users
Users crud controller for pylons
:created_on: Apr 4, 2010
from rhodecode.lib.exceptions import *
from rhodecode.model.forms import UserForm
class UsersController(BaseController):
# map.resource('user', 'users')
super(UsersController, self).__before__()
"""GET /users: All items in the collection"""
# url('users')
c.users_list = self.sa.query(User).all()
return render('admin/users/users.html')
"""POST /users: Create a new item"""
login_form = UserForm()()
form_result = login_form.to_python(dict(request.POST))
user_model.create(form_result)
h.flash(_('created user %s') % form_result['username'],
#action_logger(self.rhodecode_user, 'new_user', '', '', self.sa)
render('admin/users/user_add.html'),
h.flash(_('error occurred during creation of user %s') \
% request.POST.get('username'), category='error')
"""GET /users/new: Form to create a new item"""
# url('new_user')
return render('admin/users/user_add.html')
"""PUT /users/id: Update an existing item"""
# h.form(url('user', id=ID),
# url('user', id=ID)
c.user = user_model.get(id)
_form = UserForm(edit=True, old_data={'user_id':id,
'email':c.user.email})()
user_model.update(id, form_result)
h.flash(_('User updated succesfully'), category='success')
render('admin/users/user_edit.html'),
"""DELETE /users/id: Delete an existing item"""
user_model.delete(id)
h.flash(_('sucessfully deleted user'), category='success')
except (UserOwnsReposException, DefaultUserException), e:
h.flash(str(e), category='warning')
h.flash(_('An error occurred during deletion of user'),
"""GET /users/id: Show a specific item"""
"""GET /users/id/edit: Form to edit an existing item"""
# url('edit_user', id=ID)
c.user = self.sa.query(User).get(id)
if not c.user:
h.flash(_("You can't edit this user"), category='warning')
rhodecode.controllers.branches
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
branches controller for rhodecode
:created_on: Apr 21, 2010
from pylons import tmpl_context as c
from rhodecode.lib.auth import LoginRequired, HasRepoPermissionAnyDecorator
from rhodecode.lib.utils import OrderedDict
class BranchesController(BaseController):
@HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
'repository.admin')
super(BranchesController, self).__before__()
hg_model = ScmModel()
c.repo_info = hg_model.get_repo(c.repo_name)
c.repo_branches = OrderedDict()
for name, hash_ in c.repo_info.branches.items():
c.repo_branches[name] = c.repo_info.get_changeset(hash_)
return render('branches/branches.html')
rhodecode.controllers.changelog
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
changelog controller for rhodecode
import json
except ImportError:
#python 2.5 compatibility
import simplejson as json
from mercurial.graphmod import colored, CHANGESET, revisions as graph_rev
from pylons import request, session, tmpl_context as c
class ChangelogController(BaseController):
super(ChangelogController, self).__before__()
limit = 100
default = 20
if request.params.get('size'):
int_size = int(request.params.get('size'))
except ValueError:
int_size = default
int_size = int_size if int_size <= limit else limit
c.size = int_size
session['changelog_size'] = c.size
session.save()
c.size = int(session.get('changelog_size', default))
changesets = ScmModel().get_repo(c.repo_name)
c.total_cs = len(changesets)
c.pagination = Page(changesets, page=p, item_count=c.total_cs,
items_per_page=c.size)
self._graph(changesets, c.size, p)
return render('changelog/changelog.html')
def _graph(self, repo, size, p):
revcount = size
if not repo.revisions or repo.alias == 'git':
c.jsdata = json.dumps([])
return
max_rev = repo.revisions[-1]
offset = 1 if p == 1 else ((p - 1) * revcount + 1)
rev_start = repo.revisions[(-1 * offset)]
revcount = min(max_rev, revcount)
rev_end = max(0, rev_start - revcount)
dag = graph_rev(repo.repo, rev_start, rev_end)
c.dag = tree = list(colored(dag))
data = []
for (id, type, ctx, vtx, edges) in tree:
if type != CHANGESET:
continue
data.append(('', vtx, edges))
c.jsdata = json.dumps(data)
rhodecode.controllers.changeset
changeset controller for pylons
:created_on: Apr 25, 2010
from pylons import tmpl_context as c, url, request, response
from pylons.controllers.util import redirect
import rhodecode.lib.helpers as h
from rhodecode.lib.utils import EmptyChangeset
from vcs.exceptions import RepositoryError, ChangesetError
from vcs.nodes import FileNode
from vcs.utils import diffs as differ
class ChangesetController(BaseController):
super(ChangesetController, self).__before__()
def index(self, revision):
def wrap_to_table(str):
return '''<table class="code-difftable">
<tr class="line">
<td class="lineno new"></td>
<td class="code"><pre>%s</pre></td>
</tr>
</table>''' % str
c.changeset = hg_model.get_repo(c.repo_name).get_changeset(revision)
except RepositoryError, e:
c.changeset_old = c.changeset.parents[0]
except IndexError:
c.changeset_old = None
c.changes = []
#===================================================================
# ADDED FILES
c.sum_added = 0
for node in c.changeset.added:
filenode_old = FileNode(node.path, '', EmptyChangeset())
if filenode_old.is_binary or node.is_binary:
diff = wrap_to_table(_('binary file'))
c.sum_added += node.size
if c.sum_added < self.cut_off_limit:
f_udiff = differ.get_udiff(filenode_old, node)
diff = differ.DiffProcessor(f_udiff).as_html()
diff = wrap_to_table(_('Changeset is to big and was cut'
' off, see raw changeset instead'))
cs1 = None
cs2 = node.last_changeset.raw_id
c.changes.append(('added', node, diff, cs1, cs2))
# CHANGED FILES
c.sum_removed = 0
for node in c.changeset.changed:
filenode_old = c.changeset_old.get_node(node.path)
except ChangesetError:
if c.sum_removed < self.cut_off_limit:
if diff:
c.sum_removed += len(diff)
cs1 = filenode_old.last_changeset.raw_id
c.changes.append(('changed', node, diff, cs1, cs2))
# REMOVED FILES
for node in c.changeset.removed:
c.changes.append(('removed', node, None, None, None))
return render('changeset/changeset.html')
def raw_changeset(self, revision):
method = request.GET.get('diff', 'show')
r = hg_model.get_repo(c.repo_name)
c.scm_type = r.alias
c.changeset = r.get_changeset(revision)
except RepositoryError:
filenode_old = FileNode(node.path, '')
diff = _('binary file') + '\n'
diff = differ.DiffProcessor(f_udiff).raw_diff()
diff = _('binary file')
response.content_type = 'text/plain'
if method == 'download':
response.content_disposition = 'attachment; filename=%s.patch' % revision
parent = True if len(c.changeset.parents) > 0 else False
c.parent_tmpl = 'Parent %s' % c.changeset.parents[0].raw_id if parent else ''
c.diffs = ''
for x in c.changes:
c.diffs += x[2]
return render('changeset/raw_changeset.html')
package.rhodecode.controllers.error
~~~~~~~~~~~~~~
RhodeCode error controller
:created_on: Dec 8, 2010
import os
import cgi
import paste.fileapp
from pylons import tmpl_context as c, request, config
from pylons.middleware import media_path
class ErrorController(BaseController):
"""Generates error documents as and when they are required.
The ErrorDocuments middleware forwards to ErrorController when error
related status codes are returned from the application.
This behavior can be altered by changing the parameters to the
ErrorDocuments middleware in your config/middleware.py file.
c.rhodecode_name = config.get('rhodecode_title')
def document(self):
resp = request.environ.get('pylons.original_response')
log.debug('### %s ###', resp.status)
e = request.environ
c.serv_p = r'%(protocol)s://%(host)s/' % {
'protocol': e.get('wsgi.url_scheme'),
'host':e.get('HTTP_HOST'),
}
c.error_message = cgi.escape(request.GET.get('code', str(resp.status)))
c.error_explanation = self.get_error_explanation(resp.status_int)
#redirect to when error with given seconds
c.redirect_time = 0
c.redirect_module = _('Home page')# name to what your going to be redirected
c.url_redirect = "/"
return render('/errors/error_document.html')
def img(self, id):
"""Serve Pylons' stock images"""
return self._serve_file(os.path.join(media_path, 'img', id))
def style(self, id):
"""Serve Pylons' stock stylesheets"""
return self._serve_file(os.path.join(media_path, 'style', id))
def _serve_file(self, path):
"""Call Paste's FileApp (a WSGI application) to serve the file
at the specified path
fapp = paste.fileapp.FileApp(path)
return fapp(request.environ, self.start_response)
def get_error_explanation(self, code):
''' get the error explanations of int codes
[400, 401, 403, 404, 500]'''
code = int(code)
code = 500
if code == 400:
return _('The request could not be understood by the server due to malformed syntax.')
if code == 401:
return _('Unauthorized access to resource')
if code == 403:
return _("You don't have permission to view this page")
if code == 404:
return _('The resource could not be found')
if code == 500:
return _('The server encountered an unexpected condition which prevented it from fulfilling the request.')
rhodecode.controllers.feed
~~~~~~~~~~~~~~~~~~~~~~~~~~
Feed controller for rhodecode
:created_on: Apr 23, 2010
from pylons import url, response
from rhodecode.lib.base import BaseController
from webhelpers.feedgenerator import Atom1Feed, Rss201rev2Feed
class FeedController(BaseController):
super(FeedController, self).__before__()
#common values for feeds
self.description = 'Changes on %s repository'
self.title = "%s feed"
self.language = 'en-us'
self.ttl = "5"
self.feed_nr = 10
def atom(self, repo_name):
"""Produce an atom-1.0 feed via feedgenerator module"""
feed = Atom1Feed(title=self.title % repo_name,
link=url('summary_home', repo_name=repo_name, qualified=True),
description=self.description % repo_name,
language=self.language,
ttl=self.ttl)
changesets = ScmModel().get_repo(repo_name)
for cs in changesets[:self.feed_nr]:
feed.add_item(title=cs.message,
link=url('changeset_home', repo_name=repo_name,
revision=cs.raw_id, qualified=True),
description=str(cs.date))
response.content_type = feed.mime_type
return feed.writeString('utf-8')
def rss(self, repo_name):
"""Produce an rss2 feed via feedgenerator module"""
feed = Rss201rev2Feed(title=self.title % repo_name,
rhodecode.controllers.files
~~~~~~~~~~~~~~~~~~~~~~~~~~~
Files controller for RhodeCode
import tempfile
from mercurial import archival
from vcs.exceptions import RepositoryError, ChangesetError, \
ChangesetDoesNotExistError, EmptyRepositoryError
class FilesController(BaseController):
super(FilesController, self).__before__()
c.cut_off_limit = self.cut_off_limit
def __get_cs_or_redirect(self, rev, repo_name):
Safe way to get changeset if error occur it redirects to tip with
proper message
:param rev: revision to fetch
:param repo_name: repo name to redirect after
_repo = ScmModel().get_repo(c.repo_name)
return _repo.get_changeset(rev)
except EmptyRepositoryError, e:
h.flash(_('There are no files yet'), category='warning')
redirect(h.url('summary_home', repo_name=repo_name))
redirect(h.url('files_home', repo_name=repo_name, revision='tip'))
def index(self, repo_name, revision, f_path):
cs = self.__get_cs_or_redirect(revision, repo_name)
c.repo = ScmModel().get_repo(c.repo_name)
revision = request.POST.get('at_rev', None) or revision
def get_next_rev(cur):
max_rev = len(c.repo.revisions) - 1
r = cur + 1
if r > max_rev:
r = max_rev
return r
def get_prev_rev(cur):
r = cur - 1
c.f_path = f_path
c.changeset = cs
cur_rev = c.changeset.revision
prev_rev = c.repo.get_changeset(get_prev_rev(cur_rev)).raw_id
next_rev = c.repo.get_changeset(get_next_rev(cur_rev)).raw_id
c.url_prev = url('files_home', repo_name=c.repo_name,
revision=prev_rev, f_path=f_path)
c.url_next = url('files_home', repo_name=c.repo_name,
revision=next_rev, f_path=f_path)
c.files_list = c.changeset.get_node(f_path)
c.file_history = self._get_history(c.repo, c.files_list, f_path)
redirect(h.url('files_home', repo_name=repo_name,
revision=revision))
return render('files/files.html')
def rawfile(self, repo_name, revision, f_path):
file_node = cs.get_node(f_path)
revision=cs.raw_id))
fname = f_path.split('/')[-1].encode('utf8', 'replace')
response.content_disposition = 'attachment; filename=%s' % fname
response.content_type = file_node.mimetype
return file_node.content
def raw(self, repo_name, revision, f_path):
def annotate(self, repo_name, revision, f_path):
c.file = cs.get_node(f_path)
redirect(h.url('files_home', repo_name=repo_name, revision=cs.raw_id))
c.file_history = self._get_history(ScmModel().get_repo(c.repo_name), c.file, f_path)
c.cs = cs
return render('files/files_annotate.html')
def archivefile(self, repo_name, revision, fileformat):
archive_specs = {
'.tar.bz2': ('application/x-tar', 'tbz2'),
'.tar.gz': ('application/x-tar', 'tgz'),
'.zip': ('application/zip', 'zip'),
if not archive_specs.has_key(fileformat):
return 'Unknown archive type %s' % fileformat
def read_in_chunks(file_object, chunk_size=1024 * 40):
"""Lazy function (generator) to read a file piece by piece.
Default chunk size: 40k."""
while True:
data = file_object.read(chunk_size)
if not data:
break
yield data
archive = tempfile.TemporaryFile()
repo = ScmModel().get_repo(repo_name).repo
fname = '%s-%s%s' % (repo_name, revision, fileformat)
archival.archive(repo, archive, revision, archive_specs[fileformat][1],
prefix='%s-%s' % (repo_name, revision))
response.content_type = archive_specs[fileformat][0]
archive.seek(0)
return read_in_chunks(archive)
def diff(self, repo_name, f_path):
diff1 = request.GET.get('diff1')
diff2 = request.GET.get('diff2')
c.action = request.GET.get('diff')
c.no_changes = diff1 == diff2
c.repo = hg_model.get_repo(c.repo_name)
if diff1 not in ['', None, 'None', '0' * 12, '0' * 40]:
c.changeset_1 = c.repo.get_changeset(diff1)
node1 = c.changeset_1.get_node(f_path)
c.changeset_1 = EmptyChangeset()
node1 = FileNode('.', '', changeset=c.changeset_1)
if diff2 not in ['', None, 'None', '0' * 12, '0' * 40]:
c.changeset_2 = c.repo.get_changeset(diff2)
node2 = c.changeset_2.get_node(f_path)
c.changeset_2 = EmptyChangeset()
node2 = FileNode('.', '', changeset=c.changeset_2)
return redirect(url('files_home',
repo_name=c.repo_name, f_path=f_path))
f_udiff = differ.get_udiff(node1, node2)
diff = differ.DiffProcessor(f_udiff)
if c.action == 'download':
diff_name = '%s_vs_%s.diff' % (diff1, diff2)
response.content_disposition = 'attachment; filename=%s' \
% diff_name
if node1.is_binary or node2.is_binary:
return _('binary file changed')
return diff.raw_diff()
elif c.action == 'raw':
elif c.action == 'diff':
if node1.size > self.cut_off_limit or node2.size > self.cut_off_limit:
c.cur_diff = _('Diff is to big to display')
elif node1.is_binary or node2.is_binary:
c.cur_diff = _('Binary file')
c.cur_diff = diff.as_html()
#default option
if not c.cur_diff:
c.no_changes = True
return render('files/file_diff.html')
def _get_history(self, repo, node, f_path):
from vcs.nodes import NodeKind
if not node.kind is NodeKind.FILE:
return []
changesets = node.history
hist_l = []
changesets_group = ([], _("Changesets"))
branches_group = ([], _("Branches"))
tags_group = ([], _("Tags"))
for chs in changesets:
n_desc = 'r%s:%s' % (chs.revision, chs.short_id)
changesets_group[0].append((chs.raw_id, n_desc,))
hist_l.append(changesets_group)
for name, chs in c.repository_branches.items():
#chs = chs.split(':')[-1]
branches_group[0].append((chs, name),)
hist_l.append(branches_group)
for name, chs in c.repository_tags.items():
tags_group[0].append((chs, name),)
hist_l.append(tags_group)
return hist_l
rhodecode.controllers.home
Home controller for Rhodecode
:created_on: Feb 18, 2010
from pylons import tmpl_context as c, request
from rhodecode.lib.auth import LoginRequired
class HomeController(BaseController):
super(HomeController, self).__before__()
sortables = ['name', 'description', 'last_change', 'tip', 'owner']
current_sort = request.GET.get('sort', 'name')
current_sort_slug = current_sort.replace('-', '')
if current_sort_slug not in sortables:
c.sort_by = 'name'
current_sort_slug = c.sort_by
c.sort_by = current_sort
c.sort_slug = current_sort_slug
sort_key = current_sort_slug + '_sort'
if c.sort_by.startswith('-'):
c.repos_list = sorted(cached_repo_list, key=itemgetter(sort_key),
reverse=True)
reverse=False)
return render('/index.html')
rhodecode.controllers.journal
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Journal controller for pylons
:created_on: Nov 21, 2010
from paste.httpexceptions import HTTPInternalServerError, HTTPBadRequest
from sqlalchemy import or_
from rhodecode.lib.auth import LoginRequired, NotAnonymous
from rhodecode.lib.helpers import get_token
from rhodecode.model.db import UserLog, UserFollowing
class JournalController(BaseController):
super(JournalController, self).__before__()
# Return a rendered template
c.following = self.sa.query(UserFollowing)\
.filter(UserFollowing.user_id == c.rhodecode_user.user_id).all()
repo_ids = [x.follows_repository.repo_id for x in c.following
if x.follows_repository is not None]
user_ids = [x.follows_user.user_id for x in c.following
if x.follows_user is not None]
c.journal = self.sa.query(UserLog)\
.filter(or_(
UserLog.repository_id.in_(repo_ids),
UserLog.user_id.in_(user_ids),
))\
.order_by(UserLog.action_date.desc())\
.limit(20)\
return render('/journal.html')
def toggle_following(self):
if request.POST.get('auth_token') == get_token():
scm_model = ScmModel()
user_id = request.POST.get('follows_user_id')
if user_id:
scm_model.toggle_following_user(user_id,
c.rhodecode_user.user_id)
return 'ok'
repo_id = request.POST.get('follows_repo_id')
if repo_id:
scm_model.toggle_following_repo(repo_id,
raise HTTPBadRequest()
rhodecode.controllers.login
Login controller for rhodeocode
:created_on: Apr 22, 2010
from rhodecode.lib.auth import AuthUser, HasPermissionAnyDecorator
from rhodecode.model.forms import LoginForm, RegisterForm, PasswordResetForm
class LoginController(BaseController):
super(LoginController, self).__before__()
#redirect if already logged in
c.came_from = request.GET.get('came_from', None)
if c.rhodecode_user.is_authenticated \
and c.rhodecode_user.username != 'default':
if request.POST:
#import Login Form validator class
login_form = LoginForm()
c.form_result = login_form.to_python(dict(request.POST))
username = c.form_result['username']
user = UserModel().get_by_username(username, case_insensitive=True)
auth_user = AuthUser()
auth_user.username = user.username
auth_user.is_authenticated = True
auth_user.is_admin = user.admin
auth_user.user_id = user.user_id
auth_user.name = user.name
auth_user.lastname = user.lastname
session['rhodecode_user'] = auth_user
log.info('user %s is now authenticated', username)
user.update_lastlogin()
if c.came_from:
return redirect(c.came_from)
render('/login.html'),
return render('/login.html')
@HasPermissionAnyDecorator('hg.admin', 'hg.register.auto_activate',
'hg.register.manual_activate')
def register(self):
c.auto_active = False
for perm in user_model.get_by_username('default', cache=False).user_perms:
if perm.permission.permission_name == 'hg.register.auto_activate':
c.auto_active = True
register_form = RegisterForm()()
form_result = register_form.to_python(dict(request.POST))
form_result['active'] = c.auto_active
user_model.create_registration(form_result)
h.flash(_('You have successfully registered into rhodecode'),
return redirect(url('login_home'))
render('/register.html'),
return render('/register.html')
def password_reset(self):
password_reset_form = PasswordResetForm()()
form_result = password_reset_form.to_python(dict(request.POST))
user_model.reset_password(form_result)
h.flash(_('Your new password was sent'),
render('/password_reset.html'),
return render('/password_reset.html')
def logout(self):
session['rhodecode_user'] = AuthUser()
log.info('Logging out and setting user as Empty')
redirect(url('home'))
rhodecode.controllers.search
~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Search controller for rhodecode
:created_on: Aug 7, 2010
from pylons import request, response, config, session, tmpl_context as c, url
from rhodecode.lib.indexers import SCHEMA, IDX_NAME, ResultWrapper
from webhelpers.util import update_params
from whoosh.index import open_dir, EmptyIndexError
from whoosh.qparser import QueryParser, QueryParserError
from whoosh.query import Phrase
class SearchController(BaseController):
super(SearchController, self).__before__()
def index(self, search_repo=None):
c.repo_name = search_repo
c.formated_results = []
c.runtime = ''
c.cur_query = request.GET.get('q', None)
c.cur_type = request.GET.get('type', 'source')
c.cur_search = search_type = {'content':'content',
'commit':'content',
'path':'path',
'repository':'repository'}\
.get(c.cur_type, 'content')
if c.cur_query:
cur_query = c.cur_query.lower()
highlight_items = set()
idx = open_dir(config['app_conf']['index_dir']
, indexname=IDX_NAME)
searcher = idx.searcher()
qp = QueryParser(search_type, schema=SCHEMA)
if c.repo_name:
cur_query = u'repository:%s %s' % (c.repo_name, cur_query)
query = qp.parse(unicode(cur_query))
if isinstance(query, Phrase):
highlight_items.update(query.words)
for i in query.all_terms():
if i[0] == 'content':
highlight_items.add(i[1])
matcher = query.matcher(searcher)
log.debug(query)
log.debug(highlight_items)
results = searcher.search(query)
res_ln = len(results)
c.runtime = '%s results (%.3f seconds)' \
% (res_ln, results.runtime)
def url_generator(**kw):
return update_params("?q=%s&type=%s" \
% (c.cur_query, c.cur_search), **kw)
c.formated_results = Page(
ResultWrapper(search_type, searcher, matcher,
highlight_items),
page=p, item_count=res_ln,
items_per_page=10, url=url_generator)
except QueryParserError:
c.runtime = _('Invalid search query. Try quoting it.')
searcher.close()
except (EmptyIndexError, IOError):
log.error('Empty Index data')
c.runtime = _('There is no index to search in. '
'Please run whoosh indexer')
return render('/search/search.html')
rhodecode.controllers.settings
Settings controller for rhodecode
:created_on: Jun 30, 2010
from pylons import tmpl_context as c, request, url
from rhodecode.lib.auth import LoginRequired, HasRepoPermissionAllDecorator, \
HasRepoPermissionAnyDecorator, NotAnonymous
from rhodecode.model.forms import RepoSettingsForm, RepoForkForm
@HasRepoPermissionAllDecorator('repository.admin')
def index(self, repo_name):
c.repo_info = repo = repo_model.get_by_repo_name(repo_name)
' it was created or renamed from the file system'
render('settings/repo_settings.html'),
_form = RepoSettingsForm(edit=True, old_data={'repo_name':repo_name})()
action_logger(self.rhodecode_user, 'user_updated_repo',
return redirect(url('repo_settings_home', repo_name=changed_name))
# h.form(url('repo_settings_delete', repo_name=ID),
# url('repo_settings_delete', repo_name=ID)
action_logger(self.rhodecode_user, 'user_deleted_repo',
def fork(self, repo_name):
return render('settings/repo_fork.html')
def fork_create(self, repo_name):
_form = RepoForkForm(old_data={'repo_type':c.repo_info.repo_type})()
form_result.update({'repo_name':repo_name})
repo_model.create_fork(form_result, c.rhodecode_user)
h.flash(_('forked %s repository as %s') \
% (repo_name, form_result['fork_name']),
action_logger(self.rhodecode_user,
'user_forked_repo:%s' % form_result['fork_name'],
c.new_repo = errors.value['fork_name']
r = render('settings/repo_fork.html')
rhodecode.controllers.shortlog
Shortlog controller for rhodecode
:created_on: Apr 18, 2010
class ShortlogController(BaseController):
super(ShortlogController, self).__before__()
repo = ScmModel().get_repo(c.repo_name)
c.repo_changesets = Page(repo, page=p, items_per_page=20)
c.shortlog_data = render('shortlog/shortlog_data.html')
return c.shortlog_data
r = render('shortlog/shortlog.html')
rhodecode.controllers.summary
Summary controller for Rhodecode
import calendar
from time import mktime
from datetime import datetime, timedelta, date
from vcs.exceptions import ChangesetError
from rhodecode.model.db import Statistics
from rhodecode.lib.utils import OrderedDict, EmptyChangeset
from rhodecode.lib.celerylib import run_task
from rhodecode.lib.celerylib.tasks import get_commits_stats
class SummaryController(BaseController):
super(SummaryController, self).__before__()
c.repo_info = scm_model.get_repo(c.repo_name)
c.following = scm_model.is_following_repo(c.repo_name,
return url('shortlog_home', repo_name=c.repo_name, **kw)
c.repo_changesets = Page(c.repo_info, page=1, items_per_page=10,
url=url_generator)
if self.rhodecode_user.username == 'default':
password = ':default'
password = ''
uri = u'%(protocol)s://%(user)s%(password)s@%(host)s%(prefix)s/%(repo_name)s' % {
'user':str(c.rhodecode_user.username),
'password':password,
'prefix':e.get('SCRIPT_NAME'),
'repo_name':c.repo_name, }
c.clone_repo_url = uri
c.repo_tags = OrderedDict()
for name, hash in c.repo_info.tags.items()[:10]:
c.repo_tags[name] = c.repo_info.get_changeset(hash)
c.repo_tags[name] = EmptyChangeset(hash)
for name, hash in c.repo_info.branches.items()[:10]:
c.repo_branches[name] = c.repo_info.get_changeset(hash)
c.repo_branches[name] = EmptyChangeset(hash)
td = date.today() + timedelta(days=1)
td_1m = td - timedelta(days=calendar.mdays[td.month])
td_1y = td - timedelta(days=365)
ts_min_m = mktime(td_1m.timetuple())
ts_min_y = mktime(td_1y.timetuple())
ts_max_y = mktime(td.timetuple())
if c.repo_info.dbrepo.enable_statistics:
c.no_data_msg = _('No data loaded yet')
run_task(get_commits_stats, c.repo_info.name, ts_min_y, ts_max_y)
c.no_data_msg = _('Statistics update are disabled for this repository')
c.ts_min = ts_min_m
c.ts_max = ts_max_y
stats = self.sa.query(Statistics)\
.filter(Statistics.repository == c.repo_info.dbrepo)\
.scalar()
if stats and stats.languages:
c.no_data = False is c.repo_info.dbrepo.enable_statistics
lang_stats = json.loads(stats.languages)
c.commit_data = stats.commit_activity
c.overview_data = stats.commit_activity_combined
c.trending_languages = json.dumps(OrderedDict(
sorted(lang_stats.items(), reverse=True,
key=lambda k: k[1])[:10]
c.commit_data = json.dumps({})
c.overview_data = json.dumps([[ts_min_y, 0], [ts_max_y, 10] ])
c.trending_languages = json.dumps({})
c.no_data = True
return render('summary/summary.html')
rhodecode.controllers.tags
Tags controller for rhodecode
class TagsController(BaseController):
super(TagsController, self).__before__()
for name, hash_ in c.repo_info.tags.items():
c.repo_tags[name] = c.repo_info.get_changeset(hash_)
return render('tags/tags.html')
rhodecode.lib.auth
~~~~~~~~~~~~~~~~~~
authentication and permission libraries
:copyright: (c) 2010 by marcink.
:license: LICENSE_NAME, see LICENSE_FILE for more details.
import random
from decorator import decorator
from pylons import config, session, url, request
from rhodecode import __platform__, PLATFORM_WIN, PLATFORM_OTHERS
if __platform__ in PLATFORM_WIN:
from hashlib import sha256
if __platform__ in PLATFORM_OTHERS:
import bcrypt
from rhodecode.lib import str2bool
from rhodecode.lib.exceptions import LdapPasswordError, LdapUsernameError
from rhodecode.lib.utils import get_repo_slug
from rhodecode.lib.auth_ldap import AuthLdap
from rhodecode.model import meta
from rhodecode.model.db import Permission, RepoToPerm, Repository, \
User, UserToPerm
class PasswordGenerator(object):
"""This is a simple class for generating password from
different sets of characters
usage:
passwd_gen = PasswordGenerator()
#print 8-letter password containing only big and small letters of alphabet
print passwd_gen.gen_password(8, passwd_gen.ALPHABETS_BIG_SMALL)
ALPHABETS_NUM = r'''1234567890'''#[0]
ALPHABETS_SMALL = r'''qwertyuiopasdfghjklzxcvbnm'''#[1]
ALPHABETS_BIG = r'''QWERTYUIOPASDFGHJKLZXCVBNM'''#[2]
ALPHABETS_SPECIAL = r'''`-=[]\;',./~!@#$%^&*()_+{}|:"<>?''' #[3]
ALPHABETS_FULL = ALPHABETS_BIG + ALPHABETS_SMALL + ALPHABETS_NUM + ALPHABETS_SPECIAL#[4]
ALPHABETS_ALPHANUM = ALPHABETS_BIG + ALPHABETS_SMALL + ALPHABETS_NUM#[5]
ALPHABETS_BIG_SMALL = ALPHABETS_BIG + ALPHABETS_SMALL
ALPHABETS_ALPHANUM_BIG = ALPHABETS_BIG + ALPHABETS_NUM#[6]
ALPHABETS_ALPHANUM_SMALL = ALPHABETS_SMALL + ALPHABETS_NUM#[7]
def __init__(self, passwd=''):
self.passwd = passwd
def gen_password(self, len, type):
self.passwd = ''.join([random.choice(type) for _ in xrange(len)])
return self.passwd
class RhodeCodeCrypto(object):
@classmethod
def hash_string(cls, str_):
Cryptographic function used for password hashing based on pybcrypt
or pycrypto in windows
:param password: password to hash
return sha256(str_).hexdigest()
elif __platform__ in PLATFORM_OTHERS:
return bcrypt.hashpw(str_, bcrypt.gensalt(10))
raise Exception('Unknown or unsupported platform %s' % __platform__)
def hash_check(cls, password, hashed):
Checks matching password with it's hashed value, runs different
implementation based on platform it runs on
:param password: password
:param hashed: password in hashed form
return sha256(password).hexdigest() == hashed
return bcrypt.hashpw(password, hashed) == hashed
def get_crypt_password(password):
return RhodeCodeCrypto.hash_string(password)
def check_password(password, hashed):
return RhodeCodeCrypto.hash_check(password, hashed)
def authfunc(environ, username, password):
Dummy authentication function used in Mercurial/Git/ and access control,
:param environ: needed only for using in Basic auth
return authenticate(username, password)
def authenticate(username, password):
Authentication function used for access control,
firstly checks for db authentication then if ldap is enabled for ldap
authentication, also creates ldap user if not in database
:param username: username
user = user_model.get_by_username(username, cache=False)
log.debug('Authenticating user using RhodeCode account')
if user is not None and user.is_ldap is False:
if user.active:
if user.username == 'default' and user.active:
log.info('user %s authenticated correctly as anonymous user',
username)
return True
elif user.username == username and check_password(password, user.password):
log.info('user %s authenticated correctly', username)
log.warning('user %s is disabled', username)
log.debug('Regular authentication failed')
user_obj = user_model.get_by_username(username, cache=False,
case_insensitive=True)
if user_obj is not None and user_obj.is_ldap is False:
log.debug('this user already exists as non ldap')
return False
ldap_settings = SettingsModel().get_ldap_settings()
#======================================================================
# FALLBACK TO LDAP AUTH IN ENABLE
if str2bool(ldap_settings.get('ldap_active')):
log.debug("Authenticating user using ldap")
kwargs = {
'server':ldap_settings.get('ldap_host', ''),
'base_dn':ldap_settings.get('ldap_base_dn', ''),
'port':ldap_settings.get('ldap_port'),
'bind_dn':ldap_settings.get('ldap_dn_user'),
'bind_pass':ldap_settings.get('ldap_dn_pass'),
'use_ldaps':str2bool(ldap_settings.get('ldap_ldaps')),
'ldap_version':3,
log.debug('Checking for ldap authentication')
aldap = AuthLdap(**kwargs)
res = aldap.authenticate_ldap(username, password)
log.debug('Got ldap response %s', res)
if user_model.create_ldap(username, password):
log.info('created new ldap user')
except (LdapUsernameError, LdapPasswordError,):
pass
except (Exception,):
class AuthUser(object):
A simple object that handles a mercurial username for authentication
def __init__(self):
self.username = 'None'
self.name = ''
self.lastname = ''
self.email = ''
self.user_id = None
self.is_authenticated = False
self.is_admin = False
self.permissions = {}
def __repr__(self):
return "<AuthUser('id:%s:%s')>" % (self.user_id, self.username)
def set_available_permissions(config):
This function will propagate pylons globals with all available defined
permission given in db. We don't wannt to check each time from db for new
permissions since adding a new permission also requires application restart
ie. to decorate new views with the newly created permission
:param config:
log.info('getting information about all available permissions')
sa = meta.Session()
all_perms = sa.query(Permission).all()
finally:
meta.Session.remove()
config['available_permissions'] = [x.permission_name for x in all_perms]
def set_base_path(config):
config['base_path'] = config['pylons.app_globals'].base_path
def fill_perms(user):
Fills user permission attribute with permissions taken from database
:param user:
user.permissions['repositories'] = {}
user.permissions['global'] = set()
#===========================================================================
# fetch default permissions
default_user = UserModel().get_by_username('default', cache=True)
default_perms = sa.query(RepoToPerm, Repository, Permission)\
.join((Repository, RepoToPerm.repository_id == Repository.repo_id))\
.join((Permission, RepoToPerm.permission_id == Permission.permission_id))\
.filter(RepoToPerm.user == default_user).all()
if user.is_admin:
#=======================================================================
# #admin have all default rights set to admin
user.permissions['global'].add('hg.admin')
for perm in default_perms:
p = 'repository.admin'
user.permissions['repositories'][perm.RepoToPerm.repository.repo_name] = p
# set default permissions
#default global
default_global_perms = sa.query(UserToPerm)\
.filter(UserToPerm.user == sa.query(User)\
.filter(User.username == 'default').one())
for perm in default_global_perms:
user.permissions['global'].add(perm.permission.permission_name)
#default repositories
if perm.Repository.private and not perm.Repository.user_id == user.user_id:
#disable defaults for private repos,
p = 'repository.none'
elif perm.Repository.user_id == user.user_id:
#set admin if owner
p = perm.Permission.permission_name
# #overwrite default with user permissions if any
user_perms = sa.query(RepoToPerm, Permission, Repository)\
.filter(RepoToPerm.user_id == user.user_id).all()
for perm in user_perms:
if perm.Repository.user_id == user.user_id:#set admin if owner
return user
def get_user(session):
Gets user from session, and wraps permissions into user
:param session:
user = session.get('rhodecode_user', AuthUser())
#if the user is not logged in we check for anonymous access
#if user is logged and it's a default user check if we still have anonymous
#access enabled
if user.user_id is None or user.username == 'default':
anonymous_user = UserModel().get_by_username('default', cache=True)
if anonymous_user.active is True:
#then we set this user is logged in
user.is_authenticated = True
user.user_id = anonymous_user.user_id
user.is_authenticated = False
if user.is_authenticated:
user = UserModel().fill_data(user)
user = fill_perms(user)
session['rhodecode_user'] = user
#===============================================================================
# CHECK DECORATORS
class LoginRequired(object):
"""Must be logged in to execute this function else
redirect to login page"""
def __call__(self, func):
return decorator(self.__wrapper, func)
def __wrapper(self, func, *fargs, **fkwargs):
log.debug('Checking login required for user:%s', user.username)
log.debug('user %s is authenticated', user.username)
return func(*fargs, **fkwargs)
log.warn('user %s not authenticated', user.username)
p = ''
if request.environ.get('SCRIPT_NAME') != '/':
p += request.environ.get('SCRIPT_NAME')
p += request.environ.get('PATH_INFO')
if request.environ.get('QUERY_STRING'):
p += '?' + request.environ.get('QUERY_STRING')
log.debug('redirecting to login page with %s', p)
return redirect(url('login_home', came_from=p))
class NotAnonymous(object):
log.debug('Checking if user is not anonymous')
anonymous = user.username == 'default'
if anonymous:
class PermsDecorator(object):
"""Base class for decorators"""
def __init__(self, *required_perms):
available_perms = config['available_permissions']
for perm in required_perms:
if perm not in available_perms:
raise Exception("'%s' permission is not defined" % perm)
self.required_perms = set(required_perms)
self.user_perms = None
# _wrapper.__name__ = func.__name__
# _wrapper.__dict__.update(func.__dict__)
# _wrapper.__doc__ = func.__doc__
self.user = session.get('rhodecode_user', AuthUser())
self.user_perms = self.user.permissions
log.debug('checking %s permissions %s for %s %s',
self.__class__.__name__, self.required_perms, func.__name__,
self.user)
if self.check_permissions():
log.debug('Permission granted for %s %s', func.__name__, self.user)
log.warning('Permission denied for %s %s', func.__name__, self.user)
#redirect with forbidden ret code
return abort(403)
def check_permissions(self):
"""Dummy function for overriding"""
raise Exception('You have to write this function in child class')
class HasPermissionAllDecorator(PermsDecorator):
"""Checks for access permission for all given predicates. All of them
have to be meet in order to fulfill the request
if self.required_perms.issubset(self.user_perms.get('global')):
class HasPermissionAnyDecorator(PermsDecorator):
"""Checks for access permission for any of given predicates. In order to
fulfill the request any of predicates must be meet
if self.required_perms.intersection(self.user_perms.get('global')):
class HasRepoPermissionAllDecorator(PermsDecorator):
"""Checks for access permission for all given predicates for specific
repository. All of them have to be meet in order to fulfill the request
repo_name = get_repo_slug(request)
user_perms = set([self.user_perms['repositories'][repo_name]])
except KeyError:
if self.required_perms.issubset(user_perms):
class HasRepoPermissionAnyDecorator(PermsDecorator):
"""Checks for access permission for any of given predicates for specific
repository. In order to fulfill the request any of predicates must be meet
if self.required_perms.intersection(user_perms):
# CHECK FUNCTIONS
class PermsFunction(object):
"""Base function for other check functions"""
def __init__(self, *perms):
for perm in perms:
raise Exception("'%s' permission in not defined" % perm)
self.required_perms = set(perms)
self.granted_for = ''
self.repo_name = None
def __call__(self, check_Location=''):
user = session.get('rhodecode_user', False)
if not user:
self.user_perms = user.permissions
self.granted_for = user.username
log.debug('checking %s %s %s', self.__class__.__name__,
self.required_perms, user)
log.debug('Permission granted for %s @ %s %s', self.granted_for,
check_Location, user)
log.warning('Permission denied for %s @ %s %s', self.granted_for,
class HasPermissionAll(PermsFunction):
class HasPermissionAny(PermsFunction):
class HasRepoPermissionAll(PermsFunction):
def __call__(self, repo_name=None, check_Location=''):
self.repo_name = repo_name
return super(HasRepoPermissionAll, self).__call__(check_Location)
if not self.repo_name:
self.repo_name = get_repo_slug(request)
self.user_perms = set([self.user_perms['repositories']\
[self.repo_name]])
self.granted_for = self.repo_name
if self.required_perms.issubset(self.user_perms):
class HasRepoPermissionAny(PermsFunction):
return super(HasRepoPermissionAny, self).__call__(check_Location)
if self.required_perms.intersection(self.user_perms):
# SPECIAL VERSION TO HANDLE MIDDLEWARE AUTH
class HasPermissionAnyMiddleware(object):
def __call__(self, user, repo_name):
usr = AuthUser()
usr.user_id = user.user_id
usr.username = user.username
usr.is_admin = user.admin
self.user_perms = set([fill_perms(usr)\
.permissions['repositories'][repo_name]])
self.user_perms = set()
self.username = user.username
return self.check_permissions()
log.debug('checking mercurial protocol '
'permissions for user:%s repository:%s',
self.username, self.repo_name)
log.debug('permission granted')
log.debug('permission denied')
#!/usr/bin/env python
# encoding: utf-8
# ldap authentication lib
# Copyright (C) 2009-2011 Marcin Kuzminski <marcin@python-works.com>
Created on Nov 17, 2010
@author: marcink
import ldap
class AuthLdap(object):
def __init__(self, server, base_dn, port=389, bind_dn='', bind_pass='',
use_ldaps=False, ldap_version=3):
self.ldap_version = ldap_version
if use_ldaps:
port = port or 689
self.LDAP_USE_LDAPS = use_ldaps
self.LDAP_SERVER_ADDRESS = server
self.LDAP_SERVER_PORT = port
#USE FOR READ ONLY BIND TO LDAP SERVER
self.LDAP_BIND_DN = bind_dn
self.LDAP_BIND_PASS = bind_pass
ldap_server_type = 'ldap'
if self.LDAP_USE_LDAPS:ldap_server_type = ldap_server_type + 's'
self.LDAP_SERVER = "%s://%s:%s" % (ldap_server_type,
self.LDAP_SERVER_ADDRESS,
self.LDAP_SERVER_PORT)
self.BASE_DN = base_dn
def authenticate_ldap(self, username, password):
"""Authenticate a user via LDAP and return his/her LDAP properties.
Raises AuthenticationError if the credentials are rejected, or
EnvironmentError if the LDAP server can't be reached.
from rhodecode.lib.helpers import chop_at
uid = chop_at(username, "@%s" % self.LDAP_SERVER_ADDRESS)
if "," in username:
raise LdapUsernameError("invalid character in username: ,")
ldap.set_option(ldap.OPT_X_TLS_CACERTDIR, '/etc/openldap/cacerts')
ldap.set_option(ldap.OPT_NETWORK_TIMEOUT, 10)
server = ldap.initialize(self.LDAP_SERVER)
if self.ldap_version == 2:
server.protocol = ldap.VERSION2
server.protocol = ldap.VERSION3
if self.LDAP_BIND_DN and self.LDAP_BIND_PASS:
server.simple_bind_s(self.LDAP_BIND_DN, self.LDAP_BIND_PASS)
dn = self.BASE_DN % {'user':uid}
log.debug("Authenticating %r at %s", dn, self.LDAP_SERVER)
server.simple_bind_s(dn, password)
properties = server.search_s(dn, ldap.SCOPE_SUBTREE)
if not properties:
raise ldap.NO_SUCH_OBJECT()
except ldap.NO_SUCH_OBJECT, e:
log.debug("LDAP says no such user '%s' (%s)", uid, username)
raise LdapUsernameError()
except ldap.INVALID_CREDENTIALS, e:
log.debug("LDAP rejected password for user '%s' (%s)", uid, username)
raise LdapPasswordError()
except ldap.SERVER_DOWN, e:
raise LdapConnectionError("LDAP can't access authentication server")
return properties[0]
# mercurial repository backup manager
# Copyright (C) 2009-2010 Marcin Kuzminski <marcin@python-works.com>
Created on Feb 28, 2010
Mercurial repositories backup manager
import tarfile
import datetime
import sys
import subprocess
logging.basicConfig(level=logging.DEBUG,
format="%(asctime)s %(levelname)-5.5s %(message)s")
class BackupManager(object):
def __init__(self, repos_location, rsa_key, backup_server):
today = datetime.datetime.now().weekday() + 1
self.backup_file_name = "mercurial_repos.%s.tar.gz" % today
self.id_rsa_path = self.get_id_rsa(rsa_key)
self.repos_path = self.get_repos_path(repos_location)
self.backup_server = backup_server
self.backup_file_path = '/tmp'
logging.info('starting backup for %s', self.repos_path)
logging.info('backup target %s', self.backup_file_path)
def get_id_rsa(self, rsa_key):
if not os.path.isfile(rsa_key):
logging.error('Could not load id_rsa key file in %s', rsa_key)
sys.exit()
return rsa_key
def get_repos_path(self, path):
if not os.path.isdir(path):
logging.error('Wrong location for repositories in %s', path)
return path
def backup_repos(self):
bckp_file = os.path.join(self.backup_file_path, self.backup_file_name)
tar = tarfile.open(bckp_file, "w:gz")
for dir_name in os.listdir(self.repos_path):
logging.info('backing up %s', dir_name)
tar.add(os.path.join(self.repos_path, dir_name), dir_name)
tar.close()
logging.info('finished backup of mercurial repositories')
def transfer_files(self):
params = {
'id_rsa_key': self.id_rsa_path,
'backup_file':os.path.join(self.backup_file_path,
self.backup_file_name),
'backup_server':self.backup_server
cmd = ['scp', '-l', '40000', '-i', '%(id_rsa_key)s' % params,
'%(backup_file)s' % params,
'%(backup_server)s' % params]
subprocess.call(cmd)
logging.info('Transfered file %s to %s', self.backup_file_name, cmd[4])
def rm_file(self):
logging.info('Removing file %s', self.backup_file_name)
os.remove(os.path.join(self.backup_file_path, self.backup_file_name))
if __name__ == "__main__":
repo_location = '/home/repo_path'
backup_server = 'root@192.168.1.100:/backups/mercurial'
rsa_key = '/home/id_rsa'
B_MANAGER = BackupManager(repo_location, rsa_key, backup_server)
B_MANAGER.backup_repos()
B_MANAGER.transfer_files()
B_MANAGER.rm_file()
package.rhodecode.lib.celerylib.__init__
celery libs for RhodeCode
:created_on: Nov 27, 2010
import socket
from hashlib import md5
from vcs.utils.lazy import LazyProperty
from rhodecode.lib.pidlock import DaemonLock, LockHeld
from pylons import config
def str2bool(v):
return v.lower() in ["yes", "true", "t", "1"] if v else None
CELERY_ON = str2bool(config['app_conf'].get('use_celery'))
CELERY_ON = False
class ResultWrapper(object):
def __init__(self, task):
self.task = task
@LazyProperty
def result(self):
return self.task
def run_task(task, *args, **kwargs):
if CELERY_ON:
t = task.delay(*args, **kwargs)
log.info('running task %s:%s', t.task_id, task)
return t
except socket.error, e:
if e.errno == 111:
log.debug('Unable to connect to celeryd. Sync execution')
except KeyError, e:
log.debug('executing task %s in sync mode', task)
return ResultWrapper(task(*args, **kwargs))
def locked_task(func):
def __wrapper(func, *fargs, **fkwargs):
params = list(fargs)
params.extend(['%s-%s' % ar for ar in fkwargs.items()])
lockkey = 'task_%s' % \
md5(str(func.__name__) + '-' + \
'-'.join(map(str, params))).hexdigest()
log.info('running task with lockkey %s', lockkey)
l = DaemonLock(lockkey)
ret = func(*fargs, **fkwargs)
l.release()
return ret
except LockHeld:
log.info('LockHeld')
return 'Task with key %s already running' % lockkey
return decorator(__wrapper, func)
rhodecode.lib.celerylib.tasks
RhodeCode task modules, containing all task that suppose to be run
by celery daemon
:created_on: Oct 6, 2010
from celery.decorators import task
from rhodecode.lib.celerylib import run_task, locked_task, str2bool
from rhodecode.lib.helpers import person
from rhodecode.lib.smtp_mailer import SmtpMailer
from rhodecode.lib.utils import OrderedDict, add_cache
from rhodecode.model import init_model
from rhodecode.model.db import RhodeCodeUi
from vcs.backends import get_repo
from sqlalchemy import engine_from_config
add_cache(config)
__all__ = ['whoosh_index', 'get_commits_stats',
'reset_user_password', 'send_email']
def get_session():
engine = engine_from_config(config, 'sqlalchemy.db1.')
init_model(engine)
return sa
def get_repos_path():
sa = get_session()
q = sa.query(RhodeCodeUi).filter(RhodeCodeUi.ui_key == '/').one()
return q.ui_value
@task(ignore_result=True)
@locked_task
def whoosh_index(repo_location, full_index):
#log = whoosh_index.get_logger()
from rhodecode.lib.indexers.daemon import WhooshIndexingDaemon
index_location = config['index_dir']
WhooshIndexingDaemon(index_location=index_location,
repo_location=repo_location, sa=get_session())\
.run(full_index=full_index)
def get_commits_stats(repo_name, ts_min_y, ts_max_y):
log = get_commits_stats.get_logger()
from rhodecode.model.db import Statistics, Repository
#for js data compatibilty
author_key_cleaner = lambda k: person(k).replace('"', "")
commits_by_day_author_aggregate = {}
commits_by_day_aggregate = {}
repos_path = get_repos_path()
p = os.path.join(repos_path, repo_name)
repo = get_repo(p)
skip_date_limit = True
parse_limit = 250 #limit for single task changeset parsing optimal for
last_cs = None
timegetter = itemgetter('time')
dbrepo = sa.query(Repository)\
.filter(Repository.repo_name == repo_name).scalar()
cur_stats = sa.query(Statistics)\
.filter(Statistics.repository == dbrepo).scalar()
if cur_stats:
last_rev = cur_stats.stat_on_revision
if not repo.revisions:
if last_rev == repo.revisions[-1] and len(repo.revisions) > 1:
#pass silently without any work if we're not on first revision or
#current state of parsing revision(from db marker) is the last revision
commits_by_day_aggregate = OrderedDict(
json.loads(
cur_stats.commit_activity_combined))
commits_by_day_author_aggregate = json.loads(cur_stats.commit_activity)
log.debug('starting parsing %s', parse_limit)
lmktime = mktime
last_rev = last_rev + 1 if last_rev > 0 else last_rev
for rev in repo.revisions[last_rev:last_rev + parse_limit]:
last_cs = cs = repo.get_changeset(rev)
k = lmktime([cs.date.timetuple()[0], cs.date.timetuple()[1],
cs.date.timetuple()[2], 0, 0, 0, 0, 0, 0])
if commits_by_day_author_aggregate.has_key(author_key_cleaner(cs.author)):
l = [timegetter(x) for x in commits_by_day_author_aggregate\
[author_key_cleaner(cs.author)]['data']]
time_pos = l.index(k)
time_pos = False
if time_pos >= 0 and time_pos is not False:
datadict = commits_by_day_author_aggregate\
[author_key_cleaner(cs.author)]['data'][time_pos]
datadict["commits"] += 1
datadict["added"] += len(cs.added)
datadict["changed"] += len(cs.changed)
datadict["removed"] += len(cs.removed)
if k >= ts_min_y and k <= ts_max_y or skip_date_limit:
datadict = {"time":k,
"commits":1,
"added":len(cs.added),
"changed":len(cs.changed),
"removed":len(cs.removed),
commits_by_day_author_aggregate\
[author_key_cleaner(cs.author)]['data'].append(datadict)
commits_by_day_author_aggregate[author_key_cleaner(cs.author)] = {
"label":author_key_cleaner(cs.author),
"data":[{"time":k,
}],
"schema":["commits"],
#gather all data by day
if commits_by_day_aggregate.has_key(k):
commits_by_day_aggregate[k] += 1
commits_by_day_aggregate[k] = 1
overview_data = sorted(commits_by_day_aggregate.items(), key=itemgetter(0))
if not commits_by_day_author_aggregate:
commits_by_day_author_aggregate[author_key_cleaner(repo.contact)] = {
"label":author_key_cleaner(repo.contact),
"data":[0, 1],
stats = cur_stats if cur_stats else Statistics()
stats.commit_activity = json.dumps(commits_by_day_author_aggregate)
stats.commit_activity_combined = json.dumps(overview_data)
log.debug('last revison %s', last_rev)
leftovers = len(repo.revisions[last_rev:])
log.debug('revisions to parse %s', leftovers)
if last_rev == 0 or leftovers < parse_limit:
log.debug('getting code trending stats')
stats.languages = json.dumps(__get_codes_stats(repo_name))
stats.repository = dbrepo
stats.stat_on_revision = last_cs.revision
sa.add(stats)
sa.commit()
sa.rollback()
if len(repo.revisions) > 1:
run_task(get_commits_stats, repo_name, ts_min_y, ts_max_y)
def reset_user_password(user_email):
log = reset_user_password.get_logger()
from rhodecode.lib import auth
user = sa.query(User).filter(User.email == user_email).scalar()
new_passwd = auth.PasswordGenerator().gen_password(8,
auth.PasswordGenerator.ALPHABETS_BIG_SMALL)
if user:
user.password = auth.get_crypt_password(new_passwd)
sa.add(user)
log.info('change password for %s', user_email)
if new_passwd is None:
raise Exception('unable to generate new password')
run_task(send_email, user_email,
"Your new rhodecode password",
'Your new rhodecode password:%s' % (new_passwd))
log.info('send new password mail to %s', user_email)
log.error('Failed to update user password')
def send_email(recipients, subject, body):
Sends an email with defined parameters from the .ini files.
:param recipients: list of recipients, it this is empty the defined email
address from field 'email_to' is used instead
:param subject: subject of the mail
:param body: body of the mail
log = send_email.get_logger()
email_config = config
if not recipients:
recipients = [email_config.get('email_to')]
mail_from = email_config.get('app_email_from')
user = email_config.get('smtp_username')
passwd = email_config.get('smtp_password')
mail_server = email_config.get('smtp_server')
mail_port = email_config.get('smtp_port')
tls = str2bool(email_config.get('smtp_use_tls'))
ssl = str2bool(email_config.get('smtp_use_ssl'))
m = SmtpMailer(mail_from, user, passwd, mail_server,
mail_port, ssl, tls)
m.send(recipients, subject, body)
log.error('Mail sending failed')
def create_repo_fork(form_data, cur_user):
log = create_repo_fork.get_logger()
from vcs import get_backend
repo_model = RepoModel(get_session())
repo_model.create(form_data, cur_user, just_db=True, fork=True)
repo_name = form_data['repo_name']
repo_path = os.path.join(repos_path, repo_name)
repo_fork_path = os.path.join(repos_path, form_data['fork_name'])
alias = form_data['repo_type']
log.info('creating repo fork %s as %s', repo_name, repo_path)
backend = get_backend(alias)
backend(str(repo_fork_path), create=True, src_url=str(repo_path))
def __get_codes_stats(repo_name):
LANGUAGES_EXTENSIONS_MAP = {'scm': 'Scheme', 'asmx': 'VbNetAspx', 'Rout':
'RConsole', 'rest': 'Rst', 'abap': 'ABAP', 'go': 'Go', 'phtml': 'HtmlPhp',
'ns2': 'Newspeak', 'xml': 'EvoqueXml', 'sh-session': 'BashSession', 'ads':
'Ada', 'clj': 'Clojure', 'll': 'Llvm', 'ebuild': 'Bash', 'adb': 'Ada',
'ada': 'Ada', 'c++-objdump': 'CppObjdump', 'aspx':
'VbNetAspx', 'ksh': 'Bash', 'coffee': 'CoffeeScript', 'vert': 'GLShader',
'Makefile.*': 'Makefile', 'di': 'D', 'dpatch': 'DarcsPatch', 'rake':
'Ruby', 'moo': 'MOOCode', 'erl-sh': 'ErlangShell', 'geo': 'GLShader',
'pov': 'Povray', 'bas': 'VbNet', 'bat': 'Batch', 'd': 'D', 'lisp':
'CommonLisp', 'h': 'C', 'rbx': 'Ruby', 'tcl': 'Tcl', 'c++': 'Cpp', 'md':
'MiniD', '.vimrc': 'Vim', 'xsd': 'Xml', 'ml': 'Ocaml', 'el': 'CommonLisp',
'befunge': 'Befunge', 'xsl': 'Xslt', 'pyx': 'Cython', 'cfm':
'ColdfusionHtml', 'evoque': 'Evoque', 'cfg': 'Ini', 'htm': 'Html',
'Makefile': 'Makefile', 'cfc': 'ColdfusionHtml', 'tex': 'Tex', 'cs':
'CSharp', 'mxml': 'Mxml', 'patch': 'Diff', 'apache.conf': 'ApacheConf',
'scala': 'Scala', 'applescript': 'AppleScript', 'GNUmakefile': 'Makefile',
'c-objdump': 'CObjdump', 'lua': 'Lua', 'apache2.conf': 'ApacheConf', 'rb':
'Ruby', 'gemspec': 'Ruby', 'rl': 'RagelObjectiveC', 'vala': 'Vala', 'tmpl':
'Cheetah', 'bf': 'Brainfuck', 'plt': 'Gnuplot', 'G': 'AntlrRuby', 'xslt':
'Xslt', 'flxh': 'Felix', 'asax': 'VbNetAspx', 'Rakefile': 'Ruby', 'S': 'S',
'wsdl': 'Xml', 'js': 'Javascript', 'autodelegate': 'Myghty', 'properties':
'Ini', 'bash': 'Bash', 'c': 'C', 'g': 'AntlrRuby', 'r3': 'Rebol', 's':
'Gas', 'ashx': 'VbNetAspx', 'cxx': 'Cpp', 'boo': 'Boo', 'prolog': 'Prolog',
'sqlite3-console': 'SqliteConsole', 'cl': 'CommonLisp', 'cc': 'Cpp', 'pot':
'Gettext', 'vim': 'Vim', 'pxi': 'Cython', 'yaml': 'Yaml', 'SConstruct':
'Python', 'diff': 'Diff', 'txt': 'Text', 'cw': 'Redcode', 'pxd': 'Cython',
'plot': 'Gnuplot', 'java': 'Java', 'hrl': 'Erlang', 'py': 'Python',
'makefile': 'Makefile', 'squid.conf': 'SquidConf', 'asm': 'Nasm', 'toc':
'Tex', 'kid': 'Genshi', 'rhtml': 'Rhtml', 'po': 'Gettext', 'pl': 'Prolog',
'pm': 'Perl', 'hx': 'Haxe', 'ascx': 'VbNetAspx', 'ooc': 'Ooc', 'asy':
'Asymptote', 'hs': 'Haskell', 'SConscript': 'Python', 'pytb':
'PythonTraceback', 'myt': 'Myghty', 'hh': 'Cpp', 'R': 'S', 'aux': 'Tex',
'rst': 'Rst', 'cpp-objdump': 'CppObjdump', 'lgt': 'Logtalk', 'rss': 'Xml',
'flx': 'Felix', 'b': 'Brainfuck', 'f': 'Fortran', 'rbw': 'Ruby',
'.htaccess': 'ApacheConf', 'cxx-objdump': 'CppObjdump', 'j': 'ObjectiveJ',
'mll': 'Ocaml', 'yml': 'Yaml', 'mu': 'MuPAD', 'r': 'Rebol', 'ASM': 'Nasm',
'erl': 'Erlang', 'mly': 'Ocaml', 'mo': 'Modelica', 'def': 'Modula2', 'ini':
'Ini', 'control': 'DebianControl', 'vb': 'VbNet', 'vapi': 'Vala', 'pro':
'Prolog', 'spt': 'Cheetah', 'mli': 'Ocaml', 'as': 'ActionScript3', 'cmd':
'Batch', 'cpp': 'Cpp', 'io': 'Io', 'tac': 'Python', 'haml': 'Haml', 'rkt':
'Racket', 'st':'Smalltalk', 'inc': 'Povray', 'pas': 'Delphi', 'cmake':
'CMake', 'csh':'Tcsh', 'hpp': 'Cpp', 'feature': 'Gherkin', 'html': 'Html',
'php':'Php', 'php3':'Php', 'php4':'Php', 'php5':'Php', 'xhtml': 'Html',
'hxx': 'Cpp', 'eclass': 'Bash', 'css': 'Css',
'frag': 'GLShader', 'd-objdump': 'DObjdump', 'weechatlog': 'IrcLogs',
'tcsh': 'Tcsh', 'objdump': 'Objdump', 'pyw': 'Python', 'h++': 'Cpp',
'py3tb': 'Python3Traceback', 'jsp': 'Jsp', 'sql': 'Sql', 'mak': 'Makefile',
'php': 'Php', 'mao': 'Mako', 'man': 'Groff', 'dylan': 'Dylan', 'sass':
'Sass', 'cfml': 'ColdfusionHtml', 'darcspatch': 'DarcsPatch', 'tpl':
'Smarty', 'm': 'ObjectiveC', 'f90': 'Fortran', 'mod': 'Modula2', 'sh':
'Bash', 'lhs': 'LiterateHaskell', 'sources.list': 'SourcesList', 'axd':
'VbNetAspx', 'sc': 'Python'}
tip = repo.get_changeset()
code_stats = {}
def aggregate(cs):
for f in cs[2]:
ext = f.extension
key = LANGUAGES_EXTENSIONS_MAP.get(ext, ext)
key = key or ext
if ext in LANGUAGES_EXTENSIONS_MAP.keys() and not f.is_binary:
if code_stats.has_key(key):
code_stats[key] += 1
code_stats[key] = 1
map(aggregate, tip.walk('/'))
return code_stats or {}
rhodecode.lib.dbmigrate.__init__
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Database migration modules
:created_on: Dec 11, 2010
from rhodecode.lib.utils import BasePasterCommand, Command, add_cache
from rhodecode.lib.db_manage import DbManage
class UpgradeDb(BasePasterCommand):
"""Command used for paster to upgrade our database to newer version
max_args = 1
min_args = 1
usage = "CONFIG_FILE"
summary = "Upgrades current db to newer version given configuration file"
group_name = "RhodeCode"
parser = Command.standard_parser(verbose=True)
def command(self):
db_uri = config['sqlalchemy.db1.url']
dbmanage = DbManage(log_sql=True, dbconf=db_uri,
root=config['here'], tests=False)
dbmanage.upgrade()
def update_parser(self):
self.parser.add_option('--sql',
action='store_true',
dest='just_sql',
help="Prints upgrade sql for further investigation",
default=False)
rhodecode.lib.dbmigrate.versions.__init__
Package containing new versions of database models
# Custom Exceptions modules
Custom Exceptions modules
class LdapUsernameError(Exception):pass
class LdapPasswordError(Exception):pass
class LdapConnectionError(Exception):pass
class LdapImportError(Exception):pass
class DefaultUserException(Exception):pass
class UserOwnsReposException(Exception):pass
rhodecode.lib.hooks
~~~~~~~~~~~~~~~~~~~
Hooks runned by rhodecode
:created_on: Aug 6, 2010
import getpass
from mercurial.cmdutil import revrange
from mercurial.node import nullrev
from rhodecode.lib.utils import action_logger
def repo_size(ui, repo, hooktype=None, **kwargs):
"""Presents size of repository after push
:param ui:
:param repo:
:param hooktype:
if hooktype != 'changegroup':
size_hg, size_root = 0, 0
for path, dirs, files in os.walk(repo.root):
if path.find('.hg') != -1:
for f in files:
size_hg += os.path.getsize(os.path.join(path, f))
except OSError:
size_root += os.path.getsize(os.path.join(path, f))
size_hg_f = h.format_byte_size(size_hg)
size_root_f = h.format_byte_size(size_root)
size_total_f = h.format_byte_size(size_root + size_hg)
sys.stdout.write('Repository size .hg:%s repo:%s total:%s\n' \
% (size_hg_f, size_root_f, size_total_f))
def log_pull_action(ui, repo, **kwargs):
"""Logs user last pull action
extra_params = dict(repo.ui.configitems('rhodecode_extras'))
username = extra_params['username']
repository = extra_params['repository']
action = 'pull'
action_logger(username, action, repository, extra_params['ip'])
return 0
def log_push_action(ui, repo, **kwargs):
"""Maps user last push action to new changeset id, from mercurial
action = 'push:%s'
node = kwargs['node']
def get_revs(repo, rev_opt):
if rev_opt:
revs = revrange(repo, rev_opt)
if len(revs) == 0:
return (nullrev, nullrev)
return (max(revs), min(revs))
return (len(repo) - 1, 0)
stop, start = get_revs(repo, [node + ':'])
revs = (str(repo[r]) for r in xrange(start, stop + 1))
action = action % ','.join(revs)
# whoosh indexer daemon for rhodecode
Created on Jan 26, 2010
A deamon will read from task table and run tasks
from os.path import dirname as dn
from os.path import join as jn
#to get the rhodecode import
project_path = dn(dn(dn(dn(os.path.realpath(__file__)))))
sys.path.append(project_path)
from rhodecode.lib.helpers import safe_unicode
from whoosh.index import create_in, open_dir
from shutil import rmtree
from rhodecode.lib.indexers import INDEX_EXTENSIONS, SCHEMA, IDX_NAME
from vcs.exceptions import ChangesetError, RepositoryError
log = logging.getLogger('whooshIndexer')
# create logger
log.setLevel(logging.DEBUG)
log.propagate = False
# create console handler and set level to debug
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
# create formatter
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
# add formatter to ch
ch.setFormatter(formatter)
# add ch to logger
log.addHandler(ch)
class WhooshIndexingDaemon(object):
Deamon for atomic jobs
def __init__(self, indexname='HG_INDEX', index_location=None,
repo_location=None, sa=None):
self.indexname = indexname
self.index_location = index_location
if not index_location:
raise Exception('You have to provide index location')
self.repo_location = repo_location
if not repo_location:
raise Exception('You have to provide repositories location')
self.repo_paths = ScmModel(sa).repo_scan(self.repo_location, None)
self.initial = False
if not os.path.isdir(self.index_location):
os.makedirs(self.index_location)
log.info('Cannot run incremental index since it does not'
' yet exist running full build')
self.initial = True
def get_paths(self, repo):
"""recursive walk in root dir and return a set of all path in that dir
based on repository walk function
index_paths_ = set()
for topnode, dirs, files in repo.walk('/', 'tip'):
index_paths_.add(jn(repo.path, f.path))
for dir in dirs:
return index_paths_
def get_node(self, repo, path):
n_path = path[len(repo.path) + 1:]
node = repo.get_changeset().get_node(n_path)
return node
def get_node_mtime(self, node):
return mktime(node.last_changeset.date.timetuple())
def add_doc(self, writer, path, repo):
"""Adding doc to writer this function itself fetches data from
the instance of vcs backend"""
node = self.get_node(repo, path)
#we just index the content of chosen files, and skip binary files
if node.extension in INDEX_EXTENSIONS and not node.is_binary:
u_content = node.content
if not isinstance(u_content, unicode):
log.warning(' >> %s Could not get this content as unicode '
'replacing with empty content', path)
u_content = u''
log.debug(' >> %s [WITH CONTENT]' % path)
log.debug(' >> %s' % path)
#just index file name without it's content
writer.add_document(owner=unicode(repo.contact),
repository=safe_unicode(repo.name),
path=safe_unicode(path),
content=u_content,
modtime=self.get_node_mtime(node),
extension=node.extension)
def build_index(self):
if os.path.exists(self.index_location):
log.debug('removing previous index')
rmtree(self.index_location)
if not os.path.exists(self.index_location):
os.mkdir(self.index_location)
idx = create_in(self.index_location, SCHEMA, indexname=IDX_NAME)
writer = idx.writer()
for cnt, repo in enumerate(self.repo_paths.values()):
log.debug('building index @ %s' % repo.path)
for idx_path in self.get_paths(repo):
self.add_doc(writer, idx_path, repo)
log.debug('>> COMMITING CHANGES <<')
writer.commit(merge=True)
log.debug('>>> FINISHED BUILDING INDEX <<<')
def update_index(self):
log.debug('STARTING INCREMENTAL INDEXING UPDATE')
idx = open_dir(self.index_location, indexname=self.indexname)
# The set of all paths in the index
indexed_paths = set()
# The set of all paths we need to re-index
to_index = set()
reader = idx.reader()
# Loop over the stored fields in the index
for fields in reader.all_stored_fields():
indexed_path = fields['path']
indexed_paths.add(indexed_path)
repo = self.repo_paths[fields['repository']]
node = self.get_node(repo, indexed_path)
# This file was deleted since it was indexed
log.debug('removing from index %s' % indexed_path)
writer.delete_by_term('path', indexed_path)
# Check if this file was changed since it was indexed
indexed_time = fields['modtime']
mtime = self.get_node_mtime(node)
if mtime > indexed_time:
# The file has changed, delete it and add it to the list of
# files to reindex
log.debug('adding to reindex list %s' % indexed_path)
to_index.add(indexed_path)
# Loop over the files in the filesystem
# Assume we have a function that gathers the filenames of the
# documents to be indexed
for repo in self.repo_paths.values():
for path in self.get_paths(repo):
if path in to_index or path not in indexed_paths:
# This is either a file that's changed, or a new file
# that wasn't indexed before. So index it!
self.add_doc(writer, path, repo)
log.debug('re indexing %s' % path)
log.debug('>>> FINISHED REBUILDING INDEX <<<')
def run(self, full_index=False):
"""Run daemon"""
if full_index or self.initial:
self.build_index()
self.update_index()
rhodecode.lib.middleware.https_fixup
middleware to handle https correctly
:created_on: May 23, 2010
class HttpsFixup(object):
def __init__(self, app, config):
self.application = app
self.config = config
def __call__(self, environ, start_response):
self.__fixup(environ)
return self.application(environ, 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.
proto = environ.get('HTTP_X_URL_SCHEME')
if str2bool(self.config.get('force_https')):
proto = 'https'
if proto == 'https':
environ['wsgi.url_scheme'] = proto
environ['wsgi.url_scheme'] = 'http'
return None
# middleware to handle git api calls
Created on 2010-04-28
SimpleGit middleware for handling git protocol request (push/clone etc.)
It's implemented with basic auth function
from dulwich import server as dulserver
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)
# Do they want any objects?
if len(objects_iter) == 0:
self.progress("counting objects: %d, done.\n" % len(objects_iter))
dulserver.write_pack_data(dulserver.ProtocolFile(None, write), objects_iter,
len(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-upload-pack': SimpleGitUploadPackHandler,
'git-receive-pack': dulserver.ReceivePackHandler,
from dulwich.repo import Repo
from dulwich.web import HTTPGitApplication
from paste.auth.basic import AuthBasicAuthenticator
from paste.httpheaders import REMOTE_USER, AUTH_TYPE
from rhodecode.lib.auth import authfunc, HasPermissionAnyMiddleware
from rhodecode.lib.utils import invalidate_cache, check_repo_fast
from webob.exc import HTTPNotFound, HTTPForbidden, HTTPInternalServerError
def is_git(environ):
Returns True if request's target is git server. ``HTTP_USER_AGENT`` would
then have git client version given.
:param environ:
http_user_agent = environ.get('HTTP_USER_AGENT')
if http_user_agent and http_user_agent.startswith('git'):
class SimpleGit(object):
def __init__(self, application, config):
self.application = application
#authenticate this git request using
self.authenticate = AuthBasicAuthenticator('', authfunc)
self.ipaddr = '0.0.0.0'
self.repository = None
self.username = None
self.action = None
if not is_git(environ):
proxy_key = 'HTTP_X_REAL_IP'
def_key = 'REMOTE_ADDR'
self.ipaddr = environ.get(proxy_key, environ.get(def_key, '0.0.0.0'))
# AUTHENTICATE THIS GIT REQUEST
username = REMOTE_USER(environ)
if not username:
self.authenticate.realm = self.config['rhodecode_realm']
result = self.authenticate(environ)
if isinstance(result, str):
AUTH_TYPE.update(environ, 'basic')
REMOTE_USER.update(environ, result)
return result.wsgi_application(environ, start_response)
# GET REPOSITORY
repo_name = '/'.join(environ['PATH_INFO'].split('/')[1:])
if repo_name.endswith('/'):
repo_name = repo_name.rstrip('/')
self.repository = repo_name
return HTTPInternalServerError()(environ, start_response)
# CHECK PERMISSIONS FOR THIS REQUEST
self.action = self.__get_action(environ)
if self.action:
username = self.__get_environ_user(environ)
user = self.__get_user(username)
#check permissions for this repository
if self.action == 'push':
if not HasPermissionAnyMiddleware('repository.write',
'repository.admin')\
(user, repo_name):
return HTTPForbidden()(environ, start_response)
#any other action need at least read permission
if not HasPermissionAnyMiddleware('repository.read',
'repository.write',
self.extras = {'ip':self.ipaddr,
'username':self.username,
'action':self.action,
'repository':self.repository}
# GIT REQUEST HANDLING
self.basepath = self.config['base_path']
self.repo_path = os.path.join(self.basepath, self.repo_name)
#quick check if that dir exists...
if check_repo_fast(self.repo_name, self.basepath):
return HTTPNotFound()(environ, start_response)
app = self.__make_app()
#invalidate cache on push
self.__invalidate_cache(self.repo_name)
return app(environ, start_response)
def __make_app(self):
backend = dulserver.DictBackend({'/' + self.repo_name: Repo(self.repo_path)})
gitserve = HTTPGitApplication(backend)
return gitserve
def __get_environ_user(self, environ):
return environ.get('REMOTE_USER')
def __get_user(self, username):
return UserModel().get_by_username(username, cache=True)
def __get_action(self, environ):
Maps git request commands into a pull or push command.
service = environ['QUERY_STRING'].split('=')
if len(service) > 1:
service_cmd = service[1]
mapping = {'git-receive-pack': 'push',
'git-upload-pack': 'pull',
return mapping.get(service_cmd, service_cmd if service_cmd else 'other')
return 'other'
def __invalidate_cache(self, repo_name):
"""we know that some change was made to repositories and we should
invalidate the cache to see the changes right away but only for
push requests"""
# middleware to handle mercurial api calls
SimpleHG middleware for handling mercurial protocol request (push/clone etc.)
from mercurial.error import RepoError
from mercurial.hgweb import hgweb
from mercurial.hgweb.request import wsgiapplication
from rhodecode.lib.utils import make_ui, invalidate_cache, \
check_repo_fast, ui_sections
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'):
class SimpleHg(object):
#authenticate this mercurial request using authfunc
if not is_mercurial(environ):
# skip passing error to error controller
environ['pylons.status_code_redirect'] = True
# AUTHENTICATE THIS MERCURIAL REQUEST
self.authenticate.realm = str(self.config['rhodecode_realm'])
# MERCURIAL REQUEST HANDLING
environ['PATH_INFO'] = '/'#since we wrap into hgweb, reset the path
self.baseui = make_ui('db')
self.repo_path = os.path.join(self.basepath, repo_name)
if check_repo_fast(repo_name, self.basepath):
app = wsgiapplication(self.__make_app)
except RepoError, e:
if str(e).find('not found') != -1:
self.__invalidate_cache(repo_name)
hgserve = hgweb(str(self.repo_path), baseui=self.baseui)
return self.__load_web_settings(hgserve, self.extras)
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 mapping.has_key(cmd):
return mapping[cmd]
return cmd
def __load_web_settings(self, hgserve, extras={}):
#set the global ui for hgserve instance passed
hgserve.repo.ui = self.baseui
hgrc = os.path.join(self.repo_path, '.hg', 'hgrc')
#inject some additional parameters that will be available in ui
#for hooks
for k, v in extras.items():
hgserve.repo.ui.setconfig('rhodecode_extras', k, v)
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):
hgserve.repo.ui.setconfig(section, k, v)
return hgserve
rhodecode.lib.utils
Utilities library for RhodeCode
from UserDict import DictMixin
from mercurial import ui, config, hg
import paste
import beaker
from paste.script.command import Command, BadCommand
from vcs.backends.base import BaseChangeset
from rhodecode.model.caching_query import FromCache
from rhodecode.model.db import Repository, User, RhodeCodeUi, UserLog
def get_repo_slug(request):
return request.environ['pylons.routes_dict'].get('repo_name')
def action_logger(user, action, repo, ipaddr='', sa=None):
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
:param sa: optional sqlalchemy session
if not sa:
um = UserModel()
if hasattr(user, 'user_id'):
user_obj = user
elif isinstance(user, basestring):
user_obj = um.get_by_username(user, cache=False)
raise Exception('You have to provide user object or username')
rm = RepoModel()
if hasattr(repo, 'repo_id'):
repo_obj = rm.get(repo.repo_id, cache=False)
repo_name = repo_obj.repo_name
elif isinstance(repo, basestring):
repo_name = repo.lstrip('/')
repo_obj = rm.get_by_repo_name(repo_name, cache=False)
raise Exception('You have to provide repository to action logger')
user_log = UserLog()
user_log.user_id = user_obj.user_id
user_log.action = action
user_log.repository_id = repo_obj.repo_id
user_log.repository_name = repo_name
user_log.action_date = datetime.datetime.now()
user_log.user_ip = ipaddr
sa.add(user_log)
log.info('Adding user %s, action %s on %s', user_obj, action, repo)
def get_repos(path, recursive=False, initial=False):
Scans given path for repos and return (name,(type,path)) tuple
:param prefix:
:param path:
:param recursive:
:param initial:
from vcs.utils.helpers import get_scm
from vcs.exceptions import VCSError
scm = get_scm(path)
raise Exception('The given path %s should not be a repository got %s',
path, scm)
for dirpath in os.listdir(path):
yield dirpath, get_scm(os.path.join(path, dirpath))
except VCSError:
def check_repo_fast(repo_name, base_path):
Check given path for existence of directory
:param base_path:
:return False: if this directory is present
if os.path.isdir(os.path.join(base_path, repo_name)):return False
def check_repo(repo_name, base_path, verify=True):
repo_path = os.path.join(base_path, repo_name)
if not check_repo_fast(repo_name, base_path):
r = hg.repository(ui.ui(), repo_path)
if verify:
hg.verify(r)
#here we hnow that repo exists it was verified
log.info('%s repo is already created', repo_name)
except RepoError:
#it means that there is no valid repo there...
log.info('%s repo is free for creation', repo_name)
def ask_ok(prompt, retries=4, complaint='Yes or no, please!'):
ok = raw_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
#propagated from mercurial documentation
ui_sections = ['alias', 'auth',
'decode/encode', 'defaults',
'diff', 'email',
'extensions', 'format',
'merge-patterns', 'merge-tools',
'hooks', 'http_proxy',
'smtp', 'patch',
'paths', 'profiling',
'server', 'trusted',
'ui', 'web', ]
def make_ui(read_from='file', path=None, checkpaths=True):
A function that will read python rc files or database
and make an mercurial ui object from read options
:param path: path to mercurial config file
:param checkpaths: check the path
:param read_from: read from 'file' or 'db'
baseui = ui.ui()
#clean the baseui object
baseui._ocfg = config.config()
baseui._ucfg = config.config()
baseui._tcfg = config.config()
if read_from == 'file':
if not os.path.isfile(path):
log.warning('Unable to read config file %s' % path)
log.debug('reading hgrc from %s', path)
cfg = config.config()
cfg.read(path)
for k, v in cfg.items(section):
log.debug('settings ui from file[%s]%s:%s', section, k, v)
baseui.setconfig(section, k, v)
elif read_from == 'db':
ret = sa.query(RhodeCodeUi)\
.options(FromCache("sql_cache_short",
"get_hg_ui_settings")).all()
hg_ui = ret
for ui_ in hg_ui:
if ui_.ui_active:
log.debug('settings ui from db[%s]%s:%s', ui_.ui_section,
ui_.ui_key, ui_.ui_value)
baseui.setconfig(ui_.ui_section, ui_.ui_key, ui_.ui_value)
return baseui
def set_rhodecode_config(config):
"""Updates pylons config with new settings from database
hgsettings = SettingsModel().get_app_settings()
for k, v in hgsettings.items():
config[k] = v
def invalidate_cache(cache_key, *args):
"""Puts cache invalidation task into db for
further global cache invalidation
if cache_key.startswith('get_repo_cached_'):
name = cache_key.split('get_repo_cached_')[-1]
ScmModel().mark_for_invalidation(name)
class EmptyChangeset(BaseChangeset):
An dummy empty changeset. It's possible to pass hash when creating
an EmptyChangeset
def __init__(self, cs='0' * 40):
self._empty_cs = cs
self.revision = -1
self.message = ''
self.author = ''
self.date = ''
def raw_id(self):
"""Returns raw string identifying this changeset, useful for web
representation.
return self._empty_cs
def short_id(self):
return self.raw_id[:12]
def get_file_changeset(self, path):
return self
def get_file_content(self, path):
return u''
def get_file_size(self, path):
def repo2db_mapper(initial_repo_list, remove_obsolete=False):
"""maps all found repositories into db
user = sa.query(User).filter(User.admin == True).first()
for name, repo in initial_repo_list.items():
if not rm.get_by_repo_name(name, cache=False):
log.info('repository %s not found creating default', name)
form_data = {
'repo_name':name,
'repo_type':repo.alias,
'description':repo.description \
if repo.description != 'unknown' else \
'%s repository' % name,
'private':False
rm.create(form_data, user, just_db=True)
if remove_obsolete:
#remove from database those repositories that are not in the filesystem
for repo in sa.query(Repository).all():
if repo.repo_name not in initial_repo_list.keys():
sa.delete(repo)
class OrderedDict(dict, DictMixin):
def __init__(self, *args, **kwds):
if len(args) > 1:
raise TypeError('expected at most 1 arguments, got %d' % len(args))
self.__end
except AttributeError:
self.clear()
self.update(*args, **kwds)
def clear(self):
self.__end = end = []
end += [None, end, end] # sentinel node for doubly linked list
self.__map = {} # key --> [key, prev, next]
dict.clear(self)
def __setitem__(self, key, value):
if key not in self:
end = self.__end
curr = end[1]
curr[2] = end[1] = self.__map[key] = [key, curr, end]
dict.__setitem__(self, key, value)
def __delitem__(self, key):
dict.__delitem__(self, key)
key, prev, next = self.__map.pop(key)
prev[2] = next
next[1] = prev
def __iter__(self):
curr = end[2]
while curr is not end:
yield curr[0]
curr = curr[2]
def __reversed__(self):
curr = curr[1]
def popitem(self, last=True):
if not self:
raise KeyError('dictionary is empty')
if last:
key = reversed(self).next()
key = iter(self).next()
value = self.pop(key)
return key, value
def __reduce__(self):
items = [[k, self[k]] for k in self]
tmp = self.__map, self.__end
del self.__map, self.__end
inst_dict = vars(self).copy()
self.__map, self.__end = tmp
if inst_dict:
return (self.__class__, (items,), inst_dict)
return self.__class__, (items,)
def keys(self):
return list(self)
setdefault = DictMixin.setdefault
update = DictMixin.update
pop = DictMixin.pop
values = DictMixin.values
items = DictMixin.items
iterkeys = DictMixin.iterkeys
itervalues = DictMixin.itervalues
iteritems = DictMixin.iteritems
return '%s()' % (self.__class__.__name__,)
return '%s(%r)' % (self.__class__.__name__, self.items())
def copy(self):
return self.__class__(self)
def fromkeys(cls, iterable, value=None):
d = cls()
for key in iterable:
d[key] = value
return d
def __eq__(self, other):
if isinstance(other, OrderedDict):
return len(self) == len(other) and self.items() == other.items()
return dict.__eq__(self, other)
def __ne__(self, other):
return not self == other
#set cache regions for beaker so celery can utilise it
def add_cache(settings):
cache_settings = {'regions':None}
for key in settings.keys():
for prefix in ['beaker.cache.', 'cache.']:
if key.startswith(prefix):
name = key.split(prefix)[1].strip()
cache_settings[name] = settings[key].strip()
if cache_settings['regions']:
for region in cache_settings['regions'].split(','):
region = region.strip()
region_settings = {}
for key, value in cache_settings.items():
if key.startswith(region):
region_settings[key.split('.')[1]] = value
region_settings['expire'] = int(region_settings.get('expire',
60))
region_settings.setdefault('lock_dir',
cache_settings.get('lock_dir'))
if 'type' not in region_settings:
region_settings['type'] = cache_settings.get('type',
'memory')
beaker.cache.cache_regions[region] = region_settings
def get_current_revision():
"""Returns tuple of (number, id) from repository containing this package
or None if repository could not be found.
from vcs import get_repo
from vcs.exceptions import RepositoryError, VCSError
repopath = os.path.join(os.path.dirname(__file__), '..', '..')
scm = get_scm(repopath)[0]
repo = get_repo(path=repopath, alias=scm)
return (tip.revision, tip.short_id)
except (ImportError, RepositoryError, VCSError), err:
logging.debug("Cannot retrieve rhodecode's revision. Original error "
"was: %s" % err)
# TEST FUNCTIONS AND CREATORS
def create_test_index(repo_location, full_index):
"""Makes default test index
:param repo_location:
:param full_index:
import shutil
index_location = os.path.join(repo_location, 'index')
if os.path.exists(index_location):
shutil.rmtree(index_location)
l = DaemonLock()
repo_location=repo_location)\
def create_test_env(repos_test_path, config):
"""Makes a fresh database and
install test repository into tmp dir
from rhodecode.tests import HG_REPO, GIT_REPO, NEW_HG_REPO, NEW_GIT_REPO, \
HG_FORK, GIT_FORK, TESTS_TMP_PATH
from os.path import dirname as dn, join as jn, abspath
log = logging.getLogger('TestEnvCreator')
log.propagate = True
#PART ONE create db
dbconf = config['sqlalchemy.db1.url']
log.debug('making test db %s', dbconf)
dbmanage = DbManage(log_sql=True, dbconf=dbconf, root=config['here'],
tests=True)
dbmanage.create_tables(override=True)
dbmanage.config_prompt(repos_test_path)
dbmanage.create_default_user()
dbmanage.admin_prompt()
dbmanage.create_permissions()
dbmanage.populate_default_permissions()
#PART TWO make test repo
log.debug('making test vcs repositories')
#remove old one from previos tests
for r in [HG_REPO, GIT_REPO, NEW_HG_REPO, NEW_GIT_REPO, HG_FORK, GIT_FORK]:
if os.path.isdir(jn(TESTS_TMP_PATH, r)):
log.debug('removing %s', r)
shutil.rmtree(jn(TESTS_TMP_PATH, r))
#CREATE DEFAULT HG REPOSITORY
cur_dir = dn(dn(abspath(__file__)))
tar = tarfile.open(jn(cur_dir, 'tests', "vcs_test_hg.tar.gz"))
tar.extractall(jn(TESTS_TMP_PATH, HG_REPO))
#==============================================================================
# PASTER COMMANDS
class BasePasterCommand(Command):
Abstract Base Class for paster commands.
The celery commands are somewhat aggressive about loading
celery.conf, and since our module sets the `CELERY_LOADER`
environment variable to our loader, we have to bootstrap a bit and
make sure we've had a chance to load the pylons config off of the
command line, otherwise everything fails.
min_args_error = "Please provide a paster config file as an argument."
takes_config_file = 1
requires_config_file = True
def notify_msg(self, msg, log=False):
"""Make a notification to user, additionally if logger is passed
it logs this action using given logger
:param msg: message that will be printed to user
:param log: logging instance, to use to additionally log this message
print msg
if log and isinstance(log, logging):
log(msg)
def run(self, args):
Overrides Command.run
Checks for a config file argument and loads it.
if len(args) < self.min_args:
raise BadCommand(
self.min_args_error % {'min_args': self.min_args,
'actual_args': len(args)})
# Decrement because we're going to lob off the first argument.
# @@ This is hacky
self.min_args -= 1
self.bootstrap_config(args[0])
self.update_parser()
return super(BasePasterCommand, self).run(args[1:])
Abstract method. Allows for the class's parser to be updated
before the superclass's `run` method is called. Necessary to
allow options/arguments to be passed through to the underlying
celery command.
raise NotImplementedError("Abstract Method.")
def bootstrap_config(self, conf):
Loads the pylons configuration.
from pylons import config as pylonsconfig
path_to_ini_file = os.path.realpath(conf)
conf = paste.deploy.appconfig('config:' + path_to_ini_file)
pylonsconfig.init_app(conf.global_conf, conf.local_conf)
rhodecode.model.__init__
~~~~~~~~~~~~~~~~~~~~~~~~
The application's model objects
:created_on: Nov 25, 2010
:example:
.. code-block:: python
from paste.deploy import appconfig
from rhodecode.config.environment import load_environment
conf = appconfig('config:development.ini', relative_to = './../../')
load_environment(conf.global_conf, conf.local_conf)
engine = engine_from_config(config, 'sqlalchemy.')
# RUN YOUR CODE HERE
def init_model(engine):
"""Initializes db session, bind the engine with the metadata,
Call this before using any of the tables or classes in the model, preferably
once in application start
:param engine: engine to bind to
log.info("initializing db models for %s", engine)
meta.Base.metadata.bind = engine
class BaseModel(object):
"""Base Model for all RhodeCode models, it adds sql alchemy session
into instance of model
:param sa: If passed it reuses this session instead of creating a new one
def __init__(self, sa=None):
if sa is not None:
self.sa = sa
self.sa = meta.Session()
rhodecode.model.db
Database Models for RhodeCode
:created_on: Apr 08, 2010
from datetime import date
from sqlalchemy import *
from sqlalchemy.orm import relationship, backref
from sqlalchemy.orm.interfaces import MapperExtension
from rhodecode.model.meta import Base, Session
class RhodeCodeSettings(Base):
__tablename__ = 'rhodecode_settings'
__table_args__ = (UniqueConstraint('app_settings_name'), {'useexisting':True})
app_settings_id = Column("app_settings_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
app_settings_name = Column("app_settings_name", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
app_settings_value = Column("app_settings_value", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
def __init__(self, k='', v=''):
self.app_settings_name = k
self.app_settings_value = v
return "<%s('%s:%s')>" % (self.__class__.__name__,
self.app_settings_name, self.app_settings_value)
class RhodeCodeUi(Base):
__tablename__ = 'rhodecode_ui'
__table_args__ = {'useexisting':True}
ui_id = Column("ui_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
ui_section = Column("ui_section", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
ui_key = Column("ui_key", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
ui_value = Column("ui_value", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
ui_active = Column("ui_active", Boolean(), nullable=True, unique=None, default=True)
class User(Base):
__tablename__ = 'users'
__table_args__ = (UniqueConstraint('username'), UniqueConstraint('email'), {'useexisting':True})
user_id = Column("user_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
username = Column("username", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
password = Column("password", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
active = Column("active", Boolean(), nullable=True, unique=None, default=None)
admin = Column("admin", Boolean(), nullable=True, unique=None, default=False)
name = Column("name", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
lastname = Column("lastname", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
email = Column("email", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
last_login = Column("last_login", DateTime(timezone=False), nullable=True, unique=None, default=None)
is_ldap = Column("is_ldap", Boolean(), nullable=False, unique=None, default=False)
user_log = relationship('UserLog', cascade='all')
user_perms = relationship('UserToPerm', primaryjoin="User.user_id==UserToPerm.user_id", cascade='all')
repositories = relationship('Repository')
user_followers = relationship('UserFollowing', primaryjoin='UserFollowing.follows_user_id==User.user_id', cascade='all')
@property
def full_contact(self):
return '%s %s <%s>' % (self.name, self.lastname, self.email)
def short_contact(self):
return '%s %s' % (self.name, self.lastname)
def is_admin(self):
return self.admin
return "<%s('id:%s:%s')>" % (self.__class__.__name__,
self.user_id, self.username)
def by_username(cls, username):
return Session.query(cls).filter(cls.username == username).one()
def update_lastlogin(self):
"""Update user lastlogin"""
session = Session.object_session(self)
self.last_login = datetime.datetime.now()
session.add(self)
session.commit()
log.debug('updated user %s lastlogin', self.username)
session.rollback()
class UserLog(Base):
__tablename__ = 'user_logs'
user_log_id = Column("user_log_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
user_id = Column("user_id", Integer(), ForeignKey('users.user_id'), nullable=False, unique=None, default=None)
repository_id = Column("repository_id", Integer(length=None, convert_unicode=False, assert_unicode=None), ForeignKey('repositories.repo_id'), nullable=False, unique=None, default=None)
repository_name = Column("repository_name", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
user_ip = Column("user_ip", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
action = Column("action", UnicodeText(length=1200000, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
action_date = Column("action_date", DateTime(timezone=False), nullable=True, unique=None, default=None)
def action_as_day(self):
return date(*self.action_date.timetuple()[:3])
user = relationship('User')
repository = relationship('Repository')
class Repository(Base):
__tablename__ = 'repositories'
__table_args__ = (UniqueConstraint('repo_name'), {'useexisting':True},)
repo_id = Column("repo_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
repo_name = Column("repo_name", String(length=255, convert_unicode=False, assert_unicode=None), nullable=False, unique=True, default=None)
repo_type = Column("repo_type", String(length=255, convert_unicode=False, assert_unicode=None), nullable=False, unique=False, default='hg')
user_id = Column("user_id", Integer(), ForeignKey('users.user_id'), nullable=False, unique=False, default=None)
private = Column("private", Boolean(), nullable=True, unique=None, default=None)
enable_statistics = Column("statistics", Boolean(), nullable=True, unique=None, default=True)
description = Column("description", String(length=10000, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
fork_id = Column("fork_id", Integer(), ForeignKey('repositories.repo_id'), nullable=True, unique=False, default=None)
fork = relationship('Repository', remote_side=repo_id)
repo_to_perm = relationship('RepoToPerm', cascade='all')
stats = relationship('Statistics', cascade='all', uselist=False)
repo_followers = relationship('UserFollowing', primaryjoin='UserFollowing.follows_repo_id==Repository.repo_id', cascade='all')
logs = relationship('UserLog', cascade='all')
self.repo_id, self.repo_name)
class Permission(Base):
__tablename__ = 'permissions'
permission_id = Column("permission_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
permission_name = Column("permission_name", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
permission_longname = Column("permission_longname", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
self.permission_id, self.permission_name)
class RepoToPerm(Base):
__tablename__ = 'repo_to_perm'
__table_args__ = (UniqueConstraint('user_id', 'repository_id'), {'useexisting':True})
repo_to_perm_id = Column("repo_to_perm_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
permission_id = Column("permission_id", Integer(), ForeignKey('permissions.permission_id'), nullable=False, unique=None, default=None)
repository_id = Column("repository_id", Integer(), ForeignKey('repositories.repo_id'), nullable=False, unique=None, default=None)
permission = relationship('Permission')
class UserToPerm(Base):
__tablename__ = 'user_to_perm'
__table_args__ = (UniqueConstraint('user_id', 'permission_id'), {'useexisting':True})
user_to_perm_id = Column("user_to_perm_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
class Statistics(Base):
__tablename__ = 'statistics'
__table_args__ = (UniqueConstraint('repository_id'), {'useexisting':True})
stat_id = Column("stat_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
repository_id = Column("repository_id", Integer(), ForeignKey('repositories.repo_id'), nullable=False, unique=True, default=None)
stat_on_revision = Column("stat_on_revision", Integer(), nullable=False)
commit_activity = Column("commit_activity", LargeBinary(1000000), nullable=False)#JSON data
commit_activity_combined = Column("commit_activity_combined", LargeBinary(), nullable=False)#JSON data
languages = Column("languages", LargeBinary(1000000), nullable=False)#JSON data
repository = relationship('Repository', single_parent=True)
class UserFollowing(Base):
__tablename__ = 'user_followings'
__table_args__ = (UniqueConstraint('user_id', 'follows_repository_id'),
UniqueConstraint('user_id', 'follows_user_id')
, {'useexisting':True})
user_following_id = Column("user_following_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
follows_repo_id = Column("follows_repository_id", Integer(), ForeignKey('repositories.repo_id'), nullable=True, unique=None, default=None)
follows_user_id = Column("follows_user_id", Integer(), ForeignKey('users.user_id'), nullable=True, unique=None, default=None)
user = relationship('User', primaryjoin='User.user_id==UserFollowing.user_id')
follows_user = relationship('User', primaryjoin='User.user_id==UserFollowing.follows_user_id')
follows_repository = relationship('Repository', order_by='Repository.repo_name')
class CacheInvalidation(Base):
__tablename__ = 'cache_invalidation'
__table_args__ = (UniqueConstraint('cache_key'), {'useexisting':True})
cache_id = Column("cache_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
cache_key = Column("cache_key", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
cache_args = Column("cache_args", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
cache_active = Column("cache_active", Boolean(), nullable=True, unique=None, default=False)
def __init__(self, cache_key, cache_args=''):
self.cache_key = cache_key
self.cache_args = cache_args
self.cache_active = False
self.cache_id, self.cache_key)
class DbMigrateVersion(Base):
__tablename__ = 'db_migrate_version'
repository_id = Column('repository_id', String(255), primary_key=True)
repository_path = Column('repository_path', Text)
version = Column('version', Integer)
rhodecode.model.permission
permissions model for RhodeCode
:created_on: Aug 20, 2010
from rhodecode.model import BaseModel
from rhodecode.model.db import User, Permission, UserToPerm, RepoToPerm
class PermissionModel(BaseModel):
"""Permissions model for RhodeCode
def get_permission(self, permission_id, cache=False):
"""Get's permissions by id
:param permission_id: id of permission to get from database
:param cache: use Cache for this query
perm = self.sa.query(Permission)
if cache:
perm = perm.options(FromCache("sql_cache_short",
"get_permission_%s" % permission_id))
return perm.get(permission_id)
def get_permission_by_name(self, name, cache=False):
"""Get's permissions by given name
:param name: name to fetch
:param cache: Use cache for this query
perm = self.sa.query(Permission)\
.filter(Permission.permission_name == name)
"get_permission_%s" % name))
return perm.scalar()
def update(self, form_result):
perm_user = self.sa.query(User)\
.filter(User.username == form_result['perm_user_name']).scalar()
u2p = self.sa.query(UserToPerm).filter(UserToPerm.user == perm_user).all()
if len(u2p) != 3:
raise Exception('Defined: %s should be 3 permissions for default'
' user. This should not happen please verify'
' your database' % len(u2p))
#stage 1 change defaults
for p in u2p:
p.permission = self.get_permission_by_name(
form_result['default_perm'])
self.sa.add(p)
form_result['default_register'])
form_result['default_create'])
#stage 2 update all default permissions for repos if checked
if form_result['overwrite_default'] == True:
for r2p in self.sa.query(RepoToPerm)\
.filter(RepoToPerm.user == perm_user).all():
r2p.permission = self.get_permission_by_name(
self.sa.add(r2p)
#stage 3 set anonymous access
if perm_user.username == 'default':
perm_user.active = bool(form_result['anonymous'])
self.sa.add(perm_user)
rhodecode.model.repo
~~~~~~~~~~~~~~~~~~~~
Repository model for rhodecode
:created_on: Jun 5, 2010
from datetime import datetime
from sqlalchemy.orm import joinedload, make_transient
from vcs.backends import get_backend
from rhodecode.model.db import Repository, RepoToPerm, User, Permission, \
Statistics, RhodeCodeUi
class RepoModel(BaseModel):
def repos_path(self):
"""Get's the repositories root path from database
q = self.sa.query(RhodeCodeUi).filter(RhodeCodeUi.ui_key == '/').one()
def get(self, repo_id, cache=False):
repo = self.sa.query(Repository)\
.filter(Repository.repo_id == repo_id)
repo = repo.options(FromCache("sql_cache_short",
"get_repo_%s" % repo_id))
return repo.scalar()
def get_by_repo_name(self, repo_name, cache=False):
.filter(Repository.repo_name == repo_name)
"get_repo_%s" % repo_name))
def get_users_js(self):
users = self.sa.query(User).filter(User.active == True).all()
u_tmpl = '''{id:%s, fname:"%s", lname:"%s", nname:"%s"},'''
users_array = '[%s];' % '\n'.join([u_tmpl % (u.user_id, u.name,
u.lastname, u.username)
for u in users])
return users_array
def update(self, repo_name, form_data):
cur_repo = self.get_by_repo_name(repo_name, cache=False)
user_model = UserModel(self.sa)
#update permissions
for username, perm in form_data['perms_updates']:
r2p = self.sa.query(RepoToPerm)\
.filter(RepoToPerm.user == user_model.get_by_username(username))\
.filter(RepoToPerm.repository == cur_repo)\
.one()
r2p.permission = self.sa.query(Permission)\
.filter(Permission.permission_name == perm)\
#set new permissions
for username, perm in form_data['perms_new']:
r2p = RepoToPerm()
r2p.repository = cur_repo
r2p.user = user_model.get_by_username(username, cache=False)
#update current repo
for k, v in form_data.items():
if k == 'user':
cur_repo.user = user_model.get(v)
setattr(cur_repo, k, v)
self.sa.add(cur_repo)
if repo_name != form_data['repo_name']:
#rename our data
self.__rename_repo(repo_name, form_data['repo_name'])
def create(self, form_data, cur_user, just_db=False, fork=False):
if fork:
#force str since hg doesn't go with unicode
repo_name = str(form_data['fork_name'])
org_name = str(form_data['repo_name'])
org_name = repo_name = str(form_data['repo_name'])
new_repo = Repository()
new_repo.enable_statistics = True
if k == 'repo_name':
v = repo_name
setattr(new_repo, k, v)
parent_repo = self.sa.query(Repository)\
.filter(Repository.repo_name == org_name).scalar()
new_repo.fork = parent_repo
new_repo.user_id = cur_user.user_id
self.sa.add(new_repo)
#create default permission
repo_to_perm = RepoToPerm()
default = 'repository.read'
for p in UserModel(self.sa).get_by_username('default', cache=False).user_perms:
default = p.permission.permission_name
default_perm = 'repository.none' if form_data['private'] else default
repo_to_perm.permission_id = self.sa.query(Permission)\
.filter(Permission.permission_name == default_perm)\
.one().permission_id
repo_to_perm.repository = new_repo
repo_to_perm.user_id = UserModel(self.sa)\
.get_by_username('default', cache=False).user_id
self.sa.add(repo_to_perm)
if not just_db:
self.__create_repo(repo_name, form_data['repo_type'])
#now automatically start following this repository as owner
ScmModel(self.sa).toggle_following_repo(new_repo.repo_id,
cur_user.user_id)
def create_fork(self, form_data, cur_user):
run_task(tasks.create_repo_fork, form_data, cur_user)
def delete(self, repo):
self.sa.delete(repo)
self.__delete_repo(repo)
def delete_perm_user(self, form_data, repo_name):
self.sa.query(RepoToPerm)\
.filter(RepoToPerm.repository \
== self.get_by_repo_name(repo_name))\
.filter(RepoToPerm.user_id == form_data['user_id']).delete()
def delete_stats(self, repo_name):
self.sa.query(Statistics)\
.filter(Statistics.repository == \
self.get_by_repo_name(repo_name)).delete()
def __create_repo(self, repo_name, alias):
makes repository on filesystem
:param alias:
from rhodecode.lib.utils import check_repo
repo_path = os.path.join(self.repos_path, repo_name)
if check_repo(repo_name, self.repos_path):
log.info('creating repo %s in %s', repo_name, repo_path)
backend(repo_path, create=True)
def __rename_repo(self, old, new):
renames repository on filesystem
:param old: old name
:param new: new name
log.info('renaming repo from %s to %s', old, new)
old_path = os.path.join(self.repos_path, old)
new_path = os.path.join(self.repos_path, new)
if os.path.isdir(new_path):
raise Exception('Was trying to rename to already existing dir %s',
new_path)
shutil.move(old_path, new_path)
def __delete_repo(self, repo):
removes repo from filesystem, the removal is acctually made by
added rm__ prefix into dir, and rename internat .hg/.git dirs so this
repository is no longer valid for rhodecode, can be undeleted later on
by reverting the renames on this repository
:param repo: repo object
rm_path = os.path.join(self.repos_path, repo.repo_name)
log.info("Removing %s", rm_path)
#disable hg/git
alias = repo.repo_type
shutil.move(os.path.join(rm_path, '.%s' % alias),
os.path.join(rm_path, 'rm__.%s' % alias))
#disable repo
shutil.move(rm_path, os.path.join(self.repos_path, 'rm__%s__%s' \
% (datetime.today().isoformat(),
repo.repo_name)))
rhodecode.model.settings
Settings model for RhodeCode
:created on Nov 17, 2010
from rhodecode.model.db import RhodeCodeSettings
class SettingsModel(BaseModel):
Settings model
def get(self, settings_key, cache=False):
r = self.sa.query(RhodeCodeSettings)\
.filter(RhodeCodeSettings.app_settings_name == settings_key).scalar()
r = r.options(FromCache("sql_cache_short",
"get_setting_%s" % settings_key))
def get_app_settings(self, cache=False):
"""Get's config from database, each config key is prefixed with
'rhodecode_' prefix, than global pylons config is updated with such
keys
ret = self.sa.query(RhodeCodeSettings)
ret = ret.options(FromCache("sql_cache_short", "get_hg_settings"))
raise Exception('Could not get application settings !')
settings['rhodecode_' + each.app_settings_name] = each.app_settings_value
def get_ldap_settings(self):
Returns ldap settings from database
:returns:
ldap_active
ldap_host
ldap_port
ldap_ldaps
ldap_tls_reqcert
ldap_dn_user
ldap_dn_pass
ldap_base_dn
ldap_filter
ldap_search_scope
ldap_attr_login
ldap_attr_firstname
ldap_attr_lastname
ldap_attr_email
# ldap_search_scope
.filter(RhodeCodeSettings.app_settings_name\
.startswith('ldap_'))\
fd = {}
for row in r:
v = row.app_settings_value
if v in ['true', 'yes', 'on', 'y', 't', '1']:
v = True
elif v in ['false', 'no', 'off', 'n', 'f', '0']:
v = False
fd.update({row.app_settings_name:v})
return fd
rhodecode.model.user
users model for RhodeCode
:created_on: Apr 9, 2010
from rhodecode.lib.exceptions import DefaultUserException, UserOwnsReposException
class UserModel(BaseModel):
def get(self, user_id, cache=False):
user = self.sa.query(User)
user = user.options(FromCache("sql_cache_short",
"get_user_%s" % user_id))
return user.get(user_id)
def get_by_username(self, username, cache=False, case_insensitive=False):
if case_insensitive:
user = self.sa.query(User).filter(User.username.ilike(username))
user = self.sa.query(User)\
.filter(User.username == username)
"get_user_%s" % username))
return user.scalar()
def create(self, form_data):
new_user = User()
setattr(new_user, k, v)
self.sa.add(new_user)
def create_ldap(self, username, password):
Checks if user is in database, if not creates this user marked
as ldap user
:param username:
:param password:
from rhodecode.lib.auth import get_crypt_password
log.debug('Checking for such ldap account in RhodeCode database')
if self.get_by_username(username, case_insensitive=True) is None:
new_user.username = username.lower()#add ldap account always lowercase
new_user.password = get_crypt_password(password)
new_user.email = '%s@ldap.server' % username
new_user.active = True
new_user.is_ldap = True
new_user.name = '%s@ldap' % username
new_user.lastname = ''
log.debug('this %s user exists skipping creation of ldap account',
def create_registration(self, form_data):
if k != 'admin':
body = ('New user registration\n'
'username: %s\n'
'email: %s\n')
body = body % (form_data['username'], form_data['email'])
run_task(tasks.send_email, None,
_('[RhodeCode] New User registration'),
body)
def update(self, user_id, form_data):
new_user = self.get(user_id, cache=False)
if new_user.username == 'default':
raise DefaultUserException(
_("You can't Edit this user since it's"
" crucial for entire application"))
if k == 'new_password' and v != '':
new_user.password = v
def update_my_account(self, user_id, form_data):
if k not in ['admin', 'active']:
def delete(self, user_id):
user = self.get(user_id, cache=False)
if user.username == 'default':
_("You can't remove this user since it's"
if user.repositories:
raise UserOwnsReposException(_('This user still owns %s '
'repositories and cannot be '
'removed. Switch owners or '
'remove those repositories') \
% user.repositories)
self.sa.delete(user)
def reset_password(self, data):
run_task(tasks.reset_user_password, data['email'])
def fill_data(self, user):
Fills user data with those from database and log out user if not
present in database
if not hasattr(user, 'user_id') or user.user_id is None:
raise Exception('passed in user has to have the user_id attribute')
log.debug('filling auth user data')
dbuser = self.get(user.user_id)
user.username = dbuser.username
user.is_admin = dbuser.admin
user.name = dbuser.name
user.lastname = dbuser.lastname
user.email = dbuser.email
Status change: