kallithea Changeset - f6be760c786c

Changeset - f6be760c786c

Parent rev.

Child rev.

[Not reviewed]

default

0 2 0

Mads Kiilerich - 6 years ago 2020-04-12 02:43:56
mads@kiilerich.com

Grafted from: 2e4a6119cd6f

auth: simplify AuthUser.permissions and how it actually never use cache

2 files changed with 12 insertions and 22 deletions:

kallithea/lib/auth.py

kallithea/lib/utils.py

0 comments (0 inline, 0 general)

kallithea/lib/auth.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.lib.auth
 ~~~~~~~~~~~~~~~~~~
 authentication and permission libraries
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 4, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import hashlib
 import itertools
 import logging
 import os
 import string
 import bcrypt
 import ipaddr
 from decorator import decorator
 from sqlalchemy.orm import joinedload
 from sqlalchemy.orm.exc import ObjectDeletedError
 from tg import request
 from tg.i18n import ugettext as _
 from webob.exc import HTTPForbidden, HTTPFound
 from kallithea.config.routing import url
-from kallithea.lib.utils import conditional_cache, get_repo_group_slug, get_repo_slug, get_user_group_slug
 from kallithea.lib.utils import get_repo_group_slug, get_repo_slug, get_user_group_slug
 from kallithea.lib.utils2 import ascii_bytes, ascii_str, safe_bytes
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 from kallithea.model.db import (Permission, User, UserApiKeys, UserGroup, UserGroupMember, UserGroupRepoGroupToPerm, UserGroupRepoToPerm, UserGroupToPerm,
                                 UserGroupUserGroupToPerm, UserIpMap, UserToPerm)
 from kallithea.model.meta import Session
 from kallithea.model.user import UserModel
 log = logging.getLogger(__name__)
 class PasswordGenerator(object):
     """
     This is a simple class for generating password from different sets of
     characters
     usage::
         passwd_gen = PasswordGenerator()
         #print 8-letter password containing only big and small letters
             of alphabet
         passwd_gen.gen_password(8, passwd_gen.ALPHABETS_BIG_SMALL)
     """
     ALPHABETS_NUM = r'''1234567890'''
     ALPHABETS_SMALL = r'''qwertyuiopasdfghjklzxcvbnm'''
     ALPHABETS_BIG = r'''QWERTYUIOPASDFGHJKLZXCVBNM'''
     ALPHABETS_SPECIAL = r'''`-=[]\;',./~!@#$%^&*()_+{}|:"<>?'''
     ALPHABETS_FULL = ALPHABETS_BIG + ALPHABETS_SMALL \
         + ALPHABETS_NUM + ALPHABETS_SPECIAL
     ALPHABETS_ALPHANUM = ALPHABETS_BIG + ALPHABETS_SMALL + ALPHABETS_NUM
     ALPHABETS_BIG_SMALL = ALPHABETS_BIG + ALPHABETS_SMALL
     ALPHABETS_ALPHANUM_BIG = ALPHABETS_BIG + ALPHABETS_NUM
     ALPHABETS_ALPHANUM_SMALL = ALPHABETS_SMALL + ALPHABETS_NUM
     def gen_password(self, length, alphabet=ALPHABETS_FULL):
         assert len(alphabet) <= 256, alphabet
         l = []
         while len(l) < length:
             i = ord(os.urandom(1))
             if i < len(alphabet):
                 l.append(alphabet[i])
         return ''.join(l)
 def get_crypt_password(password):
     """
     Cryptographic function used for bcrypt password hashing.
     :param password: password to hash
     """
     return ascii_str(bcrypt.hashpw(safe_bytes(password), bcrypt.gensalt(10)))
 def check_password(password, hashed):
     """
     Checks password match the hashed value using bcrypt.
     Remains backwards compatible and accept plain sha256 hashes which used to
     be used on Windows.
     :param password: password
     :param hashed: password in hashed form
     """
     # sha256 hashes will always be 64 hex chars
     # bcrypt hashes will always contain $ (and be shorter)
     if len(hashed) == 64 and all(x in string.hexdigits for x in hashed):
         return hashlib.sha256(password).hexdigest() == hashed
     try:
         return bcrypt.checkpw(safe_bytes(password), ascii_bytes(hashed))
     except ValueError as e:
         # bcrypt will throw ValueError 'Invalid hashed_password salt' on all password errors
         log.error('error from bcrypt checking password: %s', e)
         return False
     log.error('check_password failed - no method found for hash length %s', len(hashed))
     return False
 def _cached_perms_data(user_id, user_is_admin):
     RK = 'repositories'
     GK = 'repositories_groups'
     UK = 'user_groups'
     GLOBAL = 'global'
     PERM_WEIGHTS = Permission.PERM_WEIGHTS
     permissions = {RK: {}, GK: {}, UK: {}, GLOBAL: set()}
     def bump_permission(kind, key, new_perm):
         """Add a new permission for kind and key.
         Assuming the permissions are comparable, set the new permission if it
         has higher weight, else drop it and keep the old permission.
         """
         cur_perm = permissions[kind][key]
         new_perm_val = PERM_WEIGHTS[new_perm]
         cur_perm_val = PERM_WEIGHTS[cur_perm]
         if new_perm_val > cur_perm_val:
             permissions[kind][key] = new_perm
     #======================================================================
     # fetch default permissions
@@ @@ -340,250 +340,242 @@ def _cached_perms_data(user_id, user_is_ @@
     user_user_groups_perms = Permission.get_default_user_group_perms(user_id)
     for perm in user_user_groups_perms:
         bump_permission(UK,
             perm.user_group.users_group_name,
             perm.permission.permission_name)
     return permissions
 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 a database `User` object, a user ID or cookie dict,
     it looks up the user (if needed) 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. It's up to other parts of
     the code to check e.g. if a supplied password is correct, and if so, trust
     the AuthUser object as an authenticated user.
     However, `AuthUser` does refuse to load a user that is not `active`.
     Note that Kallithea distinguishes between the default user (an actual
     user in the database with username "default") and "no user" (no actual
     User object, AuthUser filled with blank values and username "None").
     If the default user is active, that will always be used instead of
     "no user". On the other hand, if the default user is disabled (and
     there is no login information), we instead get "no user"; this should
     only happen on the login page (as all other requests are redirected).
     `is_default_user` specifically checks if the AuthUser is the user named
     "default". Use `is_anonymous` to check for both "default" and "no user".
     """
     @classmethod
     def make(cls, dbuser=None, is_external_auth=False, ip_addr=None):
         """Create an AuthUser to be authenticated ... or return None if user for some reason can't be authenticated.
         Checks that a non-None dbuser is provided, is active, and that the IP address is ok.
         """
         assert ip_addr is not None
         if dbuser is None:
             log.info('No db user for authentication')
             return None
         if not dbuser.active:
             log.info('Db user %s not active', dbuser.username)
             return None
         allowed_ips = AuthUser.get_allowed_ips(dbuser.user_id)
         if not check_ip_access(source_ip=ip_addr, allowed_ips=allowed_ips):
             log.info('Access for %s from %s forbidden - not in %s', dbuser.username, ip_addr, allowed_ips)
             return None
         return cls(dbuser=dbuser, is_external_auth=is_external_auth)
     def __init__(self, user_id=None, dbuser=None, is_external_auth=False):
         self.is_external_auth = is_external_auth # container auth - don't show logout option
         # These attributes will be overridden below if the requested user is
         # found or anonymous access (using the default user) is enabled.
         self.user_id = None
         self.username = None
         self.api_key = None
         self.name = ''
         self.lastname = ''
         self.email = ''
         self.admin = False
         # Look up database user, if necessary.
         if user_id is not None:
             assert dbuser is None
             log.debug('Auth User lookup by USER ID %s', user_id)
             dbuser = UserModel().get(user_id)
             assert dbuser is not None
         else:
             assert dbuser is not None
             log.debug('Auth User lookup by database user %s', dbuser)
         log.debug('filling %s data', dbuser)
         self.is_anonymous = dbuser.is_default_user
         if dbuser.is_default_user and not dbuser.active:
             self.username = 'None'
             self.is_default_user = False
         else:
             # copy non-confidential database fields from a `db.User` to this `AuthUser`.
             for k, v in dbuser.get_dict().items():
                 assert k not in ['api_keys', 'permissions']
                 setattr(self, k, v)
             self.is_default_user = dbuser.is_default_user
         log.debug('Auth User is now %s', self)
     @LazyProperty
     def permissions(self):
         return self.__get_perms(user=self, 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
         """
         log.debug('Getting PERMISSION tree for %s', self)
         return _cached_perms_data(self.user_id, self.is_admin)
     def has_repository_permission_level(self, repo_name, level, purpose=None):
         required_perms = {
             'read': ['repository.read', 'repository.write', 'repository.admin'],
             'write': ['repository.write', 'repository.admin'],
             'admin': ['repository.admin'],
         }[level]
         actual_perm = self.permissions['repositories'].get(repo_name)
         ok = actual_perm in required_perms
         log.debug('Checking if user %r can %r repo %r (%s): %s (has %r)',
             self.username, level, repo_name, purpose, ok, actual_perm)
         return ok
     def has_repository_group_permission_level(self, repo_group_name, level, purpose=None):
         required_perms = {
             'read': ['group.read', 'group.write', 'group.admin'],
             'write': ['group.write', 'group.admin'],
             'admin': ['group.admin'],
         }[level]
         actual_perm = self.permissions['repositories_groups'].get(repo_group_name)
         ok = actual_perm in required_perms
         log.debug('Checking if user %r can %r repo group %r (%s): %s (has %r)',
             self.username, level, repo_group_name, purpose, ok, actual_perm)
         return ok
     def has_user_group_permission_level(self, user_group_name, level, purpose=None):
         required_perms = {
             'read': ['usergroup.read', 'usergroup.write', 'usergroup.admin'],
             'write': ['usergroup.write', 'usergroup.admin'],
             'admin': ['usergroup.admin'],
         }[level]
         actual_perm = self.permissions['user_groups'].get(user_group_name)
         ok = actual_perm in required_perms
         log.debug('Checking if user %r can %r user group %r (%s): %s (has %r)',
             self.username, level, user_group_name, purpose, ok, actual_perm)
         return ok
     @property
     def api_keys(self):
         return self._get_api_keys()
     def __get_perms(self, user, 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
         """
         user_id = user.user_id
         user_is_admin = user.is_admin
         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)
     def _get_api_keys(self):
         api_keys = [self.api_key]
         for api_key in UserApiKeys.query() \
                 .filter_by(user_id=self.user_id, is_expired=False):
             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'].items()
                 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'].items()
                 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'].items()
                 if x[1] == 'usergroup.admin']
     def __repr__(self):
         return "<%s %s: %r>" % (self.__class__.__name__, self.user_id, self.username)
     def to_cookie(self):
         """ Serializes this login session to a cookie `dict`. """
         return {
             'user_id': self.user_id,
             'is_external_auth': self.is_external_auth,
+        }
     @staticmethod
     def from_cookie(cookie, ip_addr):
         """
         Deserializes an `AuthUser` from a cookie `dict` ... or return None.
         """
         return AuthUser.make(
             dbuser=UserModel().get(cookie.get('user_id')),
             is_external_auth=cookie.get('is_external_auth', False),
             ip_addr=ip_addr,
+        )
     @classmethod
     def get_allowed_ips(cls, user_id):
         _set = set()
         default_ips = UserIpMap.query().filter(UserIpMap.user_id ==
                                         User.get_default_user().user_id)
         for ip in default_ips:
             try:
                 _set.add(ip.ip_addr)
             except ObjectDeletedError:
                 # since we use heavy caching sometimes it happens that we get
                 # deleted objects here, we just skip them
                 pass
         user_ips = UserIpMap.query().filter(UserIpMap.user_id == user_id)
         for ip in user_ips:
             try:
                 _set.add(ip.ip_addr)
             except ObjectDeletedError:
                 # since we use heavy caching sometimes it happens that we get
                 # deleted objects here, we just skip them
                 pass
         return _set or set(['0.0.0.0/0', '::/0'])
 #==============================================================================
 # CHECK DECORATORS
 #==============================================================================
 def _redirect_to_login(message=None):
     """Return an exception that must be raised. It will redirect to the login
     page which will redirect back to the current URL after authentication.
     The optional message will be shown in a flash message."""
     from kallithea.lib import helpers as h
     if message:
         h.flash(message, category='warning')
     p = request.path_qs
     log.debug('Redirecting to login page, origin: %s', p)
     return HTTPFound(location=url('login_home', came_from=p))

kallithea/lib/utils.py

➞

Show inline comments

@@ @@ -517,119 +517,117 @@ def repo2db_mapper(initial_repo_dict, re @@
                     #don't hold further removals on error
                     log.error(traceback.format_exc())
                     sa.rollback()
             removed.append(repo.repo_name)
     return added, removed
 def load_rcextensions(root_path):
     path = os.path.join(root_path, 'rcextensions', '__init__.py')
     if os.path.isfile(path):
         rcext = create_module('rc', path)
         EXT = kallithea.EXTENSIONS = rcext
         log.debug('Found rcextensions now loading %s...', rcext)
         # Additional mappings that are not present in the pygments lexers
         kallithea.config.conf.LANGUAGES_EXTENSIONS_MAP.update(getattr(EXT, 'EXTRA_MAPPINGS', {}))
         # OVERRIDE OUR EXTENSIONS FROM RC-EXTENSIONS (if present)
         if getattr(EXT, 'INDEX_EXTENSIONS', []):
             log.debug('settings custom INDEX_EXTENSIONS')
             kallithea.config.conf.INDEX_EXTENSIONS = getattr(EXT, 'INDEX_EXTENSIONS', [])
         # ADDITIONAL MAPPINGS
         log.debug('adding extra into INDEX_EXTENSIONS')
         kallithea.config.conf.INDEX_EXTENSIONS.extend(getattr(EXT, 'EXTRA_INDEX_EXTENSIONS', []))
         # auto check if the module is not missing any data, set to default if is
         # this will help autoupdate new feature of rcext module
         #from kallithea.config import rcextensions
         #for k in dir(rcextensions):
         #    if not k.startswith('_') and not hasattr(EXT, k):
         #        setattr(EXT, k, getattr(rcextensions, k))
 #==============================================================================
 # MISC
 #==============================================================================
 git_req_ver = StrictVersion('1.7.4')
 def check_git_version():
     """
     Checks what version of git is installed on the system, and raise a system exit
     if it's too old for Kallithea to work properly.
     """
     if 'git' not in kallithea.BACKENDS:
         return None
     if not settings.GIT_EXECUTABLE_PATH:
         log.warning('No git executable configured - check "git_path" in the ini file.')
         return None
     try:
         stdout, stderr = GitRepository._run_git_command(['--version'])
     except RepositoryError as e:
         # message will already have been logged as error
         log.warning('No working git executable found - check "git_path" in the ini file.')
         return None
     if stderr:
         log.warning('Error/stderr from "%s --version":\n%s', settings.GIT_EXECUTABLE_PATH, safe_str(stderr))
     if not stdout:
         log.warning('No working git executable found - check "git_path" in the ini file.')
         return None
     output = safe_str(stdout).strip()
     m = re.search(r"\d+.\d+.\d+", output)
     if m:
         ver = StrictVersion(m.group(0))
         log.debug('Git executable: "%s", version %s (parsed from: "%s")',
                   settings.GIT_EXECUTABLE_PATH, ver, output)
         if ver < git_req_ver:
             log.error('Kallithea detected %s version %s, which is too old '
                       'for the system to function properly. '
                       'Please upgrade to version %s or later. '
                       'If you strictly need Mercurial repositories, you can '
                       'clear the "git_path" setting in the ini file.',
                       settings.GIT_EXECUTABLE_PATH, ver, git_req_ver)
             log.error("Terminating ...")
             sys.exit(1)
     else:
         ver = StrictVersion('0.0.0')
         log.warning('Error finding version number in "%s --version" stdout:\n%s',
                     settings.GIT_EXECUTABLE_PATH, output)
     return ver
 #===============================================================================
 # CACHE RELATED METHODS
 #===============================================================================
 def conditional_cache(region, prefix, condition, func):
     """
     Conditional caching function use like::
-        def _c(arg):
+        def func(arg):
             #heavy computation function
             return data
         # depending from condition the compute is wrapped in cache or not
         compute = conditional_cache('short_term', 'cache_desc', condition=True, func=func)
         return compute(arg)
     :param region: name of cache region
     :param prefix: cache region prefix
     :param condition: condition for cache to be triggered, and return data cached
     :param func: wrapped heavy function to compute
     :param func: heavy function to compute
     """
     wrapped = func
     if condition:
         log.debug('conditional_cache: True, wrapping call of '
                   'func: %s into %s region cache' % (region, func))
         wrapped = beaker.cache._cache_decorate((prefix,), None, None, region)(func)
     return wrapped

0 comments (0 inline, 0 general)