Changeset - 021012521a10
[Not reviewed]
default
0 1 0
Ronny Pfannschmidt - 11 years ago 2015-02-02 20:38:57
opensource@ronnypfannschmidt.de
user model: saner user permission revoking
1 file changed with 4 insertions and 6 deletions:
0 comments (0 inline, 0 general)
kallithea/model/user.py
Show inline comments
 
@@ -317,162 +317,160 @@ class UserModel(BaseModel):
 
            log.debug("password reset email %s not found" % user_email)
 

	
 
        return True
 

	
 
    def reset_password(self, data):
 
        from kallithea.lib.celerylib import tasks, run_task
 
        from kallithea.lib import auth
 
        user_email = data['email']
 
        user = User.get_by_email(user_email)
 
        new_passwd = auth.PasswordGenerator().gen_password(
 
            8, auth.PasswordGenerator.ALPHABETS_BIG_SMALL)
 
        if user:
 
            user.password = auth.get_crypt_password(new_passwd)
 
            Session().add(user)
 
            Session().commit()
 
            log.info('change password for %s' % user_email)
 
        if new_passwd is None:
 
            raise Exception('unable to generate new password')
 

	
 
        run_task(tasks.send_email, [user_email],
 
                 _('Your new password'),
 
                 _('Your new Kallithea password:%s') % (new_passwd,))
 
        log.info('send new password mail to %s' % user_email)
 

	
 
        return True
 

	
 
    def fill_data(self, auth_user, user_id=None, api_key=None, username=None):
 
        """
 
        Fetches auth_user by user_id,or api_key if present.
 
        Fills auth_user attributes with those taken from database.
 
        Additionally sets is_authenticated 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
 
        :param username: username to fetch by
 
        """
 
        if user_id is None and api_key is None and username is None:
 
            raise Exception('You need to pass user_id, api_key or username')
 

	
 
        dbuser = None
 
        if user_id is not None:
 
            dbuser = self.get(user_id)
 
        elif api_key is not None:
 
            dbuser = self.get_by_api_key(api_key)
 
        elif username is not None:
 
            dbuser = self.get_by_username(username)
 

	
 
        if dbuser is not None and dbuser.active:
 
            log.debug('filling %s data' % dbuser)
 
            for k, v in dbuser.get_dict().iteritems():
 
                if k not in ['api_keys', 'permissions']:
 
                    setattr(auth_user, k, v)
 
            return True
 
        return False
 

	
 
    def has_perm(self, user, perm):
 
        perm = self._get_perm(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)
 
        return 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)
 
        UserToPerm.query().filter(
 
            UserToPerm.user == user,
 
            UserToPerm.permission == perm,
 
        ).delete()
 

	
 
    def add_extra_email(self, user, email):
 
        """
 
        Adds email address to UserEmailMap
 

	
 
        :param user:
 
        :param email:
 
        """
 
        from kallithea.model import forms
 
        form = forms.UserExtraEmailForm()()
 
        data = form.to_python(dict(email=email))
 
        user = self._get_user(user)
 

	
 
        obj = UserEmailMap()
 
        obj.user = user
 
        obj.email = data['email']
 
        self.sa.add(obj)
 
        return obj
 

	
 
    def delete_extra_email(self, user, email_id):
 
        """
 
        Removes email address from UserEmailMap
 

	
 
        :param user:
 
        :param email_id:
 
        """
 
        user = self._get_user(user)
 
        obj = UserEmailMap.query().get(email_id)
 
        if obj:
 
            self.sa.delete(obj)
 

	
 
    def add_extra_ip(self, user, ip):
 
        """
 
        Adds ip address to UserIpMap
 

	
 
        :param user:
 
        :param ip:
 
        """
 
        from kallithea.model import forms
 
        form = forms.UserExtraIpForm()()
 
        data = form.to_python(dict(ip=ip))
 
        user = self._get_user(user)
 

	
 
        obj = UserIpMap()
 
        obj.user = user
 
        obj.ip_addr = data['ip']
 
        self.sa.add(obj)
 
        return obj
 

	
 
    def delete_extra_ip(self, user, ip_id):
 
        """
 
        Removes ip address from UserIpMap
 

	
 
        :param user:
 
        :param ip_id:
 
        """
 
        user = self._get_user(user)
 
        obj = UserIpMap.query().get(ip_id)
 
        if obj:
 
            self.sa.delete(obj)
0 comments (0 inline, 0 general)