kallithea Changeset - 837fc99ca140

Changeset - 837fc99ca140

Parent rev.

Child rev.

[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:

kallithea/lib/auth.py

kallithea/model/user.py

0 comments (0 inline, 0 general)

kallithea/lib/auth.py

➞

Show inline comments

@@ @@ -463,114 +463,114 @@ class AuthUser(object): @@
     Constructed from user ID, username, API key or cookie dict, it looks
     up the matching database `User` and copies all attributes to itself,
     adding various non-persistent data. If lookup fails but anonymous
     access to Kallithea is enabled, the default user is loaded instead.
     `AuthUser` does not by itself authenticate users and the constructor
     sets the `is_authenticated` field to False, except when falling back
     to the default anonymous user (if enabled). It's up to other parts
     of the code to check e.g. if a supplied password is correct, and if
     so, set `is_authenticated` to True.
     However, `AuthUser` does refuse to load a user that is not `active`.
     """
     def __init__(self, user_id=None, api_key=None, username=None,
             is_external_auth=False):
         self.user_id = user_id
         self._api_key = api_key # API key passed as parameter
         self.api_key = None # API key set by user_model.fill_data
         self.username = username
         self.name = ''
         self.lastname = ''
         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
             it's multiple defined, eg user in two different groups. It also
             decides if explicit flag is turned off how to specify the permission
             for case when user is in a group + have defined separate permission
         """
         user_id = user.user_id
         user_is_admin = user.is_admin
         user_inherit_default_permissions = user.inherit_default_permissions
         log.debug('Getting PERMISSION tree')
         compute = conditional_cache('short_term', 'cache_desc',
                                     condition=cache, func=_cached_perms_data)
         return compute(user_id, user_is_admin,
                        user_inherit_default_permissions, explicit, algo)
     def _get_api_keys(self):
         api_keys = [self.api_key]
         for api_key in UserApiKeys.query()\
                 .filter(UserApiKeys.user_id == self.user_id)\
                 .filter(or_(UserApiKeys.expires == -1,
                             UserApiKeys.expires >= time.time())).all():
             api_keys.append(api_key.api_key)
         return api_keys

kallithea/model/user.py

➞

Show inline comments

@@ @@ -277,117 +277,107 @@ class UserModel(BaseModel): @@
         import kallithea.lib.helpers as h
         user_email = data['email']
         user = User.get_by_email(user_email)
         if user is not None:
             log.debug('password reset user found %s' % user)
             link = h.canonical_url('reset_password_confirmation',
                                    key=user.api_key)
             reg_type = EmailNotificationModel.TYPE_PASSWORD_RESET
             body = EmailNotificationModel().get_email_tmpl(
                 reg_type, 'txt',
                 user=user.short_contact,
                 reset_url=link)
             html_body = EmailNotificationModel().get_email_tmpl(
                 reg_type, 'html',
                 user=user.short_contact,
                 reset_url=link)
             log.debug('sending email')
             run_task(tasks.send_email, [user_email],
                      _("Password reset link"), body, html_body)
             log.info('send new password mail to %s' % user_email)
         else:
             log.debug("password reset email %s not found" % user_email)
         return 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(
 , 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)
         # 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)
         UserToPerm.query().filter(

0 comments (0 inline, 0 general)