Merge pull request #3465 from luhn/security-policy

Security policy implementation

1 files changed, 124 insertions, 51 deletions

diff --git a/tests/test_security.py b/tests/test_security.py
index 8b8028f61..2a8847f3b 100644
--- a/tests/test_security.py
+++ b/tests/test_security.py
@@ -187,32 +187,22 @@ class TestRemember(unittest.TestCase):
 
         return remember(*arg, **kwarg)
 
-    def test_no_authentication_policy(self):
+    def test_no_security_policy(self):
         request = _makeRequest()
         result = self._callFUT(request, 'me')
         self.assertEqual(result, [])
 
-    def test_with_authentication_policy(self):
+    def test_with_security_policy(self):
         request = _makeRequest()
         registry = request.registry
-        _registerAuthenticationPolicy(registry, 'yo')
-        result = self._callFUT(request, 'me')
-        self.assertEqual(result, [('X-Pyramid-Test', 'me')])
-
-    def test_with_authentication_policy_no_reg_on_request(self):
-        from pyramid.threadlocal import get_current_registry
-
-        registry = get_current_registry()
-        request = _makeRequest()
-        del request.registry
-        _registerAuthenticationPolicy(registry, 'yo')
+        _registerSecurityPolicy(registry, 'yo')
         result = self._callFUT(request, 'me')
         self.assertEqual(result, [('X-Pyramid-Test', 'me')])
 
     def test_with_missing_arg(self):
         request = _makeRequest()
         registry = request.registry
-        _registerAuthenticationPolicy(registry, 'yo')
+        _registerSecurityPolicy(registry, 'yo')
         self.assertRaises(TypeError, lambda: self._callFUT(request))
 
 
@@ -228,24 +218,14 @@ class TestForget(unittest.TestCase):
 
         return forget(*arg)
 
-    def test_no_authentication_policy(self):
+    def test_no_security_policy(self):
         request = _makeRequest()
         result = self._callFUT(request)
         self.assertEqual(result, [])
 
-    def test_with_authentication_policy(self):
+    def test_with_security_policy(self):
         request = _makeRequest()
-        _registerAuthenticationPolicy(request.registry, 'yo')
-        result = self._callFUT(request)
-        self.assertEqual(result, [('X-Pyramid-Test', 'logout')])
-
-    def test_with_authentication_policy_no_reg_on_request(self):
-        from pyramid.threadlocal import get_current_registry
-
-        registry = get_current_registry()
-        request = _makeRequest()
-        del request.registry
-        _registerAuthenticationPolicy(registry, 'yo')
+        _registerSecurityPolicy(request.registry, 'yo')
         result = self._callFUT(request)
         self.assertEqual(result, [('X-Pyramid-Test', 'logout')])
 
@@ -338,6 +318,23 @@ class TestViewExecutionPermitted(unittest.TestCase):
         self.assertTrue(result)
 
 
+class TestAuthenticatedIdentity(unittest.TestCase):
+    def setUp(self):
+        testing.setUp()
+
+    def tearDown(self):
+        testing.tearDown()
+
+    def test_identity_no_security_policy(self):
+        request = _makeRequest()
+        self.assertEquals(request.authenticated_identity, None)
+
+    def test_identity(self):
+        request = _makeRequest()
+        _registerSecurityPolicy(request.registry, 'yo')
+        self.assertEqual(request.authenticated_identity, 'yo')
+
+
 class TestAuthenticatedUserId(unittest.TestCase):
     def setUp(self):
         testing.setUp()
@@ -352,8 +349,15 @@ class TestAuthenticatedUserId(unittest.TestCase):
     def test_with_authentication_policy(self):
         request = _makeRequest()
         _registerAuthenticationPolicy(request.registry, 'yo')
+        _registerSecurityPolicy(request.registry, 'wat')
         self.assertEqual(request.authenticated_userid, 'yo')
 
+    def test_with_security_policy(self):
+        request = _makeRequest()
+        # Ensure the identity is stringified.
+        _registerSecurityPolicy(request.registry, 123)
+        self.assertEqual(request.authenticated_userid, '123')
+
     def test_with_authentication_policy_no_reg_on_request(self):
         from pyramid.threadlocal import get_current_registry
 
@@ -378,6 +382,12 @@ class TestUnAuthenticatedUserId(unittest.TestCase):
     def test_with_authentication_policy(self):
         request = _makeRequest()
         _registerAuthenticationPolicy(request.registry, 'yo')
+        _registerSecurityPolicy(request.registry, 'wat')
+        self.assertEqual(request.unauthenticated_userid, 'yo')
+
+    def test_with_security_policy(self):
+        request = _makeRequest()
+        _registerSecurityPolicy(request.registry, 'yo')
         self.assertEqual(request.unauthenticated_userid, 'yo')
 
     def test_with_authentication_policy_no_reg_on_request(self):
@@ -426,43 +436,25 @@ class TestHasPermission(unittest.TestCase):
         testing.tearDown()
 
     def _makeOne(self):
-        from pyramid.security import AuthorizationAPIMixin
+        from pyramid.security import SecurityAPIMixin
         from pyramid.registry import Registry
 
-        mixin = AuthorizationAPIMixin()
+        mixin = SecurityAPIMixin()
         mixin.registry = Registry()
         mixin.context = object()
         return mixin
 
-    def test_no_authentication_policy(self):
+    def test_no_security_policy(self):
         request = self._makeOne()
         result = request.has_permission('view')
         self.assertTrue(result)
-        self.assertEqual(result.msg, 'No authentication policy in use.')
+        self.assertEqual(result.msg, 'No security policy in use.')
 
-    def test_with_no_authorization_policy(self):
+    def test_with_security_registered(self):
         request = self._makeOne()
-        _registerAuthenticationPolicy(request.registry, None)
-        self.assertRaises(
-            ValueError, request.has_permission, 'view', context=None
-        )
-
-    def test_with_authn_and_authz_policies_registered(self):
-        request = self._makeOne()
-        _registerAuthenticationPolicy(request.registry, None)
-        _registerAuthorizationPolicy(request.registry, 'yo')
+        _registerSecurityPolicy(request.registry, 'yo')
         self.assertEqual(request.has_permission('view', context=None), 'yo')
 
-    def test_with_no_reg_on_request(self):
-        from pyramid.threadlocal import get_current_registry
-
-        registry = get_current_registry()
-        request = self._makeOne()
-        del request.registry
-        _registerAuthenticationPolicy(registry, None)
-        _registerAuthorizationPolicy(registry, 'yo')
-        self.assertEqual(request.has_permission('view'), 'yo')
-
     def test_with_no_context_passed(self):
         request = self._makeOne()
         self.assertTrue(request.has_permission('view'))
@@ -473,6 +465,58 @@ class TestHasPermission(unittest.TestCase):
         self.assertRaises(AttributeError, request.has_permission, 'view')
 
 
+class TestLegacySecurityPolicy(unittest.TestCase):
+    def setUp(self):
+        testing.setUp()
+
+    def tearDown(self):
+        testing.tearDown()
+
+    def test_identity(self):
+        from pyramid.security import LegacySecurityPolicy
+
+        request = _makeRequest()
+        policy = LegacySecurityPolicy()
+        _registerAuthenticationPolicy(request.registry, 'userid')
+
+        self.assertEqual(policy.identify(request), 'userid')
+
+    def test_remember(self):
+        from pyramid.security import LegacySecurityPolicy
+
+        request = _makeRequest()
+        policy = LegacySecurityPolicy()
+        _registerAuthenticationPolicy(request.registry, None)
+
+        self.assertEqual(
+            policy.remember(request, 'userid'), [('X-Pyramid-Test', 'userid')]
+        )
+
+    def test_forget(self):
+        from pyramid.security import LegacySecurityPolicy
+
+        request = _makeRequest()
+        policy = LegacySecurityPolicy()
+        _registerAuthenticationPolicy(request.registry, None)
+
+        self.assertEqual(
+            policy.forget(request), [('X-Pyramid-Test', 'logout')]
+        )
+
+    def test_permits(self):
+        from pyramid.security import LegacySecurityPolicy
+
+        request = _makeRequest()
+        policy = LegacySecurityPolicy()
+        _registerAuthenticationPolicy(request.registry, ['p1', 'p2'])
+        _registerAuthorizationPolicy(request.registry, True)
+
+        self.assertIs(
+            policy.permits(request, request.context, 'userid', 'permission'),
+            True,
+        )
+
+
 _TEST_HEADER = 'X-Pyramid-Test'
 
 
@@ -481,6 +525,27 @@ class DummyContext:
         self.__dict__.update(kw)
 
 
+class DummySecurityPolicy:
+    def __init__(self, result):
+        self.result = result
+
+    def identify(self, request):
+        return self.result
+
+    def permits(self, request, context, identity, permission):
+        return self.result
+
+    def remember(self, request, userid, **kw):
+        headers = [(_TEST_HEADER, userid)]
+        self._header_remembered = headers[0]
+        return headers
+
+    def forget(self, request):
+        headers = [(_TEST_HEADER, 'logout')]
+        self._header_forgotten = headers[0]
+        return headers
+
+
 class DummyAuthenticationPolicy:
     def __init__(self, result):
         self.result = result
@@ -516,6 +581,14 @@ class DummyAuthorizationPolicy:
         return self.result
 
 
+def _registerSecurityPolicy(reg, result):
+    from pyramid.interfaces import ISecurityPolicy
+
+    policy = DummySecurityPolicy(result)
+    reg.registerUtility(policy, ISecurityPolicy)
+    return policy
+
+
 def _registerAuthenticationPolicy(reg, result):
     from pyramid.interfaces import IAuthenticationPolicy