fix: 管理系 API の担当範囲を確認し、SWORD の登録可能ロールの設定を効かせる

modules/weko-admin/tests/test_views.py

@@ -617,7 +617,7 @@ def test_resend_failed_mail(api,users,mocker):
 @pytest.mark.parametrize("index,is_permission",[
                          (0,True),# sysadmin
                          (1,True),# repoadmin
-                         (2,True),# comadmin
+                         (2,False),# comadmin (Root Index is not in scope)
                          (3,False),# contributor
                          (4,False),# generaluser
                          ])
@@ -629,6 +629,32 @@ def test_manual_send_site_license_mail_acl(api,users,site_license,index,is_permi
             res = api.post(url,data={"repo_id": "Root Index"})
             assert_role(res, is_permission)
 
+# .tox/c1/bin/pytest --cov=weko_admin tests/test_views.py::test_manual_send_site_license_mail_repository_scope -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp
+@pytest.mark.parametrize("index,repo_id,is_permission",[
+                         (0,"comm1",True),# sysadmin, any repository
+                         (1,"comm1",True),# repoadmin, any repository
+                         (1,"other",True),# repoadmin, any repository
+                         (2,"comm1",True),# comadmin, own repository
+                         (2,"other",False),# comadmin, other repository
+                         (2,"",False),# comadmin, no repository
+                         ])
+def test_manual_send_site_license_mail_repository_scope(api,db,users,site_license,community,index,repo_id,is_permission):
+    if repo_id:
+        site_license[0]["Info"].repository_id = repo_id
+        db.session.commit()
+    url = url_for("weko_admin.manual_send_site_license_mail",start_month="202201",end_month="202203")
+    login_user_via_session(client=api, email=users[index]["email"])
+    with patch("weko_admin.views.QueryCommonReportsHelper.get", return_value={"institution_name":[]}):
+        with patch("weko_admin.views.send_site_license_mail") as mock_send:
+            res = api.post(url,data={"repo_id": repo_id})
+            if is_permission:
+                assert res.status_code == 200
+                assert res.data == b"finished"
+                mock_send.assert_called_once()
+            else:
+                assert res.status_code == 403
+                mock_send.assert_not_called()
+
 # .tox/c1/bin/pytest --cov=weko_admin tests/test_views.py::test_manual_send_site_license_mail_guest -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp
 def test_manual_send_site_license_mail_guest(api, site_license):
     url = url_for("weko_admin.manual_send_site_license_mail",start_month="202201",end_month="202203")
@@ -871,14 +897,41 @@ def test_get_ogp_image(api, db, site_info, file_instance, mocker):
 
 #def get_search_init_display_index(selected_index=None):
 # .tox/c1/bin/pytest --cov=weko_admin tests/test_views.py::test_get_search_init_display_index -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp
-def test_get_search_init_display_index(api):
+def test_get_search_init_display_index(api, users):
     url = url_for("weko_admin.get_search_init_display_index",selected_index=1)
+    login_user_via_session(client=api, email=users[0]["email"])
     data = [{"id":"0","parent":"#","text":"Root Index","state":{"opened":True}}]
     with patch("weko_admin.views.get_init_display_index",return_value=data):
         res = api.get(url)
         assert response_data(res) == {"indexes":data}
 
 
+# .tox/c1/bin/pytest --cov=weko_admin tests/test_views.py::test_get_search_init_display_index_acl -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp
+@pytest.mark.parametrize("index,is_permission",[
+                         (0,True),# sysadmin
+                         (1,True),# repoadmin
+                         (2,False),# comadmin
+                         (3,False),# contributor
+                         (4,False),# generaluser
+                         ])
+def test_get_search_init_display_index_acl(api,users,index,is_permission):
+    url = url_for("weko_admin.get_search_init_display_index",selected_index=1)
+    login_user_via_session(client=api, email=users[index]["email"])
+    with patch("weko_admin.views.get_init_display_index",return_value=[]) as mock_get:
+        res = api.get(url)
+        assert_role(res, is_permission)
+        if not is_permission:
+            mock_get.assert_not_called()
+
+# .tox/c1/bin/pytest --cov=weko_admin tests/test_views.py::test_get_search_init_display_index_guest -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp
+def test_get_search_init_display_index_guest(api):
+    url = url_for("weko_admin.get_search_init_display_index",selected_index=1)
+    with patch("weko_admin.views.get_init_display_index",return_value=[]) as mock_get:
+        res = api.get(url)
+        assert res.status_code == 302
+        mock_get.assert_not_called()
+
+
 #def save_restricted_access():
 # .tox/c1/bin/pytest --cov=weko_admin tests/test_views.py::test_get_search_init_display_index -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp
 @pytest.mark.parametrize("index,is_permission",[

modules/weko-admin/weko_admin/views.py

@@ -21,10 +21,12 @@
 """Views for weko-admin."""
 
 import calendar
+import inspect
 import json
 import sys
 import time
 from datetime import timedelta, datetime
+from functools import wraps
 import traceback
 
 from flask import Blueprint, Response, abort, current_app, flash, json, \
@@ -541,12 +543,54 @@ def resend_failed_mail():
     return jsonify(result)
 
 
+def _is_repository_in_user_scope(repo_id):
+    """Check whether the current user administers the given repository.
+
+    System and Repository Administrators can handle any repository.
+    Other administrators are limited to the repositories they are assigned
+    to, in the same way as the repository selector of the admin screens.
+
+    :param repo_id: Repository id.
+    :return: True if the current user can handle the repository.
+    """
+    from invenio_communities.models import Community
+    super_roles = current_app.config.get(
+        'WEKO_PERMISSION_SUPER_ROLE_USER',
+        [WEKO_ADMIN_PERMISSION_ROLE_SYSTEM, WEKO_ADMIN_PERMISSION_ROLE_REPO])
+    if any(role.name in super_roles for role in current_user.roles):
+        return True
+    if not repo_id:
+        return False
+    return any(repo.id == repo_id
+               for repo in Community.get_repositories_by_user(current_user))
+
+
+def _form_repository_scope_required(func):
+    """Require ``repo_id`` sent in the form to be in the user's scope.
+
+    The check applies only when the view reads ``repo_id`` from the form;
+    internal callers that pass ``repo_id`` explicitly are not affected.
+    """
+    signature = inspect.signature(func)
+
+    @wraps(func)
+    def decorated_view(*args, **kwargs):
+        bound = signature.bind_partial(*args, **kwargs)
+        if not bound.arguments.get('repo_id'):
+            if not _is_repository_in_user_scope(request.form.get('repo_id')):
+                abort(403)
+        return func(*args, **kwargs)
+
+    return decorated_view
+
+
 @blueprint_api.route('/sitelicensesendmail/send/<start_month>/<end_month>',
                      methods=['POST'])
 @login_required
 @roles_required([WEKO_ADMIN_PERMISSION_ROLE_SYSTEM,
                  WEKO_ADMIN_PERMISSION_ROLE_REPO,
                  WEKO_ADMIN_PERMISSION_ROLE_COMMUNITY])
+@_form_repository_scope_required
 def manual_send_site_license_mail(start_month, end_month, repo_id=None):
     """Send site license mail by manual."""
     if not repo_id:
@@ -721,6 +765,9 @@ def get_ogp_image():
 
 @blueprint_api.route('/search/init_display_index/<string:selected_index>',
                      methods=['GET'])
+@login_required
+@roles_required([WEKO_ADMIN_PERMISSION_ROLE_SYSTEM,
+                 WEKO_ADMIN_PERMISSION_ROLE_REPO])
 def get_search_init_display_index(selected_index=None):
     """Get search init display index.
 

modules/weko-swordserver/tests/test_decorators.py

@@ -8,7 +8,9 @@
 from invenio_oauth2server.ext import verify_oauth_token_and_set_current_user
 from unittest.mock import MagicMock
 from weko_swordserver.errors import ErrorType, WekoSwordserverException
+from werkzeug.exceptions import Forbidden, Unauthorized
 from weko_swordserver.decorators import (
+    check_deposit_role,
     check_oauth,
     check_on_behalf_of,
     check_package_contents,
@@ -47,6 +49,50 @@ def test_check_oauth(app, client, users, tokens):
         assert e.value.message == "Authentication is failed."
 
 
+# def check_deposit_role():
+# .tox/c1/bin/pytest --cov=weko_swordserver tests/test_decorators.py::test_check_deposit_role -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-swordserver/.tox/c1/tmp
+def test_check_deposit_role(app, users):
+    func = check_deposit_role()(lambda x, y: x + y)
+    contributor = users[3]["obj"]
+    generaluser = users[4]["obj"]
+    default_roles = app.config["WEKO_SWORDSERVER_DEPOSIT_ROLE_ENABLE"]
+    try:
+        # default roles
+        assert "Contributor" in default_roles
+        assert "General" not in default_roles
+        with app.test_request_context(method="POST"):
+            login_user(contributor)
+            assert func(x=1, y=2) == 3
+        with app.test_request_context(method="POST"):
+            login_user(generaluser)
+            with pytest.raises(Forbidden):
+                func(x=1, y=2)
+
+        # the application config is applied at request time
+        app.config["WEKO_SWORDSERVER_DEPOSIT_ROLE_ENABLE"] = ["General"]
+        with app.test_request_context(method="POST"):
+            login_user(contributor)
+            with pytest.raises(Forbidden):
+                func(x=1, y=2)
+        with app.test_request_context(method="POST"):
+            login_user(generaluser)
+            assert func(x=1, y=2) == 3
+
+        # no roles are allowed
+        app.config["WEKO_SWORDSERVER_DEPOSIT_ROLE_ENABLE"] = []
+        with app.test_request_context(method="POST"):
+            login_user(contributor)
+            with pytest.raises(Forbidden):
+                func(x=1, y=2)
+
+        # not logged in
+        with app.test_request_context(method="POST"):
+            with pytest.raises(Unauthorized):
+                func(x=1, y=2)
+    finally:
+        app.config["WEKO_SWORDSERVER_DEPOSIT_ROLE_ENABLE"] = default_roles
+
+
 # def check_on_behalf_of():
 # .tox/c1/bin/pytest --cov=weko_swordserver tests/test_decorators.py::test_check_on_behalf_of -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-swordserver/.tox/c1/tmp
 def test_check_on_behalf_of(app):

modules/weko-swordserver/tests/test_views.py

@@ -387,6 +387,30 @@ def update_location_size():
     assert result.status_code == 412
     assert result.json.get("error") == "Failed to verify request body and digest."
 
+# .tox/c1/bin/pytest --cov=weko_swordserver tests/test_views.py::test_post_service_document_deposit_role_config -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-swordserver/.tox/c1/tmp
+def test_post_service_document_deposit_role_config(app, client, users, make_zip, tokens, mocker):
+    token_direct = tokens[0]["token"].access_token
+    url = url_for("weko_swordserver.post_service_document")
+    mocker_check_item = mocker.patch("weko_swordserver.views.check_import_items")
+    default_roles = app.config["WEKO_SWORDSERVER_DEPOSIT_ROLE_ENABLE"]
+    app.config["WEKO_SWORDSERVER_DIGEST_VERIFICATION"] = False
+    # the configured roles do not include the user's role
+    app.config["WEKO_SWORDSERVER_DEPOSIT_ROLE_ENABLE"] = ["Contributor"]
+    try:
+        login_user_via_session(client=client, email=users[0]["email"])
+        headers = {
+            "Authorization": "Bearer {}".format(token_direct),
+            "Content-Disposition": "attachment; filename=payload.zip",
+            "Packaging": "http://purl.org/net/sword/3.0/package/SimpleZip",
+        }
+        storage = FileStorage(filename="payload.zip", stream=make_zip())
+        result = client.post(url, data={"file": storage}, content_type="multipart/form-data", headers=headers)
+        assert result.status_code == 403
+        assert result.json.get("error") == "Not allowed operation in your role or token scope."
+        mocker_check_item.assert_not_called()
+    finally:
+        app.config["WEKO_SWORDSERVER_DEPOSIT_ROLE_ENABLE"] = default_roles
+
 # .tox/c1/bin/pytest --cov=weko_swordserver tests/test_views.py::test_post_service_document_multi_recid -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-swordserver/.tox/c1/tmp
 def test_post_service_document_multi_recid(app, client, db, users, make_zip, tokens, mocker):
     mocker.patch("invenio_pidstore.resolver.Resolver.resolve", return_value=(MagicMock(), MagicMock()))

modules/weko-swordserver/weko_swordserver/decorators.py

@@ -13,6 +13,7 @@
 from invenio_oauth2server.decorators import (
     require_api_auth, require_oauth_scopes
 )
+from weko_accounts.utils import roles_required
 
 from .errors import ErrorType, WekoSwordserverException
 
@@ -44,6 +45,24 @@ def decorated(*args, **kwargs):
         return decorated
     return wrapper
 
+def check_deposit_role():
+    """Decorator to check the roles allowed to deposit items.
+
+    The allowed roles are read from
+    ``WEKO_SWORDSERVER_DEPOSIT_ROLE_ENABLE`` on each request, so that the
+    application configuration is applied.
+    """
+    def wrapper(f):
+        @wraps(f)
+        def decorated(*args, **kwargs):
+            roles = current_app.config.get(
+                "WEKO_SWORDSERVER_DEPOSIT_ROLE_ENABLE", []
+            )
+            return roles_required(roles)(f)(*args, **kwargs)
+        return decorated
+    return wrapper
+
+
 def check_on_behalf_of():
     """Decorator to check onBehalfOf header."""
     def wrapper(f):

modules/weko-swordserver/weko_swordserver/views.py

@@ -35,7 +35,6 @@
 from invenio_pidstore.resolver import Resolver
 from werkzeug.utils import import_string
 
-from weko_accounts.utils import roles_required
 from weko_admin.api import TempDirInfo
 from weko_deposit.api import WekoRecord
 from weko_items_ui.scopes import item_create_scope, item_update_scope, item_delete_scope
@@ -50,8 +49,9 @@
 from weko_workflow.utils import get_site_info_name
 from weko_workflow.scopes import activity_scope
 
-from .config import WEKO_SWORDSERVER_DEPOSIT_ROLE_ENABLE
-from .decorators import check_on_behalf_of, check_package_contents
+from .decorators import (
+    check_deposit_role, check_on_behalf_of, check_package_contents
+)
 from .errors import ErrorType, WekoSwordserverException
 from .utils import (
     check_import_file_format,
@@ -172,7 +172,7 @@ def get_service_document():
 @limiter.limit("")
 @require_oauth_scopes(write_scope.id, actions_scope.id)
 @require_oauth_scopes(item_create_scope.id)
-@roles_required(WEKO_SWORDSERVER_DEPOSIT_ROLE_ENABLE)
+@check_deposit_role()
 @check_on_behalf_of()
 @check_package_contents()
 def post_service_document():
@@ -488,7 +488,7 @@ def process_item(item, request_info):
 @limiter.limit("")
 @require_oauth_scopes(write_scope.id, actions_scope.id)
 @require_oauth_scopes(item_update_scope.id)
-@roles_required(WEKO_SWORDSERVER_DEPOSIT_ROLE_ENABLE)
+@check_deposit_role()
 @check_on_behalf_of()
 @check_package_contents()
 def put_object(recid):
@@ -1202,7 +1202,7 @@ def link_key(link):
 @limiter.limit("")
 @require_oauth_scopes(write_scope.id, actions_scope.id)
 @require_oauth_scopes(item_delete_scope.id)
-@roles_required(WEKO_SWORDSERVER_DEPOSIT_ROLE_ENABLE)
+@check_deposit_role()
 @check_on_behalf_of()
 def delete_object(recid):
     """Deleting the entire Object