kallithea Changeset - 7b52c2351231

Changeset - 7b52c2351231

Parent rev.

Child rev.

[Not reviewed]

beta

0 1 0

Marcin Kuzminski - 14 years ago 2012-04-13 21:12:42
marcin@python-works.com

permission comments + out identation for better readability

1 file changed with 124 insertions and 121 deletions:

rhodecode/model/user.py

124

121

0 comments (0 inline, 0 general)

rhodecode/model/user.py

➞

Show inline comments

@@ @@ -305,285 +305,288 @@ class UserModel(BaseModel): @@
             if user.repositories:
                 repos = [x.repo_name for x in user.repositories]
                 raise UserOwnsReposException(
                     _(u'user "%s" still owns %s repositories and cannot be '
                       'removed. Switch owners or remove those repositories. %s')
                     % (user.username, len(repos), ', '.join(repos))
+                )
             self.sa.delete(user)
         except:
             log.error(traceback.format_exc())
             raise
     def reset_password_link(self, data):
         from rhodecode.lib.celerylib import tasks, run_task
         run_task(tasks.send_password_link, data['email'])
     def reset_password(self, data):
         from rhodecode.lib.celerylib import tasks, run_task
         run_task(tasks.reset_user_password, data['email'])
     def fill_data(self, auth_user, user_id=None, api_key=None):
         """
         Fetches auth_user by user_id,or api_key if present.
         Fills auth_user attributes with those taken from database.
         Additionally set's is_authenitated if lookup fails
         present in database
         :param auth_user: instance of user to set attributes
         :param user_id: user id to fetch by
         :param api_key: api key to fetch by
         """
         if user_id is None and api_key is None:
             raise Exception('You need to pass user_id or api_key')
         try:
             if api_key:
                 dbuser = self.get_by_api_key(api_key)
             else:
                 dbuser = self.get(user_id)
             if dbuser is not None and dbuser.active:
                 log.debug('filling %s data' % dbuser)
                 for k, v in dbuser.get_dict().items():
                     setattr(auth_user, k, v)
             else:
                 return False
         except:
             log.error(traceback.format_exc())
             auth_user.is_authenticated = False
             return False
         return True
     def fill_perms(self, user):
         """
         Fills user permission attribute with permissions taken from database
         works for permissions given for repositories, and for permissions that
         are granted to groups
         :param user: user instance to fill his perms
         """
         RK = 'repositories'
         GK = 'repositories_groups'
         GLOBAL = 'global'
         user.permissions[RK] = {}
         user.permissions[GK] = {}
         user.permissions[GLOBAL] = set()
         #======================================================================
         # fetch default permissions
         #======================================================================
         default_user = User.get_by_username('default', cache=True)
         default_user_id = default_user.user_id
         default_repo_perms = Permission.get_default_perms(default_user_id)
         default_repo_groups_perms = Permission.get_default_group_perms(default_user_id)
         if user.is_admin:
             #==================================================================
             # admin user have all default rights for repositories
             # and groups set to admin
             #==================================================================
             user.permissions[GLOBAL].add('hg.admin')
             # repositories
             for perm in default_repo_perms:
                 r_k = perm.UserRepoToPerm.repository.repo_name
                 p = 'repository.admin'
                 user.permissions[RK][r_k] = p
             # repositories groups
             for perm in default_repo_groups_perms:
                 rg_k = perm.UserRepoGroupToPerm.group.group_name
                 p = 'group.admin'
                 user.permissions[GK][rg_k] = p
             return user
         else:
             #==================================================================
             # set default permissions first for repositories and groups
             #==================================================================
             uid = user.user_id
         #==================================================================
         # set default permissions first for repositories and groups
         #==================================================================
         uid = user.user_id
             # default global permissions
             default_global_perms = self.sa.query(UserToPerm)\
                 .filter(UserToPerm.user_id == default_user_id)
         # default global permissions
         default_global_perms = self.sa.query(UserToPerm)\
             .filter(UserToPerm.user_id == default_user_id)
             for perm in default_global_perms:
                 user.permissions[GLOBAL].add(perm.permission.permission_name)
         for perm in default_global_perms:
             user.permissions[GLOBAL].add(perm.permission.permission_name)
             # defaults for repositories, taken from default user
             for perm in default_repo_perms:
                 r_k = perm.UserRepoToPerm.repository.repo_name
                 if perm.Repository.private and not (perm.Repository.user_id == uid):
                     # disable defaults for private repos,
                     p = 'repository.none'
                 elif perm.Repository.user_id == uid:
                     # set admin if owner
                     p = 'repository.admin'
                 else:
                     p = perm.Permission.permission_name
         # defaults for repositories, taken from default user
         for perm in default_repo_perms:
             r_k = perm.UserRepoToPerm.repository.repo_name
             if perm.Repository.private and not (perm.Repository.user_id == uid):
                 # disable defaults for private repos,
                 p = 'repository.none'
             elif perm.Repository.user_id == uid:
                 # set admin if owner
                 p = 'repository.admin'
             else:
                 p = perm.Permission.permission_name
             user.permissions[RK][r_k] = p
                 user.permissions[RK][r_k] = p
         # defaults for repositories groups taken from default user permission
         # on given group
         for perm in default_repo_groups_perms:
             rg_k = perm.UserRepoGroupToPerm.group.group_name
             p = perm.Permission.permission_name
             user.permissions[GK][rg_k] = p
         #==================================================================
         # overwrite defaults with user permissions if any found
         #==================================================================
         # user global permissions
         user_perms = self.sa.query(UserToPerm)\
                 .options(joinedload(UserToPerm.permission))\
                 .filter(UserToPerm.user_id == uid).all()
         for perm in user_perms:
             user.permissions[GLOBAL].add(perm.permission.permission_name)
             # defaults for repositories groups taken from default user permission
             # on given group
             for perm in default_repo_groups_perms:
                 rg_k = perm.UserRepoGroupToPerm.group.group_name
                 p = perm.Permission.permission_name
                 user.permissions[GK][rg_k] = p
         # user explicit permissions for repositories
         user_repo_perms = \
          self.sa.query(UserRepoToPerm, Permission, Repository)\
          .join((Repository, UserRepoToPerm.repository_id == Repository.repo_id))\
          .join((Permission, UserRepoToPerm.permission_id == Permission.permission_id))\
          .filter(UserRepoToPerm.user_id == uid)\
          .all()
             #==================================================================
             # overwrite defaults with user permissions if any found
             #==================================================================
         for perm in user_repo_perms:
             # set admin if owner
             r_k = perm.UserRepoToPerm.repository.repo_name
             if perm.Repository.user_id == uid:
                 p = 'repository.admin'
             else:
                 p = perm.Permission.permission_name
             user.permissions[RK][r_k] = p
             # user global permissions
             user_perms = self.sa.query(UserToPerm)\
                     .options(joinedload(UserToPerm.permission))\
                     .filter(UserToPerm.user_id == uid).all()
         # USER GROUP
         #==================================================================
         # check if user is part of user groups for this repository and
         # fill in (or replace with higher) permissions
         #==================================================================
             for perm in user_perms:
                 user.permissions[GLOBAL].add(perm.permission.permission_name)
         # users group global
         user_perms_from_users_groups = self.sa.query(UsersGroupToPerm)\
             .options(joinedload(UsersGroupToPerm.permission))\
             .join((UsersGroupMember, UsersGroupToPerm.users_group_id ==
                    UsersGroupMember.users_group_id))\
             .filter(UsersGroupMember.user_id == uid).all()
         for perm in user_perms_from_users_groups:
             user.permissions[GLOBAL].add(perm.permission.permission_name)
             # user explicit permissions for repositories
             user_repo_perms = \
              self.sa.query(UserRepoToPerm, Permission, Repository)\
              .join((Repository, UserRepoToPerm.repository_id == Repository.repo_id))\
              .join((Permission, UserRepoToPerm.permission_id == Permission.permission_id))\
              .filter(UserRepoToPerm.user_id == uid)\
              .all()
         # users group for repositories permissions
         user_repo_perms_from_users_groups = \
          self.sa.query(UsersGroupRepoToPerm, Permission, Repository,)\
          .join((Repository, UsersGroupRepoToPerm.repository_id == Repository.repo_id))\
          .join((Permission, UsersGroupRepoToPerm.permission_id == Permission.permission_id))\
          .join((UsersGroupMember, UsersGroupRepoToPerm.users_group_id == UsersGroupMember.users_group_id))\
          .filter(UsersGroupMember.user_id == uid)\
          .all()
             for perm in user_repo_perms:
                 # set admin if owner
                 r_k = perm.UserRepoToPerm.repository.repo_name
                 if perm.Repository.user_id == uid:
                     p = 'repository.admin'
                 else:
                     p = perm.Permission.permission_name
         for perm in user_repo_perms_from_users_groups:
             r_k = perm.UsersGroupRepoToPerm.repository.repo_name
             p = perm.Permission.permission_name
             cur_perm = user.permissions[RK][r_k]
             # overwrite permission only if it's greater than permission
             # given from other sources
             if PERM_WEIGHTS[p] > PERM_WEIGHTS[cur_perm]:
                 user.permissions[RK][r_k] = p
             #==================================================================
             # check if user is part of user groups for this repository and
             # fill in (or replace with higher) permissions
             #==================================================================
             # users group global
             user_perms_from_users_groups = self.sa.query(UsersGroupToPerm)\
                 .options(joinedload(UsersGroupToPerm.permission))\
                 .join((UsersGroupMember, UsersGroupToPerm.users_group_id ==
                        UsersGroupMember.users_group_id))\
                 .filter(UsersGroupMember.user_id == uid).all()
             for perm in user_perms_from_users_groups:
                 user.permissions[GLOBAL].add(perm.permission.permission_name)
         # REPO GROUP
         #==================================================================
         # get access for this user for repos group and override defaults
         #==================================================================
             # users group for repositories permissions
             user_repo_perms_from_users_groups = \
              self.sa.query(UsersGroupRepoToPerm, Permission, Repository,)\
              .join((Repository, UsersGroupRepoToPerm.repository_id == Repository.repo_id))\
              .join((Permission, UsersGroupRepoToPerm.permission_id == Permission.permission_id))\
              .join((UsersGroupMember, UsersGroupRepoToPerm.users_group_id == UsersGroupMember.users_group_id))\
              .filter(UsersGroupMember.user_id == uid)\
              .all()
         # user explicit permissions for repository
         user_repo_groups_perms = \
          self.sa.query(UserRepoGroupToPerm, Permission, RepoGroup)\
          .join((RepoGroup, UserRepoGroupToPerm.group_id == RepoGroup.group_id))\
          .join((Permission, UserRepoGroupToPerm.permission_id == Permission.permission_id))\
          .filter(UserRepoGroupToPerm.user_id == uid)\
          .all()
             for perm in user_repo_perms_from_users_groups:
                 r_k = perm.UsersGroupRepoToPerm.repository.repo_name
                 p = perm.Permission.permission_name
                 cur_perm = user.permissions[RK][r_k]
                 # overwrite permission only if it's greater than permission
                 # given from other sources
                 if PERM_WEIGHTS[p] > PERM_WEIGHTS[cur_perm]:
                     user.permissions[RK][r_k] = p
             #==================================================================
             # get access for this user for repos group and override defaults
             #==================================================================
         for perm in user_repo_groups_perms:
             rg_k = perm.UserRepoGroupToPerm.group.group_name
             p = perm.Permission.permission_name
             cur_perm = user.permissions[GK][rg_k]
             if PERM_WEIGHTS[p] > PERM_WEIGHTS[cur_perm]:
                 user.permissions[GK][rg_k] = p
             # user explicit permissions for repository
             user_repo_groups_perms = \
              self.sa.query(UserRepoGroupToPerm, Permission, RepoGroup)\
              .join((RepoGroup, UserRepoGroupToPerm.group_id == RepoGroup.group_id))\
              .join((Permission, UserRepoGroupToPerm.permission_id == Permission.permission_id))\
              .filter(UserRepoGroupToPerm.user_id == uid)\
              .all()
             for perm in user_repo_groups_perms:
                 rg_k = perm.UserRepoGroupToPerm.group.group_name
                 p = perm.Permission.permission_name
                 cur_perm = user.permissions[GK][rg_k]
                 if PERM_WEIGHTS[p] > PERM_WEIGHTS[cur_perm]:
                     user.permissions[GK][rg_k] = p
         # REPO GROUP + USER GROUP
         #==================================================================
         # check if user is part of user groups for this repo group and
         # fill in (or replace with higher) permissions
         #==================================================================
             #==================================================================
             # check if user is part of user groups for this repo group and
             # fill in (or replace with higher) permissions
             #==================================================================
         # users group for repositories permissions
         user_repo_group_perms_from_users_groups = \
          self.sa.query(UsersGroupRepoGroupToPerm, Permission, RepoGroup)\
          .join((RepoGroup, UsersGroupRepoGroupToPerm.group_id == RepoGroup.group_id))\
          .join((Permission, UsersGroupRepoGroupToPerm.permission_id == Permission.permission_id))\
          .join((UsersGroupMember, UsersGroupRepoGroupToPerm.users_group_id == UsersGroupMember.users_group_id))\
          .filter(UsersGroupMember.user_id == uid)\
          .all()
             # users group for repositories permissions
             user_repo_group_perms_from_users_groups = \
              self.sa.query(UsersGroupRepoGroupToPerm, Permission, RepoGroup)\
              .join((RepoGroup, UsersGroupRepoGroupToPerm.group_id == RepoGroup.group_id))\
              .join((Permission, UsersGroupRepoGroupToPerm.permission_id == Permission.permission_id))\
              .join((UsersGroupMember, UsersGroupRepoGroupToPerm.users_group_id == UsersGroupMember.users_group_id))\
              .filter(UsersGroupMember.user_id == uid)\
              .all()
             for perm in user_repo_group_perms_from_users_groups:
                 g_k = perm.UsersGroupRepoGroupToPerm.group.group_name
                 print perm, g_k
                 p = perm.Permission.permission_name
                 cur_perm = user.permissions[GK][g_k]
                 # overwrite permission only if it's greater than permission
                 # given from other sources
                 if PERM_WEIGHTS[p] > PERM_WEIGHTS[cur_perm]:
                     user.permissions[GK][g_k] = p
         for perm in user_repo_group_perms_from_users_groups:
             g_k = perm.UsersGroupRepoGroupToPerm.group.group_name
             print perm, g_k
             p = perm.Permission.permission_name
             cur_perm = user.permissions[GK][g_k]
             # overwrite permission only if it's greater than permission
             # given from other sources
             if PERM_WEIGHTS[p] > PERM_WEIGHTS[cur_perm]:
                 user.permissions[GK][g_k] = p
         return user
     def has_perm(self, user, perm):
         if not isinstance(perm, Permission):
             raise Exception('perm needs to be an instance of Permission class '
                             'got %s instead' % type(perm))
         user = self.__get_user(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 = self.__get_user(user)
         perm = self.__get_perm(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)
     def revoke_perm(self, user, perm):
         """
         Revoke users global permissions
         :param user:
         :param perm:
         """
         user = self.__get_user(user)
         perm = self.__get_perm(perm)
         obj = UserToPerm.query()\
                 .filter(UserToPerm.user == user)\
                 .filter(UserToPerm.permission == perm)\
                 .scalar()
         if obj:
             self.sa.delete(obj)

0 comments (0 inline, 0 general)