Merge pull request #194 from mmerickel/session_authn

Session-based authentication policy

4 files changed, 141 insertions, 3 deletions

diff --git a/CONTRIBUTORS.txt b/CONTRIBUTORS.txt
index f2195de70..2873ce1b6 100644
--- a/CONTRIBUTORS.txt
+++ b/CONTRIBUTORS.txt
@@ -138,3 +138,5 @@ Contributors
 - Juliusz Gonera, 2011/04/17
 
 - Philip Jenvey, 2011/04/24
+
+- Michael Merickel, 2011/5/25
diff --git a/docs/api/authentication.rst b/docs/api/authentication.rst
index bf7f8f8d5..5d4dbd9e3 100644
--- a/docs/api/authentication.rst
+++ b/docs/api/authentication.rst
@@ -14,6 +14,8 @@ Authentication Policies
 
   .. autoclass:: RemoteUserAuthenticationPolicy
 
+  .. autoclass:: SessionAuthenticationPolicy
+
 Helper Classes
 ~~~~~~~~~~~~~~
 
diff --git a/pyramid/authentication.py b/pyramid/authentication.py
index a6c74e549..e1384e0b8 100644
--- a/pyramid/authentication.py
+++ b/pyramid/authentication.py
@@ -524,4 +524,44 @@ class AuthTktCookieHelper(object):
 
         cookie_value = ticket.cookie_value()
         return self._get_cookies(environ, cookie_value, max_age)
-    
+
+class SessionAuthenticationPolicy(CallbackAuthenticationPolicy):
+    """ A :app:`Pyramid` authentication policy which gets its data from
+    the configured session.
+
+    Constructor Arguments
+
+    ``prefix``
+
+       A prefix used when storing the authentication parameters in the
+       session. Defaults to 'auth.'. Optional.
+
+    ``callback``
+
+       Default: ``None``.  A callback passed the userid and the
+       request, expected to return ``None`` if the userid doesn't
+       exist or a sequence of principal identifiers (possibly empty) if
+       the user does exist.  If ``callback`` is ``None``, the userid
+       will be assumed to exist with no principals.  Optional.
+    """
+    implements(IAuthenticationPolicy)
+
+    def __init__(self, prefix='auth.', callback=None):
+        self.callback = callback
+        self.prefix = prefix or ''
+        self.userid_key = prefix + 'userid'
+
+    def remember(self, request, principal, **kw):
+        """ Store a principal in the session."""
+        request.session[self.userid_key] = principal
+        return []
+
+    def forget(self, request):
+        """ Remove the stored principal from the session."""
+        if self.userid_key in request.session:
+            del request.session[self.userid_key]
+        return []
+
+    def unauthenticated_userid(self, request):
+        return request.session.get(self.userid_key)
+
diff --git a/pyramid/tests/test_authentication.py b/pyramid/tests/test_authentication.py
index ecd76a71c..8acc2b31c 100644
--- a/pyramid/tests/test_authentication.py
+++ b/pyramid/tests/test_authentication.py
@@ -751,12 +751,106 @@ class TestAuthTktCookieHelper(unittest.TestCase):
                          'auth_tkt=""; Path=/; Domain=.localhost; Max-Age=0; '
                          'Expires=Wed, 31-Dec-97 23:59:59 GMT')
 
+
+class TestSessionAuthenticationPolicy(unittest.TestCase):
+    def _getTargetClass(self):
+        from pyramid.authentication import SessionAuthenticationPolicy
+        return SessionAuthenticationPolicy
+
+    def _makeOne(self, callback=None, prefix=''):
+        return self._getTargetClass()(prefix=prefix, callback=callback)
+
+    def test_class_implements_IAuthenticationPolicy(self):
+        from zope.interface.verify import verifyClass
+        from pyramid.interfaces import IAuthenticationPolicy
+        verifyClass(IAuthenticationPolicy, self._getTargetClass())
+
+    def test_instance_implements_IAuthenticationPolicy(self):
+        from zope.interface.verify import verifyObject
+        from pyramid.interfaces import IAuthenticationPolicy
+        verifyObject(IAuthenticationPolicy, self._makeOne())
+
+    def test_unauthenticated_userid_returns_None(self):
+        request = DummyRequest()
+        policy = self._makeOne()
+        self.assertEqual(policy.unauthenticated_userid(request), None)
+
+    def test_unauthenticated_userid(self):
+        request = DummyRequest(session={'userid':'fred'})
+        policy = self._makeOne()
+        self.assertEqual(policy.unauthenticated_userid(request), 'fred')
+
+    def test_authenticated_userid_no_cookie_identity(self):
+        request = DummyRequest()
+        policy = self._makeOne()
+        self.assertEqual(policy.authenticated_userid(request), None)
+
+    def test_authenticated_userid_callback_returns_None(self):
+        request = DummyRequest(session={'userid':'fred'})
+        def callback(userid, request):
+            return None
+        policy = self._makeOne(callback)
+        self.assertEqual(policy.authenticated_userid(request), None)
+
+    def test_authenticated_userid(self):
+        request = DummyRequest(session={'userid':'fred'})
+        def callback(userid, request):
+            return True
+        policy = self._makeOne(callback)
+        self.assertEqual(policy.authenticated_userid(request), 'fred')
+
+    def test_effective_principals_no_identity(self):
+        from pyramid.security import Everyone
+        request = DummyRequest()
+        policy = self._makeOne()
+        self.assertEqual(policy.effective_principals(request), [Everyone])
+
+    def test_effective_principals_callback_returns_None(self):
+        from pyramid.security import Everyone
+        request = DummyRequest(session={'userid':'fred'})
+        def callback(userid, request):
+            return None
+        policy = self._makeOne(callback)
+        self.assertEqual(policy.effective_principals(request), [Everyone])
+
+    def test_effective_principals(self):
+        from pyramid.security import Everyone
+        from pyramid.security import Authenticated
+        request = DummyRequest(session={'userid':'fred'})
+        def callback(userid, request):
+            return ['group.foo']
+        policy = self._makeOne(callback)
+        self.assertEqual(policy.effective_principals(request),
+                         [Everyone, Authenticated, 'fred', 'group.foo'])
+
+    def test_remember(self):
+        request = DummyRequest()
+        policy = self._makeOne()
+        result = policy.remember(request, 'fred')
+        self.assertEqual(request.session.get('userid'), 'fred')
+        self.assertEqual(result, [])
+
+    def test_forget(self):
+        request = DummyRequest(session={'userid':'fred'})
+        policy = self._makeOne()
+        result = policy.forget(request)
+        self.assertEqual(request.session.get('userid'), None)
+        self.assertEqual(result, [])
+
+    def test_forget_no_identity(self):
+        request = DummyRequest()
+        policy = self._makeOne()
+        result = policy.forget(request)
+        self.assertEqual(request.session.get('userid'), None)
+        self.assertEqual(result, [])
+
 class DummyContext:
     pass
 
 class DummyRequest:
-    def __init__(self, environ):
-        self.environ = environ
+    def __init__(self, environ=None, session=None):
+        self.environ = environ or {}
+        self.session = session or {}
         self.callbacks = []
 
     def add_response_callback(self, callback):