Changeset - 837fc99ca140
[Not reviewed]
default
0 2 0
Søren Løvborg - 10 years ago 2015-07-26 13:58:50
kwi@kwi.dk
auth: have fill_data take User object, not lookup key

Simplify the UserModel.fill_data interface by passing the actual db.User
object instead of a key by which it can be looked up. This also saves a
lookup in case the db.User is already loaded (which for now is only the
case in one place, but that'll change in the following changesets).

Also enhance fill_data docstring.
2 files changed with 11 insertions and 21 deletions:
0 comments (0 inline, 0 general)
kallithea/lib/auth.py
Show inline comments
 
@@ -487,66 +487,66 @@ class AuthUser(object):
 
        self.email = ''
 
        self.is_authenticated = False
 
        self.admin = False
 
        self.inherit_default_permissions = False
 
        self.is_external_auth = is_external_auth
 

	
 
        self._propagate_data()
 

	
 
    @LazyProperty
 
    def permissions(self):
 
        return self.__get_perms(user=self, cache=False)
 

	
 
    @property
 
    def api_keys(self):
 
        return self._get_api_keys()
 

	
 
    def _propagate_data(self):
 
        user_model = UserModel()
 
        self.anonymous_user = User.get_default_user(cache=True)
 
        is_user_loaded = False
 

	
 
        # lookup by userid
 
        if self.user_id is not None and self.user_id != self.anonymous_user.user_id:
 
            log.debug('Auth User lookup by USER ID %s' % self.user_id)
 
            is_user_loaded = user_model.fill_data(self, user_id=self.user_id)
 
            is_user_loaded = user_model.fill_data(self, user_model.get(self.user_id))
 

	
 
        # try go get user by API key
 
        elif self._api_key and self._api_key != self.anonymous_user.api_key:
 
            log.debug('Auth User lookup by API key %s' % self._api_key)
 
            is_user_loaded = user_model.fill_data(self, api_key=self._api_key)
 
            is_user_loaded = user_model.fill_data(self, User.get_by_api_key(self._api_key))
 

	
 
        # lookup by username
 
        elif self.username:
 
            log.debug('Auth User lookup by USER NAME %s' % self.username)
 
            is_user_loaded = user_model.fill_data(self, username=self.username)
 
            is_user_loaded = user_model.fill_data(self, User.get_by_username(self.username))
 
        else:
 
            log.debug('No data in %s that could been used to log in' % self)
 

	
 
        if not is_user_loaded:
 
            # if we cannot authenticate user try anonymous
 
            if self.anonymous_user.active:
 
                user_model.fill_data(self, user_id=self.anonymous_user.user_id)
 
                user_model.fill_data(self, self.anonymous_user)
 
                # then we set this user is logged in
 
                self.is_authenticated = True
 
            else:
 
                self.user_id = None
 
                self.username = None
 
                self.is_authenticated = False
 

	
 
        if not self.username:
 
            self.username = 'None'
 

	
 
        log.debug('Auth User is now %s' % self)
 

	
 
    def __get_perms(self, user, explicit=True, algo='higherwin', cache=False):
 
        """
 
        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: `AuthUser` instance
 
        :param explicit: In case there are permissions both for user and a group
 
            that user is part of, explicit flag will define if user will
 
            explicitly override permissions from group, if it's False it will
 
            make decision based on the algo
 
        :param algo: algorithm to decide what permission should be choose if
kallithea/model/user.py
Show inline comments
 
@@ -301,69 +301,59 @@ class UserModel(BaseModel):
 
        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 is not None:
 
            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):
 
    def fill_data(self, auth_user, dbuser):
 
        """
 
        Fetches auth_user by user_id, api_key or username, if present.
 
        Fills auth_user attributes with those taken from database.
 
        Copies database fields from a `db.User` to an `AuthUser`. Does
 
        not copy `api_keys` and `permissions` attributes.
 

	
 
        Checks that `dbuser` is `active` (and not None) before copying;
 
        returns True on success.
 

	
 
        :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
 
        :param dbuser: `db.User` instance to copy from
 
        """
 
        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 = User.get_by_api_key(api_key)
 
        elif username is not None:
 
            dbuser = User.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)
0 comments (0 inline, 0 general)