Fix/issue62782/issue62796(No.557, 560, 618, 1016)

modules/weko-search-ui/tests/test_utils.py

@@ -59,6 +59,7 @@
     check_tsv_import_items,
     check_xml_import_items,
     check_index_access_permissions,
+    check_index_permission,
     check_permission,
     check_provide_in_system,
     check_sub_item_is_system,
@@ -4842,6 +4843,40 @@ def test_function():
             with pytest.raises(BadRequest):
                 test_function()
 
+# .tox/c1/bin/pytest --cov=weko_search_ui tests/test_utils.py::test_check_index_permission -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-search_ui/.tox/c1/tmp
+def test_check_index_permission():
+    @check_index_permission
+    def test_function(**kwargs):
+        return kwargs
+
+    with pytest.raises(NotFound):
+        test_function()
+
+    with patch("weko_search_ui.utils.Indexes.get_index") as get_index:
+        with pytest.raises(NotFound):
+            test_function(path_str="1_invalid")
+        get_index.assert_called_once_with(index_id="1")
+
+    with patch("weko_search_ui.utils.Indexes.get_index", return_value=None):
+        with pytest.raises(NotFound):
+            test_function(index_id=1)
+
+    index = MagicMock(id=1)
+    with patch("weko_search_ui.utils.Indexes.get_index", return_value=index), \
+            patch("weko_search_ui.utils.filter_index_list_by_role", return_value=[]):
+        with pytest.raises(Forbidden):
+            test_function(index_id=1)
+
+    first_index = MagicMock(id=1)
+    second_index = MagicMock(id=2)
+    with patch("weko_search_ui.utils.Indexes.get_index", side_effect=[first_index, second_index]), \
+            patch("weko_search_ui.utils.filter_index_list_by_role", return_value=[second_index]):
+        assert test_function(path_str="1_2") == {"path_str": "2"}
+
+    with patch("weko_search_ui.utils.Indexes.get_index", return_value=first_index), \
+            patch("weko_search_ui.utils.filter_index_list_by_role", return_value=[first_index]):
+        assert test_function(index_id="1") == {"index_id": 1}
+
 
 # def handle_check_file_metadata(list_record, data_path):
 def test_handle_check_file_metadata(i18n_app, record_with_metadata):

modules/weko-search-ui/tests/test_views.py

@@ -148,6 +148,17 @@ def test_journal_detail(i18n_app, users, indices):
         assert journal_detail(33)
 
 
+def test_journal_detail_forbidden(client, users, indices):
+    url = url_for("weko_search_ui.journal_detail", index_id=33)
+    with patch(
+        "flask_login.utils._get_user",
+        side_effect=lambda: _fresh_user(users[3]),
+    ), patch("weko_search_ui.utils.filter_index_list_by_role", return_value=[]):
+        response = client.get(url)
+
+    assert response.status_code == 403
+
+
 # def search_feedback_mail_list():
 def test_search_feedback_mail_list(i18n_app, users):
     with patch("flask_login.utils._get_user", return_value=users[3]['obj']):

modules/weko-search-ui/weko_search_ui/utils.py

@@ -85,8 +85,10 @@
 from weko_index_tree.utils import (
     check_index_permissions,
     check_restrict_doi_with_indexes,
+    filter_index_list_by_role
 )
 from weko_index_tree.models import Index
+from weko_index_tree.api import Indexes
 from weko_indextree_journal.api import Journals
 from weko_logging.activity_logger import UserActivityLogger
 from weko_records.api import FeedbackMailList, JsonldMapping, RequestMailList, ItemTypes, ItemLink, ItemApplication
@@ -5440,6 +5442,43 @@ def decorated_view(*args, **kwargs):
 
     return decorated_view
 
+def check_index_permission(view):
+    """Require access to indexes supplied as ``path_str`` or ``index_id``."""
+
+    @wraps(view)
+    def decorated_view(*args, **kwargs):
+        path_str = kwargs.get("path_str")
+        index_id = kwargs.get("index_id")
+        if path_str is not None:
+            index_ids = path_str.split("_")
+        elif index_id is not None:
+            index_ids = [str(index_id)]
+        else:
+            abort(404)
+
+        index_list = []
+        for index_id in index_ids:
+            if not index_id.isdigit():
+                abort(404)
+
+            index = Indexes.get_index(index_id=index_id)
+            if index is None:
+                abort(404)
+
+            index_list.append(index)
+
+        allowed_index_list = filter_index_list_by_role(index_list)
+        if not allowed_index_list:
+            abort(403)
+
+        if path_str is not None:
+            kwargs["path_str"] = "_".join(str(index.id) for index in allowed_index_list)
+        else:
+            kwargs["index_id"] = allowed_index_list[0].id
+
+        return view(*args, **kwargs)
+
+    return decorated_view
 
 def handle_check_file_metadata(list_record, data_path):
     """Check file contents, thumbnails metadata.

modules/weko-search-ui/weko_search_ui/views.py

@@ -59,6 +59,7 @@
     check_index_access_permissions,
     check_permission,
     get_journal_info,
+    check_index_permission
 )
 
 _signals = Namespace()
@@ -347,6 +348,7 @@ def opensearch_description():
 
 
 @blueprint.route("/journal_info/<int:index_id>", methods=["GET"])
+@check_index_permission
 def journal_detail(index_id=0):
     """Render a check view."""
     result = get_journal_info(index_id)

modules/weko-workflow/tests/test_views.py

@@ -4354,7 +4354,8 @@ def test_get_feedback_maillist_acl_nologin(client,db_register2):
     (5, 200),
     (6, 200),
 ])
-def test_get_feedback_maillist_acl_users(client, users, users_index, status_code):
+def test_get_feedback_maillist_acl_users(
+        client, users, db_register_full_action, users_index, status_code):
     """Test of get feedback maillist."""
     login(client=client, email=users[users_index]['email'])
     url = url_for('weko_workflow.get_feedback_maillist', activity_id='1')

modules/weko-workflow/weko_workflow/views.py

@@ -2880,6 +2880,7 @@ def save_item_application(activity_id='0', action_id='0'):
 @workflow_blueprint.route('/get_feedback_maillist/<string:activity_id>',
                  methods=['GET'])
 @login_required
+@check_authority
 def get_feedback_maillist(activity_id='0'):
     """アクティビティに設定されているフィードバックメール送信先の情報を取得して返す