From d4b0ef5327c2b205a6b3a448ab5076d3ca96b38e Mon Sep 17 00:00:00 2001 From: Masaharu Hayashi Date: Sun, 27 Sep 2026 21:40:19 +0000 Subject: [PATCH 1/3] =?UTF-8?q?fix(weko-admin):=20=E3=82=B5=E3=82=A4?= =?UTF-8?q?=E3=83=88=E3=83=A9=E3=82=A4=E3=82=BB=E3=83=B3=E3=82=B9=E3=83=A1?= =?UTF-8?q?=E3=83=BC=E3=83=AB=E6=89=8B=E5=8B=95=E9=80=81=E4=BF=A1=E3=81=A7?= =?UTF-8?q?=E5=AF=BE=E8=B1=A1=E3=83=AA=E3=83=9D=E3=82=B8=E3=83=88=E3=83=AA?= =?UTF-8?q?=E3=81=AE=E6=A8=A9=E9=99=90=E3=82=92=E7=A2=BA=E8=AA=8D=E3=81=99?= =?UTF-8?q?=E3=82=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit フォームで指定されたリポジトリが操作者の担当範囲に含まれることを デコレータで確認する。System/Repository Administrator は従来どおり。 Co-Authored-By: Claude Opus 5.5 --- modules/weko-admin/tests/test_views.py | 28 +++++++++++++++- modules/weko-admin/weko_admin/views.py | 44 ++++++++++++++++++++++++++ 2 files changed, 71 insertions(+), 1 deletion(-) diff --git a/modules/weko-admin/tests/test_views.py b/modules/weko-admin/tests/test_views.py index 6ecc03d2a2..290afe9827 100644 --- a/modules/weko-admin/tests/test_views.py +++ b/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") diff --git a/modules/weko-admin/weko_admin/views.py b/modules/weko-admin/weko_admin/views.py index 69c5f96e01..c191d20cdd 100644 --- a/modules/weko-admin/weko_admin/views.py +++ b/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//', 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: From 7f681e2fea4dfd5ba15def583bc96252e95f609a Mon Sep 17 00:00:00 2001 From: Masaharu Hayashi Date: Sun, 27 Sep 2026 21:41:03 +0000 Subject: [PATCH 2/3] =?UTF-8?q?fix(weko-admin):=20=E6=A4=9C=E7=B4=A2?= =?UTF-8?q?=E8=A8=AD=E5=AE=9A=E7=94=A8=E3=81=AE=E3=82=A4=E3=83=B3=E3=83=87?= =?UTF-8?q?=E3=83=83=E3=82=AF=E3=82=B9=E5=8F=96=E5=BE=97=20API=20=E3=82=92?= =?UTF-8?q?=E7=AE=A1=E7=90=86=E8=80=85=E3=81=AB=E9=99=90=E5=AE=9A=E3=81=99?= =?UTF-8?q?=E3=82=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 検索設定画面と同じく System/Repository Administrator のみ利用できるよう login_required と roles_required を付ける。 Co-Authored-By: Claude Opus 5.5 --- modules/weko-admin/tests/test_views.py | 29 +++++++++++++++++++++++++- modules/weko-admin/weko_admin/views.py | 3 +++ 2 files changed, 31 insertions(+), 1 deletion(-) diff --git a/modules/weko-admin/tests/test_views.py b/modules/weko-admin/tests/test_views.py index 290afe9827..e73855b37c 100644 --- a/modules/weko-admin/tests/test_views.py +++ b/modules/weko-admin/tests/test_views.py @@ -897,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",[ diff --git a/modules/weko-admin/weko_admin/views.py b/modules/weko-admin/weko_admin/views.py index c191d20cdd..0408a2bd19 100644 --- a/modules/weko-admin/weko_admin/views.py +++ b/modules/weko-admin/weko_admin/views.py @@ -765,6 +765,9 @@ def get_ogp_image(): @blueprint_api.route('/search/init_display_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. From 229ad779af1a06aa6f2f090c05bd03f5b4160e2a Mon Sep 17 00:00:00 2001 From: Masaharu Hayashi Date: Sun, 27 Sep 2026 21:43:40 +0000 Subject: [PATCH 3/3] =?UTF-8?q?fix(weko-swordserver):=20=E7=99=BB=E9=8C=B2?= =?UTF-8?q?=E5=8F=AF=E8=83=BD=E3=83=AD=E3=83=BC=E3=83=AB=E3=81=AE=E8=A8=AD?= =?UTF-8?q?=E5=AE=9A=E3=82=92=E5=AE=9F=E8=A1=8C=E6=99=82=E3=81=AB=E5=8F=82?= =?UTF-8?q?=E7=85=A7=E3=81=99=E3=82=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit WEKO_SWORDSERVER_DEPOSIT_ROLE_ENABLE をモジュール定数として import せず、 リクエストごとに current_app.config から読むデコレータ check_deposit_role を 追加し、POST /sword/service-document・PUT/DELETE /sword/deposit に適用する。 ロールの判定自体は従来どおり weko_accounts の roles_required に委ねる。 Co-Authored-By: Claude Opus 5.5 --- .../weko-swordserver/tests/test_decorators.py | 46 +++++++++++++++++++ modules/weko-swordserver/tests/test_views.py | 24 ++++++++++ .../weko_swordserver/decorators.py | 19 ++++++++ .../weko_swordserver/views.py | 12 ++--- 4 files changed, 95 insertions(+), 6 deletions(-) diff --git a/modules/weko-swordserver/tests/test_decorators.py b/modules/weko-swordserver/tests/test_decorators.py index 74647f404a..182919921b 100644 --- a/modules/weko-swordserver/tests/test_decorators.py +++ b/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): diff --git a/modules/weko-swordserver/tests/test_views.py b/modules/weko-swordserver/tests/test_views.py index d12fee8a4f..71501b77d1 100644 --- a/modules/weko-swordserver/tests/test_views.py +++ b/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())) diff --git a/modules/weko-swordserver/weko_swordserver/decorators.py b/modules/weko-swordserver/weko_swordserver/decorators.py index f55a1bec8b..f81d62fe3b 100644 --- a/modules/weko-swordserver/weko_swordserver/decorators.py +++ b/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): diff --git a/modules/weko-swordserver/weko_swordserver/views.py b/modules/weko-swordserver/weko_swordserver/views.py index 08d521391e..9da7c45f9c 100644 --- a/modules/weko-swordserver/weko_swordserver/views.py +++ b/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