kallithea Changeset - 6edba875451d

Changeset - 6edba875451d

Parent rev.

Child rev.

[Not reviewed]

default

0 9 0

Søren Løvborg - 9 years ago 2017-02-27 20:36:49
sorenl@unity3d.com

cleanup: drop superfluous Session.add calls

Session.add should only be called on newly created database objects.

Per the Kallithea contribution guidelines:

When getting an object from the session (via Session().query or
any of the utility functions that look up objects in the database),
it's already part of the session, and should not be added again.

9 files changed with 6 insertions and 22 deletions:

kallithea/lib/db_manage.py

kallithea/model/changeset_status.py

kallithea/model/gist.py

kallithea/model/permission.py

kallithea/model/repo.py

kallithea/model/repo_group.py

kallithea/model/scm.py

kallithea/model/user.py

kallithea/model/user_group.py

0 comments (0 inline, 0 general)

kallithea/lib/db_manage.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.lib.db_manage
 ~~~~~~~~~~~~~~~~~~~~~~~
 Database creation, and setup module for Kallithea. Used for creation
 of database as well as for migration operations
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 10, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import sys
 import time
 import uuid
 import logging
 from os.path import dirname
 import alembic.config
 import alembic.command
 from kallithea.lib.paster_commands.common import ask_ok
 from kallithea.model.user import UserModel
 from kallithea.model.base import init_model
 from kallithea.model.db import User, Permission, Ui, \
     Setting, UserToPerm, RepoGroup, \
     UserRepoGroupToPerm, CacheInvalidation, Repository
 from sqlalchemy.engine import create_engine
 from kallithea.model.repo_group import RepoGroupModel
 #from kallithea.model import meta
 from kallithea.model.meta import Session, Base
 from kallithea.model.repo import RepoModel
 from kallithea.model.permission import PermissionModel
 log = logging.getLogger(__name__)
 def notify(msg):
     """
     Notification for migrations messages
     """
     ml = len(msg) + (4 * 2)
     print('\n%s\n*** %s ***\n%s' % ('*' * ml, msg, '*' * ml)).upper()
 class DbManage(object):
     def __init__(self, log_sql, dbconf, root, tests=False, SESSION=None, cli_args=None):
         self.dbname = dbconf.split('/')[-1]
         self.tests = tests
         self.root = root
         self.dburi = dbconf
         self.log_sql = log_sql
         self.db_exists = False
         self.cli_args = cli_args or {}
         self.init_db(SESSION=SESSION)
     def _ask_ok(self, msg):
         """Invoke ask_ok unless the force_ask option provides the answer"""
         force_ask = self.cli_args.get('force_ask')
         if force_ask is not None:
             return force_ask
         return ask_ok(msg)
     def init_db(self, SESSION=None):
         if SESSION:
             self.sa = SESSION
         else:
             #init new sessions
             engine = create_engine(self.dburi, echo=self.log_sql)
             init_model(engine)
             self.sa = Session()
     def create_tables(self, override=False):
         """
         Create a auth database
         """
         log.info("Any existing database is going to be destroyed")
         if self.tests:
             destroy = True
         else:
             destroy = self._ask_ok('Are you sure to destroy old database ? [y/n]')
         if not destroy:
             print 'Nothing done.'
             sys.exit(0)
         if destroy:
             Base.metadata.drop_all()
         checkfirst = not override
         Base.metadata.create_all(checkfirst=checkfirst)
         # Create an Alembic configuration and generate the version table,
         # "stamping" it with the most recent Alembic migration revision, to
         # tell Alembic that all the schema upgrades are already in effect.
         alembic_cfg = alembic.config.Config()
         alembic_cfg.set_main_option('script_location', 'kallithea:alembic')
         alembic_cfg.set_main_option('sqlalchemy.url', self.dburi)
         # This command will give an error in an Alembic multi-head scenario,
         # but in practice, such a scenario should not come up during database
         # creation, even during development.
         alembic.command.stamp(alembic_cfg, 'head')
         log.info('Created tables for %s', self.dbname)
     def fix_repo_paths(self):
         """
         Fixes a old kallithea version path into new one without a '*'
         """
         paths = self.sa.query(Ui) \
                 .filter(Ui.ui_key == '/') \
                 .scalar()
         paths.ui_value = paths.ui_value.replace('*', '')
         self.sa.add(paths)
         self.sa.commit()
     def fix_default_user(self):
         """
         Fixes a old default user with some 'nicer' default values,
         used mostly for anonymous access
         """
         def_user = self.sa.query(User).filter_by(is_default_user=True).one()
         def_user.name = 'Anonymous'
         def_user.lastname = 'User'
         def_user.email = 'anonymous@kallithea-scm.org'
         self.sa.add(def_user)
         self.sa.commit()
     def fix_settings(self):
         """
         Fixes kallithea settings adds ga_code key for google analytics
         """
         hgsettings3 = Setting('ga_code', '')
         self.sa.add(hgsettings3)
         self.sa.commit()
     def admin_prompt(self, second=False):
         if not self.tests:
             import getpass
             username = self.cli_args.get('username')
             password = self.cli_args.get('password')
             email = self.cli_args.get('email')
             def get_password():
                 password = getpass.getpass('Specify admin password '
                                            '(min 6 chars):')
                 confirm = getpass.getpass('Confirm password:')
                 if password != confirm:
                     log.error('passwords mismatch')
                     return False
                 if len(password) < 6:
                     log.error('password is to short use at least 6 characters')
                     return False
                 return password
             if username is None:
                 username = raw_input('Specify admin username:')
             if password is None:
                 password = get_password()
                 if not password:
                     #second try
                     password = get_password()
                     if not password:
                         sys.exit()
             if email is None:
                 email = raw_input('Specify admin email:')
             self.create_user(username, password, email, True)
         else:
             log.info('creating admin and regular test users')
             from kallithea.tests.base import TEST_USER_ADMIN_LOGIN, \
             TEST_USER_ADMIN_PASS, TEST_USER_ADMIN_EMAIL, \
             TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS, \
             TEST_USER_REGULAR_EMAIL, TEST_USER_REGULAR2_LOGIN, \
             TEST_USER_REGULAR2_PASS, TEST_USER_REGULAR2_EMAIL
             self.create_user(TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS,
                              TEST_USER_ADMIN_EMAIL, True)
             self.create_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS,
                              TEST_USER_REGULAR_EMAIL, False)
             self.create_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS,
                              TEST_USER_REGULAR2_EMAIL, False)
     def create_ui_settings(self, repo_store_path):
         """
         Creates ui settings, fills out hooks
         """
         #HOOKS
         hooks1_key = Ui.HOOK_UPDATE
         hooks1_ = self.sa.query(Ui) \
             .filter(Ui.ui_key == hooks1_key).scalar()
         hooks1 = Ui() if hooks1_ is None else hooks1_
         hooks1.ui_section = 'hooks'
         hooks1.ui_key = hooks1_key
         hooks1.ui_value = 'hg update >&2'
         hooks1.ui_active = False
         self.sa.add(hooks1)
         hooks2_key = Ui.HOOK_REPO_SIZE
         hooks2_ = self.sa.query(Ui) \
             .filter(Ui.ui_key == hooks2_key).scalar()
         hooks2 = Ui() if hooks2_ is None else hooks2_
         hooks2.ui_section = 'hooks'
         hooks2.ui_key = hooks2_key
         hooks2.ui_value = 'python:kallithea.lib.hooks.repo_size'
         self.sa.add(hooks2)
         hooks3 = Ui()
         hooks3.ui_section = 'hooks'
         hooks3.ui_key = Ui.HOOK_PUSH
         hooks3.ui_value = 'python:kallithea.lib.hooks.log_push_action'
         self.sa.add(hooks3)
         hooks4 = Ui()
         hooks4.ui_section = 'hooks'
         hooks4.ui_key = Ui.HOOK_PRE_PUSH
         hooks4.ui_value = 'python:kallithea.lib.hooks.pre_push'
         self.sa.add(hooks4)
         hooks5 = Ui()
         hooks5.ui_section = 'hooks'
         hooks5.ui_key = Ui.HOOK_PULL
         hooks5.ui_value = 'python:kallithea.lib.hooks.log_pull_action'
         self.sa.add(hooks5)
         hooks6 = Ui()
         hooks6.ui_section = 'hooks'
         hooks6.ui_key = Ui.HOOK_PRE_PULL
         hooks6.ui_value = 'python:kallithea.lib.hooks.pre_pull'
         self.sa.add(hooks6)
         # enable largefiles
         largefiles = Ui()
         largefiles.ui_section = 'extensions'
         largefiles.ui_key = 'largefiles'
         largefiles.ui_value = ''
         self.sa.add(largefiles)
         # set default largefiles cache dir, defaults to
         # /repo location/.cache/largefiles
         largefiles = Ui()
         largefiles.ui_section = 'largefiles'
         largefiles.ui_key = 'usercache'
         largefiles.ui_value = os.path.join(repo_store_path, '.cache',
                                            'largefiles')
         self.sa.add(largefiles)
         # enable hgsubversion disabled by default
         hgsubversion = Ui()
         hgsubversion.ui_section = 'extensions'
         hgsubversion.ui_key = 'hgsubversion'
         hgsubversion.ui_value = ''
         hgsubversion.ui_active = False
         self.sa.add(hgsubversion)
         # enable hggit disabled by default
         hggit = Ui()
         hggit.ui_section = 'extensions'
         hggit.ui_key = 'hggit'
         hggit.ui_value = ''
         hggit.ui_active = False
         self.sa.add(hggit)
     def create_auth_plugin_options(self, skip_existing=False):
         """
         Create default auth plugin settings, and make it active
         :param skip_existing:
         """
         for k, v, t in [('auth_plugins', 'kallithea.lib.auth_modules.auth_internal', 'list'),
                      ('auth_internal_enabled', 'True', 'bool')]:
             if skip_existing and Setting.get_by_name(k) != None:
                 log.debug('Skipping option %s', k)
                 continue
             setting = Setting(k, v, t)
             self.sa.add(setting)
     def create_default_options(self, skip_existing=False):
         """Creates default settings"""
         for k, v, t in [
             ('default_repo_enable_locking',  False, 'bool'),
             ('default_repo_enable_downloads', False, 'bool'),
             ('default_repo_enable_statistics', False, 'bool'),
             ('default_repo_private', False, 'bool'),
             ('default_repo_type', 'hg', 'unicode')]:
             if skip_existing and Setting.get_by_name(k) is not None:
                 log.debug('Skipping option %s', k)
                 continue
             setting = Setting(k, v, t)
             self.sa.add(setting)
     def fixup_groups(self):
         def_usr = User.get_default_user()
         for g in RepoGroup.query().all():
             g.group_name = g.get_new_name(g.name)
             self.sa.add(g)
             # get default perm
             default = UserRepoGroupToPerm.query() \
                 .filter(UserRepoGroupToPerm.group == g) \
                 .filter(UserRepoGroupToPerm.user == def_usr) \
                 .scalar()
             if default is None:
                 log.debug('missing default permission for group %s adding', g)
                 RepoGroupModel()._create_default_perms(g)
     def reset_permissions(self, username):
         """
         Resets permissions to default state, useful when old systems had
         bad permissions, we must clean them up
         :param username:
         """
         default_user = User.get_by_username(username)
         if not default_user:
             return
         u2p = UserToPerm.query() \
             .filter(UserToPerm.user == default_user).all()
         fixed = False
         if len(u2p) != len(Permission.DEFAULT_USER_PERMISSIONS):
             for p in u2p:
                 Session().delete(p)
             fixed = True
             self.populate_default_permissions()
         return fixed
     def update_repo_info(self):
         for repo in Repository.query():
             repo.update_changeset_cache()
     def config_prompt(self, test_repo_path='', retries=3):
         _path = self.cli_args.get('repos_location')
         if retries == 3:
             log.info('Setting up repositories config')
         if _path is not None:
             path = _path
         elif not self.tests and not test_repo_path:
             path = raw_input(
                  'Enter a valid absolute path to store repositories. '
                  'All repositories in that path will be added automatically:'
+            )
         else:
             path = test_repo_path
         path_ok = True
         # check proper dir
         if not os.path.isdir(path):
             path_ok = False
             log.error('Given path %s is not a valid directory', path)
         elif not os.path.isabs(path):
             path_ok = False
             log.error('Given path %s is not an absolute path', path)
         # check if path is at least readable.
         if not os.access(path, os.R_OK):
             path_ok = False
             log.error('Given path %s is not readable', path)
         # check write access, warn user about non writeable paths
         elif not os.access(path, os.W_OK) and path_ok:
             log.warning('No write permission to given path %s', path)
             if not self._ask_ok('Given path %s is not writeable, do you want to '
                           'continue with read only mode ? [y/n]' % (path,)):
                 log.error('Canceled by user')
                 sys.exit(-1)
         if retries == 0:
             sys.exit('max retries reached')
         if not path_ok:
             if _path is not None:
                 sys.exit('Invalid repo path: %s' % _path)
             retries -= 1
             return self.config_prompt(test_repo_path, retries) # recursing!!!
         real_path = os.path.normpath(os.path.realpath(path))
         if real_path != os.path.normpath(path):
             log.warning('Using normalized path %s instead of %s', real_path, path)
         return real_path
     def create_settings(self, path):
         self.create_ui_settings(path)
         ui_config = [
             ('web', 'allow_archive', 'gz zip bz2'),
             ('web', 'baseurl', '/'),
             ('paths', '/', path),
             #('phases', 'publish', 'false')
+        ]
         for section, key, value in ui_config:
             ui_conf = Ui()
             setattr(ui_conf, 'ui_section', section)
             setattr(ui_conf, 'ui_key', key)
             setattr(ui_conf, 'ui_value', value)
             self.sa.add(ui_conf)
         settings = [
             ('realm', 'Kallithea', 'unicode'),
             ('title', '', 'unicode'),
             ('ga_code', '', 'unicode'),
             ('show_public_icon', True, 'bool'),
             ('show_private_icon', True, 'bool'),
             ('stylify_metatags', False, 'bool'),
             ('dashboard_items', 100, 'int'), # TODO: call it page_size
             ('admin_grid_items', 25, 'int'),
             ('show_version', True, 'bool'),
             ('use_gravatar', True, 'bool'),
             ('gravatar_url', User.DEFAULT_GRAVATAR_URL, 'unicode'),
             ('clone_uri_tmpl', Repository.DEFAULT_CLONE_URI, 'unicode'),
             ('update_url', Setting.DEFAULT_UPDATE_URL, 'unicode'),
+        ]
         for key, val, type_ in settings:
             sett = Setting(key, val, type_)
             self.sa.add(sett)
         self.create_auth_plugin_options()
         self.create_default_options()
         log.info('created ui config')
     def create_user(self, username, password, email='', admin=False):
         log.info('creating user %s', username)
         UserModel().create_or_update(username, password, email,
                                      firstname=u'Kallithea', lastname=u'Admin',
                                      active=True, admin=admin,
                                      extern_type=User.DEFAULT_AUTH_TYPE)
     def create_default_user(self):
         log.info('creating default user')
         # create default user for handling default permissions.
         user = UserModel().create_or_update(username=User.DEFAULT_USER,
                                             password=str(uuid.uuid1())[:20],
                                             email='anonymous@kallithea-scm.org',
                                             firstname=u'Anonymous',
                                             lastname=u'User')
         # based on configuration options activate/deactivate this user which
         # controls anonymous access
         if self.cli_args.get('public_access') is False:
             log.info('Public access disabled')
             user.active = False
             Session().commit()
     def create_permissions(self):
         """
         Creates all permissions defined in the system
         """
         # module.(access|create|change|delete)_[name]
         # module.(none|read|write|admin)
         log.info('creating permissions')
         PermissionModel(self.sa).create_permissions()
     def populate_default_permissions(self):
         """
         Populate default permissions. It will create only the default
         permissions that are missing, and not alter already defined ones
         """
         log.info('creating default user permissions')
         PermissionModel(self.sa).create_default_permissions(user=User.DEFAULT_USER)

kallithea/model/changeset_status.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.changeset_status
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Changeset status controller
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 30, 2012
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 from sqlalchemy.orm import joinedload
 from kallithea.model.base import BaseModel
 from kallithea.model.db import ChangesetStatus, PullRequest, Repository, User
 from kallithea.lib.exceptions import StatusChangeOnClosedPullRequestError
 log = logging.getLogger(__name__)
 class ChangesetStatusModel(BaseModel):
     def _get_status_query(self, repo, revision, pull_request,
                           with_revisions=False):
         repo = Repository.guess_instance(repo)
         q = ChangesetStatus.query() \
             .filter(ChangesetStatus.repo == repo)
         if not with_revisions:
             # only report the latest vote across all users! TODO: be smarter!
             q = q.filter(ChangesetStatus.version == 0)
         if revision:
             q = q.filter(ChangesetStatus.revision == revision)
         elif pull_request:
             pull_request = PullRequest.guess_instance(pull_request)
             q = q.filter(ChangesetStatus.pull_request == pull_request)
         else:
             raise Exception('Please specify revision or pull_request')
         q = q.order_by(ChangesetStatus.version.asc())
         return q
     def _calculate_status(self, statuses):
         """
         Given a list of statuses, calculate the resulting status, according to
         the policy: approve if consensus, reject when at least one reject.
         """
         if not statuses:
             return ChangesetStatus.STATUS_UNDER_REVIEW
         if all(st and st.status == ChangesetStatus.STATUS_APPROVED for st in statuses):
             return ChangesetStatus.STATUS_APPROVED
         if any(st and st.status == ChangesetStatus.STATUS_REJECTED for st in statuses):
             return ChangesetStatus.STATUS_REJECTED
         return ChangesetStatus.STATUS_UNDER_REVIEW
     def calculate_pull_request_result(self, pull_request):
         """
         Return a tuple (reviewers, pending reviewers, pull request status)
         Only approve and reject counts as valid votes.
         """
         # collect latest votes from all voters
         cs_statuses = dict()
         for st in reversed(self.get_statuses(pull_request.org_repo,
                                              pull_request=pull_request,
                                              with_revisions=True)):
             cs_statuses[st.author.username] = st
         # collect votes from official reviewers
         pull_request_reviewers = []
         pull_request_pending_reviewers = []
         relevant_statuses = []
         for user in pull_request.get_reviewer_users():
             st = cs_statuses.get(user.username)
             relevant_statuses.append(st)
             status = ChangesetStatus.STATUS_NOT_REVIEWED if st is None else st.status
             if status in (ChangesetStatus.STATUS_NOT_REVIEWED,
                           ChangesetStatus.STATUS_UNDER_REVIEW):
                 pull_request_pending_reviewers.append(user)
             pull_request_reviewers.append((user, status))
         result = self._calculate_status(relevant_statuses)
         return (pull_request_reviewers,
                 pull_request_pending_reviewers,
                 result)
     def get_statuses(self, repo, revision=None, pull_request=None,
                      with_revisions=False):
         q = self._get_status_query(repo, revision, pull_request,
                                    with_revisions)
         q = q.options(joinedload('author'))
         return q.all()
     def get_status(self, repo, revision=None, pull_request=None, as_str=True):
         """
         Returns latest status of changeset for given revision or for given
         pull request. Statuses are versioned inside a table itself and
         version == 0 is always the current one
         :param repo:
         :param revision: 40char hash or None
         :param pull_request: pull_request reference
         :param as_str: return status as string not object
         """
         q = self._get_status_query(repo, revision, pull_request)
         # need to use first here since there can be multiple statuses
         # returned from pull_request
         status = q.first()
         if as_str:
             return str(status.status) if status else ChangesetStatus.DEFAULT
         return status
     def set_status(self, repo, status, user, comment, revision=None,
                    pull_request=None, dont_allow_on_closed_pull_request=False):
         """
         Creates new status for changeset or updates the old ones bumping their
         version, leaving the current status at the value of 'status'.
         :param repo:
         :param status:
         :param user:
         :param comment:
         :param revision:
         :param pull_request:
         :param dont_allow_on_closed_pull_request: don't allow a status change
             if last status was for pull request and it's closed. We shouldn't
             mess around this manually
         """
         repo = Repository.guess_instance(repo)
         q = ChangesetStatus.query()
         if revision is not None:
             assert pull_request is None
             q = q.filter(ChangesetStatus.repo == repo)
             q = q.filter(ChangesetStatus.revision == revision)
             revisions = [revision]
         else:
             assert pull_request is not None
             pull_request = PullRequest.guess_instance(pull_request)
             repo = pull_request.org_repo
             q = q.filter(ChangesetStatus.repo == repo)
             q = q.filter(ChangesetStatus.revision.in_(pull_request.revisions))
             revisions = pull_request.revisions
         cur_statuses = q.all()
         #if statuses exists and last is associated with a closed pull request
         # we need to check if we can allow this status change
         if (dont_allow_on_closed_pull_request and cur_statuses
             and getattr(cur_statuses[0].pull_request, 'status', '')
                 == PullRequest.STATUS_CLOSED):
             raise StatusChangeOnClosedPullRequestError(
                 'Changing status on closed pull request is not allowed'
+            )
         #update all current statuses with older version
         for st in cur_statuses:
             st.version += 1
             self.sa.add(st)
         new_statuses = []
         for rev in revisions:
             new_status = ChangesetStatus()
             new_status.version = 0 # default
             new_status.author = User.guess_instance(user)
             new_status.repo = Repository.guess_instance(repo)
             new_status.status = status
             new_status.comment = comment
             new_status.revision = rev
             new_status.pull_request = pull_request
             new_statuses.append(new_status)
             self.sa.add(new_status)
         return new_statuses

kallithea/model/gist.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.gist
 ~~~~~~~~~~~~~~~~~~~~
 gist model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: May 9, 2013
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import random
 import time
 import logging
 import traceback
 import shutil
 from kallithea.lib.utils2 import safe_unicode, safe_int, \
     time_to_datetime, AttributeDict
 from kallithea.lib.compat import json
 from kallithea.model.base import BaseModel
 from kallithea.model.db import Gist, User
 from kallithea.model.repo import RepoModel
 from kallithea.model.scm import ScmModel
 log = logging.getLogger(__name__)
 GIST_STORE_LOC = '.rc_gist_store'
 GIST_METADATA_FILE = '.rc_gist_metadata'
 def make_gist_id():
     """Generate a random, URL safe, almost certainly unique gist identifier."""
     rnd = random.SystemRandom() # use cryptographically secure system PRNG
     alphabet = '23456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghjklmnpqrstuvwxyz'
     length = 20
     return u''.join(rnd.choice(alphabet) for _ in xrange(length))
 class GistModel(BaseModel):
     def __delete_gist(self, gist):
         """
         removes gist from filesystem
         :param gist: gist object
         """
         root_path = RepoModel().repos_path
         rm_path = os.path.join(root_path, GIST_STORE_LOC, gist.gist_access_id)
         log.info("Removing %s", rm_path)
         shutil.rmtree(rm_path)
     def _store_metadata(self, repo, gist_id, gist_access_id, user_id, gist_type,
                         gist_expires):
         """
         store metadata inside the gist, this can be later used for imports
         or gist identification
         """
         metadata = {
             'metadata_version': '1',
             'gist_db_id': gist_id,
             'gist_access_id': gist_access_id,
             'gist_owner_id': user_id,
             'gist_type': gist_type,
             'gist_expires': gist_expires,
             'gist_updated': time.time(),
+        }
         with open(os.path.join(repo.path, '.hg', GIST_METADATA_FILE), 'wb') as f:
             f.write(json.dumps(metadata))
     def get_gist(self, gist):
         return Gist.guess_instance(gist)
     def get_gist_files(self, gist_access_id, revision=None):
         """
         Get files for given gist
         :param gist_access_id:
         """
         repo = Gist.get_by_access_id(gist_access_id)
         cs = repo.scm_instance.get_changeset(revision)
         return cs, [n for n in cs.get_node('/')]
     def create(self, description, owner, gist_mapping,
                gist_type=Gist.GIST_PUBLIC, lifetime=-1):
         """
         :param description: description of the gist
         :param owner: user who created this gist
         :param gist_mapping: mapping {filename:{'content':content},...}
         :param gist_type: type of gist private/public
         :param lifetime: in minutes, -1 == forever
         """
         owner = User.guess_instance(owner)
         gist_id = make_gist_id()
         lifetime = safe_int(lifetime, -1)
         gist_expires = time.time() + (lifetime * 60) if lifetime != -1 else -1
         log.debug('set GIST expiration date to: %s',
                   time_to_datetime(gist_expires)
                    if gist_expires != -1 else 'forever')
         #create the Database version
         gist = Gist()
         gist.gist_description = description
         gist.gist_access_id = gist_id
         gist.owner_id = owner.user_id
         gist.gist_expires = gist_expires
         gist.gist_type = safe_unicode(gist_type)
         self.sa.add(gist)
         self.sa.flush() # make database assign gist.gist_id
         if gist_type == Gist.GIST_PUBLIC:
             # use DB ID for easy to use GIST ID
             gist_id = safe_unicode(gist.gist_id)
             gist.gist_access_id = gist_id
             self.sa.add(gist)
         gist_repo_path = os.path.join(GIST_STORE_LOC, gist_id)
         log.debug('Creating new %s GIST repo in %s', gist_type, gist_repo_path)
         repo = RepoModel()._create_filesystem_repo(
             repo_name=gist_id, repo_type='hg', repo_group=GIST_STORE_LOC)
         processed_mapping = {}
         for filename in gist_mapping:
             if filename != os.path.basename(filename):
                 raise Exception('Filename cannot be inside a directory')
             content = gist_mapping[filename]['content']
             #TODO: expand support for setting explicit lexers
 #             if lexer is None:
 #                 try:
 #                     guess_lexer = pygments.lexers.guess_lexer_for_filename
 #                     lexer = guess_lexer(filename,content)
 #                 except pygments.util.ClassNotFound:
 #                     lexer = 'text'
             processed_mapping[filename] = {'content': content}
         # now create single multifile commit
         message = 'added file'
         message += 's: ' if len(processed_mapping) > 1 else ': '
         message += ', '.join([x for x in processed_mapping])
         #fake Kallithea Repository object
         fake_repo = AttributeDict(dict(
             repo_name=gist_repo_path,
             scm_instance_no_cache=lambda: repo,
         ))
         ScmModel().create_nodes(
             user=owner.user_id, repo=fake_repo,
             message=message,
             nodes=processed_mapping,
             trigger_push_hook=False
+        )
         self._store_metadata(repo, gist.gist_id, gist.gist_access_id,
                              owner.user_id, gist.gist_type, gist.gist_expires)
         return gist
     def delete(self, gist, fs_remove=True):
         gist = Gist.guess_instance(gist)
         try:
             self.sa.delete(gist)
             if fs_remove:
                 self.__delete_gist(gist)
             else:
                 log.debug('skipping removal from filesystem')
         except Exception:
             log.error(traceback.format_exc())
             raise
     def update(self, gist, description, owner, gist_mapping, gist_type,
                lifetime):
         gist = Gist.guess_instance(gist)
         gist_repo = gist.scm_instance
         lifetime = safe_int(lifetime, -1)
         if lifetime == 0:  # preserve old value
             gist_expires = gist.gist_expires
         else:
             gist_expires = time.time() + (lifetime * 60) if lifetime != -1 else -1
         #calculate operation type based on given data
         gist_mapping_op = {}
         for k, v in gist_mapping.items():
             # add, mod, del
             if not v['org_filename'] and v['filename']:
                 op = 'add'
             elif v['org_filename'] and not v['filename']:
                 op = 'del'
             else:
                 op = 'mod'
             v['op'] = op
             gist_mapping_op[k] = v
         gist.gist_description = description
         gist.gist_expires = gist_expires
         gist.owner = owner
         gist.gist_type = gist_type
         message = 'updated file'
         message += 's: ' if len(gist_mapping) > 1 else ': '
         message += ', '.join([x for x in gist_mapping])
         #fake Kallithea Repository object
         fake_repo = AttributeDict(dict(
             repo_name=gist_repo.path,
             scm_instance_no_cache=lambda: gist_repo,
         ))
         self._store_metadata(gist_repo, gist.gist_id, gist.gist_access_id,
                              owner.user_id, gist.gist_type, gist.gist_expires)
         ScmModel().update_nodes(
             user=owner.user_id,
             repo=fake_repo,
             message=message,
             nodes=gist_mapping_op,
             trigger_push_hook=False
+        )
         return gist

kallithea/model/permission.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.permission
 ~~~~~~~~~~~~~~~~~~~~~~~~~~
 permissions model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Aug 20, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 from sqlalchemy.exc import DatabaseError
 from kallithea.model.base import BaseModel
 from kallithea.model.db import User, Permission, UserToPerm, UserRepoToPerm, \
     UserRepoGroupToPerm, UserUserGroupToPerm
 from kallithea.lib.utils2 import str2bool
 log = logging.getLogger(__name__)
 class PermissionModel(BaseModel):
     """
     Permissions model for Kallithea
     """
     def create_permissions(self):
         """
         Create permissions for whole system
         """
         for p in Permission.PERMS:
             if not Permission.get_by_key(p[0]):
                 new_perm = Permission()
                 new_perm.permission_name = p[0]
                 self.sa.add(new_perm)
     def create_default_permissions(self, user, force=False):
         """
         Create missing default permissions for user. If force is set, the default
         permissions for the user are reset, otherwise only missing permissions are
         created.
         :param user:
         """
         user = User.guess_instance(user)
         def _make_perm(perm):
             new_perm = UserToPerm()
             new_perm.user = user
             new_perm.permission = Permission.get_by_key(perm)
             return new_perm
         def _get_group(perm_name):
             return '.'.join(perm_name.split('.')[:1])
         perms = UserToPerm.query().filter(UserToPerm.user == user).all()
         defined_perms_groups = map(_get_group,
                                 (x.permission.permission_name for x in perms))
         log.debug('GOT ALREADY DEFINED:%s', perms)
         DEFAULT_PERMS = Permission.DEFAULT_USER_PERMISSIONS
         if force:
             for perm in perms:
                 self.sa.delete(perm)
             self.sa.commit()
             defined_perms_groups = []
         # For every default permission that needs to be created, we check if
         # its group is already defined. If it's not, we create default permission.
         for perm_name in DEFAULT_PERMS:
             gr = _get_group(perm_name)
             if gr not in defined_perms_groups:
                 log.debug('GR:%s not found, creating permission %s',
                           gr, perm_name)
                 new_perm = _make_perm(perm_name)
                 self.sa.add(new_perm)
     def update(self, form_result):
         perm_user = User.get_by_username(username=form_result['perm_user_name'])
         try:
             # stage 1 set anonymous access
             if perm_user.is_default_user:
                 perm_user.active = str2bool(form_result['anonymous'])
                 self.sa.add(perm_user)
             # stage 2 reset defaults and set them from form data
             def _make_new(usr, perm_name):
                 log.debug('Creating new permission:%s', perm_name)
                 new = UserToPerm()
                 new.user = usr
                 new.permission = Permission.get_by_key(perm_name)
                 return new
             # clear current entries, to make this function idempotent
             # it will fix even if we define more permissions or permissions
             # are somehow missing
             u2p = self.sa.query(UserToPerm) \
                 .filter(UserToPerm.user == perm_user) \
                 .all()
             for p in u2p:
                 self.sa.delete(p)
             #create fresh set of permissions
             for def_perm_key in ['default_repo_perm',
                                  'default_group_perm',
                                  'default_user_group_perm',
                                  'default_repo_create',
                                  'create_on_write', # special case for create repos on write access to group
                                  #'default_repo_group_create', #not implemented yet
                                  'default_user_group_create',
                                  'default_fork',
                                  'default_register',
                                  'default_extern_activate']:
                 p = _make_new(perm_user, form_result[def_perm_key])
                 self.sa.add(p)
             #stage 3 update all default permissions for repos if checked
             if form_result['overwrite_default_repo']:
                 _def_name = form_result['default_repo_perm'].split('repository.')[-1]
                 _def = Permission.get_by_key('repository.' + _def_name)
                 # repos
                 for r2p in self.sa.query(UserRepoToPerm) \
                                .filter(UserRepoToPerm.user == perm_user) \
                                .all():
                     #don't reset PRIVATE repositories
                     if not r2p.repository.private:
                         r2p.permission = _def
                         self.sa.add(r2p)
             if form_result['overwrite_default_group']:
                 _def_name = form_result['default_group_perm'].split('group.')[-1]
                 # groups
                 _def = Permission.get_by_key('group.' + _def_name)
                 for g2p in self.sa.query(UserRepoGroupToPerm) \
                                .filter(UserRepoGroupToPerm.user == perm_user) \
                                .all():
                     g2p.permission = _def
                     self.sa.add(g2p)
             if form_result['overwrite_default_user_group']:
                 _def_name = form_result['default_user_group_perm'].split('usergroup.')[-1]
                 # groups
                 _def = Permission.get_by_key('usergroup.' + _def_name)
                 for g2p in self.sa.query(UserUserGroupToPerm) \
                                .filter(UserUserGroupToPerm.user == perm_user) \
                                .all():
                     g2p.permission = _def
                     self.sa.add(g2p)
             self.sa.commit()
         except (DatabaseError,):
             log.error(traceback.format_exc())
             self.sa.rollback()
             raise

kallithea/model/repo.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.repo
 ~~~~~~~~~~~~~~~~~~~~
 Repository model for kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Jun 5, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import shutil
 import logging
 import traceback
 from datetime import datetime
 from sqlalchemy.orm import subqueryload
 from kallithea.lib.utils import make_ui
 from kallithea.lib.vcs.backends import get_backend
 from kallithea.lib.compat import json
 from kallithea.lib.utils2 import LazyProperty, safe_str, safe_unicode, \
     remove_prefix, obfuscate_url_pw, get_current_authuser
 from kallithea.lib.caching_query import FromCache
 from kallithea.lib.hooks import log_delete_repository
 from kallithea.model.base import BaseModel
 from kallithea.model.db import Repository, UserRepoToPerm, UserGroupRepoToPerm, \
     UserRepoGroupToPerm, UserGroupRepoGroupToPerm, User, Permission, \
     Statistics, UserGroup, Ui, RepoGroup, RepositoryField
 from kallithea.lib import helpers as h
 from kallithea.lib.auth import HasRepoPermissionLevel, HasUserGroupPermissionLevel
 from kallithea.lib.exceptions import AttachedForksError
 from kallithea.model.scm import UserGroupList
 log = logging.getLogger(__name__)
 class RepoModel(BaseModel):
     URL_SEPARATOR = Repository.url_sep()
     def _create_default_perms(self, repository, private):
         # create default permission
         default = 'repository.read'
         def_user = User.get_default_user()
         for p in def_user.user_perms:
             if p.permission.permission_name.startswith('repository.'):
                 default = p.permission.permission_name
                 break
         default_perm = 'repository.none' if private else default
         repo_to_perm = UserRepoToPerm()
         repo_to_perm.permission = Permission.get_by_key(default_perm)
         repo_to_perm.repository = repository
         repo_to_perm.user_id = def_user.user_id
         self.sa.add(repo_to_perm)
         return repo_to_perm
     @LazyProperty
     def repos_path(self):
         """
         Gets the repositories root path from database
         """
         q = self.sa.query(Ui).filter(Ui.ui_key == '/').one()
         return q.ui_value
     def get(self, repo_id, cache=False):
         repo = self.sa.query(Repository) \
             .filter(Repository.repo_id == repo_id)
         if cache:
             repo = repo.options(FromCache("sql_cache_short",
                                           "get_repo_%s" % repo_id))
         return repo.scalar()
     def get_repo(self, repository):
         return Repository.guess_instance(repository)
     def get_by_repo_name(self, repo_name, cache=False):
         repo = self.sa.query(Repository) \
             .filter(Repository.repo_name == repo_name)
         if cache:
             repo = repo.options(FromCache("sql_cache_short",
                                           "get_repo_%s" % repo_name))
         return repo.scalar()
     def get_all_user_repos(self, user):
         """
         Gets all repositories that user have at least read access
         :param user:
         """
         from kallithea.lib.auth import AuthUser
         user = User.guess_instance(user)
         repos = AuthUser(dbuser=user).permissions['repositories']
         access_check = lambda r: r[1] in ['repository.read',
                                           'repository.write',
                                           'repository.admin']
         repos = [x[0] for x in filter(access_check, repos.items())]
         return Repository.query().filter(Repository.repo_name.in_(repos))
     def get_users_js(self):
         users = self.sa.query(User) \
             .filter(User.active == True) \
             .order_by(User.name, User.lastname) \
             .all()
         return json.dumps([
+            {
                 'id': u.user_id,
                 'fname': h.escape(u.name),
                 'lname': h.escape(u.lastname),
                 'nname': u.username,
                 'gravatar_lnk': h.gravatar_url(u.email, size=28, default='default'),
                 'gravatar_size': 14,
             } for u in users]
+        )
     def get_user_groups_js(self):
         user_groups = self.sa.query(UserGroup) \
             .filter(UserGroup.users_group_active == True) \
             .order_by(UserGroup.users_group_name) \
             .options(subqueryload(UserGroup.members)) \
             .all()
         user_groups = UserGroupList(user_groups, perm_level='read')
         return json.dumps([
+            {
                 'id': gr.users_group_id,
                 'grname': gr.users_group_name,
                 'grmembers': len(gr.members),
             } for gr in user_groups]
+        )
     @classmethod
     def _render_datatable(cls, tmpl, *args, **kwargs):
         import kallithea
         from pylons import tmpl_context as c, request
         from pylons.i18n.translation import _
         _tmpl_lookup = kallithea.CONFIG['pylons.app_globals'].mako_lookup
         template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
         tmpl = template.get_def(tmpl)
         kwargs.update(dict(_=_, h=h, c=c, request=request))
         return tmpl.render(*args, **kwargs)
     def get_repos_as_dict(self, repos_list=None, admin=False, perm_check=True,
                           super_user_actions=False, short_name=False):
         _render = self._render_datatable
         from pylons import tmpl_context as c
         def repo_lnk(name, rtype, rstate, private, fork_of):
             return _render('repo_name', name, rtype, rstate, private, fork_of,
                            short_name=short_name, admin=False)
         def last_change(last_change):
             return _render("last_change", last_change)
         def rss_lnk(repo_name):
             return _render("rss", repo_name)
         def atom_lnk(repo_name):
             return _render("atom", repo_name)
         def last_rev(repo_name, cs_cache):
             return _render('revision', repo_name, cs_cache.get('revision'),
                            cs_cache.get('raw_id'), cs_cache.get('author'),
                            cs_cache.get('message'))
         def desc(desc):
             return h.urlify_text(desc, truncate=80, stylize=c.visual.stylify_metatags)
         def state(repo_state):
             return _render("repo_state", repo_state)
         def repo_actions(repo_name):
             return _render('repo_actions', repo_name, super_user_actions)
         def owner_actions(owner_id, username):
             return _render('user_name', owner_id, username)
         repos_data = []
         for repo in repos_list:
             if perm_check:
                 # check permission at this level
                 if not HasRepoPermissionLevel('read')(repo.repo_name, 'get_repos_as_dict check'):
                     continue
             cs_cache = repo.changeset_cache
             row = {
                 "raw_name": repo.repo_name,
                 "just_name": repo.just_name,
                 "name": repo_lnk(repo.repo_name, repo.repo_type,
                                  repo.repo_state, repo.private, repo.fork),
                 "last_change_iso": repo.last_db_change.isoformat(),
                 "last_change": last_change(repo.last_db_change),
                 "last_changeset": last_rev(repo.repo_name, cs_cache),
                 "last_rev_raw": cs_cache.get('revision'),
                 "desc": desc(repo.description),
                 "owner": h.person(repo.owner),
                 "state": state(repo.repo_state),
                 "rss": rss_lnk(repo.repo_name),
                 "atom": atom_lnk(repo.repo_name),
+            }
             if admin:
                 row.update({
                     "action": repo_actions(repo.repo_name),
                     "owner": owner_actions(repo.owner_id,
                                            h.person(repo.owner))
                 })
             repos_data.append(row)
         return {
             "totalRecords": len(repos_list),
             "startIndex": 0,
             "sort": "name",
             "dir": "asc",
             "records": repos_data
+        }
     def _get_defaults(self, repo_name):
         """
         Gets information about repository, and returns a dict for
         usage in forms
         :param repo_name:
         """
         repo_info = Repository.get_by_repo_name(repo_name)
         if repo_info is None:
             return None
         defaults = repo_info.get_dict()
         defaults['repo_name'] = repo_info.just_name
         defaults['repo_group'] = repo_info.group_id
         for strip, k in [(0, 'repo_type'), (1, 'repo_enable_downloads'),
                          (1, 'repo_description'), (1, 'repo_enable_locking'),
                          (1, 'repo_landing_rev'), (0, 'clone_uri'),
                          (1, 'repo_private'), (1, 'repo_enable_statistics')]:
             attr = k
             if strip:
                 attr = remove_prefix(k, 'repo_')
             val = defaults[attr]
             if k == 'repo_landing_rev':
                 val = ':'.join(defaults[attr])
             defaults[k] = val
             if k == 'clone_uri':
                 defaults['clone_uri_hidden'] = repo_info.clone_uri_hidden
         # fill owner
         if repo_info.owner:
             defaults.update({'owner': repo_info.owner.username})
         else:
             replacement_user = User.query().filter(User.admin ==
                                                    True).first().username
             defaults.update({'owner': replacement_user})
         # fill repository users
         for p in repo_info.repo_to_perm:
             defaults.update({'u_perm_%s' % p.user.username:
                                  p.permission.permission_name})
         # fill repository groups
         for p in repo_info.users_group_to_perm:
             defaults.update({'g_perm_%s' % p.users_group.users_group_name:
                                  p.permission.permission_name})
         return defaults
     def update(self, repo, **kwargs):
         try:
             cur_repo = Repository.guess_instance(repo)
             org_repo_name = cur_repo.repo_name
             if 'owner' in kwargs:
                 cur_repo.owner = User.get_by_username(kwargs['owner'])
             if 'repo_group' in kwargs:
                 cur_repo.group = RepoGroup.get(kwargs['repo_group'])
                 cur_repo.repo_name = cur_repo.get_new_name(cur_repo.just_name)
             log.debug('Updating repo %s with params:%s', cur_repo, kwargs)
             for k in ['repo_enable_downloads',
                       'repo_description',
                       'repo_enable_locking',
                       'repo_landing_rev',
                       'repo_private',
                       'repo_enable_statistics',
                       ]:
                 if k in kwargs:
                     setattr(cur_repo, remove_prefix(k, 'repo_'), kwargs[k])
             clone_uri = kwargs.get('clone_uri')
             if clone_uri is not None and clone_uri != cur_repo.clone_uri_hidden:
                 cur_repo.clone_uri = clone_uri
             if 'repo_name' in kwargs:
                 cur_repo.repo_name = cur_repo.get_new_name(kwargs['repo_name'])
             #if private flag is set, reset default permission to NONE
             if kwargs.get('repo_private'):
                 EMPTY_PERM = 'repository.none'
                 RepoModel().grant_user_permission(
                     repo=cur_repo, user='default', perm=EMPTY_PERM
+                )
                 #handle extra fields
             for field in filter(lambda k: k.startswith(RepositoryField.PREFIX),
                                 kwargs):
                 k = RepositoryField.un_prefix_key(field)
                 ex_field = RepositoryField.get_by_key_name(key=k, repo=cur_repo)
                 if ex_field:
                     ex_field.field_value = kwargs[field]
                     self.sa.add(ex_field)
             self.sa.add(cur_repo)
             if org_repo_name != cur_repo.repo_name:
                 # rename repository
                 self._rename_filesystem_repo(old=org_repo_name, new=cur_repo.repo_name)
             return cur_repo
         except Exception:
             log.error(traceback.format_exc())
             raise
     def _create_repo(self, repo_name, repo_type, description, owner,
                      private=False, clone_uri=None, repo_group=None,
                      landing_rev='rev:tip', fork_of=None,
                      copy_fork_permissions=False, enable_statistics=False,
                      enable_locking=False, enable_downloads=False,
                      copy_group_permissions=False, state=Repository.STATE_PENDING):
         """
         Create repository inside database with PENDING state. This should only be
         executed by create() repo, with exception of importing existing repos.
         """
         from kallithea.model.scm import ScmModel
         owner = User.guess_instance(owner)
         fork_of = Repository.guess_instance(fork_of)
         repo_group = RepoGroup.guess_instance(repo_group)
         try:
             repo_name = safe_unicode(repo_name)
             description = safe_unicode(description)
             # repo name is just a name of repository
             # while repo_name_full is a full qualified name that is combined
             # with name and path of group
             repo_name_full = repo_name
             repo_name = repo_name.split(self.URL_SEPARATOR)[-1]
             new_repo = Repository()
             new_repo.repo_state = state
             new_repo.enable_statistics = False
             new_repo.repo_name = repo_name_full
             new_repo.repo_type = repo_type
             new_repo.owner = owner
             new_repo.group = repo_group
             new_repo.description = description or repo_name
             new_repo.private = private
             new_repo.clone_uri = clone_uri
             new_repo.landing_rev = landing_rev
             new_repo.enable_statistics = enable_statistics
             new_repo.enable_locking = enable_locking
             new_repo.enable_downloads = enable_downloads
             if repo_group:
                 new_repo.enable_locking = repo_group.enable_locking
             if fork_of:
                 parent_repo = fork_of
                 new_repo.fork = parent_repo
             self.sa.add(new_repo)
             if fork_of and copy_fork_permissions:
                 repo = fork_of
                 user_perms = UserRepoToPerm.query() \
                     .filter(UserRepoToPerm.repository == repo).all()
                 group_perms = UserGroupRepoToPerm.query() \
                     .filter(UserGroupRepoToPerm.repository == repo).all()
                 for perm in user_perms:
                     UserRepoToPerm.create(perm.user, new_repo, perm.permission)
                 for perm in group_perms:
                     UserGroupRepoToPerm.create(perm.users_group, new_repo,
                                                perm.permission)
             elif repo_group and copy_group_permissions:
                 user_perms = UserRepoGroupToPerm.query() \
                     .filter(UserRepoGroupToPerm.group == repo_group).all()
                 group_perms = UserGroupRepoGroupToPerm.query() \
                     .filter(UserGroupRepoGroupToPerm.group == repo_group).all()
                 for perm in user_perms:
                     perm_name = perm.permission.permission_name.replace('group.', 'repository.')
                     perm_obj = Permission.get_by_key(perm_name)
                     UserRepoToPerm.create(perm.user, new_repo, perm_obj)
                 for perm in group_perms:
                     perm_name = perm.permission.permission_name.replace('group.', 'repository.')
                     perm_obj = Permission.get_by_key(perm_name)
                     UserGroupRepoToPerm.create(perm.users_group, new_repo, perm_obj)
             else:
                 self._create_default_perms(new_repo, private)
             # now automatically start following this repository as owner
             ScmModel(self.sa).toggle_following_repo(new_repo.repo_id,
                                                     owner.user_id)
             # we need to flush here, in order to check if database won't
             # throw any exceptions, create filesystem dirs at the very end
             self.sa.flush()
             return new_repo
         except Exception:
             log.error(traceback.format_exc())
             raise
     def create(self, form_data, cur_user):
         """
         Create repository using celery tasks
         :param form_data:
         :param cur_user:
         """
         from kallithea.lib.celerylib import tasks
         return tasks.create_repo(form_data, cur_user)
     def _update_permissions(self, repo, perms_new=None, perms_updates=None,
                             check_perms=True):
         if not perms_new:
             perms_new = []
         if not perms_updates:
             perms_updates = []
         # update permissions
         for member, perm, member_type in perms_updates:
             if member_type == 'user':
                 # this updates existing one
                 self.grant_user_permission(
                     repo=repo, user=member, perm=perm
+                )
             else:
                 #check if we have permissions to alter this usergroup's access
                 if not check_perms or HasUserGroupPermissionLevel('read')(member):
                     self.grant_user_group_permission(
                         repo=repo, group_name=member, perm=perm
+                    )
             # set new permissions
         for member, perm, member_type in perms_new:
             if member_type == 'user':
                 self.grant_user_permission(
                     repo=repo, user=member, perm=perm
+                )
             else:
                 #check if we have permissions to alter this usergroup's access
                 if not check_perms or HasUserGroupPermissionLevel('read')(member):
                     self.grant_user_group_permission(
                         repo=repo, group_name=member, perm=perm
+                    )
     def create_fork(self, form_data, cur_user):
         """
         Simple wrapper into executing celery task for fork creation
         :param form_data:
         :param cur_user:
         """
         from kallithea.lib.celerylib import tasks
         return tasks.create_repo_fork(form_data, cur_user)
     def delete(self, repo, forks=None, fs_remove=True, cur_user=None):
         """
         Delete given repository, forks parameter defines what do do with
         attached forks. Throws AttachedForksError if deleted repo has attached
         forks
         :param repo:
         :param forks: str 'delete' or 'detach'
         :param fs_remove: remove(archive) repo from filesystem
         """
         if not cur_user:
             cur_user = getattr(get_current_authuser(), 'username', None)
         repo = Repository.guess_instance(repo)
         if repo is not None:
             if forks == 'detach':
                 for r in repo.forks:
                     r.fork = None
                     self.sa.add(r)
             elif forks == 'delete':
                 for r in repo.forks:
                     self.delete(r, forks='delete')
             elif [f for f in repo.forks]:
                 raise AttachedForksError()
             old_repo_dict = repo.get_dict()
             try:
                 self.sa.delete(repo)
                 if fs_remove:
                     self._delete_filesystem_repo(repo)
                 else:
                     log.debug('skipping removal from filesystem')
                 log_delete_repository(old_repo_dict,
                                       deleted_by=cur_user)
             except Exception:
                 log.error(traceback.format_exc())
                 raise
     def grant_user_permission(self, repo, user, perm):
         """
         Grant permission for user on given repository, or update existing one
         if found
         :param repo: Instance of Repository, repository_id, or repository name
         :param user: Instance of User, user_id or username
         :param perm: Instance of Permission, or permission_name
         """
         user = User.guess_instance(user)
         repo = Repository.guess_instance(repo)
         permission = Permission.guess_instance(perm)
         # check if we have that permission already
         obj = self.sa.query(UserRepoToPerm) \
             .filter(UserRepoToPerm.user == user) \
             .filter(UserRepoToPerm.repository == repo) \
             .scalar()
         if obj is None:
             # create new !
             obj = UserRepoToPerm()
             self.sa.add(obj)
         obj.repository = repo
         obj.user = user
         obj.permission = permission
         self.sa.add(obj)
         log.debug('Granted perm %s to %s on %s', perm, user, repo)
         return obj
     def revoke_user_permission(self, repo, user):
         """
         Revoke permission for user on given repository
         :param repo: Instance of Repository, repository_id, or repository name
         :param user: Instance of User, user_id or username
         """
         user = User.guess_instance(user)
         repo = Repository.guess_instance(repo)
         obj = self.sa.query(UserRepoToPerm) \
             .filter(UserRepoToPerm.repository == repo) \
             .filter(UserRepoToPerm.user == user) \
             .scalar()
         if obj is not None:
             self.sa.delete(obj)
             log.debug('Revoked perm on %s on %s', repo, user)
     def grant_user_group_permission(self, repo, group_name, perm):
         """
         Grant permission for user group on given repository, or update
         existing one if found
         :param repo: Instance of Repository, repository_id, or repository name
         :param group_name: Instance of UserGroup, users_group_id,
             or user group name
         :param perm: Instance of Permission, or permission_name
         """
         repo = Repository.guess_instance(repo)
         group_name = UserGroup.guess_instance(group_name)
         permission = Permission.guess_instance(perm)
         # check if we have that permission already
         obj = self.sa.query(UserGroupRepoToPerm) \
             .filter(UserGroupRepoToPerm.users_group == group_name) \
             .filter(UserGroupRepoToPerm.repository == repo) \
             .scalar()
         if obj is None:
             # create new
             obj = UserGroupRepoToPerm()
             self.sa.add(obj)
         obj.repository = repo
         obj.users_group = group_name
         obj.permission = permission
         self.sa.add(obj)
         log.debug('Granted perm %s to %s on %s', perm, group_name, repo)
         return obj
     def revoke_user_group_permission(self, repo, group_name):
         """
         Revoke permission for user group on given repository
         :param repo: Instance of Repository, repository_id, or repository name
         :param group_name: Instance of UserGroup, users_group_id,
             or user group name
         """
         repo = Repository.guess_instance(repo)
         group_name = UserGroup.guess_instance(group_name)
         obj = self.sa.query(UserGroupRepoToPerm) \
             .filter(UserGroupRepoToPerm.repository == repo) \
             .filter(UserGroupRepoToPerm.users_group == group_name) \
             .scalar()
         if obj is not None:
             self.sa.delete(obj)
             log.debug('Revoked perm to %s on %s', repo, group_name)
     def delete_stats(self, repo_name):
         """
         removes stats for given repo
         :param repo_name:
         """
         repo = Repository.guess_instance(repo_name)
         try:
             obj = self.sa.query(Statistics) \
                 .filter(Statistics.repository == repo).scalar()
             if obj is not None:
                 self.sa.delete(obj)
         except Exception:
             log.error(traceback.format_exc())
             raise
     def _create_filesystem_repo(self, repo_name, repo_type, repo_group,
                                 clone_uri=None, repo_store_location=None):
         """
         Makes repository on filesystem. Operation is group aware, meaning that it will create
         a repository within a group, and alter the paths accordingly to the group location.
         :param repo_name:
         :param alias:
         :param parent:
         :param clone_uri:
         :param repo_store_location:
         """
         from kallithea.lib.utils import is_valid_repo, is_valid_repo_group
         from kallithea.model.scm import ScmModel
         if '/' in repo_name:
             raise ValueError('repo_name must not contain groups got `%s`' % repo_name)
         if isinstance(repo_group, RepoGroup):
             new_parent_path = os.sep.join(repo_group.full_path_splitted)
         else:
             new_parent_path = repo_group or ''
         if repo_store_location:
             _paths = [repo_store_location]
         else:
             _paths = [self.repos_path, new_parent_path, repo_name]
             # we need to make it str for mercurial
         repo_path = os.path.join(*map(lambda x: safe_str(x), _paths))
         # check if this path is not a repository
         if is_valid_repo(repo_path, self.repos_path):
             raise Exception('This path %s is a valid repository' % repo_path)
         # check if this path is a group
         if is_valid_repo_group(repo_path, self.repos_path):
             raise Exception('This path %s is a valid group' % repo_path)
         log.info('creating repo %s in %s from url: `%s`',
             repo_name, safe_unicode(repo_path),
             obfuscate_url_pw(clone_uri))
         backend = get_backend(repo_type)
         if repo_type == 'hg':
             baseui = make_ui('db', clear_session=False)
             # patch and reset hooks section of UI config to not run any
             # hooks on creating remote repo
             for k, v in baseui.configitems('hooks'):
                 baseui.setconfig('hooks', k, None)
             repo = backend(repo_path, create=True, src_url=clone_uri, baseui=baseui)
         elif repo_type == 'git':
             repo = backend(repo_path, create=True, src_url=clone_uri, bare=True)
             # add kallithea hook into this repo
             ScmModel().install_git_hooks(repo=repo)
         else:
             raise Exception('Not supported repo_type %s expected hg/git' % repo_type)
         log.debug('Created repo %s with %s backend',
                   safe_unicode(repo_name), safe_unicode(repo_type))
         return repo
     def _rename_filesystem_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 = safe_str(os.path.join(self.repos_path, old))
         new_path = safe_str(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_filesystem_repo(self, repo):
         """
         removes repo from filesystem, the removal is actually done by
         renaming dir to a 'rm__*' prefix which Kallithea will skip.
         It can be undeleted later by reverting the rename.
         :param repo: repo object
         """
         rm_path = safe_str(os.path.join(self.repos_path, repo.repo_name))
         log.info("Removing %s", rm_path)
         _now = datetime.now()
         _ms = str(_now.microsecond).rjust(6, '0')
         _d = 'rm__%s__%s' % (_now.strftime('%Y%m%d_%H%M%S_' + _ms),
                              repo.just_name)
         if repo.group:
             args = repo.group.full_path_splitted + [_d]
             _d = os.path.join(*args)
         if os.path.exists(rm_path):
             shutil.move(rm_path, safe_str(os.path.join(self.repos_path, _d)))
         else:
             log.error("Can't find repo to delete in %r", rm_path)

kallithea/model/repo_group.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.repo_group
 ~~~~~~~~~~~~~~~~~~~~~~~~~~
 repo group model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Jan 25, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import logging
 import traceback
 import shutil
 import datetime
 from kallithea.lib.utils2 import LazyProperty
 from kallithea.model.base import BaseModel
 from kallithea.model.db import RepoGroup, Ui, UserRepoGroupToPerm, \
     User, Permission, UserGroupRepoGroupToPerm, UserGroup, Repository
 log = logging.getLogger(__name__)
 class RepoGroupModel(BaseModel):
     @LazyProperty
     def repos_path(self):
         """
         Gets the repositories root path from database
         """
         q = Ui.get_by_key('paths', '/')
         return q.ui_value
     def _create_default_perms(self, new_group):
         # create default permission
         default_perm = 'group.read'
         def_user = User.get_default_user()
         for p in def_user.user_perms:
             if p.permission.permission_name.startswith('group.'):
                 default_perm = p.permission.permission_name
                 break
         repo_group_to_perm = UserRepoGroupToPerm()
         repo_group_to_perm.permission = Permission.get_by_key(default_perm)
         repo_group_to_perm.group = new_group
         repo_group_to_perm.user_id = def_user.user_id
         self.sa.add(repo_group_to_perm)
         return repo_group_to_perm
     def _create_group(self, group_name):
         """
         makes repository group on filesystem
         :param repo_name:
         :param parent_id:
         """
         create_path = os.path.join(self.repos_path, group_name)
         log.debug('creating new group in %s', create_path)
         if os.path.isdir(create_path):
             raise Exception('That directory already exists !')
         os.makedirs(create_path)
         log.debug('Created group in %s', create_path)
     def _rename_group(self, old, new):
         """
         Renames a group on filesystem
         :param group_name:
         """
         if old == new:
             log.debug('skipping group rename')
             return
         log.debug('renaming repository group from %s to %s', old, new)
         old_path = os.path.join(self.repos_path, old)
         new_path = os.path.join(self.repos_path, new)
         log.debug('renaming repos paths from %s to %s', old_path, new_path)
         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_group(self, group, force_delete=False):
         """
         Deletes a group from a filesystem
         :param group: instance of group from database
         :param force_delete: use shutil rmtree to remove all objects
         """
         paths = group.full_path.split(RepoGroup.url_sep())
         paths = os.sep.join(paths)
         rm_path = os.path.join(self.repos_path, paths)
         log.info("Removing group %s", rm_path)
         # delete only if that path really exists
         if os.path.isdir(rm_path):
             if force_delete:
                 shutil.rmtree(rm_path)
             else:
                 #archive that group`
                 _now = datetime.datetime.now()
                 _ms = str(_now.microsecond).rjust(6, '0')
                 _d = 'rm__%s_GROUP_%s' % (_now.strftime('%Y%m%d_%H%M%S_' + _ms),
                                           group.name)
                 shutil.move(rm_path, os.path.join(self.repos_path, _d))
     def create(self, group_name, group_description, owner, parent=None,
                just_db=False, copy_permissions=False):
         try:
             owner = User.guess_instance(owner)
             parent_group = RepoGroup.guess_instance(parent)
             new_repo_group = RepoGroup()
             new_repo_group.owner = owner
             new_repo_group.group_description = group_description or group_name
             new_repo_group.parent_group = parent_group
             new_repo_group.group_name = new_repo_group.get_new_name(group_name)
             self.sa.add(new_repo_group)
             # create an ADMIN permission for owner except if we're super admin,
             # later owner should go into the owner field of groups
             if not owner.is_admin:
                 self.grant_user_permission(repo_group=new_repo_group,
                                            user=owner, perm='group.admin')
             if parent_group and copy_permissions:
                 # copy permissions from parent
                 user_perms = UserRepoGroupToPerm.query() \
                     .filter(UserRepoGroupToPerm.group == parent_group).all()
                 group_perms = UserGroupRepoGroupToPerm.query() \
                     .filter(UserGroupRepoGroupToPerm.group == parent_group).all()
                 for perm in user_perms:
                     # don't copy over the permission for user who is creating
                     # this group, if he is not super admin he get's admin
                     # permission set above
                     if perm.user != owner or owner.is_admin:
                         UserRepoGroupToPerm.create(perm.user, new_repo_group, perm.permission)
                 for perm in group_perms:
                     UserGroupRepoGroupToPerm.create(perm.users_group, new_repo_group, perm.permission)
             else:
                 self._create_default_perms(new_repo_group)
             if not just_db:
                 # we need to flush here, in order to check if database won't
                 # throw any exceptions, create filesystem dirs at the very end
                 self.sa.flush()
                 self._create_group(new_repo_group.group_name)
             return new_repo_group
         except Exception:
             log.error(traceback.format_exc())
             raise
     def _update_permissions(self, repo_group, perms_new=None,
                             perms_updates=None, recursive=None,
                             check_perms=True):
         from kallithea.model.repo import RepoModel
         from kallithea.lib.auth import HasUserGroupPermissionLevel
         if not perms_new:
             perms_new = []
         if not perms_updates:
             perms_updates = []
         def _set_perm_user(obj, user, perm):
             if isinstance(obj, RepoGroup):
                 self.grant_user_permission(repo_group=obj, user=user, perm=perm)
             elif isinstance(obj, Repository):
                 user = User.guess_instance(user)
                 # private repos will not allow to change the default permissions
                 # using recursive mode
                 if obj.private and user.is_default_user:
                     return
                 # we set group permission but we have to switch to repo
                 # permission
                 perm = perm.replace('group.', 'repository.')
                 RepoModel().grant_user_permission(
                     repo=obj, user=user, perm=perm
+                )
         def _set_perm_group(obj, users_group, perm):
             if isinstance(obj, RepoGroup):
                 self.grant_user_group_permission(repo_group=obj,
                                                   group_name=users_group,
                                                   perm=perm)
             elif isinstance(obj, Repository):
                 # we set group permission but we have to switch to repo
                 # permission
                 perm = perm.replace('group.', 'repository.')
                 RepoModel().grant_user_group_permission(
                     repo=obj, group_name=users_group, perm=perm
+                )
         # start updates
         updates = []
         log.debug('Now updating permissions for %s in recursive mode:%s',
                   repo_group, recursive)
         for obj in repo_group.recursive_groups_and_repos():
             # iterated obj is an instance of a repos group or repository in
             # that group, recursive option can be: none, repos, groups, all
             if recursive == 'all':
                 pass
             elif recursive == 'repos':
                 # skip groups, other than this one
                 if isinstance(obj, RepoGroup) and not obj == repo_group:
                     continue
             elif recursive == 'groups':
                 # skip repos
                 if isinstance(obj, Repository):
                     continue
             else:  # recursive == 'none': # DEFAULT don't apply to iterated objects
                 obj = repo_group
                 # also we do a break at the end of this loop.
             # update permissions
             for member, perm, member_type in perms_updates:
                 ## set for user
                 if member_type == 'user':
                     # this updates also current one if found
                     _set_perm_user(obj, user=member, perm=perm)
                 ## set for user group
                 else:
                     #check if we have permissions to alter this usergroup's access
                     if not check_perms or HasUserGroupPermissionLevel('read')(member):
                         _set_perm_group(obj, users_group=member, perm=perm)
             # set new permissions
             for member, perm, member_type in perms_new:
                 if member_type == 'user':
                     _set_perm_user(obj, user=member, perm=perm)
                 else:
                     #check if we have permissions to alter this usergroup's access
                     if not check_perms or HasUserGroupPermissionLevel('read')(member):
                         _set_perm_group(obj, users_group=member, perm=perm)
             updates.append(obj)
             # if it's not recursive call for all,repos,groups
             # break the loop and don't proceed with other changes
             if recursive not in ['all', 'repos', 'groups']:
                 break
         return updates
     def update(self, repo_group, form_data):
         try:
             repo_group = RepoGroup.guess_instance(repo_group)
             old_path = repo_group.full_path
             # change properties
             repo_group.group_description = form_data['group_description']
             repo_group.parent_group_id = form_data['parent_group_id']
             repo_group.enable_locking = form_data['enable_locking']
             repo_group.parent_group = RepoGroup.get(form_data['parent_group_id'])
             repo_group.group_name = repo_group.get_new_name(form_data['group_name'])
             new_path = repo_group.full_path
             self.sa.add(repo_group)
             # iterate over all members of this groups and do fixes
             # set locking if given
             # if obj is a repoGroup also fix the name of the group according
             # to the parent
             # if obj is a Repo fix it's name
             # this can be potentially heavy operation
             for obj in repo_group.recursive_groups_and_repos():
                 #set the value from it's parent
                 obj.enable_locking = repo_group.enable_locking
                 if isinstance(obj, RepoGroup):
                     new_name = obj.get_new_name(obj.name)
                     log.debug('Fixing group %s to new name %s' \
                                 % (obj.group_name, new_name))
                     obj.group_name = new_name
                 elif isinstance(obj, Repository):
                     # we need to get all repositories from this new group and
                     # rename them accordingly to new group path
                     new_name = obj.get_new_name(obj.just_name)
                     log.debug('Fixing repo %s to new name %s' \
                                 % (obj.repo_name, new_name))
                     obj.repo_name = new_name
                 self.sa.add(obj)
             self._rename_group(old_path, new_path)
             return repo_group
         except Exception:
             log.error(traceback.format_exc())
             raise
     def delete(self, repo_group, force_delete=False):
         repo_group = RepoGroup.guess_instance(repo_group)
         try:
             self.sa.delete(repo_group)
             self._delete_group(repo_group, force_delete)
         except Exception:
             log.error('Error removing repo_group %s', repo_group)
             raise
     def add_permission(self, repo_group, obj, obj_type, perm, recursive):
         from kallithea.model.repo import RepoModel
         repo_group = RepoGroup.guess_instance(repo_group)
         perm = Permission.guess_instance(perm)
         for el in repo_group.recursive_groups_and_repos():
             # iterated obj is an instance of a repos group or repository in
             # that group, recursive option can be: none, repos, groups, all
             if recursive == 'all':
                 pass
             elif recursive == 'repos':
                 # skip groups, other than this one
                 if isinstance(el, RepoGroup) and not el == repo_group:
                     continue
             elif recursive == 'groups':
                 # skip repos
                 if isinstance(el, Repository):
                     continue
             else:  # recursive == 'none': # DEFAULT don't apply to iterated objects
                 el = repo_group
                 # also we do a break at the end of this loop.
             if isinstance(el, RepoGroup):
                 if obj_type == 'user':
                     RepoGroupModel().grant_user_permission(el, user=obj, perm=perm)
                 elif obj_type == 'user_group':
                     RepoGroupModel().grant_user_group_permission(el, group_name=obj, perm=perm)
                 else:
                     raise Exception('undefined object type %s' % obj_type)
             elif isinstance(el, Repository):
                 # for repos we need to hotfix the name of permission
                 _perm = perm.permission_name.replace('group.', 'repository.')
                 if obj_type == 'user':
                     RepoModel().grant_user_permission(el, user=obj, perm=_perm)
                 elif obj_type == 'user_group':
                     RepoModel().grant_user_group_permission(el, group_name=obj, perm=_perm)
                 else:
                     raise Exception('undefined object type %s' % obj_type)
             else:
                 raise Exception('el should be instance of Repository or '
                                 'RepositoryGroup got %s instead' % type(el))
             # if it's not recursive call for all,repos,groups
             # break the loop and don't proceed with other changes
             if recursive not in ['all', 'repos', 'groups']:
                 break
     def delete_permission(self, repo_group, obj, obj_type, recursive):
         """
         Revokes permission for repo_group for given obj(user or users_group),
         obj_type can be user or user group
         :param repo_group:
         :param obj: user or user group id
         :param obj_type: user or user group type
         :param recursive: recurse to all children of group
         """
         from kallithea.model.repo import RepoModel
         repo_group = RepoGroup.guess_instance(repo_group)
         for el in repo_group.recursive_groups_and_repos():
             # iterated obj is an instance of a repos group or repository in
             # that group, recursive option can be: none, repos, groups, all
             if recursive == 'all':
                 pass
             elif recursive == 'repos':
                 # skip groups, other than this one
                 if isinstance(el, RepoGroup) and not el == repo_group:
                     continue
             elif recursive == 'groups':
                 # skip repos
                 if isinstance(el, Repository):
                     continue
             else:  # recursive == 'none': # DEFAULT don't apply to iterated objects
                 el = repo_group
                 # also we do a break at the end of this loop.
             if isinstance(el, RepoGroup):
                 if obj_type == 'user':
                     RepoGroupModel().revoke_user_permission(el, user=obj)
                 elif obj_type == 'user_group':
                     RepoGroupModel().revoke_user_group_permission(el, group_name=obj)
                 else:
                     raise Exception('undefined object type %s' % obj_type)
             elif isinstance(el, Repository):
                 if obj_type == 'user':
                     RepoModel().revoke_user_permission(el, user=obj)
                 elif obj_type == 'user_group':
                     RepoModel().revoke_user_group_permission(el, group_name=obj)
                 else:
                     raise Exception('undefined object type %s' % obj_type)
             else:
                 raise Exception('el should be instance of Repository or '
                                 'RepositoryGroup got %s instead' % type(el))
             # if it's not recursive call for all,repos,groups
             # break the loop and don't proceed with other changes
             if recursive not in ['all', 'repos', 'groups']:
                 break
     def grant_user_permission(self, repo_group, user, perm):
         """
         Grant permission for user on given repository group, or update
         existing one if found
         :param repo_group: Instance of RepoGroup, repositories_group_id,
             or repositories_group name
         :param user: Instance of User, user_id or username
         :param perm: Instance of Permission, or permission_name
         """
         repo_group = RepoGroup.guess_instance(repo_group)
         user = User.guess_instance(user)
         permission = Permission.guess_instance(perm)
         # check if we have that permission already
         obj = self.sa.query(UserRepoGroupToPerm) \
             .filter(UserRepoGroupToPerm.user == user) \
             .filter(UserRepoGroupToPerm.group == repo_group) \
             .scalar()
         if obj is None:
             # create new !
             obj = UserRepoGroupToPerm()
             self.sa.add(obj)
         obj.group = repo_group
         obj.user = user
         obj.permission = permission
         self.sa.add(obj)
         log.debug('Granted perm %s to %s on %s', perm, user, repo_group)
         return obj
     def revoke_user_permission(self, repo_group, user):
         """
         Revoke permission for user on given repository group
         :param repo_group: Instance of RepoGroup, repositories_group_id,
             or repositories_group name
         :param user: Instance of User, user_id or username
         """
         repo_group = RepoGroup.guess_instance(repo_group)
         user = User.guess_instance(user)
         obj = self.sa.query(UserRepoGroupToPerm) \
             .filter(UserRepoGroupToPerm.user == user) \
             .filter(UserRepoGroupToPerm.group == repo_group) \
             .scalar()
         if obj is not None:
             self.sa.delete(obj)
             log.debug('Revoked perm on %s on %s', repo_group, user)
     def grant_user_group_permission(self, repo_group, group_name, perm):
         """
         Grant permission for user group on given repository group, or update
         existing one if found
         :param repo_group: Instance of RepoGroup, repositories_group_id,
             or repositories_group name
         :param group_name: Instance of UserGroup, users_group_id,
             or user group name
         :param perm: Instance of Permission, or permission_name
         """
         repo_group = RepoGroup.guess_instance(repo_group)
         group_name = UserGroup.guess_instance(group_name)
         permission = Permission.guess_instance(perm)
         # check if we have that permission already
         obj = self.sa.query(UserGroupRepoGroupToPerm) \
             .filter(UserGroupRepoGroupToPerm.group == repo_group) \
             .filter(UserGroupRepoGroupToPerm.users_group == group_name) \
             .scalar()
         if obj is None:
             # create new
             obj = UserGroupRepoGroupToPerm()
             self.sa.add(obj)
         obj.group = repo_group
         obj.users_group = group_name
         obj.permission = permission
         self.sa.add(obj)
         log.debug('Granted perm %s to %s on %s', perm, group_name, repo_group)
         return obj
     def revoke_user_group_permission(self, repo_group, group_name):
         """
         Revoke permission for user group on given repository group
         :param repo_group: Instance of RepoGroup, repositories_group_id,
             or repositories_group name
         :param group_name: Instance of UserGroup, users_group_id,
             or user group name
         """
         repo_group = RepoGroup.guess_instance(repo_group)
         group_name = UserGroup.guess_instance(group_name)
         obj = self.sa.query(UserGroupRepoGroupToPerm) \
             .filter(UserGroupRepoGroupToPerm.group == repo_group) \
             .filter(UserGroupRepoGroupToPerm.users_group == group_name) \
             .scalar()
         if obj is not None:
             self.sa.delete(obj)
             log.debug('Revoked perm to %s on %s', repo_group, group_name)

kallithea/model/scm.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.scm
 ~~~~~~~~~~~~~~~~~~~
 Scm model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 9, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import sys
 import posixpath
 import re
 import time
 import traceback
 import logging
 import cStringIO
 import pkg_resources
 from sqlalchemy import func
 from pylons.i18n.translation import _
 import kallithea
 from kallithea.lib.vcs import get_backend
 from kallithea.lib.vcs.exceptions import RepositoryError
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 from kallithea.lib.vcs.nodes import FileNode
 from kallithea.lib.vcs.backends.base import EmptyChangeset
 from kallithea import BACKENDS
 from kallithea.lib import helpers as h
 from kallithea.lib.utils2 import safe_str, safe_unicode, get_server_url, \
     _set_extras
 from kallithea.lib.auth import HasRepoPermissionLevel, HasRepoGroupPermissionLevel, \
     HasUserGroupPermissionLevel, HasPermissionAny, HasPermissionAny
 from kallithea.lib.utils import get_filesystem_repos, make_ui, \
     action_logger
 from kallithea.model.base import BaseModel
 from kallithea.model.db import Repository, Ui, CacheInvalidation, \
     UserFollowing, UserLog, User, RepoGroup, PullRequest
 from kallithea.lib.hooks import log_push_action
 from kallithea.lib.exceptions import NonRelativePathError, IMCCommitError
 log = logging.getLogger(__name__)
 class UserTemp(object):
     def __init__(self, user_id):
         self.user_id = user_id
     def __repr__(self):
         return "<%s('id:%s')>" % (self.__class__.__name__, self.user_id)
 class RepoTemp(object):
     def __init__(self, repo_id):
         self.repo_id = repo_id
     def __repr__(self):
         return "<%s('id:%s')>" % (self.__class__.__name__, self.repo_id)
 class _PermCheckIterator(object):
     def __init__(self, obj_list, obj_attr, perm_set, perm_checker, extra_kwargs=None):
         """
         Creates iterator from given list of objects, additionally
         checking permission for them from perm_set var
         :param obj_list: list of db objects
         :param obj_attr: attribute of object to pass into perm_checker
         :param perm_set: list of permissions to check
         :param perm_checker: callable to check permissions against
         """
         self.obj_list = obj_list
         self.obj_attr = obj_attr
         self.perm_set = perm_set
         self.perm_checker = perm_checker
         self.extra_kwargs = extra_kwargs or {}
     def __len__(self):
         return len(self.obj_list)
     def __repr__(self):
         return '<%s (%s)>' % (self.__class__.__name__, self.__len__())
     def __iter__(self):
         for db_obj in self.obj_list:
             # check permission at this level
             name = getattr(db_obj, self.obj_attr, None)
             if not self.perm_checker(*self.perm_set)(
                     name, self.__class__.__name__, **self.extra_kwargs):
                 continue
             yield db_obj
 class RepoList(_PermCheckIterator):
     def __init__(self, db_repo_list, perm_level, extra_kwargs=None):
         super(RepoList, self).__init__(obj_list=db_repo_list,
                     obj_attr='repo_name', perm_set=[perm_level],
                     perm_checker=HasRepoPermissionLevel,
                     extra_kwargs=extra_kwargs)
 class RepoGroupList(_PermCheckIterator):
     def __init__(self, db_repo_group_list, perm_level, extra_kwargs=None):
         super(RepoGroupList, self).__init__(obj_list=db_repo_group_list,
                     obj_attr='group_name', perm_set=[perm_level],
                     perm_checker=HasRepoGroupPermissionLevel,
                     extra_kwargs=extra_kwargs)
 class UserGroupList(_PermCheckIterator):
     def __init__(self, db_user_group_list, perm_level, extra_kwargs=None):
         super(UserGroupList, self).__init__(obj_list=db_user_group_list,
                     obj_attr='users_group_name', perm_set=[perm_level],
                     perm_checker=HasUserGroupPermissionLevel,
                     extra_kwargs=extra_kwargs)
 class ScmModel(BaseModel):
     """
     Generic Scm Model
     """
     def __get_repo(self, instance):
         cls = Repository
         if isinstance(instance, cls):
             return instance
         elif isinstance(instance, int) or safe_str(instance).isdigit():
             return cls.get(instance)
         elif isinstance(instance, basestring):
             return cls.get_by_repo_name(instance)
         elif instance is not None:
             raise Exception('given object must be int, basestr or Instance'
                             ' of %s got %s' % (type(cls), type(instance)))
     @LazyProperty
     def repos_path(self):
         """
         Gets the repositories root path from database
         """
         q = self.sa.query(Ui).filter(Ui.ui_key == '/').one()
         return q.ui_value
     def repo_scan(self, repos_path=None):
         """
         Listing of repositories in given path. This path should not be a
         repository itself. Return a dictionary of repository objects
         :param repos_path: path to directory containing repositories
         """
         if repos_path is None:
             repos_path = self.repos_path
         log.info('scanning for repositories in %s', repos_path)
         baseui = make_ui('db')
         repos = {}
         for name, path in get_filesystem_repos(repos_path):
             # name need to be decomposed and put back together using the /
             # since this is internal storage separator for kallithea
             name = Repository.normalize_repo_name(name)
             try:
                 if name in repos:
                     raise RepositoryError('Duplicate repository name %s '
                                           'found in %s' % (name, path))
                 else:
                     klass = get_backend(path[0])
                     if path[0] == 'hg' and path[0] in BACKENDS.keys():
                         repos[name] = klass(safe_str(path[1]), baseui=baseui)
                     if path[0] == 'git' and path[0] in BACKENDS.keys():
                         repos[name] = klass(path[1])
             except OSError:
                 continue
         log.debug('found %s paths with repositories', len(repos))
         return repos
     def get_repos(self, repos):
         """Return the repos the user has access to"""
         return RepoList(repos, perm_level='read')
     def get_repo_groups(self, groups=None):
         """Return the repo groups the user has access to
         If no groups are specified, use top level groups.
         """
         if groups is None:
             groups = RepoGroup.query() \
                 .filter(RepoGroup.parent_group_id == None).all()
         return RepoGroupList(groups, perm_level='read')
     def mark_for_invalidation(self, repo_name):
         """
         Mark caches of this repo invalid in the database.
         :param repo_name: the repo for which caches should be marked invalid
         """
         CacheInvalidation.set_invalidate(repo_name)
         repo = Repository.get_by_repo_name(repo_name)
         if repo is not None:
             repo.update_changeset_cache()
     def toggle_following_repo(self, follow_repo_id, user_id):
         f = self.sa.query(UserFollowing) \
             .filter(UserFollowing.follows_repository_id == follow_repo_id) \
             .filter(UserFollowing.user_id == user_id).scalar()
         if f is not None:
             try:
                 self.sa.delete(f)
                 action_logger(UserTemp(user_id),
                               'stopped_following_repo',
                               RepoTemp(follow_repo_id))
                 return
             except Exception:
                 log.error(traceback.format_exc())
                 raise
         try:
             f = UserFollowing()
             f.user_id = user_id
             f.follows_repository_id = follow_repo_id
             self.sa.add(f)
             action_logger(UserTemp(user_id),
                           'started_following_repo',
                           RepoTemp(follow_repo_id))
         except Exception:
             log.error(traceback.format_exc())
             raise
     def toggle_following_user(self, follow_user_id, user_id):
         f = self.sa.query(UserFollowing) \
             .filter(UserFollowing.follows_user_id == follow_user_id) \
             .filter(UserFollowing.user_id == user_id).scalar()
         if f is not None:
             try:
                 self.sa.delete(f)
                 return
             except Exception:
                 log.error(traceback.format_exc())
                 raise
         try:
             f = UserFollowing()
             f.user_id = user_id
             f.follows_user_id = follow_user_id
             self.sa.add(f)
         except Exception:
             log.error(traceback.format_exc())
             raise
     def is_following_repo(self, repo_name, user_id, cache=False):
         r = self.sa.query(Repository) \
             .filter(Repository.repo_name == repo_name).scalar()
         f = self.sa.query(UserFollowing) \
             .filter(UserFollowing.follows_repository == r) \
             .filter(UserFollowing.user_id == user_id).scalar()
         return f is not None
     def is_following_user(self, username, user_id, cache=False):
         u = User.get_by_username(username)
         f = self.sa.query(UserFollowing) \
             .filter(UserFollowing.follows_user == u) \
             .filter(UserFollowing.user_id == user_id).scalar()
         return f is not None
     def get_followers(self, repo):
         repo = Repository.guess_instance(repo)
         return self.sa.query(UserFollowing) \
                 .filter(UserFollowing.follows_repository == repo).count()
     def get_forks(self, repo):
         repo = Repository.guess_instance(repo)
         return self.sa.query(Repository) \
                 .filter(Repository.fork == repo).count()
     def get_pull_requests(self, repo):
         repo = Repository.guess_instance(repo)
         return self.sa.query(PullRequest) \
                 .filter(PullRequest.other_repo == repo) \
                 .filter(PullRequest.status != PullRequest.STATUS_CLOSED).count()
     def mark_as_fork(self, repo, fork, user):
         repo = self.__get_repo(repo)
         fork = self.__get_repo(fork)
         if fork and repo.repo_id == fork.repo_id:
             raise Exception("Cannot set repository as fork of itself")
         if fork and repo.repo_type != fork.repo_type:
             raise RepositoryError("Cannot set repository as fork of repository with other type")
         repo.fork = fork
         self.sa.add(repo)
         return repo
     def _handle_rc_scm_extras(self, username, repo_name, repo_alias,
                               action=None):
         from kallithea import CONFIG
         from kallithea.lib.base import _get_ip_addr
         try:
             from pylons import request
             environ = request.environ
         except TypeError:
             # we might use this outside of request context, let's fake the
             # environ data
             from webob import Request
             environ = Request.blank('').environ
         extras = {
             'ip': _get_ip_addr(environ),
             'username': username,
             'action': action or 'push_local',
             'repository': repo_name,
             'scm': repo_alias,
             'config': CONFIG['__file__'],
             'server_url': get_server_url(environ),
             'make_lock': None,
             'locked_by': [None, None]
+        }
         _set_extras(extras)
     def _handle_push(self, repo, username, action, repo_name, revisions):
         """
         Triggers push action hooks
         :param repo: SCM repo
         :param username: username who pushes
         :param action: push/push_local/push_remote
         :param repo_name: name of repo
         :param revisions: list of revisions that we pushed
         """
         self._handle_rc_scm_extras(username, repo_name, repo_alias=repo.alias)
         _scm_repo = repo._repo
         # trigger push hook
         if repo.alias == 'hg':
             log_push_action(_scm_repo.ui, _scm_repo, node=revisions[0])
         elif repo.alias == 'git':
             log_push_action(None, _scm_repo, _git_revs=revisions)
     def _get_IMC_module(self, scm_type):
         """
         Returns InMemoryCommit class based on scm_type
         :param scm_type:
         """
         if scm_type == 'hg':
             from kallithea.lib.vcs.backends.hg import MercurialInMemoryChangeset
             return MercurialInMemoryChangeset
         if scm_type == 'git':
             from kallithea.lib.vcs.backends.git import GitInMemoryChangeset
             return GitInMemoryChangeset
         raise Exception('Invalid scm_type, must be one of hg,git got %s'
                         % (scm_type,))
     def pull_changes(self, repo, username):
         """
         Pull from "clone URL".
         """
         dbrepo = self.__get_repo(repo)
         clone_uri = dbrepo.clone_uri
         if not clone_uri:
             raise Exception("This repository doesn't have a clone uri")
         repo = dbrepo.scm_instance
         repo_name = dbrepo.repo_name
         try:
             if repo.alias == 'git':
                 repo.fetch(clone_uri)
                 # git doesn't really have something like post-fetch action
                 # we fake that now. #TODO: extract fetched revisions somehow
                 # here
                 self._handle_push(repo,
                                   username=username,
                                   action='push_remote',
                                   repo_name=repo_name,
                                   revisions=[])
             else:
                 self._handle_rc_scm_extras(username, dbrepo.repo_name,
                                            repo.alias, action='push_remote')
                 repo.pull(clone_uri)
             self.mark_for_invalidation(repo_name)
         except Exception:
             log.error(traceback.format_exc())
             raise
     def commit_change(self, repo, repo_name, cs, user, author, message,
                       content, f_path):
         """
         Commit a change to a single file
         :param repo: a db_repo.scm_instance
         """
         user = User.guess_instance(user)
         IMC = self._get_IMC_module(repo.alias)
         # decoding here will force that we have proper encoded values
         # in any other case this will throw exceptions and deny commit
         content = safe_str(content)
         path = safe_str(f_path)
         # message and author needs to be unicode
         # proper backend should then translate that into required type
         message = safe_unicode(message)
         author = safe_unicode(author)
         imc = IMC(repo)
         imc.change(FileNode(path, content, mode=cs.get_file_mode(f_path)))
         try:
             tip = imc.commit(message=message, author=author,
                              parents=[cs], branch=cs.branch)
         except Exception as e:
             log.error(traceback.format_exc())
             raise IMCCommitError(str(e))
         finally:
             # always clear caches, if commit fails we want fresh object also
             self.mark_for_invalidation(repo_name)
         self._handle_push(repo,
                           username=user.username,
                           action='push_local',
                           repo_name=repo_name,
                           revisions=[tip.raw_id])
         return tip
     def _sanitize_path(self, f_path):
         if f_path.startswith('/') or f_path.startswith('.') or '../' in f_path:
             raise NonRelativePathError('%s is not an relative path' % f_path)
         if f_path:
             f_path = posixpath.normpath(f_path)
         return f_path
     def get_nodes(self, repo_name, revision, root_path='/', flat=True):
         """
         Recursively walk root dir and return a set of all paths found.
         :param repo_name: name of repository
         :param revision: revision for which to list nodes
         :param root_path: root path to list
         :param flat: return as a list, if False returns a dict with description
         """
         _files = list()
         _dirs = list()
         try:
             _repo = self.__get_repo(repo_name)
             changeset = _repo.scm_instance.get_changeset(revision)
             root_path = root_path.lstrip('/')
             for topnode, dirs, files in changeset.walk(root_path):
                 for f in files:
                     _files.append(f.path if flat else {"name": f.path,
                                                        "type": "file"})
                 for d in dirs:
                     _dirs.append(d.path if flat else {"name": d.path,
                                                       "type": "dir"})
         except RepositoryError:
             log.debug(traceback.format_exc())
             raise
         return _dirs, _files
     def create_nodes(self, user, repo, message, nodes, parent_cs=None,
                      author=None, trigger_push_hook=True):
         """
         Commits specified nodes to repo.
         :param user: Kallithea User object or user_id, the committer
         :param repo: Kallithea Repository object
         :param message: commit message
         :param nodes: mapping {filename:{'content':content},...}
         :param parent_cs: parent changeset, can be empty than it's initial commit
         :param author: author of commit, cna be different that committer only for git
         :param trigger_push_hook: trigger push hooks
         :returns: new committed changeset
         """
         user = User.guess_instance(user)
         scm_instance = repo.scm_instance_no_cache()
         processed_nodes = []
         for f_path in nodes:
             content = nodes[f_path]['content']
             f_path = self._sanitize_path(f_path)
             f_path = safe_str(f_path)
             # decoding here will force that we have proper encoded values
             # in any other case this will throw exceptions and deny commit
             if isinstance(content, (basestring,)):
                 content = safe_str(content)
             elif isinstance(content, (file, cStringIO.OutputType,)):
                 content = content.read()
             else:
                 raise Exception('Content is of unrecognized type %s' % (
                     type(content)
                 ))
             processed_nodes.append((f_path, content))
         message = safe_unicode(message)
         committer = user.full_contact
         author = safe_unicode(author) if author else committer
         IMC = self._get_IMC_module(scm_instance.alias)
         imc = IMC(scm_instance)
         if not parent_cs:
             parent_cs = EmptyChangeset(alias=scm_instance.alias)
         if isinstance(parent_cs, EmptyChangeset):
             # EmptyChangeset means we we're editing empty repository
             parents = None
         else:
             parents = [parent_cs]
         # add multiple nodes
         for path, content in processed_nodes:
             imc.add(FileNode(path, content=content))
         tip = imc.commit(message=message,
                          author=author,
                          parents=parents,
                          branch=parent_cs.branch)
         self.mark_for_invalidation(repo.repo_name)
         if trigger_push_hook:
             self._handle_push(scm_instance,
                               username=user.username,
                               action='push_local',
                               repo_name=repo.repo_name,
                               revisions=[tip.raw_id])
         return tip
     def update_nodes(self, user, repo, message, nodes, parent_cs=None,
                      author=None, trigger_push_hook=True):
         """
         Commits specified nodes to repo. Again.
         """
         user = User.guess_instance(user)
         scm_instance = repo.scm_instance_no_cache()
         message = safe_unicode(message)
         committer = user.full_contact
         author = safe_unicode(author) if author else committer
         imc_class = self._get_IMC_module(scm_instance.alias)
         imc = imc_class(scm_instance)
         if not parent_cs:
             parent_cs = EmptyChangeset(alias=scm_instance.alias)
         if isinstance(parent_cs, EmptyChangeset):
             # EmptyChangeset means we we're editing empty repository
             parents = None
         else:
             parents = [parent_cs]
         # add multiple nodes
         for _filename, data in nodes.items():
             # new filename, can be renamed from the old one
             filename = self._sanitize_path(data['filename'])
             old_filename = self._sanitize_path(_filename)
             content = data['content']
             filenode = FileNode(old_filename, content=content)
             op = data['op']
             if op == 'add':
                 imc.add(filenode)
             elif op == 'del':
                 imc.remove(filenode)
             elif op == 'mod':
                 if filename != old_filename:
                     #TODO: handle renames, needs vcs lib changes
                     imc.remove(filenode)
                     imc.add(FileNode(filename, content=content))
                 else:
                     imc.change(filenode)
         # commit changes
         tip = imc.commit(message=message,
                          author=author,
                          parents=parents,
                          branch=parent_cs.branch)
         self.mark_for_invalidation(repo.repo_name)
         if trigger_push_hook:
             self._handle_push(scm_instance,
                               username=user.username,
                               action='push_local',
                               repo_name=repo.repo_name,
                               revisions=[tip.raw_id])
     def delete_nodes(self, user, repo, message, nodes, parent_cs=None,
                      author=None, trigger_push_hook=True):
         """
         Deletes specified nodes from repo.
         :param user: Kallithea User object or user_id, the committer
         :param repo: Kallithea Repository object
         :param message: commit message
         :param nodes: mapping {filename:{'content':content},...}
         :param parent_cs: parent changeset, can be empty than it's initial commit
         :param author: author of commit, cna be different that committer only for git
         :param trigger_push_hook: trigger push hooks
         :returns: new committed changeset after deletion
         """
         user = User.guess_instance(user)
         scm_instance = repo.scm_instance_no_cache()
         processed_nodes = []
         for f_path in nodes:
             f_path = self._sanitize_path(f_path)
             # content can be empty but for compatibility it allows same dicts
             # structure as add_nodes
             content = nodes[f_path].get('content')
             processed_nodes.append((f_path, content))
         message = safe_unicode(message)
         committer = user.full_contact
         author = safe_unicode(author) if author else committer
         IMC = self._get_IMC_module(scm_instance.alias)
         imc = IMC(scm_instance)
         if not parent_cs:
             parent_cs = EmptyChangeset(alias=scm_instance.alias)
         if isinstance(parent_cs, EmptyChangeset):
             # EmptyChangeset means we we're editing empty repository
             parents = None
         else:
             parents = [parent_cs]
         # add multiple nodes
         for path, content in processed_nodes:
             imc.remove(FileNode(path, content=content))
         tip = imc.commit(message=message,
                          author=author,
                          parents=parents,
                          branch=parent_cs.branch)
         self.mark_for_invalidation(repo.repo_name)
         if trigger_push_hook:
             self._handle_push(scm_instance,
                               username=user.username,
                               action='push_local',
                               repo_name=repo.repo_name,
                               revisions=[tip.raw_id])
         return tip
     def get_unread_journal(self):
         return self.sa.query(UserLog).count()
     def get_repo_landing_revs(self, repo=None):
         """
         Generates select option with tags branches and bookmarks (for hg only)
         grouped by type
         :param repo:
         """
         hist_l = []
         choices = []
         repo = self.__get_repo(repo)
         hist_l.append(['rev:tip', _('latest tip')])
         choices.append('rev:tip')
         if repo is None:
             return choices, hist_l
         repo = repo.scm_instance
         branches_group = ([(u'branch:%s' % k, k) for k, v in
                            repo.branches.iteritems()], _("Branches"))
         hist_l.append(branches_group)
         choices.extend([x[0] for x in branches_group[0]])
         if repo.alias == 'hg':
             bookmarks_group = ([(u'book:%s' % k, k) for k, v in
                                 repo.bookmarks.iteritems()], _("Bookmarks"))
             hist_l.append(bookmarks_group)
             choices.extend([x[0] for x in bookmarks_group[0]])
         tags_group = ([(u'tag:%s' % k, k) for k, v in
                        repo.tags.iteritems()], _("Tags"))
         hist_l.append(tags_group)
         choices.extend([x[0] for x in tags_group[0]])
         return choices, hist_l
     def install_git_hooks(self, repo, force_create=False):
         """
         Creates a kallithea hook inside a git repository
         :param repo: Instance of VCS repo
         :param force_create: Create even if same name hook exists
         """
         loc = os.path.join(repo.path, 'hooks')
         if not repo.bare:
             loc = os.path.join(repo.path, '.git', 'hooks')
         if not os.path.isdir(loc):
             os.makedirs(loc)
         tmpl_post = "#!/usr/bin/env %s\n" % sys.executable or 'python2'
         tmpl_post += pkg_resources.resource_string(
             'kallithea', os.path.join('config', 'post_receive_tmpl.py')
+        )
         tmpl_pre = "#!/usr/bin/env %s\n" % sys.executable or 'python2'
         tmpl_pre += pkg_resources.resource_string(
             'kallithea', os.path.join('config', 'pre_receive_tmpl.py')
+        )
         for h_type, tmpl in [('pre', tmpl_pre), ('post', tmpl_post)]:
             _hook_file = os.path.join(loc, '%s-receive' % h_type)
             has_hook = False
             log.debug('Installing git hook in repo %s', repo)
             if os.path.exists(_hook_file):
                 # let's take a look at this hook, maybe it's kallithea ?
                 log.debug('hook exists, checking if it is from kallithea')
                 with open(_hook_file, 'rb') as f:
                     data = f.read()
                     matches = re.compile(r'(?:%s)\s*=\s*(.*)'
                                          % 'KALLITHEA_HOOK_VER').search(data)
                     if matches:
                         try:
                             ver = matches.groups()[0]
                             log.debug('got %s it is kallithea', ver)
                             has_hook = True
                         except Exception:
                             log.error(traceback.format_exc())
             else:
                 # there is no hook in this dir, so we want to create one
                 has_hook = True
             if has_hook or force_create:
                 log.debug('writing %s hook file !', h_type)
                 try:
                     with open(_hook_file, 'wb') as f:
                         tmpl = tmpl.replace('_TMPL_', kallithea.__version__)
                         f.write(tmpl)
                     os.chmod(_hook_file, 0755)
                 except IOError as e:
                     log.error('error writing %s: %s', _hook_file, e)
             else:
                 log.debug('skipping writing hook file')
 def AvailableRepoGroupChoices(top_perms, repo_group_perm_level, extras=()):
     """Return group_id,string tuples with choices for all the repo groups where
     the user has the necessary permissions.
     Top level is -1.
     """
     groups = RepoGroup.query().all()
     if HasPermissionAny('hg.admin')('available repo groups'):
         groups.append(None)
     else:
         groups = list(RepoGroupList(groups, perm_level=repo_group_perm_level))
         if top_perms and HasPermissionAny(*top_perms)('available repo groups'):
             groups.append(None)
         for extra in extras:
             if not any(rg == extra for rg in groups):
                 groups.append(extra)
     return RepoGroup.groups_choices(groups=groups)

kallithea/model/user.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.user
 ~~~~~~~~~~~~~~~~~~~~
 users model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 9, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import hashlib
 import hmac
 import logging
 import time
 import traceback
 from pylons import config
 from pylons.i18n.translation import _
 from sqlalchemy.exc import DatabaseError
 from kallithea.lib.utils2 import safe_str, generate_api_key, get_current_authuser
 from kallithea.lib.caching_query import FromCache
 from kallithea.model.base import BaseModel
 from kallithea.model.db import Permission, User, UserToPerm, Notification, \
     UserEmailMap, UserIpMap
 from kallithea.lib.exceptions import DefaultUserException, \
     UserOwnsReposException
 from kallithea.model.meta import Session
 log = logging.getLogger(__name__)
 class UserModel(BaseModel):
     password_reset_token_lifetime = 86400 # 24 hours
     def get(self, user_id, cache=False):
         user = self.sa.query(User)
         if cache:
             user = user.options(FromCache("sql_cache_short",
                                           "get_user_%s" % user_id))
         return user.get(user_id)
     def get_user(self, user):
         return User.guess_instance(user)
     def create(self, form_data, cur_user=None):
         if not cur_user:
             cur_user = getattr(get_current_authuser(), 'username', None)
         from kallithea.lib.hooks import log_create_user, \
             check_allowed_create_user
         _fd = form_data
         user_data = {
             'username': _fd['username'],
             'password': _fd['password'],
             'email': _fd['email'],
             'firstname': _fd['firstname'],
             'lastname': _fd['lastname'],
             'active': _fd['active'],
             'admin': False
+        }
         # raises UserCreationError if it's not allowed
         check_allowed_create_user(user_data, cur_user)
         from kallithea.lib.auth import get_crypt_password
         new_user = User()
         for k, v in form_data.items():
             if k == 'password':
                 v = get_crypt_password(v)
             if k == 'firstname':
                 k = 'name'
             setattr(new_user, k, v)
         new_user.api_key = generate_api_key()
         Session().add(new_user)
         Session().flush() # make database assign new_user.user_id
         log_create_user(new_user.get_dict(), cur_user)
         return new_user
     def create_or_update(self, username, password, email, firstname=u'',
                          lastname=u'', active=True, admin=False,
                          extern_type=None, extern_name=None, cur_user=None):
         """
         Creates a new instance if not found, or updates current one
         :param username:
         :param password:
         :param email:
         :param active:
         :param firstname:
         :param lastname:
         :param active:
         :param admin:
         :param extern_name:
         :param extern_type:
         :param cur_user:
         """
         if not cur_user:
             cur_user = getattr(get_current_authuser(), 'username', None)
         from kallithea.lib.auth import get_crypt_password, check_password
         from kallithea.lib.hooks import log_create_user, \
             check_allowed_create_user
         user_data = {
             'username': username, 'password': password,
             'email': email, 'firstname': firstname, 'lastname': lastname,
             'active': active, 'admin': admin
+        }
         # raises UserCreationError if it's not allowed
         check_allowed_create_user(user_data, cur_user)
         log.debug('Checking for %s account in Kallithea database', username)
         user = User.get_by_username(username, case_insensitive=True)
         if user is None:
             log.debug('creating new user %s', username)
             new_user = User()
             edit = False
         else:
             log.debug('updating user %s', username)
             new_user = user
             edit = True
         try:
             new_user.username = username
             new_user.admin = admin
             new_user.email = email
             new_user.active = active
             new_user.extern_name = safe_str(extern_name) \
                 if extern_name else None
             new_user.extern_type = safe_str(extern_type) \
                 if extern_type else None
             new_user.name = firstname
             new_user.lastname = lastname
             if not edit:
                 new_user.api_key = generate_api_key()
             # set password only if creating an user or password is changed
             password_change = new_user.password and \
                 not check_password(password, new_user.password)
             if not edit or password_change:
                 reason = 'new password' if edit else 'new user'
                 log.debug('Updating password reason=>%s', reason)
                 new_user.password = get_crypt_password(password) \
                     if password else ''
             if user is None:
                 Session().add(new_user)
                 Session().flush() # make database assign new_user.user_id
             if not edit:
                 log_create_user(new_user.get_dict(), cur_user)
             return new_user
         except (DatabaseError,):
             log.error(traceback.format_exc())
             raise
     def create_registration(self, form_data):
         from kallithea.model.notification import NotificationModel
         import kallithea.lib.helpers as h
         form_data['admin'] = False
         form_data['extern_type'] = User.DEFAULT_AUTH_TYPE
         form_data['extern_name'] = ''
         new_user = self.create(form_data)
         # notification to admins
         subject = _('New user registration')
         body = (
             u'New user registration\n'
             '---------------------\n'
             '- Username: {user.username}\n'
             '- Full Name: {user.full_name}\n'
             '- Email: {user.email}\n'
             ).format(user=new_user)
         edit_url = h.canonical_url('edit_user', id=new_user.user_id)
         email_kwargs = {
             'registered_user_url': edit_url,
             'new_username': new_user.username,
             'new_email': new_user.email,
             'new_full_name': new_user.full_name}
         NotificationModel().create(created_by=new_user, subject=subject,
                                    body=body, recipients=None,
                                    type_=Notification.TYPE_REGISTRATION,
                                    email_kwargs=email_kwargs)
     def update(self, user_id, form_data, skip_attrs=None):
         from kallithea.lib.auth import get_crypt_password
         skip_attrs = skip_attrs or []
         user = self.get(user_id, cache=False)
         if user.is_default_user:
             raise DefaultUserException(
                             _("You can't edit this user since it's "
                               "crucial for entire application"))
         for k, v in form_data.items():
             if k in skip_attrs:
                 continue
             if k == 'new_password' and v:
                 user.password = get_crypt_password(v)
             else:
                 # old legacy thing orm models store firstname as name,
                 # need proper refactor to username
                 if k == 'firstname':
                     k = 'name'
                 setattr(user, k, v)
         self.sa.add(user)
     def update_user(self, user, **kwargs):
         from kallithea.lib.auth import get_crypt_password
         user = User.guess_instance(user)
         if user.is_default_user:
             raise DefaultUserException(
                 _("You can't edit this user since it's"
                   " crucial for entire application")
+            )
         for k, v in kwargs.items():
             if k == 'password' and v:
                 v = get_crypt_password(v)
             setattr(user, k, v)
         self.sa.add(user)
         return user
     def delete(self, user, cur_user=None):
         if cur_user is None:
             cur_user = getattr(get_current_authuser(), 'username', None)
         user = User.guess_instance(user)
         if user.is_default_user:
             raise DefaultUserException(
                 _("You can't remove this user since it is"
                   " crucial for the entire application"))
         if user.repositories:
             repos = [x.repo_name for x in user.repositories]
             raise UserOwnsReposException(
                 _('User "%s" still owns %s repositories and cannot be '
                   'removed. Switch owners or remove those repositories: %s')
                 % (user.username, len(repos), ', '.join(repos)))
         if user.repo_groups:
             repogroups = [x.group_name for x in user.repo_groups]
             raise UserOwnsReposException(_(
                 'User "%s" still owns %s repository groups and cannot be '
                 'removed. Switch owners or remove those repository groups: %s')
                 % (user.username, len(repogroups), ', '.join(repogroups)))
         if user.user_groups:
             usergroups = [x.users_group_name for x in user.user_groups]
             raise UserOwnsReposException(
                 _('User "%s" still owns %s user groups and cannot be '
                   'removed. Switch owners or remove those user groups: %s')
                 % (user.username, len(usergroups), ', '.join(usergroups)))
         self.sa.delete(user)
         from kallithea.lib.hooks import log_delete_user
         log_delete_user(user.get_dict(), cur_user)
     def can_change_password(self, user):
         from kallithea.lib import auth_modules
         managed_fields = auth_modules.get_managed_fields(user)
         return 'password' not in managed_fields
     def get_reset_password_token(self, user, timestamp, session_id):
         """
         The token is a 40-digit hexstring, calculated as a HMAC-SHA1.
         In a traditional HMAC scenario, an attacker is unable to know or
         influence the secret key, but can know or influence the message
         and token. This scenario is slightly different (in particular
         since the message sender is also the message recipient), but
         sufficiently similar to use an HMAC. Benefits compared to a plain
         SHA1 hash includes resistance against a length extension attack.
         The HMAC key consists of the following values (known only to the
         server and authorized users):
         * per-application secret (the `app_instance_uuid` setting), without
           which an attacker cannot counterfeit tokens
         * hashed user password, invalidating the token upon password change
         The HMAC message consists of the following values (potentially known
         to an attacker):
         * session ID (the anti-CSRF token), requiring an attacker to have
           access to the browser session in which the token was created
         * numeric user ID, limiting the token to a specific user (yet allowing
           users to be renamed)
         * user email address
         * time of token issue (a Unix timestamp, to enable token expiration)
         The key and message values are separated by NUL characters, which are
         guaranteed not to occur in any of the values.
         """
         app_secret = config.get('app_instance_uuid')
         return hmac.HMAC(
             key=u'\0'.join([app_secret, user.password]).encode('utf-8'),
             msg=u'\0'.join([session_id, str(user.user_id), user.email, str(timestamp)]).encode('utf-8'),
             digestmod=hashlib.sha1,
         ).hexdigest()
     def send_reset_password_email(self, data):
         """
         Sends email with a password reset token and link to the password
         reset confirmation page with all information (including the token)
         pre-filled. Also returns URL of that page, only without the token,
         allowing users to copy-paste or manually enter the token from the
         email.
         """
         from kallithea.lib.celerylib import tasks
         from kallithea.model.notification import EmailNotificationModel
         import kallithea.lib.helpers as h
         user_email = data['email']
         user = User.get_by_email(user_email)
         timestamp = int(time.time())
         if user is not None:
             if self.can_change_password(user):
                 log.debug('password reset user %s found', user)
                 token = self.get_reset_password_token(user,
                                                       timestamp,
                                                       h.authentication_token())
                 # URL must be fully qualified; but since the token is locked to
                 # the current browser session, we must provide a URL with the
                 # current scheme and hostname, rather than the canonical_url.
                 link = h.url('reset_password_confirmation', qualified=True,
                              email=user_email,
                              timestamp=timestamp,
                              token=token)
             else:
                 log.debug('password reset user %s found but was managed', user)
                 token = link = None
             reg_type = EmailNotificationModel.TYPE_PASSWORD_RESET
             body = EmailNotificationModel().get_email_tmpl(
                 reg_type, 'txt',
                 user=user.short_contact,
                 reset_token=token,
                 reset_url=link)
             html_body = EmailNotificationModel().get_email_tmpl(
                 reg_type, 'html',
                 user=user.short_contact,
                 reset_token=token,
                 reset_url=link)
             log.debug('sending email')
             tasks.send_email([user_email], _("Password reset link"), body, html_body)
             log.info('send new password mail to %s', user_email)
         else:
             log.debug("password reset email %s not found", user_email)
         return h.url('reset_password_confirmation',
                      email=user_email,
                      timestamp=timestamp)
     def verify_reset_password_token(self, email, timestamp, token):
         from kallithea.lib.celerylib import tasks
         from kallithea.lib import auth
         import kallithea.lib.helpers as h
         user = User.get_by_email(email)
         if user is None:
             log.debug("user with email %s not found", email)
             return False
         token_age = int(time.time()) - int(timestamp)
         if token_age < 0:
             log.debug('timestamp is from the future')
             return False
         if token_age > UserModel.password_reset_token_lifetime:
             log.debug('password reset token expired')
             return False
         expected_token = self.get_reset_password_token(user,
                                                        timestamp,
                                                        h.authentication_token())
         log.debug('computed password reset token: %s', expected_token)
         log.debug('received password reset token: %s', token)
         return expected_token == token
     def reset_password(self, user_email, new_passwd):
         from kallithea.lib.celerylib import tasks
         from kallithea.lib import auth
         user = User.get_by_email(user_email)
         if user is not None:
             if not self.can_change_password(user):
                 raise Exception('trying to change password for external user')
             user.password = auth.get_crypt_password(new_passwd)
             Session().commit()
             log.info('change password for %s', user_email)
         if new_passwd is None:
             raise Exception('unable to set new password')
         tasks.send_email([user_email],
                  _('Password reset notification'),
                  _('The password to your account %s has been changed using password reset form.') % (user.username,))
         log.info('send password reset mail to %s', user_email)
         return True
     def has_perm(self, user, perm):
         perm = Permission.guess_instance(perm)
         user = User.guess_instance(user)
         return UserToPerm.query().filter(UserToPerm.user == user) \
             .filter(UserToPerm.permission == perm).scalar() is not None
     def grant_perm(self, user, perm):
         """
         Grant user global permissions
         :param user:
         :param perm:
         """
         user = User.guess_instance(user)
         perm = Permission.guess_instance(perm)
         # if this permission is already granted skip it
         _perm = UserToPerm.query() \
             .filter(UserToPerm.user == user) \
             .filter(UserToPerm.permission == perm) \
             .scalar()
         if _perm:
             return
         new = UserToPerm()
         new.user = user
         new.permission = perm
         self.sa.add(new)
         return new
     def revoke_perm(self, user, perm):
         """
         Revoke users global permissions
         :param user:
         :param perm:
         """
         user = User.guess_instance(user)
         perm = Permission.guess_instance(perm)
         UserToPerm.query().filter(
             UserToPerm.user == user,
             UserToPerm.permission == perm,
         ).delete()
     def add_extra_email(self, user, email):
         """
         Adds email address to UserEmailMap
         :param user:
         :param email:
         """
         from kallithea.model import forms
         form = forms.UserExtraEmailForm()()
         data = form.to_python(dict(email=email))
         user = User.guess_instance(user)
         obj = UserEmailMap()
         obj.user = user
         obj.email = data['email']
         self.sa.add(obj)
         return obj
     def delete_extra_email(self, user, email_id):
         """
         Removes email address from UserEmailMap
         :param user:
         :param email_id:
         """
         user = User.guess_instance(user)
         obj = UserEmailMap.query().get(email_id)
         if obj is not None:
             self.sa.delete(obj)
     def add_extra_ip(self, user, ip):
         """
         Adds IP address to UserIpMap
         :param user:
         :param ip:
         """
         from kallithea.model import forms
         form = forms.UserExtraIpForm()()
         data = form.to_python(dict(ip=ip))
         user = User.guess_instance(user)
         obj = UserIpMap()
         obj.user = user
         obj.ip_addr = data['ip']
         self.sa.add(obj)
         return obj
     def delete_extra_ip(self, user, ip_id):
         """
         Removes IP address from UserIpMap
         :param user:
         :param ip_id:
         """
         user = User.guess_instance(user)
         obj = UserIpMap.query().get(ip_id)
         if obj:
             self.sa.delete(obj)

kallithea/model/user_group.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.user_group
 ~~~~~~~~~~~~~~~~~~~~~~~~~~
 user group model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Oct 1, 2011
 :author: nvinot, marcink
 """
 import logging
 import traceback
 from kallithea.model.base import BaseModel
 from kallithea.model.db import UserGroupMember, UserGroup, \
     UserGroupRepoToPerm, Permission, UserGroupToPerm, User, UserUserGroupToPerm, \
     UserGroupUserGroupToPerm
 from kallithea.lib.exceptions import UserGroupsAssignedException, \
     RepoGroupAssignmentError
 log = logging.getLogger(__name__)
 class UserGroupModel(BaseModel):
     def _create_default_perms(self, user_group):
         # create default permission
         default_perm = 'usergroup.read'
         def_user = User.get_default_user()
         for p in def_user.user_perms:
             if p.permission.permission_name.startswith('usergroup.'):
                 default_perm = p.permission.permission_name
                 break
         user_group_to_perm = UserUserGroupToPerm()
         user_group_to_perm.permission = Permission.get_by_key(default_perm)
         user_group_to_perm.user_group = user_group
         user_group_to_perm.user_id = def_user.user_id
         self.sa.add(user_group_to_perm)
         return user_group_to_perm
     def _update_permissions(self, user_group, perms_new=None,
                             perms_updates=None):
         from kallithea.lib.auth import HasUserGroupPermissionLevel
         if not perms_new:
             perms_new = []
         if not perms_updates:
             perms_updates = []
         # update permissions
         for member, perm, member_type in perms_updates:
             if member_type == 'user':
                 # this updates existing one
                 self.grant_user_permission(
                     user_group=user_group, user=member, perm=perm
+                )
             else:
                 #check if we have permissions to alter this usergroup's access
                 if HasUserGroupPermissionLevel('read')(member):
                     self.grant_user_group_permission(
                         target_user_group=user_group, user_group=member, perm=perm
+                    )
         # set new permissions
         for member, perm, member_type in perms_new:
             if member_type == 'user':
                 self.grant_user_permission(
                     user_group=user_group, user=member, perm=perm
+                )
             else:
                 #check if we have permissions to alter this usergroup's access
                 if HasUserGroupPermissionLevel('read')(member):
                     self.grant_user_group_permission(
                         target_user_group=user_group, user_group=member, perm=perm
+                    )
     def get(self, user_group_id, cache=False):
         return UserGroup.get(user_group_id)
     def get_group(self, user_group):
         return UserGroup.guess_instance(user_group)
     def get_by_name(self, name, cache=False, case_insensitive=False):
         return UserGroup.get_by_group_name(name, cache, case_insensitive)
     def create(self, name, description, owner, active=True, group_data=None):
         try:
             new_user_group = UserGroup()
             new_user_group.owner = User.guess_instance(owner)
             new_user_group.users_group_name = name
             new_user_group.user_group_description = description
             new_user_group.users_group_active = active
             if group_data:
                 new_user_group.group_data = group_data
             self.sa.add(new_user_group)
             self._create_default_perms(new_user_group)
             self.grant_user_permission(user_group=new_user_group,
                                        user=owner, perm='usergroup.admin')
             return new_user_group
         except Exception:
             log.error(traceback.format_exc())
             raise
     def update(self, user_group, form_data):
         try:
             user_group = UserGroup.guess_instance(user_group)
             for k, v in form_data.items():
                 if k == 'users_group_members':
                     members_list = []
                     if v:
                         v = [v] if isinstance(v, basestring) else v
                         for u_id in set(v):
                             member = UserGroupMember(user_group.users_group_id, u_id)
                             members_list.append(member)
                             self.sa.add(member)
                     user_group.members = members_list
                 setattr(user_group, k, v)
             # Flush to make db assign users_group_member_id to newly
             # created UserGroupMembers.
             self.sa.flush()
         except Exception:
             log.error(traceback.format_exc())
             raise
     def delete(self, user_group, force=False):
         """
         Deletes user group, unless force flag is used
         raises exception if there are members in that group, else deletes
         group and users
         :param user_group:
         :param force:
         """
         user_group = UserGroup.guess_instance(user_group)
         try:
             # check if this group is not assigned to repo
             assigned_groups = UserGroupRepoToPerm.query() \
                 .filter(UserGroupRepoToPerm.users_group == user_group).all()
             assigned_groups = [x.repository.repo_name for x in assigned_groups]
             if assigned_groups and not force:
                 raise UserGroupsAssignedException(
                     'User Group assigned to %s' % ", ".join(assigned_groups))
             self.sa.delete(user_group)
         except Exception:
             log.error(traceback.format_exc())
             raise
     def add_user_to_group(self, user_group, user):
         user_group = UserGroup.guess_instance(user_group)
         user = User.guess_instance(user)
         for m in user_group.members:
             u = m.user
             if u.user_id == user.user_id:
                 # user already in the group, skip
                 return True
         try:
             user_group_member = UserGroupMember()
             user_group_member.user = user
             user_group_member.users_group = user_group
             user_group.members.append(user_group_member)
             user.group_member.append(user_group_member)
             self.sa.add(user_group_member)
             return user_group_member
         except Exception:
             log.error(traceback.format_exc())
             raise
     def remove_user_from_group(self, user_group, user):
         user_group = UserGroup.guess_instance(user_group)
         user = User.guess_instance(user)
         user_group_member = None
         for m in user_group.members:
             if m.user_id == user.user_id:
                 # Found this user's membership row
                 user_group_member = m
                 break
         if user_group_member:
             try:
                 self.sa.delete(user_group_member)
                 return True
             except Exception:
                 log.error(traceback.format_exc())
                 raise
         else:
             # User isn't in that group
             return False
     def has_perm(self, user_group, perm):
         user_group = UserGroup.guess_instance(user_group)
         perm = Permission.guess_instance(perm)
         return UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group == user_group) \
             .filter(UserGroupToPerm.permission == perm).scalar() is not None
     def grant_perm(self, user_group, perm):
         user_group = UserGroup.guess_instance(user_group)
         perm = Permission.guess_instance(perm)
         # if this permission is already granted skip it
         _perm = UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group == user_group) \
             .filter(UserGroupToPerm.permission == perm) \
             .scalar()
         if _perm:
             return
         new = UserGroupToPerm()
         new.users_group = user_group
         new.permission = perm
         self.sa.add(new)
         return new
     def revoke_perm(self, user_group, perm):
         user_group = UserGroup.guess_instance(user_group)
         perm = Permission.guess_instance(perm)
         obj = UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group == user_group) \
             .filter(UserGroupToPerm.permission == perm).scalar()
         if obj is not None:
             self.sa.delete(obj)
     def grant_user_permission(self, user_group, user, perm):
         """
         Grant permission for user on given user group, or update
         existing one if found
         :param user_group: Instance of UserGroup, users_group_id,
             or users_group_name
         :param user: Instance of User, user_id or username
         :param perm: Instance of Permission, or permission_name
         """
         user_group = UserGroup.guess_instance(user_group)
         user = User.guess_instance(user)
         permission = Permission.guess_instance(perm)
         # check if we have that permission already
         obj = self.sa.query(UserUserGroupToPerm) \
             .filter(UserUserGroupToPerm.user == user) \
             .filter(UserUserGroupToPerm.user_group == user_group) \
             .scalar()
         if obj is None:
             # create new !
             obj = UserUserGroupToPerm()
             self.sa.add(obj)
         obj.user_group = user_group
         obj.user = user
         obj.permission = permission
         self.sa.add(obj)
         log.debug('Granted perm %s to %s on %s', perm, user, user_group)
         return obj
     def revoke_user_permission(self, user_group, user):
         """
         Revoke permission for user on given repository group
         :param user_group: Instance of RepoGroup, repositories_group_id,
             or repositories_group name
         :param user: Instance of User, user_id or username
         """
         user_group = UserGroup.guess_instance(user_group)
         user = User.guess_instance(user)
         obj = self.sa.query(UserUserGroupToPerm) \
             .filter(UserUserGroupToPerm.user == user) \
             .filter(UserUserGroupToPerm.user_group == user_group) \
             .scalar()
         if obj is not None:
             self.sa.delete(obj)
             log.debug('Revoked perm on %s on %s', user_group, user)
     def grant_user_group_permission(self, target_user_group, user_group, perm):
         """
         Grant user group permission for given target_user_group
         :param target_user_group:
         :param user_group:
         :param perm:
         """
         target_user_group = UserGroup.guess_instance(target_user_group)
         user_group = UserGroup.guess_instance(user_group)
         permission = Permission.guess_instance(perm)
         # forbid assigning same user group to itself
         if target_user_group == user_group:
             raise RepoGroupAssignmentError('target repo:%s cannot be '
                                            'assigned to itself' % target_user_group)
         # check if we have that permission already
         obj = self.sa.query(UserGroupUserGroupToPerm) \
             .filter(UserGroupUserGroupToPerm.target_user_group == target_user_group) \
             .filter(UserGroupUserGroupToPerm.user_group == user_group) \
             .scalar()
         if obj is None:
             # create new !
             obj = UserGroupUserGroupToPerm()
             self.sa.add(obj)
         obj.user_group = user_group
         obj.target_user_group = target_user_group
         obj.permission = permission
         self.sa.add(obj)
         log.debug('Granted perm %s to %s on %s', perm, target_user_group, user_group)
         return obj
     def revoke_user_group_permission(self, target_user_group, user_group):
         """
         Revoke user group permission for given target_user_group
         :param target_user_group:
         :param user_group:
         """
         target_user_group = UserGroup.guess_instance(target_user_group)
         user_group = UserGroup.guess_instance(user_group)
         obj = self.sa.query(UserGroupUserGroupToPerm) \
             .filter(UserGroupUserGroupToPerm.target_user_group == target_user_group) \
             .filter(UserGroupUserGroupToPerm.user_group == user_group) \
             .scalar()
         if obj is not None:
             self.sa.delete(obj)
             log.debug('Revoked perm on %s on %s', target_user_group, user_group)
     def enforce_groups(self, user, groups, extern_type=None):
         user = User.guess_instance(user)
         log.debug('Enforcing groups %s on user %s', user, groups)
         current_groups = user.group_member
         # find the external created groups
         externals = [x.users_group for x in current_groups
                      if 'extern_type' in x.users_group.group_data]
         # calculate from what groups user should be removed
         # externals that are not in groups
         for gr in externals:
             if gr.users_group_name not in groups:
                 log.debug('Removing user %s from user group %s', user, gr)
                 self.remove_user_from_group(gr, user)
         # now we calculate in which groups user should be == groups params
         owner = User.get_first_admin().username
         for gr in set(groups):
             existing_group = UserGroup.get_by_group_name(gr)
             if not existing_group:
                 desc = u'Automatically created from plugin:%s' % extern_type
                 # we use first admin account to set the owner of the group
                 existing_group = UserGroupModel().create(gr, desc, owner,
                                         group_data={'extern_type': extern_type})
             # we can only add users to special groups created via plugins
             managed = 'extern_type' in existing_group.group_data
             if managed:
                 log.debug('Adding user %s to user group %s', user, gr)
                 UserGroupModel().add_user_to_group(existing_group, user)
             else:
                 log.debug('Skipping addition to group %s since it is '
                           'not managed by auth plugins' % gr)

0 comments (0 inline, 0 general)