kallithea Changeset - c0da0ef508da

Changeset - c0da0ef508da

Parent rev.

Child rev.

[Not reviewed]

stable

0 2 0

Mads Kiilerich - 10 years ago 2015-07-07 02:09:35
madski@unity3d.com

auth: only API keys with 40 alpha-numeric characters are valid

This makes it easy to disable API keys in the database without violating the
uniqueness constraint, using something like:

UPDATE users SET api_key='-'||api_key;
UPDATE user_api_keys SET api_key='-'||api_key;

2 files changed with 5 insertions and 0 deletions:

kallithea/model/db.py

kallithea/tests/functional/test_login.py

0 comments (0 inline, 0 general)

kallithea/model/db.py

➞

Show inline comments

@@ @@ -497,96 +497,99 @@ class User(Base, BaseModel): @@
     @property
     def AuthUser(self):
         """
         Returns instance of AuthUser for this user
         """
         from kallithea.lib.auth import AuthUser
         return AuthUser(user_id=self.user_id, api_key=self.api_key,
                         username=self.username)
     @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
     def user_data(self, val):
         try:
             self._user_data = json.dumps(val)
         except Exception:
             log.error(traceback.format_exc())
     def __unicode__(self):
         return u"<%s('id:%s:%s')>" % (self.__class__.__name__,
                                       self.user_id, self.username)
     @classmethod
     def get_by_username(cls, username, case_insensitive=False, cache=False):
         if case_insensitive:
             q = cls.query().filter(cls.username.ilike(username))
         else:
             q = cls.query().filter(cls.username == username)
         if cache:
             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(UserApiKeys.api_key == api_key)\
                 .filter(or_(UserApiKeys.expires == -1,
                             UserApiKeys.expires >= time.time()))\
                 .first()
             if _res:
                 res = _res.user
         return res
     @classmethod
     def get_by_email(cls, email, case_insensitive=False, cache=False):
         if case_insensitive:
             q = cls.query().filter(cls.email.ilike(email))
         else:
             q = cls.query().filter(cls.email == 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
             if case_insensitive:
                 q = q.filter(UserEmailMap.email.ilike(email))
             else:
                 q = q.filter(UserEmailMap.email == 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
     def get_from_cs_author(cls, author):
         """

kallithea/tests/functional/test_login.py

➞

Show inline comments

@@ @@ -280,96 +280,98 @@ class TestLoginController(TestController @@
                                     key=key))
         self.assertEqual(response.status, '302 Found')
         self.assertTrue(response.location.endswith(url('reset_password')))
         # GOOD KEY
         key = User.get_by_username(username).api_key
         response = self.app.get(url(controller='login',
                                     action='password_reset_confirmation',
                                     key=key))
         self.assertEqual(response.status, '302 Found')
         self.assertTrue(response.location.endswith(url('login_home')))
         self.checkSessionFlash(response,
                                ('Your password reset was successful, '
                                 'new password has been sent to your email'))
         response = response.follow()
     def _get_api_whitelist(self, values=None):
         config = {'api_access_controllers_whitelist': values or []}
         return config
     @parameterized.expand([
         ('none', None),
         ('empty_string', ''),
         ('fake_number', '123456'),
         ('proper_api_key', None)
     ])
     def test_access_not_whitelisted_page_via_api_key(self, test_name, api_key):
         whitelist = self._get_api_whitelist([])
         with mock.patch('kallithea.CONFIG', whitelist):
             self.assertEqual([],
                              whitelist['api_access_controllers_whitelist'])
             if test_name == 'proper_api_key':
                 #use builtin if api_key is None
                 api_key = User.get_first_admin().api_key
             with fixture.anon_access(False):
                 self.app.get(url(controller='changeset',
                                  action='changeset_raw',
                                  repo_name=HG_REPO, revision='tip', api_key=api_key),
                              status=302)
     @parameterized.expand([
         ('none', None, 302),
         ('empty_string', '', 302),
         ('fake_number', '123456', 302),
         ('fake_not_alnum', 'a-z', 302),
         ('fake_api_key', '0123456789abcdef0123456789ABCDEF01234567', 302),
         ('proper_api_key', None, 200)
     ])
     def test_access_whitelisted_page_via_api_key(self, test_name, api_key, code):
         whitelist = self._get_api_whitelist(['ChangesetController:changeset_raw'])
         with mock.patch('kallithea.CONFIG', whitelist):
             self.assertEqual(['ChangesetController:changeset_raw'],
                              whitelist['api_access_controllers_whitelist'])
             if test_name == 'proper_api_key':
                 api_key = User.get_first_admin().api_key
             with fixture.anon_access(False):
                 self.app.get(url(controller='changeset',
                                  action='changeset_raw',
                                  repo_name=HG_REPO, revision='tip', api_key=api_key),
                              status=code)
     def test_access_page_via_extra_api_key(self):
         whitelist = self._get_api_whitelist(['ChangesetController:changeset_raw'])
         with mock.patch('kallithea.CONFIG', whitelist):
             self.assertEqual(['ChangesetController:changeset_raw'],
                              whitelist['api_access_controllers_whitelist'])
             new_api_key = ApiKeyModel().create(TEST_USER_ADMIN_LOGIN, u'test')
             Session().commit()
             with fixture.anon_access(False):
                 self.app.get(url(controller='changeset',
                                  action='changeset_raw',
                                  repo_name=HG_REPO, revision='tip', api_key=new_api_key.api_key),
                              status=200)
     def test_access_page_via_expired_api_key(self):
         whitelist = self._get_api_whitelist(['ChangesetController:changeset_raw'])
         with mock.patch('kallithea.CONFIG', whitelist):
             self.assertEqual(['ChangesetController:changeset_raw'],
                              whitelist['api_access_controllers_whitelist'])
             new_api_key = ApiKeyModel().create(TEST_USER_ADMIN_LOGIN, u'test')
             Session().commit()
             #patch the api key and make it expired
             new_api_key.expires = 0
             Session().add(new_api_key)
             Session().commit()
             with fixture.anon_access(False):
                 self.app.get(url(controller='changeset',
                                  action='changeset_raw',
                                  repo_name=HG_REPO, revision='tip',
                                  api_key=new_api_key.api_key),
                              status=302)

0 comments (0 inline, 0 general)