Fix/issue62783

modules/weko-admin/tests/conftest.py

@@ -74,7 +74,8 @@
 from weko_index_tree.models import Index, IndexStyle
 from weko_items_ui.config import WEKO_ITEMS_UI_CRIS_LINKAGE_RESEARCHMAP_MERGE_MODE_DEFAULT
 from weko_records_ui import WekoRecordsUI
-from weko_records_ui.config import WEKO_PERMISSION_SUPER_ROLE_USER
+from weko_records_ui.config import WEKO_PERMISSION_SUPER_ROLE_USER, \
+    WEKO_PERMISSION_ROLE_COMMUNITY
 from weko_records import WekoRecords
 from weko_records.models import SiteLicenseInfo, SiteLicenseIpAddress,ItemType,ItemTypeName,ItemTypeJsonldMapping
 from weko_redis.redis import RedisConnection
@@ -175,6 +176,7 @@ def base_app(instance_path, cache_config,request ,search_class):
         WEKO_ADMIN_RESTRICTED_ACCESS_SETTINGS = WEKO_ADMIN_RESTRICTED_ACCESS_SETTINGS,
         WEKO_WORKFLOW_USAGE_REPORT_WORKFLOW_NAME = 'test workflow31001',
         WEKO_PERMISSION_SUPER_ROLE_USER=WEKO_PERMISSION_SUPER_ROLE_USER,
+        WEKO_PERMISSION_ROLE_COMMUNITY=WEKO_PERMISSION_ROLE_COMMUNITY,
         WEKO_ITEMS_UI_CRIS_LINKAGE_RESEARCHMAP_MERGE_MODE_DEFAULT=WEKO_ITEMS_UI_CRIS_LINKAGE_RESEARCHMAP_MERGE_MODE_DEFAULT
     )
     app_.testing = True

modules/weko-admin/tests/test_views.py

@@ -517,9 +517,10 @@ def test_get_feedback_mail(api, users):
 
 #def get_send_mail_history():
 # .tox/c1/bin/pytest --cov=weko_admin tests/test_views.py::test_get_send_mail_history -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp
-def test_get_send_mail_history(api, mocker):
+def test_get_send_mail_history(api, users, mocker):
     mocker.patch("weko_admin.views.FeedbackMail.load_feedback_mail_history",side_effect=lambda x, y:{"page":x})
     url = url_for("weko_admin.get_send_mail_history")
+    login_user_via_session(client=api, email=users[0]["email"])  # sysadmin
     input = {"page":2, "repo_id":"Root Index"}
     res = api.get(url,query_string=input)
     assert response_data(res) == {"page":2}
@@ -528,6 +529,29 @@ def test_get_send_mail_history(api, mocker):
     res = api.get(url,query_string=input)
     assert response_data(res) == {"page":1}
 
+# .tox/c1/bin/pytest --cov=weko_admin tests/test_views.py::test_get_send_mail_history_guest -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp
+def test_get_send_mail_history_guest(api, mocker):
+    """未ログインの場合は401になること。"""
+    mocker.patch("weko_admin.views.FeedbackMail.load_feedback_mail_history",side_effect=lambda x, y:{"page":x})
+    url = url_for("weko_admin.get_send_mail_history")
+    res = api.get(url, query_string={"page":1, "repo_id":"Root Index"})
+    assert res.status_code == 401
+
+# .tox/c1/bin/pytest --cov=weko_admin tests/test_views.py::test_get_send_mail_history_scope -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp
+@pytest.mark.parametrize("index,repo_id,is_permission",[
+                         (2, "comm1", True),        # comadmin(担当コミュニティ)
+                         (2, "other_repo", False),  # comadmin(担当外コミュニティ)
+                         (3, "comm1", False),       # contributor(コミュニティ管理者ロールなし)
+                         (4, "comm1", False),       # generaluser(ロールなし)
+                         ])
+def test_get_send_mail_history_scope(api, users, community, index, repo_id, is_permission, mocker):
+    """repository_scope_requiredによるコミュニティ管理者のスコープ制御を確認する。"""
+    mocker.patch("weko_admin.views.FeedbackMail.load_feedback_mail_history",side_effect=lambda x, y:{"page":x})
+    login_user_via_session(client=api, email=users[index]["email"])
+    url = url_for("weko_admin.get_send_mail_history")
+    res = api.get(url, query_string={"page":1, "repo_id":repo_id})
+    assert_role(res, is_permission)
+
 
 #def get_failed_mail():
 # .tox/c1/bin/pytest --cov=weko_admin tests/test_views.py::test_get_failed_mail_acl -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp

modules/weko-admin/weko_admin/permissions.py

@@ -20,7 +20,11 @@
 
 """WEKO3 module docstring."""
 
+from functools import wraps
+
 import pkg_resources
+from flask import abort, current_app, request
+from flask_login import current_user
 from flask_principal import ActionNeed
 from invenio_access import Permission, action_factory
 
@@ -55,3 +59,88 @@ def admin_permission_factory(action):
         from flask_principal import Permission
 
     return Permission(action_class)
+
+
+def _is_super_user(user):
+    """Check whether the user has a system/repository administrator role."""
+    supers = current_app.config['WEKO_PERMISSION_SUPER_ROLE_USER']
+    return any(role.name in supers for role in (user.roles or []))
+
+
+def _is_community_admin(user):
+    """Check whether the user has a community administrator role."""
+    comadmin = current_app.config['WEKO_PERMISSION_ROLE_COMMUNITY']
+    return any(role.name in comadmin for role in (user.roles or []))
+
+
+def _lookup_param(data, kwargs, path):
+    """リクエストデータまたはURL変数からパラメータ値を取得する。
+
+    pathにドットが含まれる場合はネストしたdictを辿る
+    (例: 'data.repository')。途中がdictでない、またはキーが
+    存在しない場合はNoneを返す。
+    """
+    if not path:
+        return None
+    if '.' not in path:
+        return data.get(path) or kwargs.get(path)
+    value = data
+    for key in path.split('.'):
+        if not isinstance(value, dict) or key not in value:
+            return None
+        value = value[key]
+    return value
+
+
+def repository_scope_required(repository_id_param=None,
+                               id_param=None, id_model=None, id_attr='repository_id',
+                               pk_attr='id'):
+    """repository_id(またはid_param経由でid_modelから解決したid_attr)が
+    current_userの担当範囲内であることを要求する。
+
+    - System/Repository Administrator: 無条件許可
+    - Community Administrator: 担当コミュニティ(Community.get_repositories_by_user)のみ許可
+    - id_param(既存レコード=移動元)とrepository_id_param(リクエスト指定=移動先)の
+      両方の値が取得できた場合は、双方が担当範囲内であることを要求する
+      (担当外コミュニティへのレコード移動を防ぐ)。片方しか取得できない場合は
+      取得できた方だけで判定する
+    - repository_id_param / id_param にはドット記法でネストしたキーを指定できる
+      (例: 'data.repository')
+    - pk_attr: id_modelの主キー列名。デフォルトは'id'。主キーが異なる場合は
+      明示的に指定する(例: WidgetItemはid列を持たず主キーはwidget_id)
+    """
+    def decorator(f):
+        @wraps(f)
+        def wrapped(*args, **kwargs):
+            if not current_user.is_authenticated:
+                abort(401)
+            if _is_super_user(current_user):
+                return f(*args, **kwargs)
+
+            data = request.get_json(silent=True) or request.form or request.args
+            repository_ids = []
+
+            target_id = _lookup_param(data, kwargs, id_param)
+            if target_id:
+                record = id_model.query.filter_by(**{pk_attr: target_id}).one_or_none()
+                if record is None:
+                    abort(404)
+                # 移動元(既存レコードの現在の所属)
+                repository_ids.append(getattr(record, id_attr, None))
+            requested_repository_id = _lookup_param(data, kwargs, repository_id_param)
+            if requested_repository_id:
+                # 移動先(リクエストで指定された所属)
+                repository_ids.append(requested_repository_id)
+
+            if _is_community_admin(current_user) and repository_ids:
+                # weko_admin初期化時のimportで循環参照が発生するため、
+                # get_repository_list()と同様に関数内でimportする。
+                from invenio_communities.models import Community
+                communities = Community.get_repositories_by_user(current_user)
+                community_ids = {str(c.id) for c in communities}
+                if all(str(rid) in community_ids for rid in repository_ids):
+                    return f(*args, **kwargs)
+
+            abort(403)
+        return wrapped
+    return decorator

modules/weko-admin/weko_admin/views.py

@@ -48,6 +48,7 @@
 from .api import send_site_license_mail
 from .config import WEKO_ADMIN_PERMISSION_ROLE_REPO, \
     WEKO_ADMIN_PERMISSION_ROLE_SYSTEM, WEKO_ADMIN_PERMISSION_ROLE_COMMUNITY
+from .permissions import repository_scope_required
 from .models import FacetSearchSetting, SessionLifetime, SiteInfo, AdminSettings
 from .utils import FeedbackMail, StatisticMail, UsageReport, \
     format_site_info_data, get_admin_lang_setting, \
@@ -471,6 +472,7 @@ def get_feedback_mail():
 
 
 @blueprint_api.route('/get_send_mail_history', methods=['GET'])
+@repository_scope_required(repository_id_param='repo_id')
 def get_send_mail_history():
     """API allow to get send mail history.
 

modules/weko-gridlayout/tests/conftest.py

@@ -46,7 +46,8 @@
 from weko_records.models import ItemTypeProperty
 from weko_records.models import ItemType, ItemTypeMapping, ItemTypeName
 from weko_records.api import Mapping
-from weko_records_ui.config import WEKO_PERMISSION_SUPER_ROLE_USER
+from weko_records_ui.config import WEKO_PERMISSION_SUPER_ROLE_USER, \
+    WEKO_PERMISSION_ROLE_COMMUNITY
 from weko_index_tree.models import Index
 from weko_gridlayout import WekoGridLayout
 #from weko_admin import WekoAdmin
@@ -119,7 +120,8 @@ def base_app(instance_path):
         FILES_REST_DEFAULT_MAX_FILE_SIZE=None,
         FILES_REST_OBJECT_KEY_MAX_LEN=255,
         BABEL_DEFAULT_TIMEZONE='Asia/Tokyo',
-        WEKO_PERMISSION_SUPER_ROLE_USER=WEKO_PERMISSION_SUPER_ROLE_USER
+        WEKO_PERMISSION_SUPER_ROLE_USER=WEKO_PERMISSION_SUPER_ROLE_USER,
+        WEKO_PERMISSION_ROLE_COMMUNITY=WEKO_PERMISSION_ROLE_COMMUNITY
     )
     Babel(app_)
     InvenioDB(app_)