Merge pull request #5326 from hypothesis/add-flag-permission

Add `flag` permission to Annotation resource (context)

h/traversal/contexts.py

@@ -22,7 +22,7 @@
 from __future__ import unicode_literals
 
 from pyramid.security import DENY_ALL
-from pyramid.security import Allow
+from pyramid.security import Allow, Everyone, Authenticated
 from pyramid.security import principals_allowed_by_permission
 
 from h.models.organization import ORGANIZATION_DEFAULT_PUBID
@@ -58,8 +58,16 @@ def __acl__(self):
         if self.annotation.shared:
             for principal in self._group_principals(self.group):
                 acl.append((Allow, principal, 'read'))
+                # If this annotation is readable by everyone, it should be
+                # flaggable by any authenticated user, i.e. replace
+                # ``security.Everyone`` with ``security.Authenticated``
+                if principal == Everyone:
+                    acl.append((Allow, Authenticated, 'flag'))
+                else:
+                    acl.append((Allow, principal, 'flag'))
         else:
             acl.append((Allow, self.annotation.userid, 'read'))
+            acl.append((Allow, self.annotation.userid, 'flag'))
 
         for action in ['admin', 'update', 'delete']:
             acl.append((Allow, self.annotation.userid, action))

h/views/api/flags.py

@@ -2,7 +2,6 @@
 
 from __future__ import unicode_literals
 
-from pyramid import security
 from pyramid.httpexceptions import HTTPNoContent
 
 from h.views.api.config import api_config
@@ -16,8 +15,7 @@
             request_method='PUT',
             link_name='annotation.flag',
             description='Flag an annotation for review.',
-            effective_principals=security.Authenticated,
-            permission='read')
+            permission='flag')
 def create(context, request):
     svc = request.find_service(name='flag')
     svc.create(request.user, context.annotation)

tests/functional/api/test_flags.py

@@ -0,0 +1,107 @@
+# -*- coding: utf-8 -*-
+
+from __future__ import unicode_literals
+
+import pytest
+
+# String type for request/response headers and metadata in WSGI.
+#
+# Per PEP-3333, this is intentionally `str` under both Python 2 and 3, even
+# though it has different meanings.
+#
+# See https://www.python.org/dev/peps/pep-3333/#a-note-on-string-types
+native_str = str
+
+
+@pytest.mark.functional
+class TestPutFlag(object):
+    def test_it_returns_http_204_if_user_allowed_to_flag_shared_annotation(self,
+                                                                           app,
+                                                                           annotation,
+                                                                           user_with_token):
+        user, token = user_with_token
+        headers = {'Authorization': str('Bearer {}'.format(token.value))}
+
+        res = app.put('/api/annotations/{id}/flag'.format(id=annotation.id), headers=headers)
+
+        # This annotation was not created by this user but it is a shared annotation
+        assert res.status_code == 204
+
+    def test_it_returns_http_204_if_user_allowed_to_flag_private_annotation(self,
+                                                                            app,
+                                                                            private_annotation,
+                                                                            user_with_token):
+
+        user, token = user_with_token
+        headers = {'Authorization': str('Bearer {}'.format(token.value))}
+
+        res = app.put('/api/annotations/{id}/flag'.format(id=private_annotation.id), headers=headers)
+
+        # This annotation was created by this user
+        assert res.status_code == 204
+
+    def test_it_returns_http_404_if_user_not_allowed_to_flag_private_annotation(self,
+                                                                                app,
+                                                                                unreadable_annotation,
+                                                                                user_with_token):
+
+        user, token = user_with_token
+        headers = {'Authorization': str('Bearer {}'.format(token.value))}
+
+        res = app.put('/api/annotations/{id}/flag'.format(id=unreadable_annotation.id),
+                                                          headers=headers,
+                                                          expect_errors=True)
+
+        # This private annotation was not created by this user
+        assert res.status_code == 404
+
+    def test_it_returns_http_404_if_unauthenticated(self,
+                                                    app,
+                                                    annotation):
+
+        res = app.put('/api/annotations/{id}/flag'.format(id=annotation.id), expect_errors=True)
+
+        assert res.status_code == 404
+
+
+@pytest.fixture
+def annotation(db_session, factories):
+    ann = factories.Annotation(userid='acct:testuser@example.com',
+                               groupid='__world__',
+                               shared=True)
+    db_session.commit()
+    return ann
+
+
+@pytest.fixture
+def private_annotation(db_session, factories, user):
+    ann = factories.Annotation(userid=user.userid,
+                               groupid='__world__',
+                               shared=False)
+    db_session.commit()
+    return ann
+
+
+@pytest.fixture
+def unreadable_annotation(db_session, factories):
+    user = factories.User()
+    ann = factories.Annotation(userid=user.userid,
+                               groupid='__world__',
+                               shared=False)
+    db_session.commit()
+    return ann
+
+
+@pytest.fixture
+def user(db_session, factories):
+    user = factories.User()
+    db_session.commit()
+    return user
+
+
+@pytest.fixture
+def user_with_token(user, db_session, factories):
+    token = factories.DeveloperToken(userid=user.userid)
+    db_session.add(token)
+    db_session.commit()
+    return (user, token)

tests/h/traversal/contexts_test.py

@@ -38,6 +38,7 @@ def test_acl_private(self, factories, group_service, links_service):
         res = AnnotationContext(ann, group_service, links_service)
         actual = res.__acl__()
         expect = [(security.Allow, 'saoirse', 'read'),
+                  (security.Allow, 'saoirse', 'flag'),
                   (security.Allow, 'saoirse', 'admin'),
                   (security.Allow, 'saoirse', 'update'),
                   (security.Allow, 'saoirse', 'delete'),
@@ -87,15 +88,15 @@ def test_acl_deleted(self, factories, group_service, links_service):
         ('unknown-group', 'francis', False),
         ('unknown-group', None, False),
     ])
-    def test_acl_shared(self,
-                        factories,
-                        pyramid_config,
-                        pyramid_request,
-                        groupid,
-                        userid,
-                        permitted,
-                        group_service,
-                        links_service):
+    def test_acl_read_shared(self,
+                            factories,
+                            pyramid_config,
+                            pyramid_request,
+                            groupid,
+                            userid,
+                            permitted,
+                            group_service,
+                            links_service):
         """
         Shared annotation contexts should delegate their 'read' permission to
         their containing group.
@@ -116,6 +117,51 @@ def test_acl_shared(self,
         else:
             assert not pyramid_request.has_permission('read', res)
 
+    @pytest.mark.parametrize('groupid,userid,permitted', [
+        ('freeforall', 'jim', True),
+        ('freeforall', 'saoirse', True),
+        ('freeforall', None, False),
+        ('only-saoirse', 'jim', False),
+        ('only-saoirse', 'saoirse', True),
+        ('only-saoirse', None, False),
+        ('pals', 'jim', True),
+        ('pals', 'saoirse', True),
+        ('pals', 'francis', False),
+        ('pals', None, False),
+        ('unknown-group', 'jim', False),
+        ('unknown-group', 'saoirse', False),
+        ('unknown-group', 'francis', False),
+        ('unknown-group', None, False),
+    ])
+    def test_acl_flag_shared(self,
+                            factories,
+                            pyramid_config,
+                            pyramid_request,
+                            groupid,
+                            userid,
+                            permitted,
+                            group_service,
+                            links_service):
+        """
+        Flag permissions should echo read permissions with the exception that
+        `Security.Everyone` does not get the permission
+        """
+        # Set up the test with a dummy authn policy and a real ACL authz
+        # policy:
+        policy = ACLAuthorizationPolicy()
+        pyramid_config.testing_securitypolicy(userid)
+        pyramid_config.set_authorization_policy(policy)
+
+        ann = factories.Annotation(shared=True,
+                                   userid='mioara',
+                                   groupid=groupid)
+        res = AnnotationContext(ann, group_service, links_service)
+
+        if permitted:
+            assert pyramid_request.has_permission('flag', res)
+        else:
+            assert not pyramid_request.has_permission('flag', res)
+
     @pytest.fixture
     def groups(self):
         return {