Changeset 10055 for main/waeup.kofa/trunk/src

Timestamp:

4 Apr 2013, 15:12:43 (12 years ago)

Author:

uli

Message:

Provide infrastructure to remember failed logins.

Location:

main/waeup.kofa/trunk/src/waeup/kofa

Files:

• authentication.py (modified) (9 diffs)
• interfaces.py (modified) (2 diffs)
• students/authentication.py (modified) (3 diffs)
• tests/test_authentication.py (modified) (2 diffs)

main/waeup.kofa/trunk/src/waeup/kofa/authentication.py

@@ -19 +19 @@
 """
 import grok
+import time
 from zope.event import notify
 from zope.component import getUtility, getUtilitiesFor
 from zope.component.interfaces import IFactory
-from zope.interface import Interface
+from zope.interface import Interface, implementedBy
 from zope.schema import getFields
 from zope.securitypolicy.interfaces import (
@@ -37 +38 @@
     IUserAccount, IAuthPluginUtility, IPasswordValidator,
     IKofaPrincipal, IKofaPrincipalInfo, IKofaPluggable,
-    IBatchProcessor, IGNORE_MARKER)
+    IBatchProcessor, IGNORE_MARKER, IFailedLoginInfo)
 from waeup.kofa.utils.batching import BatchProcessor
 
@@ -84 +85 @@
     grok.implements(IKofaPrincipalInfo)
 
-    def __init__(self, id, title, description, email, phone, public_name, user_type):
+    def __init__(self, id, title, description, email, phone, public_name,
+                 user_type):
         self.id = id
         self.title = title
@@ -95 +97 @@
         self.authenticatorPlugin = None
 
+    def __eq__(self, obj):
+        default = object()
+        result = []
+        for name in ('id', 'title', 'description', 'email', 'phone',
+                     'public_name', 'user_type', 'credentialsPlugin',
+                     'authenticatorPlugin'):
+            result.append(
+                getattr(self, name) == getattr(obj, name, default))
+        return False not in result
+
 class KofaPrincipal(Principal):
     """A portal principal.
 
-    Kofa principals provide an extra `email`, `phone`, `public_name` and `user_type`
-    attribute extending ordinary principals.
+    Kofa principals provide an extra `email`, `phone`, `public_name`
+    and `user_type` attribute extending ordinary principals.
     """
 
@@ -151 +163 @@
         return principal
 
+class FailedLoginInfo(grok.Model):
+    grok.implements(IFailedLoginInfo)
+
+    def __init__(self, num=0, last=None):
+        self.num = num
+        self.last = last
+        return
+
+    def as_tuple(self):
+        return (self.num, self.last)
+
+    def set_values(self, num=0, last=None):
+        self.num, self.last = num, last
+        self._p_changed = True
+        pass
+
+    def increase(self):
+        self.set_values(num=self.num + 1, last=time.time())
+        pass
+
+    def reset(self):
+        self.set_values(num=0, last=None)
+        pass
+
 class Account(grok.Model):
+    """Kofa user accounts store infos about a user.
+
+    Beside the usual data and an (encrypted) password, accounts also
+    have a persistent attribute `failed_logins` which is an instance
+    of `waeup.kofa.authentication.FailedLoginInfo`.
+
+    This attribute can be manipulated directly (set new value,
+    increase values, or reset).
+    """
     grok.implements(IUserAccount)
-
-    _local_roles = dict()
 
     def __init__(self, name, password, title=None, description=None,
@@ -168 +211 @@
         self.setPassword(password)
         self.setSiteRolesForPrincipal(roles)
+
         # We don't want to share this dict with other accounts
         self._local_roles = dict()
+        self.failed_logins = FailedLoginInfo()
 
     def setPassword(self, password):
@@ -235 +280 @@
         if account is None:
             return None
+        # The following shows how 'time penalties' could be enforced
+        # on failed logins. First three failed logins are 'for
+        # free'. After that the user has to wait for 1, 2, 4, 8, 16,
+        # 32, ... seconds before a login can succeed.
+        # There are, however, some problems to discuss, before we
+        # really use this in all authenticators.
+
+        #num, last = account.failed_logins.as_tuple()
+        #if (num > 2) and (time.time() < (last + 2**(num-3))):
+        #    # tried login while account still blocked due to previous
+        #    # login errors.
+        #    return None
         if not account.checkPassword(credentials['password']):
+            #account.failed_logins.increase()
             return None
         return KofaPrincipalInfo(
@@ -458 +516 @@
     """A plugin that updates users.
     """
-
     grok.implements(IKofaPluggable)
     grok.name('users')
@@ -478 +535 @@
                         'UsersPlugin: %s attribute %s added.' % (
                         user.name,i[0]))
+            if not hasattr(user, 'failed_logins'):
+                # add attribute `failed_logins`...
+                user.failed_logins = FailedLoginInfo()
+                logger.info(
+                    'UsersPlugin: attribute failed_logins added.')
             # Remove deprecated attributes
             for i in self.deprecated_attributes:

main/waeup.kofa/trunk/src/waeup/kofa/interfaces.py

@@ -446 +446 @@
         required = False,)
 
+class IFailedLoginInfo(IKofaObject):
+    """Info about failed logins.
+
+    Timestamps are supposed to be stored as floats using time.time()
+    or similar.
+    """
+    num = schema.Int(
+        title = _(u'Number of failed logins'),
+        description = _(u'Number of failed logins'),
+        required = True,
+        default = 0,
+        )
+
+    last = schema.Float(
+        title = _(u'Timestamp'),
+        description = _(u'Timestamp of last failed login or `None`'),
+        required = False,
+        default = None,
+        )
+
+    def as_tuple():
+        """Get login info as tuple ``<NUM>, <TIMESTAMP>``.
+        """
+
+    def set_values(num=0, last=None):
+        """Set number of failed logins and timestamp of last one.
+        """
+
+    def increase():
+        """Increase the current number of failed logins and set timestamp.
+        """
+
+    def reset():
+        """Set failed login counters back to zero.
+        """
+
+
 class IUserAccount(IKofaObject):
     """A user account.
     """
+
+    failed_logins = Attribute("""FailedLoginInfo for this account""")
+
     name = schema.TextLine(
         title = _(u'User Id'),
@@ -486 +526 @@
         required = False,
         )
+
+
 
 class IPasswordValidator(Interface):

main/waeup.kofa/trunk/src/waeup/kofa/students/authentication.py

@@ -20 +20 @@
 """
 import grok
+import time
 from zope.component import getUtility
 from zope.password.interfaces import IPasswordManager
@@ -29 +30 @@
 from zope.session.interfaces import ISession
 from waeup.kofa.authentication import (
-    KofaPrincipalInfo, get_principal_role_manager)
+    KofaPrincipalInfo, get_principal_role_manager, FailedLoginInfo)
 from waeup.kofa.interfaces import (
     IAuthPluginUtility, IUserAccount, IPasswordValidator)
@@ -69 +70 @@
     def description(self):
         return self.title
+
+    @property
+    def failed_logins(self):
+        if not hasattr(self.context, 'failed_logins'):
+            self.context.failed_logins = FailedLoginInfo()
+        return self.context.failed_logins
 
     def _get_roles(self):

main/waeup.kofa/trunk/src/waeup/kofa/tests/test_authentication.py

@@ -17 +17 @@
 ##
 import grok
+import logging
+import time
 import unittest
+from cStringIO import StringIO
+from zope.component import getGlobalSiteManager
 from zope.component.hooks import setSite, clearSite
 from zope.interface.verify import verifyClass, verifyObject
+from zope.password.testing import setUpPasswordManagers
 from zope.pluggableauth.interfaces import IAuthenticatorPlugin
 from zope.securitypolicy.interfaces import IPrincipalRoleManager
 from waeup.kofa.testing import FunctionalTestCase, FunctionalLayer
 from waeup.kofa.authentication import (
-    UserAuthenticatorPlugin, Account, KofaPrincipalInfo,
-    get_principal_role_manager)
+    UserAuthenticatorPlugin, Account, KofaPrincipalInfo, FailedLoginInfo,
+    get_principal_role_manager, UsersPlugin,)
+from waeup.kofa.interfaces import (
+    IUserAccount, IFailedLoginInfo, IKofaPrincipalInfo, IKofaPluggable)
 
 class FakeSite(grok.Site, grok.Container):
+    #def getSiteManager(self):
+    #    return None
+    #    return getGlobalSiteManager()
     pass
 
@@ -87 +97 @@
         self.assertTrue(hasattr(prm2, '_context') is False)
         return
+
+    def make_failed_logins(self, num):
+        # do `num` failed logins and a valid one afterwards
+        del self.site['users']
+        self.site['users'] = {'bob': Account('bob', 'secret')}
+        plugin = UserAuthenticatorPlugin()
+        resultlist = []
+        # reset accounts
+        for x in range(num):
+            resultlist.append(plugin.authenticateCredentials(
+                dict(login='bob', password='wrongsecret')))
+        resultlist.append(plugin.authenticateCredentials(
+            dict(login='bob', password='secret')))
+        return resultlist
+
+    def DISABLED_test_failed_logins(self):
+        # after three failed logins, an account is blocked
+        # XXX: this tests authenticator with time penalty (currently
+        # disabled)
+        results = []
+        succ_principal = KofaPrincipalInfo(
+            id='bob',
+            title='bob',
+            description=None,
+            email=None,
+            phone=None,
+            public_name=None,
+            user_type=u'user')
+        for x in range(4):
+            results.append(self.make_failed_logins(x))
+        self.assertEqual(results[2], [None, None, succ_principal])
+        # last login was blocked although correctly entered due to
+        # time penalty
+        self.assertEqual(results[3], [None, None, None, None])
+        return
+
+class KofaPrincipalInfoTests(unittest.TestCase):
+
+    def create_info(self):
+        return KofaPrincipalInfo(
+            id='bob',
+            title='bob',
+            description=None,
+            email=None,
+            phone=None,
+            public_name=None,
+            user_type=u'user')
+
+    def test_iface(self):
+        # make sure we implement the promised interfaces
+        info = self.create_info()
+        verifyClass(IKofaPrincipalInfo, KofaPrincipalInfo)
+        verifyObject(IKofaPrincipalInfo, info)
+        return
+
+    def test_equality(self):
+        # we can test two infos for equality
+        info1 = self.create_info()
+        info2 = self.create_info()
+        self.assertEqual(info1, info2)
+        self.assertTrue(info1 == info2)
+        info1.id = 'blah'
+        self.assertTrue(info1 != info2)
+        self.assertTrue((info1 == info2) is False)
+        info1.id = 'bob'
+        info2.id = 'blah'
+        self.assertTrue(info1 != info2)
+        self.assertTrue((info1 == info2) is False)
+        return
+
+class FailedLoginInfoTests(unittest.TestCase):
+
+    def test_iface(self):
+        # make sure we fullfill the promised interfaces
+        info1 = FailedLoginInfo()
+        info2 = FailedLoginInfo(num=1, last=time.time())
+        self.assertTrue(
+            verifyClass(IFailedLoginInfo, FailedLoginInfo))
+        self.assertTrue(verifyObject(IFailedLoginInfo, info1))
+        # make sure the stored values have correct type if not None
+        self.assertTrue(verifyObject(IFailedLoginInfo, info2))
+        return
+
+    def test_default_values(self):
+        # By default we get 0, None
+        info = FailedLoginInfo()
+        self.assertEqual(info.num, 0)
+        self.assertEqual(info.last, None)
+        return
+
+    def test_set_values_by_attribute(self):
+        # we can set values by attribute
+        ts = time.gmtime(0)
+        info = FailedLoginInfo()
+        info.num = 5
+        info.last = ts
+        self.assertEqual(info.num, 5)
+        self.assertEqual(info.last, ts)
+        return
+
+    def test_set_values_by_constructor(self):
+        # we can set values by constructor args
+        ts = time.gmtime(0)
+        info = FailedLoginInfo(5, ts)
+        self.assertEqual(info.num, 5)
+        self.assertEqual(info.last, ts)
+        return
+
+    def test_set_values_by_keywords(self):
+        # we can set values by constructor keywords
+        ts = time.gmtime(0)
+        info = FailedLoginInfo(last=ts, num=3)
+        self.assertEqual(info.num, 3)
+        self.assertEqual(info.last, ts)
+        return
+
+    def test_as_tuple(self):
+        # we can get the info values as tuple
+        ts = time.gmtime(0)
+        info = FailedLoginInfo(last=ts, num=3)
+        self.assertEqual(info.as_tuple(), (3, ts))
+        return
+
+    def test_set_values(self):
+        # we can set the values of a an info instance
+        ts = time.time()
+        info = FailedLoginInfo()
+        info.set_values(num=3, last=ts)
+        self.assertEqual(info.num, 3)
+        self.assertEqual(info.last, ts)
+        return
+
+    def test_increase(self):
+        # we can increase the number of failed logins
+        ts1 = time.time()
+        info = FailedLoginInfo()
+        info.increase()
+        self.assertEqual(info.num, 1)
+        self.assertTrue(info.last > ts1)
+        ts2 = info.last
+        info.increase()
+        self.assertEqual(info.num, 2)
+        self.assertTrue(info.last > ts2)
+        return
+
+    def test_reset(self):
+        # we can reset failed login infos.
+        info = FailedLoginInfo()
+        info.increase()
+        info.reset()
+        self.assertEqual(info.num, 0)
+        self.assertEqual(info.last, None)
+        return
+
+class AccountTests(unittest.TestCase):
+
+    def setUp(self):
+        setUpPasswordManagers()
+        return
+
+    def test_iface(self):
+        acct = Account('bob', 'mypasswd')
+        self.assertTrue(
+            verifyClass(IUserAccount, Account))
+        self.assertTrue(
+            verifyObject(IUserAccount, acct))
+        return
+
+    def test_failed_logins(self):
+        # we can retrieve infos about failed logins
+        ts = time.time()
+        acct = Account('bob', 'mypasswd')
+        self.assertTrue(hasattr(acct, 'failed_logins'))
+        acct.failed_logins.set_values(num=3, last=ts)
+        self.assertEqual(acct.failed_logins.last, ts)
+        self.assertEqual(acct.failed_logins.num, 3)
+        return
+
+    def test_failed_logins_per_inst(self):
+        # we get a different counter for each Account instance
+        acct1 = Account('bob', 'secret')
+        acct2 = Account('alice', 'alsosecret')
+        self.assertTrue(acct1.failed_logins is not acct2.failed_logins)
+        return
+
+class FakeUserAccount(object):
+    pass
+
+class UsersPluginTests(unittest.TestCase):
+
+    def setUp(self):
+        setUpPasswordManagers()
+        self.site = FakeSite()
+        self.site['users'] = grok.Container()
+        return
+
+    def get_logger(self):
+        logger = logging.getLogger('waeup.test')
+        stream = StringIO()
+        handler = logging.StreamHandler(stream)
+        logger.setLevel(logging.DEBUG)
+        logger.propagate = False
+        logger.addHandler(handler)
+        return logger, stream
+
+    def test_ifaces(self):
+        # make sure we implement the promised interfaces
+        plugin = UsersPlugin()
+        verifyClass(IKofaPluggable, UsersPlugin)
+        verifyObject(IKofaPluggable, plugin)
+        return
+
+    def test_update(self):
+        # make sure user accounts are updated properly.
+        plugin = UsersPlugin()
+        logger, stream = self.get_logger()
+        plugin.update(self.site, 'app', logger)
+        stream.seek(0)
+        self.assertEqual(stream.read(), '')
+        self.site['users']['bob'] = FakeUserAccount()
+        logger, stream = self.get_logger()
+        plugin.update(self.site, 'app', logger)
+        stream.seek(0)
+        log_content = stream.read()
+        self.assertTrue(hasattr(self.site['users']['bob'], 'description'))
+        self.assertTrue(hasattr(self.site['users']['bob'], 'failed_logins'))
+        self.assertTrue(
+            isinstance(self.site['users']['bob'].failed_logins,
+                       FailedLoginInfo))
+        self.assertTrue('attribute description added' in log_content)
+        self.assertTrue('attribute failed_logins added' in log_content)
+        return