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

@@ @@ -415,210 +415,210 @@ def _cached_perms_data(user_id, user_is_ @@
         p = perm.Permission.permission_name
         cur_perm = permissions[UK][g_k]
         if multiple_counter[g_k] > 1:
             p = _choose_perm(p, cur_perm)
         permissions[UK][g_k] = p
     #user explicit permission for user groups
     user_user_groups_perms = Permission.get_default_user_group_perms(user_id)
     for perm in user_user_groups_perms:
         u_k = perm.UserUserGroupToPerm.user_group.users_group_name
         p = perm.Permission.permission_name
         cur_perm = permissions[UK][u_k]
         if not explicit:
             p = _choose_perm(p, cur_perm)
         permissions[UK][u_k] = p
     return permissions
 def allowed_api_access(controller_name, whitelist=None, api_key=None):
     """
     Check if given controller_name is in whitelist API access
     """
     if not whitelist:
         from kallithea import CONFIG
         whitelist = aslist(CONFIG.get('api_access_controllers_whitelist'),
                            sep=',')
         log.debug('whitelist of API access is: %s' % (whitelist))
     api_access_valid = controller_name in whitelist
     if api_access_valid:
         log.debug('controller:%s is in API whitelist' % (controller_name))
     else:
         msg = 'controller: %s is *NOT* in API whitelist' % (controller_name)
         if api_key:
             #if we use API key and don't have access it's a warning
             log.warning(msg)
         else:
             log.debug(msg)
     return api_access_valid
 class AuthUser(object):
     """
     Represents a Kallithea user, including various authentication and
     authorization information. Typically used to store the current user,
     but is also used as a generic user information data structure in
     parts of the code, e.g. user management.
     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
     @property
     def is_admin(self):
         return self.admin
     @property
     def repositories_admin(self):
         """
         Returns list of repositories you're an admin of
         """
         return [x[0] for x in self.permissions['repositories'].iteritems()
                 if x[1] == 'repository.admin']
     @property
     def repository_groups_admin(self):
         """
         Returns list of repository groups you're an admin of
         """
         return [x[0] for x in self.permissions['repositories_groups'].iteritems()
                 if x[1] == 'group.admin']
     @property
     def user_groups_admin(self):
         """
         Returns list of user groups you're an admin of
         """
         return [x[0] for x in self.permissions['user_groups'].iteritems()
                 if x[1] == 'usergroup.admin']
     @staticmethod
     def check_ip_allowed(user, ip_addr):
         """
         Check if the given IP address (a `str`) is allowed for the given
         user (an `AuthUser` or `db.User`).
         """
         allowed_ips = AuthUser.get_allowed_ips(user.user_id, cache=True,
             inherit_from_default=user.inherit_default_permissions)
         if check_ip_access(source_ip=ip_addr, allowed_ips=allowed_ips):
             log.debug('IP:%s is in range of %s' % (ip_addr, allowed_ips))
             return True
         else:
             log.info('Access for IP:%s forbidden, '
                      'not in %s' % (ip_addr, allowed_ips))
             return False
     def __repr__(self):
         return "<AuthUser('id:%s[%s] auth:%s')>"\
             % (self.user_id, self.username, self.is_authenticated)

kallithea/model/user.py

➞

Show inline comments

@@ @@ -229,213 +229,203 @@ class UserModel(BaseModel): @@
             raise DefaultUserException(
                 _("You can't Edit this user since it's"
                   " crucial for entire application")
+            )
         for k, v in kwargs.items():
             if k == 'password' and v:
                 v = get_crypt_password(v)
             setattr(user, k, v)
         self.sa.add(user)
         return user
     def delete(self, user, cur_user=None):
         if cur_user is None:
             cur_user = getattr(get_current_authuser(), 'username', None)
         user = self._get_user(user)
         if user.username == User.DEFAULT_USER:
             raise DefaultUserException(
                 _("You can't remove this user since it is"
                   " crucial for the entire application"))
         if user.repositories:
             repos = [x.repo_name for x in user.repositories]
             raise UserOwnsReposException(
                 _('User "%s" still owns %s repositories and cannot be '
                   'removed. Switch owners or remove those repositories: %s')
                 % (user.username, len(repos), ', '.join(repos)))
         if user.repo_groups:
             repogroups = [x.group_name for x in user.repo_groups]
             raise UserOwnsReposException(_(
                 'User "%s" still owns %s repository groups and cannot be '
                 'removed. Switch owners or remove those repository groups: %s')
                 % (user.username, len(repogroups), ', '.join(repogroups)))
         if user.user_groups:
             usergroups = [x.users_group_name for x in user.user_groups]
             raise UserOwnsReposException(
                 _('User "%s" still owns %s user groups and cannot be '
                   'removed. Switch owners or remove those user groups: %s')
                 % (user.username, len(usergroups), ', '.join(usergroups)))
         self.sa.delete(user)
         from kallithea.lib.hooks import log_delete_user
         log_delete_user(user.get_dict(), cur_user)
     def reset_password_link(self, data):
         from kallithea.lib.celerylib import tasks, run_task
         from kallithea.model.notification import EmailNotificationModel
         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(
             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 is not None:
             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

0 comments (0 inline, 0 general)