Access control failures are logged with WARN level (fixes #1074)

CHANGELOG.rst

@@ -22,6 +22,7 @@ Protocol is now at version **1.16**. See `API changelog`_.
 **New features**
 
 - Enforce the permission endpoint when the admin plugin is included (fixes #1059)
+- Access control failures are logged with WARN level (fixes #1074)
 
 **Bug fixes**
 
@@ -39,6 +40,7 @@ Protocol is now at version **1.16**. See `API changelog`_.
 - Do not keep the whole Kinto Admin bundle in the repo (fixes #1012)
 - Remove the email example from the custom code event listener tutorial (fixes #420)
 - Removed useless logging info from resource (ref #603)
+- Make sure prefixed userid is always first in principals
 
 
 6.0.0 (2017-03-03)

kinto/core/authorization.py

@@ -4,7 +4,7 @@
 from pyramid.security import IAuthorizationPolicy, Authenticated
 from zope.interface import implementer
 
-from kinto.core import utils
+from kinto.core import utils, logger
 from kinto.core.storage import exceptions as storage_exceptions
 
 
@@ -83,17 +83,21 @@ def permits(self, context, principals, permission):
         # The ShareableResource class will take care of the filtering.
         is_list_operation = (context.on_collection and not permission.endswith('create'))
         if not allowed and is_list_operation:
-            shared = context.fetch_shared_records(permission,
-                                                  principals,
-                                                  self.get_bound_permissions)
-            # If allowed to create this kind of object on parent, then allow to obtain the list.
-            if len(bound_perms) > 0:
-                parent_create_perm = [(parent_uri, create_permission)]
-            else:
-                parent_create_perm = [('', 'create')]  # Root object.
-            allowed_to_create = context.check_permission(principals, parent_create_perm)
+            allowed = bool(context.fetch_shared_records(permission,
+                                                        principals,
+                                                        self.get_bound_permissions))
+            if not allowed:
+                # If allowed to create this kind of object on parent,
+                # then allow to obtain the list.
+                if len(bound_perms) > 0:
+                    bound_perms = [(parent_uri, create_permission)]
+                else:
+                    bound_perms = [('', 'create')]  # Root object.
+                allowed = context.check_permission(principals, bound_perms)
 
-            allowed = shared or allowed_to_create
+        if not allowed:
+            logger.warn("{userid} is not allowed to {perm} {uri}",
+                        userid=principals[0], uri=object_id, perm=permission)
 
         return allowed
 

kinto/core/logs.py

@@ -54,8 +54,10 @@ def __call__(self, logger, name, event_dict):
         else:
             pattern = '{event} {context}'
 
-        output = {}
-        for field in ['method', 'path', 'code', 't', 'event']:
+        output = {
+            'event': str(event_dict.pop('event', '?')).format(**event_dict)
+        }
+        for field in ['method', 'path', 'code', 't']:
             output[field] = decode_value(event_dict.pop(field, '?'))
 
         querystring = event_dict.pop('querystring', {})

kinto/core/utils.py

@@ -319,7 +319,7 @@ def prefixed_principals(request):
     principals = [p for p in principals if p != userid]
 
     if request.prefixed_userid not in principals:
-        principals.append(request.prefixed_userid)
+        principals = [request.prefixed_userid] + principals
 
     return principals
 

tests/core/test_authorization.py

@@ -170,7 +170,7 @@ def setUp(self):
         self.context = mock.MagicMock()
         self.context.permission_object_id = '/articles/43/comments/2'
         self.context.required_permission = 'read'
-        self.principals = []
+        self.principals = ["portier:yourself"]
         self.context.get_prefixed_principals.return_value = self.principals
         self.permission = 'dynamic'
 
@@ -182,6 +182,15 @@ def test_permits_return_if_authenticated_when_permission_is_private(self):
                                            ['system.Authenticated'],
                                            'private'))
 
+    def test_permits_logs_authz_failures(self):
+        self.context.on_collection = False
+        self.context.check_permission.return_value = False
+        with mock.patch('kinto.core.authorization.logger') as mocked:
+            self.authz.permits(self.context, self.principals, 'dynamic')
+        mocked.warn.assert_called_with('{userid} is not allowed to {perm} {uri}',
+                                       perm='read', uri='/articles/43/comments/2',
+                                       userid='portier:yourself')
+
     def test_permits_refers_to_context_to_check_permissions(self):
         self.context.check_permission.return_value = True
         allowed = self.authz.permits(self.context, self.principals, 'dynamic')
@@ -190,7 +199,7 @@ def test_permits_refers_to_context_to_check_permissions(self):
     def test_permits_refers_to_context_to_check_permission_principals(self):
         self.context.check_permission.return_value = False
         allowed = self.authz.permits(
-            self.context, ['fxa:user', 'system.Authenticated'], 'dynamic')
+            self.context, self.principals, 'dynamic')
         self.assertTrue(allowed)
 
     def test_permits_reads_the_context_when_permission_is_dynamic(self):
@@ -231,7 +240,7 @@ def test_permits_takes_route_factory_allowed_principals_into_account(self):
         self.context.resource_name = 'record'
         self.context.required_permission = 'create'
         self.context._settings = {'record_create_principals': 'fxa:user'}
-        allowed = self.authz.permits(self.context, ['fxa:user'], 'dynamic')
+        allowed = self.authz.permits(self.context, self.principals, 'dynamic')
         self.context._check_permission.assert_not_called()
         self.assertTrue(allowed)
 
@@ -252,7 +261,7 @@ def test_permits_returns_true_if_collection_and_shared_records(self):
         # Note: we use the list of principals from request.prefixed_principals
         self.context.fetch_shared_records.assert_called_with(
             'read',
-            ['system.Everyone', 'system.Authenticated', 'basicauth:bob'],
+            ['basicauth:bob', 'system.Everyone', 'system.Authenticated'],
             self.authz.get_bound_permissions)
         self.assertTrue(allowed)
 
@@ -274,7 +283,7 @@ def test_permits_returns_false_if_collection_is_unknown(self):
         # Note: we use the list of principals from request.prefixed_principals
         self.context.fetch_shared_records.assert_called_with(
             'read',
-            ['system.Everyone', 'system.Authenticated', 'basicauth:bob'],
+            ['basicauth:bob', 'system.Everyone', 'system.Authenticated'],
             self.authz.get_bound_permissions)
         self.assertFalse(allowed)
 

tests/core/test_utils.py

@@ -142,7 +142,7 @@ def test_removes_unprefixed_from_principals(self):
         request.effective_principals = ['foo', 'system.Authenticated']
         request.prefixed_userid = 'basic:foo'
         self.assertEqual(prefixed_principals(request),
-                         ['system.Authenticated', 'basic:foo'])
+                         ['basic:foo', 'system.Authenticated'])
 
     def test_works_if_userid_is_not_in_principals(self):
         request = DummyRequest()