permissions: disable item edit and delete buttons for librarians

* Creates separate permission file for the item resource.
* Restricts access to the API item edit, update, delete actions only to librarians of item's affiliated library.
* Creates complete units tests for item permissions.
* Closes #574

Co-Authored-by: Aly Badr <[email protected]>
Co-Authored-by: Olivier DOSSMANN <[email protected]>

rero_ils/config.py

@@ -49,6 +49,8 @@
 from .modules.holdings.api import Holding, HoldingsIndexer
 from .modules.item_types.api import ItemType
 from .modules.items.api import Item, ItemsIndexer
+from .modules.items.permissions import can_create_item_factory, \
+    can_update_delete_item_factory
 from .modules.libraries.api import Library
 from .modules.libraries.permissions import can_create_library_factory, \
     can_delete_library_factory, can_update_library_factory
@@ -397,9 +399,9 @@ def _(x):
         max_result_window=10000,
         search_factory_imp='rero_ils.query:organisation_search_factory',
         read_permission_factory_imp=can_access_organisation_records_factory,
-        create_permission_factory_imp=can_create_organisation_records_factory,
-        update_permission_factory_imp=can_update_organisation_records_factory,
-        delete_permission_factory_imp=can_delete_organisation_records_factory,
+        create_permission_factory_imp=can_create_item_factory,
+        update_permission_factory_imp=can_update_delete_item_factory,
+        delete_permission_factory_imp=can_update_delete_item_factory,
     ),
     itty=dict(
         pid_type='itty',

rero_ils/modules/items/permissions.py

@@ -0,0 +1,67 @@
+# -*- coding: utf-8 -*-
+#
+# RERO ILS
+# Copyright (C) 2019 RERO
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, version 3 of the License.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+"""Item permissions."""
+
+
+from ...permissions import staffer_is_authenticated
+
+
+def can_update_delete_item_factory(record, *args, **kwargs):
+    """Checks if logged user can update or delete its organisation items.
+
+    librarian must have librarian or system_librarian role.
+    librarian can only update, delete items of its affiliated library.
+    sys_librarian can update, delete any item of its org only.
+    """
+    def can(self):
+        patron = staffer_is_authenticated()
+        if patron and patron.organisation_pid == record.organisation_pid:
+            return check_patron_library_permissions(patron, record)
+        return False
+    return type('Check', (), {'can': can})()
+
+
+def can_create_item_factory(record, *args, **kwargs):
+    """Checks if the logged user can create items for its organisation.
+
+    librarian must have librarian or system_librarian role.
+    librarian can only create items in its affiliated library.
+    sys_librarian can create any item of its org only.
+    """
+    def can(self):
+        patron = staffer_is_authenticated()
+        if patron and not record:
+            return True
+        if patron and patron.organisation_pid == record.organisation_pid:
+            return check_patron_library_permissions(patron, record)
+        return False
+    return type('Check', (), {'can': can})()
+
+
+def check_patron_library_permissions(patron, record):
+    """Checks if the logged user can create items for its organisation.
+
+    librarian must have librarian or system_librarian role.
+    librarian can only create, update, delete items in its affiliated library.
+    sys_librarian can create, update, delete any item of its org only.
+    """
+    if not patron.is_system_librarian:
+        if patron.library_pid and \
+                record.library_pid != patron.library_pid:
+            return False
+    return True

tests/api/test_items_rest_views.py

@@ -381,7 +381,9 @@ def test_item_secure_api(client, json_header, item_lib_martigny,
 def test_item_secure_api_create(client, json_header, item_lib_martigny,
                                 librarian_martigny_no_email,
                                 librarian_sion_no_email,
-                                item_lib_martigny_data):
+                                item_lib_martigny_data,
+                                item_lib_saxon_data,
+                                system_librarian_martigny_no_email):
     """Test item secure api create."""
     # Martigny
     login_user_via_session(client, librarian_martigny_no_email.user)
@@ -393,6 +395,25 @@ def test_item_secure_api_create(client, json_header, item_lib_martigny,
         post_url,
         item_lib_martigny_data
     )
+    # librarian can create items on its affilicated library
+    assert res.status_code == 201
+
+    del item_lib_saxon_data['pid']
+    res, _ = postdata(
+        client,
+        post_url,
+        item_lib_saxon_data
+    )
+    # librarian can not create items for another library
+    assert res.status_code == 403
+
+    login_user_via_session(client, system_librarian_martigny_no_email.user)
+    res, _ = postdata(
+        client,
+        post_url,
+        item_lib_saxon_data
+    )
+    # sys_librarian can create items for any library
     assert res.status_code == 201
 
     # Sion
@@ -403,56 +424,95 @@ def test_item_secure_api_create(client, json_header, item_lib_martigny,
         post_url,
         item_lib_martigny_data
     )
+    # librarian can not create items in another organisation
     assert res.status_code == 403
 
 
 def test_item_secure_api_update(client, json_header, item_lib_saxon,
                                 librarian_martigny_no_email,
                                 librarian_sion_no_email,
-                                item_lib_saxon_data
+                                item_lib_martigny,
+                                system_librarian_martigny_no_email
                                 ):
     """Test item secure api update."""
     # Martigny
     login_user_via_session(client, librarian_martigny_no_email.user)
+    record_url = url_for('invenio_records_rest.item_item',
+                         pid_value=item_lib_martigny.pid)
+
+    item_lib_martigny['call_number'] = 'call_number'
+    res = client.put(
+        record_url,
+        data=json.dumps(item_lib_martigny),
+        headers=json_header
+    )
+    # librarian can update items of its affiliated library
+    assert res.status_code == 200
+
     record_url = url_for('invenio_records_rest.item_item',
                          pid_value=item_lib_saxon.pid)
 
-    data = item_lib_saxon
-    data['call_number'] = 'call_number'
+    item_lib_saxon['call_number'] = 'call_number'
+    res = client.put(
+        record_url,
+        data=json.dumps(item_lib_saxon),
+        headers=json_header
+    )
+    # librarian can not update items of other libraries
+    assert res.status_code == 403
+
+    login_user_via_session(client, system_librarian_martigny_no_email.user)
     res = client.put(
         record_url,
-        data=json.dumps(data),
+        data=json.dumps(item_lib_saxon),
         headers=json_header
     )
+    # sys_librarian can update items of other libraries in same organisation.
     assert res.status_code == 200
 
     # Sion
     login_user_via_session(client, librarian_sion_no_email.user)
 
     res = client.put(
         record_url,
-        data=json.dumps(data),
+        data=json.dumps(item_lib_saxon),
         headers=json_header
     )
+    # librarian can not update items of other libraries in other organisation.
     assert res.status_code == 403
 
 
 def test_item_secure_api_delete(client, item_lib_saxon,
                                 librarian_martigny_no_email,
                                 librarian_sion_no_email,
-                                item_lib_saxon_data,
-                                json_header):
+                                item_lib_martigny,
+                                json_header,
+                                system_librarian_martigny_no_email):
     """Test item secure api delete."""
     # Martigny
     login_user_via_session(client, librarian_martigny_no_email.user)
     record_url = url_for('invenio_records_rest.item_item',
-                         pid_value=item_lib_saxon.pid)
+                         pid_value=item_lib_martigny.pid)
 
     res = client.delete(record_url)
+    # librarian can delete items of its affiliated library
     assert res.status_code == 204
 
+    record_url = url_for('invenio_records_rest.item_item',
+                         pid_value=item_lib_saxon.pid)
+
+    res = client.delete(record_url)
+    # librarian can not delete items of other libraries
+    assert res.status_code == 403
+
     # Sion
     login_user_via_session(client, librarian_sion_no_email.user)
 
     res = client.delete(record_url)
-    assert res.status_code == 410
+    # librarian can not delete items of other organisations
+    assert res.status_code == 403
+
+    login_user_via_session(client, system_librarian_martigny_no_email.user)
+    res = client.delete(record_url)
+    # sys_librarian can delete items in other libraries in same org.
+    assert res.status_code == 204

tests/api/test_serializers.py

@@ -134,6 +134,15 @@ def test_items_serializers(
     assert response.status_code == 200
     data = get_json(response)
     assert 'cannot_delete' in data['permissions']
+    assert 'cannot_update' in data['permissions']
+    assert data['metadata'].get('item_type').get('$ref')
+
+    item_url = url_for(
+        'invenio_records_rest.item_item', pid_value=item_lib_martigny.pid)
+    response = client.get(item_url, headers=json_header)
+    assert response.status_code == 200
+    data = get_json(response)
+    assert 'cannot_delete' not in data['permissions']
     assert 'cannot_update' not in data['permissions']
     assert data['metadata'].get('item_type').get('$ref')