kallithea Changeset - 62e9be5eb783

Changeset - 62e9be5eb783

Parent rev.

Child rev.

[Not reviewed]

default

0 5 0

Søren Løvborg - 9 years ago 2017-02-27 21:28:12
sorenl@unity3d.com

cleanup: do Session.add directly inside _create_default_perms

Let the various _create_default_perms functions add the created object
to the SQLAlchemy database session, instead of having the caller do it.

This is in accordance with the Kallithea contribution guidelines:

When creating an object using a factory function (like create_repo),
the returned object has already (by convention) been added to the
session, and should not be added again.

5 files changed with 8 insertions and 10 deletions:

kallithea/lib/db_manage.py

kallithea/lib/utils.py

kallithea/model/repo.py

kallithea/model/repo_group.py

kallithea/model/user_group.py

0 comments (0 inline, 0 general)

kallithea/lib/db_manage.py

➞

Show inline comments

@@ @@ -146,352 +146,351 @@ class DbManage(object): @@
         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)
                 perm_obj = RepoGroupModel()._create_default_perms(g)
                 self.sa.add(perm_obj)
                 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/lib/utils.py

➞

Show inline comments

@@ @@ -259,386 +259,385 @@ def is_valid_repo(repo_name, base_path, @@
     :param repo_name:
     :param base_path:
     :param scm:
     :return True: if given path is a valid repository
     """
     full_path = os.path.join(safe_str(base_path), safe_str(repo_name))
     try:
         scm_ = get_scm(full_path)
         if scm:
             return scm_[0] == scm
         return True
     except VCSError:
         return False
 def is_valid_repo_group(repo_group_name, base_path, skip_path_check=False):
     """
     Returns True if given path is a repository group False otherwise
     :param repo_name:
     :param base_path:
     """
     full_path = os.path.join(safe_str(base_path), safe_str(repo_group_name))
     # check if it's not a repo
     if is_valid_repo(repo_group_name, base_path):
         return False
     try:
         # we need to check bare git repos at higher level
         # since we might match branches/hooks/info/objects or possible
         # other things inside bare git repo
         get_scm(os.path.dirname(full_path))
         return False
     except VCSError:
         pass
     # check if it's a valid path
     if skip_path_check or os.path.isdir(full_path):
         return True
     return False
 #propagated from mercurial documentation
 ui_sections = ['alias', 'auth',
                 'decode/encode', 'defaults',
                 'diff', 'email',
                 'extensions', 'format',
                 'merge-patterns', 'merge-tools',
                 'hooks', 'http_proxy',
                 'smtp', 'patch',
                 'paths', 'profiling',
                 'server', 'trusted',
                 'ui', 'web', ]
 def make_ui(read_from='file', path=None, checkpaths=True, clear_session=True):
     """
     A function that will read python rc files or database
     and make an mercurial ui object from read options
     :param path: path to mercurial config file
     :param checkpaths: check the path
     :param read_from: read from 'file' or 'db'
     """
     baseui = ui.ui()
     # clean the baseui object
     baseui._ocfg = config.config()
     baseui._ucfg = config.config()
     baseui._tcfg = config.config()
     if read_from == 'file':
         if not os.path.isfile(path):
             log.debug('hgrc file is not present at %s, skipping...', path)
             return False
         log.debug('reading hgrc from %s', path)
         cfg = config.config()
         cfg.read(path)
         for section in ui_sections:
             for k, v in cfg.items(section):
                 log.debug('settings ui from file: [%s] %s=%s', section, k, v)
                 baseui.setconfig(safe_str(section), safe_str(k), safe_str(v))
     elif read_from == 'db':
         sa = meta.Session()
         ret = sa.query(Ui).all()
         hg_ui = ret
         for ui_ in hg_ui:
             if ui_.ui_active:
                 ui_val = '' if ui_.ui_value is None else safe_str(ui_.ui_value)
                 log.debug('settings ui from db: [%s] %s=%r', ui_.ui_section,
                           ui_.ui_key, ui_val)
                 baseui.setconfig(safe_str(ui_.ui_section), safe_str(ui_.ui_key),
                                  ui_val)
         if clear_session:
             meta.Session.remove()
         # force set push_ssl requirement to False, Kallithea handles that
         baseui.setconfig('web', 'push_ssl', False)
         baseui.setconfig('web', 'allow_push', '*')
         # prevent interactive questions for ssh password / passphrase
         ssh = baseui.config('ui', 'ssh', default='ssh')
         baseui.setconfig('ui', 'ssh', '%s -oBatchMode=yes -oIdentitiesOnly=yes' % ssh)
     return baseui
 def set_app_settings(config):
     """
     Updates app config with new settings from database
     :param config:
     """
     hgsettings = Setting.get_app_settings()
     for k, v in hgsettings.items():
         config[k] = v
 def set_vcs_config(config):
     """
     Patch VCS config with some Kallithea specific stuff
     :param config: kallithea.CONFIG
     """
     from kallithea.lib.vcs import conf
     from kallithea.lib.utils2 import aslist
     conf.settings.BACKENDS = {
         'hg': 'kallithea.lib.vcs.backends.hg.MercurialRepository',
         'git': 'kallithea.lib.vcs.backends.git.GitRepository',
+    }
     conf.settings.GIT_EXECUTABLE_PATH = config.get('git_path', 'git')
     conf.settings.GIT_REV_FILTER = config.get('git_rev_filter', '--all').strip()
     conf.settings.DEFAULT_ENCODINGS = aslist(config.get('default_encoding',
                                                         'utf8'), sep=',')
 def set_indexer_config(config):
     """
     Update Whoosh index mapping
     :param config: kallithea.CONFIG
     """
     from kallithea.config import conf
     log.debug('adding extra into INDEX_EXTENSIONS')
     conf.INDEX_EXTENSIONS.extend(re.split('\s+', config.get('index.extensions', '')))
     log.debug('adding extra into INDEX_FILENAMES')
     conf.INDEX_FILENAMES.extend(re.split('\s+', config.get('index.filenames', '')))
 def map_groups(path):
     """
     Given a full path to a repository, create all nested groups that this
     repo is inside. This function creates parent-child relationships between
     groups and creates default perms for all new groups.
     :param paths: full path to repository
     """
     sa = meta.Session()
     groups = path.split(Repository.url_sep())
     parent = None
     group = None
     # last element is repo in nested groups structure
     groups = groups[:-1]
     rgm = RepoGroupModel(sa)
     owner = User.get_first_admin()
     for lvl, group_name in enumerate(groups):
         group_name = u'/'.join(groups[:lvl] + [group_name])
         group = RepoGroup.get_by_group_name(group_name)
         desc = '%s group' % group_name
         # skip folders that are now removed repos
         if REMOVED_REPO_PAT.match(group_name):
             break
         if group is None:
             log.debug('creating group level: %s group_name: %s',
                       lvl, group_name)
             group = RepoGroup(group_name, parent)
             group.group_description = desc
             group.owner = owner
             sa.add(group)
             perm_obj = rgm._create_default_perms(group)
             sa.add(perm_obj)
             rgm._create_default_perms(group)
             sa.flush()
         parent = group
     return group
 def repo2db_mapper(initial_repo_list, remove_obsolete=False,
                    install_git_hooks=False, user=None, overwrite_git_hooks=False):
     """
     maps all repos given in initial_repo_list, non existing repositories
     are created, if remove_obsolete is True it also check for db entries
     that are not in initial_repo_list and removes them.
     :param initial_repo_list: list of repositories found by scanning methods
     :param remove_obsolete: check for obsolete entries in database
     :param install_git_hooks: if this is True, also check and install git hook
         for a repo if missing
     :param overwrite_git_hooks: if this is True, overwrite any existing git hooks
         that may be encountered (even if user-deployed)
     """
     from kallithea.model.repo import RepoModel
     from kallithea.model.scm import ScmModel
     sa = meta.Session()
     repo_model = RepoModel()
     if user is None:
         user = User.get_first_admin()
     added = []
     ##creation defaults
     defs = Setting.get_default_repo_settings(strip_prefix=True)
     enable_statistics = defs.get('repo_enable_statistics')
     enable_locking = defs.get('repo_enable_locking')
     enable_downloads = defs.get('repo_enable_downloads')
     private = defs.get('repo_private')
     for name, repo in initial_repo_list.items():
         group = map_groups(name)
         unicode_name = safe_unicode(name)
         db_repo = repo_model.get_by_repo_name(unicode_name)
         # found repo that is on filesystem not in Kallithea database
         if not db_repo:
             log.info('repository %s not found, creating now', name)
             added.append(name)
             desc = (repo.description
                     if repo.description != 'unknown'
                     else '%s repository' % name)
             new_repo = repo_model._create_repo(
                 repo_name=name,
                 repo_type=repo.alias,
                 description=desc,
                 repo_group=getattr(group, 'group_id', None),
                 owner=user,
                 enable_locking=enable_locking,
                 enable_downloads=enable_downloads,
                 enable_statistics=enable_statistics,
                 private=private,
                 state=Repository.STATE_CREATED
+            )
             sa.commit()
             # we added that repo just now, and make sure it has githook
             # installed, and updated server info
             if new_repo.repo_type == 'git':
                 git_repo = new_repo.scm_instance
                 ScmModel().install_git_hooks(git_repo)
                 # update repository server-info
                 log.debug('Running update server info')
                 git_repo._update_server_info()
             new_repo.update_changeset_cache()
         elif install_git_hooks:
             if db_repo.repo_type == 'git':
                 ScmModel().install_git_hooks(db_repo.scm_instance, force_create=overwrite_git_hooks)
     removed = []
     # remove from database those repositories that are not in the filesystem
     unicode_initial_repo_list = set(safe_unicode(name) for name in initial_repo_list)
     for repo in sa.query(Repository).all():
         if repo.repo_name not in unicode_initial_repo_list:
             if remove_obsolete:
                 log.debug("Removing non-existing repository found in db `%s`",
                           repo.repo_name)
                 try:
                     RepoModel(sa).delete(repo, forks='detach', fs_remove=False)
                     sa.commit()
                 except Exception:
                     #don't hold further removals on error
                     log.error(traceback.format_exc())
                     sa.rollback()
             removed.append(repo.repo_name)
     return added, removed
 def load_rcextensions(root_path):
     import kallithea
     from kallithea.config import conf
     path = os.path.join(root_path, 'rcextensions', '__init__.py')
     if os.path.isfile(path):
         rcext = create_module('rc', path)
         EXT = kallithea.EXTENSIONS = rcext
         log.debug('Found rcextensions now loading %s...', rcext)
         # Additional mappings that are not present in the pygments lexers
         conf.LANGUAGES_EXTENSIONS_MAP.update(getattr(EXT, 'EXTRA_MAPPINGS', {}))
         #OVERRIDE OUR EXTENSIONS FROM RC-EXTENSIONS (if present)
         if getattr(EXT, 'INDEX_EXTENSIONS', []):
             log.debug('settings custom INDEX_EXTENSIONS')
             conf.INDEX_EXTENSIONS = getattr(EXT, 'INDEX_EXTENSIONS', [])
         #ADDITIONAL MAPPINGS
         log.debug('adding extra into INDEX_EXTENSIONS')
         conf.INDEX_EXTENSIONS.extend(getattr(EXT, 'EXTRA_INDEX_EXTENSIONS', []))
         # auto check if the module is not missing any data, set to default if is
         # this will help autoupdate new feature of rcext module
         #from kallithea.config import rcextensions
         #for k in dir(rcextensions):
         #    if not k.startswith('_') and not hasattr(EXT, k):
         #        setattr(EXT, k, getattr(rcextensions, k))
 #==============================================================================
 # MISC
 #==============================================================================
 def check_git_version():
     """
     Checks what version of git is installed in system, and issues a warning
     if it's too old for Kallithea to work properly.
     """
     from kallithea import BACKENDS
     from kallithea.lib.vcs.backends.git.repository import GitRepository
     from kallithea.lib.vcs.conf import settings
     from distutils.version import StrictVersion
     if 'git' not in BACKENDS:
         return None
     stdout, stderr = GitRepository._run_git_command(['--version'], _bare=True,
                                                     _safe=True)
     m = re.search("\d+.\d+.\d+", stdout)
     if m:
         ver = StrictVersion(m.group(0))
     else:
         ver = StrictVersion('0.0.0')
     req_ver = StrictVersion('1.7.4')
     log.debug('Git executable: "%s" version %s detected: %s',
               settings.GIT_EXECUTABLE_PATH, ver, stdout)
     if stderr:
         log.warning('Error detecting git version: %r', stderr)
     elif ver < req_ver:
         log.warning('Kallithea detected git version %s, which is too old '
                     'for the system to function properly. '
                     'Please upgrade to version %s or later.' % (ver, req_ver))
     return ver
 #===============================================================================
 # CACHE RELATED METHODS
 #===============================================================================
 # set cache regions for beaker so celery can utilise it
 def setup_cache_regions(settings):
     # Create dict with just beaker cache configs with prefix stripped
     cache_settings = {'regions': None}
     prefix = 'beaker.cache.'
     for key in settings:
         if key.startswith(prefix):
             name = key[len(prefix):]
             cache_settings[name] = settings[key]
     # Find all regions, apply defaults, and apply to beaker
     if cache_settings['regions']:
         for region in cache_settings['regions'].split(','):
             region = region.strip()
             prefix = region + '.'
             region_settings = {}
             for key in cache_settings:
                 if key.startswith(prefix):
                     name = key[len(prefix):]
                     region_settings[name] = cache_settings[key]
             region_settings.setdefault('expire',
                                        cache_settings.get('expire', '60'))
             region_settings.setdefault('lock_dir',
                                        cache_settings.get('lock_dir'))
             region_settings.setdefault('data_dir',
                                        cache_settings.get('data_dir'))
             region_settings.setdefault('type',

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:
                 perm_obj = self._create_default_perms(new_repo, private)
                 self.sa.add(perm_obj)
                 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()
         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()
         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) \

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:
                 perm_obj = self._create_default_perms(new_repo_group)
                 self.sa.add(perm_obj)
                 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)

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)
             perm_obj = self._create_default_perms(new_user_group)
             self.sa.add(perm_obj)
             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()
         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

0 comments (0 inline, 0 general)