- Add API functions for authenticated_userid and effective_principals.

3 files changed, 78 insertions, 15 deletions

diff --git a/repoze/bfg/interfaces.py b/repoze/bfg/interfaces.py
index 710b91074..fee7b8d61 100644
--- a/repoze/bfg/interfaces.py
+++ b/repoze/bfg/interfaces.py
@@ -46,6 +46,15 @@ class ISecurityPolicy(Interface):
             information in the context and the authentication data in
             the request allow the action implied by the permission """
 
+    def authenticated_userid(request):
+        """ Return the userid of the currently authenticated user or
+        None if there is no currently authenticated user """
+
+    def effective_principals(request):
+        """ Return the list of 'effective' principals for the request.
+        This must include the userid of the currently authenticated
+        user if a user is currently authenticated."""
+
 class NoAuthorizationInformation(Exception):
     pass
 
diff --git a/repoze/bfg/security.py b/repoze/bfg/security.py
index 7e0ba6ffe..5ab6ae31d 100644
--- a/repoze/bfg/security.py
+++ b/repoze/bfg/security.py
@@ -27,6 +27,25 @@ def has_permission(permission, context, request):
         return True
     return policy.permits(context, request, permission)
 
+def authenticated_userid(request):
+    """ Return the userid of the currently authenticated user or None
+    if there is no security policy in effect or there is no currently
+    authenticated user """
+    policy = queryUtility(ISecurityPolicy)
+    if policy is None:
+        return None
+    return policy.authenticated_userid(request)
+
+def effective_principals(request):
+    """ Return the list of 'effective' principals for the request.
+    This will include the userid of the currently authenticated user
+    if a user is currently authenticated. If no security policy is in
+    effect, this will return an empty sequence."""
+    policy = queryUtility(ISecurityPolicy)
+    if policy is None:
+        return []
+    return policy.effective_principals(request)
+
 class ACLAuthorizer(object):
 
     def __init__(self, context, logger=None):
@@ -60,7 +79,6 @@ class ACLAuthorizer(object):
         result = Denied(None, acl, permission, principals, self.context)
         self.logger and self.logger.debug(str(result))
         return result
-        
 
 class RemoteUserACLSecurityPolicy(object):
     """ A security policy which:
@@ -100,14 +118,9 @@ class RemoteUserACLSecurityPolicy(object):
         return False
 
     def authenticated_userid(self, request):
-        """ Return the id of the currently authenticated user or
-        None if the user is not authenticated """
         return request.environ.get('REMOTE_USER', None)
 
     def effective_principals(self, request):
-        """ Return the list of 'effective' principals for the request.
-        This will include the userid of the currently authenticated
-        user if a user is currently authenticated. """
         userid = self.authenticated_userid(request)
         effective_principals = [Everyone]
 
@@ -116,7 +129,6 @@ class RemoteUserACLSecurityPolicy(object):
             effective_principals.append(userid)
         return effective_principals
 
-
 class PermitsResult:
     def __init__(self, ace, acl, permission, principals, context):
         self.acl = acl
diff --git a/repoze/bfg/tests/test_security.py b/repoze/bfg/tests/test_security.py
index d0bb5dcaa..6d85e2160 100644
--- a/repoze/bfg/tests/test_security.py
+++ b/repoze/bfg/tests/test_security.py
@@ -225,6 +225,17 @@ class RemoteUserACLSecurityPolicy(unittest.TestCase, PlacelessSetup):
     def tearDown(self):
         PlacelessSetup.tearDown(self)
 
+    def test_instance_implements_ISecurityPolicy(self):
+        from zope.interface.verify import verifyObject
+        from repoze.bfg.interfaces import ISecurityPolicy
+        logger = DummyLogger()
+        verifyObject(ISecurityPolicy, self._makeOne(logger))
+
+    def test_class_implements_ISecurityPolicy(self):
+        from zope.interface.verify import verifyClass
+        from repoze.bfg.interfaces import ISecurityPolicy
+        verifyClass(ISecurityPolicy, self._getTargetClass())
+
     def test_authenticated_userid(self):
         context = DummyContext()
         request = DummyRequest({'REMOTE_USER':'fred'})
@@ -311,10 +322,12 @@ class RemoteUserACLSecurityPolicy(unittest.TestCase, PlacelessSetup):
         self.assertEqual(authorizer_factory.permission, 'view')
         self.assertEqual(authorizer_factory.context, context)
 
-class TestHasPermission(unittest.TestCase):
-    def _getFUT(self):
-        from repoze.bfg.security import has_permission
-        return has_permission
+class TestAPIFunctions(unittest.TestCase, PlacelessSetup):
+    def setUp(self):
+        PlacelessSetup.setUp(self)
+        
+    def tearDown(self):
+        PlacelessSetup.tearDown(self)
 
     def _registerSecurityPolicy(self, secpol):
         import zope.component
@@ -322,16 +335,39 @@ class TestHasPermission(unittest.TestCase):
         from repoze.bfg.interfaces import ISecurityPolicy
         gsm.registerUtility(secpol, ISecurityPolicy)
 
-    def test_registered(self):
+    def test_has_permission_registered(self):
         secpol = DummySecurityPolicy(False)
         self._registerSecurityPolicy(secpol)
-        has_permission = self._getFUT()
+        from repoze.bfg.security import has_permission
         self.assertEqual(has_permission('view', None, None), False)
         
-    def test_not_registered(self):
-        has_permission = self._getFUT()
+    def test_has_permission_not_registered(self):
+        from repoze.bfg.security import has_permission
         self.assertEqual(has_permission('view', None, None), True)
 
+    def test_authenticated_userid_registered(self):
+        secpol = DummySecurityPolicy(False)
+        self._registerSecurityPolicy(secpol)
+        from repoze.bfg.security import authenticated_userid
+        request = DummyRequest({})
+        self.assertEqual(authenticated_userid(request), 'fred')
+        
+    def test_authenticated_userid_not_registered(self):
+        from repoze.bfg.security import authenticated_userid
+        request = DummyRequest({})
+        self.assertEqual(authenticated_userid(request), None)
+
+    def test_effective_principals_registered(self):
+        secpol = DummySecurityPolicy(False)
+        self._registerSecurityPolicy(secpol)
+        from repoze.bfg.security import effective_principals
+        request = DummyRequest({})
+        self.assertEqual(effective_principals(request), ['fred', 'bob'])
+        
+    def test_effective_principals_not_registered(self):
+        from repoze.bfg.security import effective_principals
+        request = DummyRequest({})
+        self.assertEqual(effective_principals(request), [])
 
 class TestViewPermission(unittest.TestCase):
     def _getTargetClass(self):
@@ -384,6 +420,12 @@ class DummySecurityPolicy:
         self.checked = args
         return self.result
 
+    def authenticated_userid(self, request):
+        return 'fred'
+
+    def effective_principals(self, request):
+        return ['fred', 'bob']
+
 class DummyLogger:
     def __init__(self):
         self.messages = []