kallithea Changeset - 7977ca209b1d

Changeset - 7977ca209b1d

Parent rev.

Child rev.

[Not reviewed]

default

0 2 0

Mads Kiilerich - 7 years ago 2019-04-08 00:11:20
mads@kiilerich.com

auth: make User.get_by_api_key more strict about only returning active non-default users

Thus drop some extra checks against default user.

2 files changed with 7 insertions and 5 deletions:

kallithea/lib/base.py

kallithea/model/db.py

0 comments (0 inline, 0 general)

kallithea/lib/base.py

➞

Show inline comments

@@ @@ -372,53 +372,53 @@ class BaseController(TGController): @@
         c.repo_name = get_repo_slug(request)  # can be empty
         c.backends = BACKENDS.keys()
         self.cut_off_limit = safe_int(config.get('cut_off_limit'))
         c.my_pr_count = PullRequest.query(reviewer_id=request.authuser.user_id, include_closed=False).count()
         self.scm_model = ScmModel()
     @staticmethod
     def _determine_auth_user(api_key, bearer_token, session_authuser, ip_addr):
         """
         Create an `AuthUser` object given the API key/bearer token
         (if any) and the value of the authuser session cookie.
         Returns None if no valid user is found (like not active or no access for IP).
         """
         # Authenticate by bearer token
         if bearer_token is not None:
             api_key = bearer_token
         # Authenticate by API key
         if api_key is not None:
             dbuser = User.get_by_api_key(api_key)
             au = AuthUser.make(dbuser=dbuser, authenticating_api_key=api_key, is_external_auth=True, ip_addr=ip_addr)
             if au is None or au.is_anonymous:
                 log.warning('API key ****%s is NOT valid', api_key[-4:])
                 raise webob.exc.HTTPForbidden(_('Invalid API key'))
             return au
             if dbuser is None:
                 log.info('No db user found for authentication with API key ****%s from %s',
                          api_key[-4:], ip_addr)
                 return None
             return AuthUser.make(dbuser=dbuser, authenticating_api_key=api_key, is_external_auth=True, ip_addr=ip_addr)
         # Authenticate by session cookie
         # In ancient login sessions, 'authuser' may not be a dict.
         # In that case, the user will have to log in again.
         # v0.3 and earlier included an 'is_authenticated' key; if present,
         # this must be True.
         if isinstance(session_authuser, dict) and session_authuser.get('is_authenticated', True):
             return AuthUser.from_cookie(session_authuser, ip_addr=ip_addr)
         # Authenticate by auth_container plugin (if enabled)
         if any(
             plugin.is_container_auth
             for plugin in auth_modules.get_auth_plugins()
         ):
             try:
                 user_info = auth_modules.authenticate('', '', request.environ)
             except UserCreationError as e:
                 from kallithea.lib import helpers as h
                 h.flash(e, 'error', logf=log.error)
             else:
                 if user_info is not None:
                     username = user_info['username']
                     user = User.get_by_username(username, case_insensitive=True)
                     return log_in_user(user, remember=False, is_external_auth=True, ip_addr=ip_addr)

kallithea/model/db.py

➞

Show inline comments

@@ @@ -577,48 +577,50 @@ class User(Base, BaseDbModel): @@
             q = q.options(FromCache(
                             "sql_cache_short",
                             "get_user_%s" % _hash_key(username)
+                          )
+            )
         return q.scalar()
     @classmethod
     def get_by_api_key(cls, api_key, cache=False, fallback=True):
         if len(api_key) != 40 or not api_key.isalnum():
             return None
         q = cls.query().filter(cls.api_key == api_key)
         if cache:
             q = q.options(FromCache("sql_cache_short",
                                     "get_api_key_%s" % api_key))
         res = q.scalar()
         if fallback and not res:
             # fallback to additional keys
             _res = UserApiKeys.query().filter_by(api_key=api_key, is_expired=False).first()
             if _res:
                 res = _res.user
         if res is None or not res.active or res.is_default_user:
             return None
         return res
     @classmethod
     def get_by_email(cls, email, cache=False):
         q = cls.query().filter(func.lower(cls.email) == func.lower(email))
         if cache:
             q = q.options(FromCache("sql_cache_short",
                                     "get_email_key_%s" % email))
         ret = q.scalar()
         if ret is None:
             q = UserEmailMap.query()
             # try fetching in alternate email map
             q = q.filter(func.lower(UserEmailMap.email) == func.lower(email))
             q = q.options(joinedload(UserEmailMap.user))
             if cache:
                 q = q.options(FromCache("sql_cache_short",
                                         "get_email_map_key_%s" % email))
             ret = getattr(q.scalar(), 'user', None)
         return ret
     @classmethod

0 comments (0 inline, 0 general)