kallithea Changeset - 7557da2252a3

Changeset - 7557da2252a3

Parent rev.

Child rev.

[Not reviewed]

default

0 5 0

Søren Løvborg - 10 years ago 2015-07-26 14:10:16
kwi@kwi.dk

auth: construct AuthUser from either user_id or db.User object

If the caller already has the database User object, there's no reason
for AuthUser to look it up again.

The `api_key` lookup functionality is dropped, because 1) it's only
used in one place, and 2) it's simple enough for the caller to do the
lookup itself.

The `user_id` lookup functionality is kept, because 1) it's frequently
used, and 2) far from a simple `User.get(id)` lookup, it has a complex
interaction with UserModel. (That cleanup will have to wait for another
day.)

All calls of the form `AuthUser(user_id=x.user_id)` can be replaced with
`AuthUser(dbuser=x)`, assuming `x` is a db.User. However, verifying that
assumption requires a manual audit of every call site, since `x` might
also be another `AuthUser` object, for instance. Therefore, only the
most obvious call sites have been fixed here.

5 files changed with 17 insertions and 20 deletions:

kallithea/controllers/api/__init__.py

kallithea/lib/auth.py

kallithea/lib/base.py

kallithea/model/db.py

kallithea/tests/api/api_base.py

0 comments (0 inline, 0 general)

kallithea/controllers/api/__init__.py

➞

Show inline comments

@@ @@ -149,25 +149,25 @@ class JSONRPCController(WSGIController): @@
+            )
         except KeyError, e:
             return jsonrpc_error(retid=self._req_id,
                                  message='Incorrect JSON query missing %s' % e)
         # check if we can find this session using api_key
         try:
             u = User.get_by_api_key(self._req_api_key)
             if u is None:
                 return jsonrpc_error(retid=self._req_id,
                                      message='Invalid API key')
-            auth_u = AuthUser(u.user_id)
+            auth_u = AuthUser(dbuser=u)
             if not AuthUser.check_ip_allowed(auth_u, ip_addr):
                 return jsonrpc_error(retid=self._req_id,
                         message='request from IP:%s not allowed' % (ip_addr,))
             else:
                 log.info('Access for IP:%s allowed' % (ip_addr,))
         except Exception, e:
             return jsonrpc_error(retid=self._req_id,
                                  message='Invalid API key')
         self._error = None
         try:

kallithea/lib/auth.py

➞

Show inline comments

@@ @@ -451,72 +451,68 @@ def allowed_api_access(controller_name, @@
         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, API key or cookie dict, it looks
     up the matching database `User` and copies all attributes to itself,
     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 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,
+    def __init__(self, user_id=None, dbuser=None,
             is_external_auth=False):
         self.is_authenticated = False
         self.is_external_auth = is_external_auth
         user_model = UserModel()
         self.anonymous_user = User.get_default_user(cache=True)
         is_user_loaded = False
         # These attributes will be overriden by fill_data, below, unless the
         # requested user cannot be found and the default anonymous user is
         # not enabled.
         self.user_id = None
         self.username = None
         self.api_key = None
         self.name = ''
         self.lastname = ''
         self.email = ''
         self.admin = False
         self.inherit_default_permissions = False
-        # lookup by userid
+        # Look up database user, if necessary.
         if user_id is not None:
             log.debug('Auth User lookup by USER ID %s' % user_id)
             is_user_loaded = user_model.fill_data(self, user_model.get(user_id))
             dbuser = user_model.get(user_id)
         else:
             # Note: dbuser is allowed to be None.
             log.debug('Auth User lookup by database user %s', dbuser)
         # try go get user by API key
         elif api_key:
             log.debug('Auth User lookup by API key %s' % api_key)
             is_user_loaded = user_model.fill_data(self, User.get_by_api_key(api_key))
         else:
             log.debug('No data in %s that could been used to log in' % self)
         is_user_loaded = user_model.fill_data(self, dbuser)
         # If user cannot be found, try falling back to anonymous.
         if not is_user_loaded:
             is_user_loaded =  user_model.fill_data(self, self.anonymous_user)
         # The anonymous user is always "logged in".
         if self.user_id == self.anonymous_user.user_id:
             self.is_authenticated = True
         if not self.username:
             self.username = 'None'

kallithea/lib/base.py

➞

Show inline comments

@@ @@ -106,25 +106,25 @@ def _get_access_path(environ): @@
 def log_in_user(user, remember, is_external_auth):
     """
     Log a `User` in and update session and cookies. If `remember` is True,
     the session cookie is set to expire in a year; otherwise, it expires at
     the end of the browser session.
     Returns populated `AuthUser` object.
     """
     user.update_lastlogin()
     meta.Session().commit()
-    auth_user = AuthUser(user_id=user.user_id,
+    auth_user = AuthUser(dbuser=user,
                          is_external_auth=is_external_auth)
     auth_user.set_authenticated()
     # Start new session to prevent session fixation attacks.
     session.invalidate()
     session['authuser'] = cookie = auth_user.to_cookie()
     # If they want to be remembered, update the cookie
     if remember:
         t = datetime.datetime.now() + datetime.timedelta(days=365)
         session._set_cookie_expires(t)
@@ @@ -376,25 +376,26 @@ class BaseController(WSGIController): @@
         self.scm_model = ScmModel(self.sa)
     @staticmethod
     def _determine_auth_user(api_key, session_authuser):
         """
         Create an `AuthUser` object given the API key (if any) and the
         value of the authuser session cookie.
         """
         # Authenticate by API key
         if api_key:
             # when using API_KEY we are sure user exists.
             return AuthUser(api_key=api_key, is_external_auth=True)
             return AuthUser(dbuser=User.get_by_api_key(api_key),
                             is_external_auth=True)
         # 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.
         if isinstance(session_authuser, dict):
             try:
                 return AuthUser.from_cookie(session_authuser)
             except UserCreationError as e:
                 # container auth or other auth functions that create users on
                 # the fly can throw UserCreationError to signal issues with
                 # user creation. Explanation should be provided in the
                 # exception object.

kallithea/model/db.py

➞

Show inline comments

@@ @@ -502,25 +502,25 @@ class User(Base, BaseModel): @@
         return '%s %s' % (self.firstname, self.lastname)
     @property
     def is_admin(self):
         return self.admin
     @property
     def AuthUser(self):
         """
         Returns instance of AuthUser for this user
         """
         from kallithea.lib.auth import AuthUser
-        return AuthUser(user_id=self.user_id)
+        return AuthUser(dbuser=self)
     @hybrid_property
     def user_data(self):
         if not self._user_data:
             return {}
         try:
             return json.loads(self._user_data)
         except TypeError:
             return {}
     @user_data.setter

kallithea/tests/api/api_base.py

➞

Show inline comments

@@ @@ -226,55 +226,55 @@ class _BaseTestApi(object): @@
             ret = usr.get_api_data()
             ret_all.append(jsonify(ret))
         expected = ret_all
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user(self):
         id_, params = _build_data(self.apikey, 'get_user',
                                   userid=TEST_USER_ADMIN_LOGIN)
         response = api_call(self, params)
         usr = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = usr.get_api_data()
-        ret['permissions'] = AuthUser(usr.user_id).permissions
+        ret['permissions'] = AuthUser(dbuser=usr).permissions
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user_that_does_not_exist(self):
         id_, params = _build_data(self.apikey, 'get_user',
                                   userid='trololo')
         response = api_call(self, params)
         expected = "user `%s` does not exist" % 'trololo'
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_user_without_giving_userid(self):
         id_, params = _build_data(self.apikey, 'get_user')
         response = api_call(self, params)
         usr = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = usr.get_api_data()
-        ret['permissions'] = AuthUser(usr.user_id).permissions
+        ret['permissions'] = AuthUser(dbuser=usr).permissions
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user_without_giving_userid_non_admin(self):
         id_, params = _build_data(self.apikey_regular, 'get_user')
         response = api_call(self, params)
         usr = User.get_by_username(self.TEST_USER_LOGIN)
         ret = usr.get_api_data()
-        ret['permissions'] = AuthUser(usr.user_id).permissions
+        ret['permissions'] = AuthUser(dbuser=usr).permissions
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user_with_giving_userid_non_admin(self):
         id_, params = _build_data(self.apikey_regular, 'get_user',
                                   userid=self.TEST_USER_LOGIN)
         response = api_call(self, params)
         expected = 'userid is not the same as your user'
         self._compare_error(id_, expected, given=response.body)

0 comments (0 inline, 0 general)