#!/usr/bin/env python
# encoding: utf-8
# summary controller for pylons
# Copyright (C) 2009-2010 Marcin Kuzminski <marcin@python-works.com>
#
# 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 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.
"""
Created on April 18, 2010
summary controller for pylons
@author: marcink
from pylons import tmpl_context as c, request, url
from vcs.exceptions import ChangesetError
from rhodecode.lib.auth import LoginRequired, HasRepoPermissionAnyDecorator
from rhodecode.lib.base import BaseController, render
from rhodecode.lib.utils import OrderedDict, EmptyChangeset
from rhodecode.model.hg import HgModel
from rhodecode.model.db import Statistics
from webhelpers.paginate import Page
from rhodecode.lib.celerylib import run_task
from rhodecode.lib.celerylib.tasks import get_commits_stats
from datetime import datetime, timedelta
from time import mktime
import calendar
import logging
try:
import json
except ImportError:
#python 2.5 compatibility
import simplejson as json
log = logging.getLogger(__name__)
class SummaryController(BaseController):
@LoginRequired()
@HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
'repository.admin')
def __before__(self):
super(SummaryController, self).__before__()
def index(self):
hg_model = HgModel()
c.repo_info = hg_model.get_repo(c.repo_name)
def url_generator(**kw):
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)
e = request.environ
uri = u'%(protocol)s://%(user)s@%(host)s%(prefix)s/%(repo_name)s' % {
if self.rhodecode_user.username == 'default':
password = ':default'
else:
password = ''
uri = u'%(protocol)s://%(user)s%(password)s@%(host)s%(prefix)s/%(repo_name)s' % {
'protocol': e.get('wsgi.url_scheme'),
'user':str(c.rhodecode_user.username),
'password':password,
'host':e.get('HTTP_HOST'),
'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)
except ChangesetError:
c.repo_tags[name] = EmptyChangeset(hash)
c.repo_branches = OrderedDict()
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 = datetime.today() + timedelta(days=1)
y, m, d = td.year, td.month, td.day
ts_min_y = mktime((y - 1, (td - timedelta(days=calendar.mdays[m])).month,
d, 0, 0, 0, 0, 0, 0,))
ts_min_m = mktime((y, (td - timedelta(days=calendar.mdays[m])).month,
ts_max_y = mktime((y, m, d, 0, 0, 0, 0, 0, 0,))
run_task(get_commits_stats, c.repo_info.name, ts_min_y, ts_max_y)
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:
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])[:2]
)
c.commit_data = json.dumps({})
c.overview_data = json.dumps([[ts_min_y, 0], [ts_max_y, 0] ])
c.trending_languages = json.dumps({})
return render('summary/summary.html')
# authentication and permission libraries
Created on April 4, 2010
from pylons import config, session, url, request
from pylons.controllers.util import abort, redirect
from rhodecode.lib.utils import get_repo_slug
from rhodecode.model import meta
from rhodecode.model.user import UserModel
from rhodecode.model.caching_query import FromCache
from rhodecode.model.db import User, RepoToPerm, Repository, Permission, \
UserToPerm
import bcrypt
from decorator import decorator
import random
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
def get_crypt_password(password):
"""Cryptographic function used for password hashing based on sha1
:param password: password to hash
return bcrypt.hashpw(password, bcrypt.gensalt(10))
def check_password(password, hashed):
return bcrypt.hashpw(password, hashed) == hashed
def authfunc(environ, username, password):
user = UserModel().get_by_username(username, cache=False)
if user:
if user.active:
if user.username == username and check_password(password, user.password):
if user.username == 'default' and user.active:
log.info('user %s authenticated correctly', username)
return True
elif user.username == username and check_password(password, user.password):
log.error('user %s is disabled', username)
return False
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()
except:
pass
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(sa).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:
# middleware to handle mercurial api calls
Created on 2010-04-28
SimpleHG middleware for handling mercurial protocol request (push/clone etc.)
It's implemented with basic auth function
from itertools import chain
from mercurial.error import RepoError
from mercurial.hgweb import hgweb
from mercurial.hgweb.request import wsgiapplication
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 is_mercurial, make_ui, invalidate_cache, \
check_repo_fast, ui_sections
from webob.exc import HTTPNotFound, HTTPForbidden, HTTPInternalServerError
import os
import traceback
class SimpleHg(object):
def __init__(self, application, config):
self.application = application
self.config = config
#authenticate this mercurial request using
#authenticate this mercurial request using authfunc
self.authenticate = AuthBasicAuthenticator('', authfunc)
self.ipaddr = '0.0.0.0'
self.repository = None
self.username = None
self.action = None
def __call__(self, environ, start_response):
if not is_mercurial(environ):
return self.application(environ, start_response)
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 MERCURIAL 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
log.error(traceback.format_exc())
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)
self.username = 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}
# MERCURIAL REQUEST HANDLING
environ['PATH_INFO'] = '/'#since we wrap into hgweb, reset the path
self.baseui = make_ui('db')
self.basepath = self.config['base_path']
self.repo_path = os.path.join(self.basepath, repo_name)
#quick check if that dir exists...
if check_repo_fast(repo_name, self.basepath):
return HTTPNotFound()(environ, start_response)
app = wsgiapplication(self.__make_app)
except RepoError, e:
if str(e).find('not found') != -1:
except Exception:
#invalidate cache on push
self.__invalidate_cache(repo_name)
Status change: