diff --git a/modules/weko-admin/tests/conftest.py b/modules/weko-admin/tests/conftest.py index 9e691c5f73..d0b8416262 100644 --- a/modules/weko-admin/tests/conftest.py +++ b/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 diff --git a/modules/weko-admin/tests/test_permissions.py b/modules/weko-admin/tests/test_permissions.py index 87136f1ce5..57d76cd37b 100644 --- a/modules/weko-admin/tests/test_permissions.py +++ b/modules/weko-admin/tests/test_permissions.py @@ -1,8 +1,12 @@ import pkg_resources -from mock import patch +import pytest +from mock import MagicMock, patch +from flask_login import login_user +from werkzeug.exceptions import Forbidden, NotFound, Unauthorized -from weko_admin.permissions import admin_permission_factory +from weko_admin.permissions import _lookup_param, \ + admin_permission_factory, repository_scope_required # .tox/c1/bin/pytest --cov=weko_admin tests/test_permissions.py -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp @@ -13,8 +17,334 @@ def test_admin_permission_factory(app, users): result = admin_permission_factory(action) permission_values = [permission.value for permission in list(result.needs)] assert action in permission_values - + with patch("weko_admin.permissions.pkg_resources.get_distribution", side_effect=pkg_resources.DistributionNotFound): result = admin_permission_factory(action) permission_values = [permission.value for permission in list(result.needs)] assert action in permission_values + + +# def repository_scope_required(...): +# .tox/c1/bin/pytest --cov=weko_admin tests/test_permissions.py::test_repository_scope_required_anonymous -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp +def test_repository_scope_required_anonymous(app): + """未ログインの場合は401になること。""" + @repository_scope_required(repository_id_param='repo_id') + def view(*args, **kwargs): + return 'ok' + + with app.test_request_context('/?repo_id=repoA'): + with pytest.raises(Unauthorized): + view() + + +# .tox/c1/bin/pytest --cov=weko_admin tests/test_permissions.py::test_repository_scope_required_super_user -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp +@pytest.mark.parametrize("index", [0, 1]) # sysadmin, repoadmin +def test_repository_scope_required_super_user(app, users, index): + """System/Repository Administratorは無条件で許可されること。""" + @repository_scope_required(repository_id_param='repo_id') + def view(*args, **kwargs): + return 'ok' + + with app.test_request_context('/?repo_id=repoA'): + login_user(users[index]["obj"]) + assert view() == 'ok' + + +# .tox/c1/bin/pytest --cov=weko_admin tests/test_permissions.py::test_repository_scope_required_no_role -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp +def test_repository_scope_required_no_role(app, users): + """ロールなしログインユーザーは拒否されること。""" + @repository_scope_required(repository_id_param='repo_id') + def view(*args, **kwargs): + return 'ok' + + with app.test_request_context('/?repo_id=repoA'): + login_user(users[4]["obj"]) # generaluser + with pytest.raises(Forbidden): + view() + + +# .tox/c1/bin/pytest --cov=weko_admin tests/test_permissions.py::test_repository_scope_required_community_admin_allowed -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp +def test_repository_scope_required_community_admin_allowed(app, users): + """担当コミュニティのCommunity Administratorは許可されること。""" + @repository_scope_required(repository_id_param='repo_id') + def view(*args, **kwargs): + return 'ok' + + community = MagicMock(id='repoA') + with app.test_request_context('/?repo_id=repoA'): + login_user(users[2]["obj"]) # comadmin + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=[community]): + assert view() == 'ok' + + +# .tox/c1/bin/pytest --cov=weko_admin tests/test_permissions.py::test_repository_scope_required_community_admin_denied -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp +def test_repository_scope_required_community_admin_denied(app, users): + """担当外コミュニティのCommunity Administratorは拒否されること。""" + @repository_scope_required(repository_id_param='repo_id') + def view(*args, **kwargs): + return 'ok' + + community = MagicMock(id='repoB') + with app.test_request_context('/?repo_id=repoA'): + login_user(users[2]["obj"]) # comadmin(repoBのみ担当) + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=[community]): + with pytest.raises(Forbidden): + view() + + +# .tox/c1/bin/pytest --cov=weko_admin tests/test_permissions.py::test_repository_scope_required_both_params_required -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp +def test_repository_scope_required_both_params_required(app, users): + """DB側とbody側の両方が取得できる場合は双方が担当範囲内であることを要求すること。""" + record = MagicMock(repository_id='repoA') + id_model = MagicMock() + id_model.query.filter_by.return_value.one_or_none.return_value = record + + @repository_scope_required(repository_id_param='repository_id', + id_param='page_id', id_model=id_model) + def view(*args, **kwargs): + return 'ok' + + community = MagicMock(id='repoA') + # DB側(repoA)は担当内だが、bodyで担当外のrepoBへの移動が指定されているため拒否 + with app.test_request_context('/?repository_id=repoB&page_id=1'): + login_user(users[2]["obj"]) # comadmin(repoAを担当) + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=[community]): + with pytest.raises(Forbidden): + view() + id_model.query.filter_by.assert_called_with(id='1') + + +# .tox/c1/bin/pytest --cov=weko_admin tests/test_permissions.py::test_repository_scope_required_db_out_of_scope -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp +def test_repository_scope_required_db_out_of_scope(app, users): + """DB側が担当外の場合はbody側が担当内でも拒否されること。""" + record = MagicMock(repository_id='repoB') + id_model = MagicMock() + id_model.query.filter_by.return_value.one_or_none.return_value = record + + @repository_scope_required(repository_id_param='repository_id', + id_param='page_id', id_model=id_model) + def view(*args, **kwargs): + return 'ok' + + community = MagicMock(id='repoA') + with app.test_request_context('/?repository_id=repoA&page_id=1'): + login_user(users[2]["obj"]) # comadmin(repoAを担当) + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=[community]): + with pytest.raises(Forbidden): + view() + + +# .tox/c1/bin/pytest --cov=weko_admin tests/test_permissions.py::test_repository_scope_required_both_in_scope -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp +def test_repository_scope_required_both_in_scope(app, users): + """DB側・body側とも担当内(同一ID)の場合は許可されること。""" + record = MagicMock(repository_id='repoA') + id_model = MagicMock() + id_model.query.filter_by.return_value.one_or_none.return_value = record + + @repository_scope_required(repository_id_param='repository_id', + id_param='page_id', id_model=id_model) + def view(*args, **kwargs): + return 'ok' + + community = MagicMock(id='repoA') + with app.test_request_context('/?repository_id=repoA&page_id=1'): + login_user(users[2]["obj"]) # comadmin(repoAを担当) + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=[community]): + assert view() == 'ok' + + +# .tox/c1/bin/pytest --cov=weko_admin tests/test_permissions.py::test_repository_scope_required_move_within_scope -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp +def test_repository_scope_required_move_within_scope(app, users): + """複数コミュニティ担当者による担当内から担当内への移動は許可されること。""" + record = MagicMock(repository_id='repoA') + id_model = MagicMock() + id_model.query.filter_by.return_value.one_or_none.return_value = record + + @repository_scope_required(repository_id_param='repository_id', + id_param='page_id', id_model=id_model) + def view(*args, **kwargs): + return 'ok' + + communities = [MagicMock(id='repoA'), MagicMock(id='repoB')] + with app.test_request_context('/?repository_id=repoB&page_id=1'): + login_user(users[2]["obj"]) # comadmin(repoA/repoBを担当) + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=communities): + assert view() == 'ok' + + +# .tox/c1/bin/pytest --cov=weko_admin tests/test_permissions.py::test_repository_scope_required_id_param_only -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp +@pytest.mark.parametrize("db_repository_id, allowed", [('repoA', True), ('repoB', False)]) +def test_repository_scope_required_id_param_only(app, users, db_repository_id, allowed): + """id_paramのみ値がある場合(削除系)はDB側の値だけで判定すること。""" + record = MagicMock(repository_id=db_repository_id) + id_model = MagicMock() + id_model.query.filter_by.return_value.one_or_none.return_value = record + + @repository_scope_required(repository_id_param='repository_id', + id_param='page_id', id_model=id_model) + def view(*args, **kwargs): + return 'ok' + + community = MagicMock(id='repoA') + with app.test_request_context('/?page_id=1'): + login_user(users[2]["obj"]) # comadmin(repoAを担当) + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=[community]): + if allowed: + assert view() == 'ok' + else: + with pytest.raises(Forbidden): + view() + + +# .tox/c1/bin/pytest --cov=weko_admin tests/test_permissions.py::test_repository_scope_required_repository_id_param_only -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp +@pytest.mark.parametrize("body_repository_id, allowed", [('repoA', True), ('repoB', False)]) +def test_repository_scope_required_repository_id_param_only(app, users, + body_repository_id, allowed): + """repository_id_paramのみ値がある場合(新規作成)はbody側の値だけで判定すること。""" + id_model = MagicMock() + + @repository_scope_required(repository_id_param='repository_id', + id_param='page_id', id_model=id_model) + def view(*args, **kwargs): + return 'ok' + + community = MagicMock(id='repoA') + with app.test_request_context('/?repository_id={}'.format(body_repository_id)): + login_user(users[2]["obj"]) # comadmin(repoAを担当) + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=[community]): + if allowed: + assert view() == 'ok' + else: + with pytest.raises(Forbidden): + view() + id_model.query.filter_by.assert_not_called() + + +# .tox/c1/bin/pytest --cov=weko_admin tests/test_permissions.py::test_repository_scope_required_no_params -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp +def test_repository_scope_required_no_params(app, users): + """id_param/repository_id_paramとも値がない場合は拒否されること。""" + id_model = MagicMock() + + @repository_scope_required(repository_id_param='repository_id', + id_param='page_id', id_model=id_model) + def view(*args, **kwargs): + return 'ok' + + community = MagicMock(id='repoA') + with app.test_request_context('/'): + login_user(users[2]["obj"]) # comadmin + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=[community]): + with pytest.raises(Forbidden): + view() + + +# .tox/c1/bin/pytest --cov=weko_admin tests/test_permissions.py::test_repository_scope_required_db_repository_id_none -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp +def test_repository_scope_required_db_repository_id_none(app, users): + """DBレコードのrepository_idがNoneの場合はbody側が担当内でも拒否されること。""" + record = MagicMock(repository_id=None) + id_model = MagicMock() + id_model.query.filter_by.return_value.one_or_none.return_value = record + + @repository_scope_required(repository_id_param='repository_id', + id_param='page_id', id_model=id_model) + def view(*args, **kwargs): + return 'ok' + + community = MagicMock(id='repoA') + with app.test_request_context('/?repository_id=repoA&page_id=1'): + login_user(users[2]["obj"]) # comadmin(repoAを担当) + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=[community]): + with pytest.raises(Forbidden): + view() + + +# .tox/c1/bin/pytest --cov=weko_admin tests/test_permissions.py::test_repository_scope_required_nested_param -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp +@pytest.mark.parametrize("body_repository_id, allowed", [('repoA', True), ('repoB', False)]) +def test_repository_scope_required_nested_param(app, users, body_repository_id, allowed): + """repository_id_paramにドット記法を指定した場合はネストした値で判定すること。""" + @repository_scope_required(repository_id_param='data.repository') + def view(*args, **kwargs): + return 'ok' + + community = MagicMock(id='repoA') + with app.test_request_context( + '/', json={'data': {'repository': body_repository_id}}): + login_user(users[2]["obj"]) # comadmin(repoAを担当) + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=[community]): + if allowed: + assert view() == 'ok' + else: + with pytest.raises(Forbidden): + view() + + +# .tox/c1/bin/pytest --cov=weko_admin tests/test_permissions.py::test_repository_scope_required_id_param_not_found -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp +def test_repository_scope_required_id_param_not_found(app, users): + """id_paramで指定したレコードが存在しない場合は404になること。""" + id_model = MagicMock() + id_model.query.filter_by.return_value.one_or_none.return_value = None + + @repository_scope_required(id_param='page_id', id_model=id_model) + def view(*args, **kwargs): + return 'ok' + + with app.test_request_context('/?page_id=999'): + login_user(users[2]["obj"]) # comadmin + with pytest.raises(NotFound): + view() + + +# .tox/c1/bin/pytest --cov=weko_admin tests/test_permissions.py::test_repository_scope_required_no_repository_id -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp +def test_repository_scope_required_no_repository_id(app, users): + """repository_idが取得できない場合は拒否されること。""" + @repository_scope_required(repository_id_param='repo_id') + def view(*args, **kwargs): + return 'ok' + + with app.test_request_context('/'): + login_user(users[2]["obj"]) # comadmin + with pytest.raises(Forbidden): + view() + + +# def _lookup_param(data, kwargs, path): +# .tox/c1/bin/pytest --cov=weko_admin tests/test_permissions.py::test_lookup_param -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp +@pytest.mark.parametrize("data, kwargs, path, expected", [ + # ドット無し: dataを優先 + ({'repo_id': 'repoA'}, {'repo_id': 'repoB'}, 'repo_id', 'repoA'), + # ドット無し: dataに無ければkwargsにフォールバック + ({}, {'repo_id': 'repoB'}, 'repo_id', 'repoB'), + # ドット無し: dataの値がfalsyならkwargsにフォールバック + ({'repo_id': ''}, {'repo_id': 'repoB'}, 'repo_id', 'repoB'), + # ドット無し: どちらにも無い + ({}, {}, 'repo_id', None), + # ドット記法: ネストした値を取得 + ({'data': {'repository': 'repoA'}}, {}, 'data.repository', 'repoA'), + # ドット記法: 途中がdictでない + ({'data': 'foo'}, {}, 'data.repository', None), + ({'data': None}, {}, 'data.repository', None), + ({'data': [1, 2]}, {}, 'data.repository', None), + # ドット記法: キーが存在しない + ({'data': {}}, {}, 'data.repository', None), + ({}, {}, 'data.repository', None), + # ドット記法: 3階層 + ({'a': {'b': {'c': 'repoA'}}}, {}, 'a.b.c', 'repoA'), + ({'a': {'b': {}}}, {}, 'a.b.c', None), + # pathが未指定 + ({'repo_id': 'repoA'}, {}, None, None), + ({'repo_id': 'repoA'}, {}, '', None), +]) +def test_lookup_param(data, kwargs, path, expected): + """_lookup_paramがdata/kwargs/ドット記法から正しく値を取得すること。""" + assert _lookup_param(data, kwargs, path) == expected diff --git a/modules/weko-admin/tests/test_views.py b/modules/weko-admin/tests/test_views.py index 6ecc03d2a2..106eed46c7 100644 --- a/modules/weko-admin/tests/test_views.py +++ b/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 diff --git a/modules/weko-admin/weko_admin/permissions.py b/modules/weko-admin/weko_admin/permissions.py index f23dd0cfcc..0dd37c4102 100644 --- a/modules/weko-admin/weko_admin/permissions.py +++ b/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 diff --git a/modules/weko-admin/weko_admin/views.py b/modules/weko-admin/weko_admin/views.py index 69c5f96e01..60df4d954e 100644 --- a/modules/weko-admin/weko_admin/views.py +++ b/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. diff --git a/modules/weko-gridlayout/tests/conftest.py b/modules/weko-gridlayout/tests/conftest.py index f872aed7d9..517a80f2a6 100644 --- a/modules/weko-gridlayout/tests/conftest.py +++ b/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_) diff --git a/modules/weko-gridlayout/tests/test_views.py b/modules/weko-gridlayout/tests/test_views.py index aa7932184f..91124ad3cd 100644 --- a/modules/weko-gridlayout/tests/test_views.py +++ b/modules/weko-gridlayout/tests/test_views.py @@ -7,7 +7,7 @@ from invenio_cache import current_cache from invenio_accounts.testutils import login_user_via_session -from weko_gridlayout.models import WidgetDesignPage,WidgetDesignSetting +from weko_gridlayout.models import WidgetDesignPage,WidgetDesignSetting,WidgetItem # The endpoints these cases cover carry @login_required and nothing else # (weko_gridlayout/views.py), so every signed-in user reaches them. The 403s @@ -21,6 +21,24 @@ (4, 200), ] +# users indices: 0=contributor, 1=repoadmin, 2=sysadmin, 3=comadmin, +# 4=generaluser (see conftest.users fixture). +# +# load_widget_list_design_setting/save_widget_layout_setting/ +# save_widget_design_page/delete_widget_design_page/delete_widget_item are +# now protected by weko_admin.permissions.repository_scope_required. These +# requests carry no repository_id/page_id/data_id in the body, so +# repository_id cannot be resolved: System/Repository Administrator are +# still allowed unconditionally, but Contributor/Community Administrator/ +# no-role users are rejected because scope cannot be confirmed. +user_results_repo_scope_no_target = [ + (0, 403), + (1, 200), + (2, 200), + (3, 403), + (4, 403), +] + # def preload_pages(): def test_preload_pages(i18n_app, db): @@ -58,7 +76,7 @@ def test_unlocked_widget_guest(client, users): assert res.status_code == 302 -@pytest.mark.parametrize('id, status_code', user_results1) +@pytest.mark.parametrize('id, status_code', user_results_repo_scope_no_target) def test_save_widget_layout_setting_login(client, users, id, status_code): login_user_via_session(client=client, email=users[id]["email"]) with patch("weko_gridlayout.views.WidgetDesignServices.update_widget_design_setting", return_value={}): @@ -76,7 +94,10 @@ def test_save_widget_layout_setting_guest(client, users): assert res.status_code == 302 -@pytest.mark.parametrize('id, status_code', user_results1) +# save_widget_item is now protected by repository_scope_required as well. +# This request carries no data_id/data.repository, so the scope cannot be +# resolved and only System/Repository Administrator are allowed. +@pytest.mark.parametrize('id, status_code', user_results_repo_scope_no_target) def test_save_widget_item_login(client, users, id, status_code): login_user_via_session(client=client, email=users[id]["email"]) with patch("weko_gridlayout.views.WidgetItemServices.save_command", return_value={}): @@ -94,7 +115,99 @@ def test_save_widget_item_guest(client, users): assert res.status_code == 302 -@pytest.mark.parametrize('id, status_code', user_results1) +# save_widget_item scope check (repository_scope_required). +# data_id -> the widget being updated (source repository, taken from DB), +# data.repository -> the repository requested by the client (destination). +# .tox/c1/bin/pytest --cov=weko_gridlayout tests/test_views.py -k save_widget_item_scope -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-gridlayout/.tox/c1/tmp +def test_save_widget_item_scope_no_role(client, users, widget_item): + """ロールなしログインユーザーは拒否されること。""" + login_user_via_session(client=client, email=users[4]["email"]) + with patch("weko_gridlayout.views.WidgetItemServices.save_command", return_value={}): + res = client.post( + "/admin/save_widget_item", + data=json.dumps({"flag_edit": False, + "data": {"repository": "Root Index"}}), + content_type="application/json") + assert res.status_code == 403 + + +@pytest.mark.parametrize('id', [1, 2]) # repoadmin, sysadmin +def test_save_widget_item_scope_super_user(client, users, widget_item, id): + """System/Repository Administratorは新規作成・更新とも許可されること。""" + login_user_via_session(client=client, email=users[id]["email"]) + with patch("weko_gridlayout.views.WidgetItemServices.save_command", return_value={}): + # create + res = client.post( + "/admin/save_widget_item", + data=json.dumps({"flag_edit": False, + "data": {"repository": "Root Index"}}), + content_type="application/json") + assert res.status_code == 200 + + # update + res = client.post( + "/admin/save_widget_item", + data=json.dumps({"flag_edit": True, + "data_id": widget_item[0].widget_id, + "data": {"repository": "Root Index"}}), + content_type="application/json") + assert res.status_code == 200 + + +@pytest.mark.parametrize('repository, status_code', + [("Root Index", 200), ("other", 403)]) +def test_save_widget_item_scope_comadmin_create(client, users, repository, + status_code): + """Community Administratorの新規作成は担当コミュニティのみ許可されること。""" + login_user_via_session(client=client, email=users[3]["email"]) # comadmin + community = MagicMock(id="Root Index") + with patch("weko_gridlayout.views.WidgetItemServices.save_command", return_value={}): + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=[community]): + res = client.post( + "/admin/save_widget_item", + data=json.dumps({"flag_edit": False, + "data": {"repository": repository}}), + content_type="application/json") + assert res.status_code == status_code + + +@pytest.mark.parametrize('repository, status_code', + [("Root Index", 200), ("other", 403)]) +def test_save_widget_item_scope_comadmin_update(client, users, widget_item, + repository, status_code): + """担当内ウィジェットの更新は許可され、担当外への移動は拒否されること。""" + login_user_via_session(client=client, email=users[3]["email"]) # comadmin + community = MagicMock(id="Root Index") + with patch("weko_gridlayout.views.WidgetItemServices.save_command", return_value={}): + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=[community]): + res = client.post( + "/admin/save_widget_item", + data=json.dumps({"flag_edit": True, + "data_id": widget_item[0].widget_id, + "data": {"repository": repository}}), + content_type="application/json") + assert res.status_code == status_code + + +def test_save_widget_item_scope_not_found(client, users, widget_item): + """存在しないdata_idを指定した場合は404になること。""" + login_user_via_session(client=client, email=users[3]["email"]) # comadmin + community = MagicMock(id="Root Index") + with patch("weko_gridlayout.views.WidgetItemServices.save_command", return_value={}): + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=[community]): + res = client.post( + "/admin/save_widget_item", + data=json.dumps({"flag_edit": True, + "data_id": 999999, + "data": {"repository": "Root Index"}}), + content_type="application/json") + assert res.status_code == 404 + + +@pytest.mark.parametrize('id, status_code', user_results_repo_scope_no_target) def test_save_widget_design_page_login(client, users, id, status_code): login_user_via_session(client=client, email=users[id]["email"]) with patch("weko_gridlayout.views.WidgetDesignPageServices.add_or_update_page", return_value={}): @@ -112,7 +225,7 @@ def test_save_widget_design_page_guest(client, users): assert res.status_code == 302 -@pytest.mark.parametrize('id, status_code', user_results1) +@pytest.mark.parametrize('id, status_code', user_results_repo_scope_no_target) def test_load_widget_list_design_setting_login(client, users, id, status_code): from weko_gridlayout import views from weko_gridlayout.services import WidgetDesignServices @@ -144,7 +257,11 @@ def test_load_widget_list_design_setting_issue50978(client, users): views.get_default_language = Mock() WidgetDesignServices.get_widget_list = Mock(return_value={}) WidgetDesignServices.get_widget_preview = Mock(return_value={}) - login_user_via_session(client=client, email=users[3]["email"]) + # Use a System Administrator here: load_widget_list_design_setting is now + # protected by repository_scope_required, and this test targets the + # view's own request-body validation (400), not the scope check, so it + # must bypass the scope check via the unconditional super-user path. + login_user_via_session(client=client, email=users[2]["email"]) # no request data res = client.post("/admin/load_widget_list_design_setting") @@ -291,7 +408,7 @@ def test_load_widget_design_page_issue50978(client, users): assert res4.status_code == 400 -@pytest.mark.parametrize('id, status_code', user_results1) +@pytest.mark.parametrize('id, status_code', user_results_repo_scope_no_target) def test_delete_widget_item_login(client, users, id, status_code): login_user_via_session(client=client, email=users[id]["email"]) with patch("weko_gridlayout.views.WidgetItemServices.delete_by_id", return_value={}): @@ -311,7 +428,10 @@ def test_delete_widget_item_guest(client, users): # .tox/c1/bin/pytest --cov=weko_gridlayout tests/test_views.py::test_load_widget_design_page_issue50978 -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-items-ui/.tox/c1/tmp def test_delete_widget_item_issue50978(client, users): - login_user_via_session(client=client, email=users[3]["email"]) + # System Administrator: delete_widget_item is now protected by + # repository_scope_required and this test targets the view's own + # request-body validation (400), so it must bypass the scope check. + login_user_via_session(client=client, email=users[2]["email"]) with patch("weko_gridlayout.views.WidgetItemServices.delete_by_id", return_value={}): # no request data. The view reads request.headers['Content-Type'] # directly, so it needs the header even when there is no body. @@ -328,7 +448,7 @@ def test_delete_widget_item_issue50978(client, users): assert res4.status_code == 400 -@pytest.mark.parametrize('id, status_code', user_results1) +@pytest.mark.parametrize('id, status_code', user_results_repo_scope_no_target) def test_delete_widget_design_page_login(client, users, id, status_code): login_user_via_session(client=client, email=users[id]["email"]) with patch("weko_gridlayout.views.WidgetDesignPageServices.delete_page", return_value={}): @@ -348,7 +468,10 @@ def test_delete_widget_design_page_guest(client, users): # .tox/c1/bin/pytest --cov=weko_gridlayout tests/test_views.py::test_delete_widget_design_page_issue50978 -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-items-ui/.tox/c1/tmp def test_delete_widget_design_page_issue50978(client, users): - login_user_via_session(client=client, email=users[3]["email"]) + # System Administrator: delete_widget_design_page is now protected by + # repository_scope_required and this test targets the view's own + # request-body validation (400), so it must bypass the scope check. + login_user_via_session(client=client, email=users[2]["email"]) with patch("weko_gridlayout.views.WidgetDesignPageServices.delete_page", return_value={}): # no request data res3 = client.post( @@ -435,7 +558,10 @@ def test_save_widget_design_page(client, users): # .tox/c1/bin/pytest --cov=weko_gridlayout tests/test_views.py::test_save_widget_design_page_issue50978 -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-items-ui/.tox/c1/tmp def test_save_widget_design_page_issue50978(client, users): - login_user_via_session(client=client, email=users[3]["email"]) + # System Administrator: save_widget_design_page is now protected by + # repository_scope_required and this test targets the view's own + # request-body validation (400), so it must bypass the scope check. + login_user_via_session(client=client, email=users[2]["email"]) # no request data res3 = client.post( "/admin/save_widget_design_page", @@ -878,3 +1004,343 @@ def test_unlocked_widget_issue50978(client, users): content_type="application/json" ) assert res4.status_code == 400 + + +# --------------------------------------------------------------------------- +# repository_scope_required coverage +# +# no.283/284/288/289/292: load_widget_list_design_setting, +# save_widget_layout_setting, save_widget_design_page, +# delete_widget_design_page and delete_widget_item are now protected by +# weko_admin.permissions.repository_scope_required so that only +# System/Repository Administrator or the Community Administrator in charge +# of the target repository can operate on it. +# +# users indices (see conftest.users): 0=contributor, 1=repoadmin, +# 2=sysadmin, 3=comadmin, 4=generaluser, 7=plain user with no role. +# --------------------------------------------------------------------------- + +IN_SCOPE_COMMUNITY = [MagicMock(id='Root Index')] +OUT_OF_SCOPE_COMMUNITY = [MagicMock(id='OtherRepo')] + + +# def load_widget_list_design_setting(): +@pytest.mark.parametrize('id, status_code', [(1, 200), (2, 200)]) +def test_load_widget_list_design_setting_scope_super_user( + client, users, id, status_code): + """System/Repository Administratorは無条件で許可されること。""" + from weko_gridlayout.services import WidgetDesignServices + WidgetDesignServices.get_widget_list = Mock(return_value={}) + WidgetDesignServices.get_widget_preview = Mock(return_value={}) + login_user_via_session(client=client, email=users[id]["email"]) + res = client.post("/admin/load_widget_list_design_setting", + data=json.dumps({"repository_id": "Root Index"}), + content_type="application/json") + assert res.status_code == status_code + + +def test_load_widget_list_design_setting_scope_community_admin_allowed( + client, users): + """担当コミュニティのCommunity Administratorは許可されること。""" + from weko_gridlayout.services import WidgetDesignServices + WidgetDesignServices.get_widget_list = Mock(return_value={}) + WidgetDesignServices.get_widget_preview = Mock(return_value={}) + login_user_via_session(client=client, email=users[3]["email"]) + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=IN_SCOPE_COMMUNITY): + res = client.post("/admin/load_widget_list_design_setting", + data=json.dumps({"repository_id": "Root Index"}), + content_type="application/json") + assert res.status_code == 200 + + +def test_load_widget_list_design_setting_scope_community_admin_denied( + client, users): + """担当外コミュニティのCommunity Administratorは拒否されること。""" + login_user_via_session(client=client, email=users[3]["email"]) + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=OUT_OF_SCOPE_COMMUNITY): + res = client.post("/admin/load_widget_list_design_setting", + data=json.dumps({"repository_id": "Root Index"}), + content_type="application/json") + assert res.status_code == 403 + + +def test_load_widget_list_design_setting_scope_no_role(client, users): + """ロールなしログインユーザーは拒否されること。""" + login_user_via_session(client=client, email=users[4]["email"]) + res = client.post("/admin/load_widget_list_design_setting", + data=json.dumps({"repository_id": "Root Index"}), + content_type="application/json") + assert res.status_code == 403 + + +def test_load_widget_list_design_setting_scope_anonymous(client, users): + """未ログインの場合はログイン画面へリダイレクトされること(login_requiredが先に働く)。""" + res = client.post("/admin/load_widget_list_design_setting", + data=json.dumps({"repository_id": "Root Index"}), + content_type="application/json") + assert res.status_code == 302 + + +# def save_widget_layout_setting(): +def test_save_widget_layout_setting_scope_community_admin_allowed( + client, users): + """担当コミュニティのCommunity Administratorは許可されること。""" + login_user_via_session(client=client, email=users[3]["email"]) + with patch("weko_gridlayout.views.WidgetDesignServices.update_widget_design_setting", + return_value={}): + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=IN_SCOPE_COMMUNITY): + res = client.post( + "/admin/save_widget_layout_setting", + data=json.dumps({"repository_id": "Root Index", "page_id": 0}), + content_type="application/json") + assert res.status_code == 200 + + +def test_save_widget_layout_setting_scope_community_admin_denied( + client, users): + """担当外コミュニティのCommunity Administratorは拒否されること。""" + login_user_via_session(client=client, email=users[3]["email"]) + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=OUT_OF_SCOPE_COMMUNITY): + res = client.post( + "/admin/save_widget_layout_setting", + data=json.dumps({"repository_id": "Root Index", "page_id": 0}), + content_type="application/json") + assert res.status_code == 403 + + +def test_save_widget_layout_setting_scope_no_role(client, users): + """ロールなしログインユーザーは拒否されること。""" + login_user_via_session(client=client, email=users[4]["email"]) + res = client.post( + "/admin/save_widget_layout_setting", + data=json.dumps({"repository_id": "Root Index", "page_id": 0}), + content_type="application/json") + assert res.status_code == 403 + + +def test_save_widget_layout_setting_scope_anonymous(client, users): + """未ログインの場合はログイン画面へリダイレクトされること(login_requiredが先に働く)。""" + res = client.post( + "/admin/save_widget_layout_setting", + data=json.dumps({"repository_id": "Root Index", "page_id": 0}), + content_type="application/json") + assert res.status_code == 302 + + +def test_save_widget_layout_setting_scope_both_params_required( + client, users, db_register): + """既存ページ更新時はDB側とbody側の双方が担当範囲内であることを要求すること。 + + page_id=1の実データは'Root Index'(担当内)に属するが、bodyでは担当外の + OtherRepoが指定されているため拒否される。 + """ + login_user_via_session(client=client, email=users[3]["email"]) + with patch("weko_gridlayout.views.WidgetDesignServices.update_widget_design_setting", + return_value={}): + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=IN_SCOPE_COMMUNITY): + res = client.post( + "/admin/save_widget_layout_setting", + data=json.dumps({"repository_id": "OtherRepo", "page_id": 1}), + content_type="application/json") + assert res.status_code == 403 + + +def test_save_widget_layout_setting_scope_id_param_not_found( + client, users, db_register): + """id_paramで指定したページが存在しない場合は404になること。""" + login_user_via_session(client=client, email=users[3]["email"]) + res = client.post( + "/admin/save_widget_layout_setting", + data=json.dumps({"repository_id": "Root Index", "page_id": 999}), + content_type="application/json") + assert res.status_code == 404 + + +# def save_widget_design_page(): +def test_save_widget_design_page_scope_community_admin_allowed( + client, users): + """担当コミュニティのCommunity Administratorは許可されること。""" + login_user_via_session(client=client, email=users[3]["email"]) + with patch("weko_gridlayout.views.WidgetDesignPageServices.add_or_update_page", + return_value={}): + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=IN_SCOPE_COMMUNITY): + res = client.post( + "/admin/save_widget_design_page", + data=json.dumps({"repository_id": "Root Index", "page_id": 0}), + content_type="application/json") + assert res.status_code == 200 + + +def test_save_widget_design_page_scope_community_admin_denied(client, users): + """担当外コミュニティのCommunity Administratorは拒否されること。""" + login_user_via_session(client=client, email=users[3]["email"]) + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=OUT_OF_SCOPE_COMMUNITY): + res = client.post( + "/admin/save_widget_design_page", + data=json.dumps({"repository_id": "Root Index", "page_id": 0}), + content_type="application/json") + assert res.status_code == 403 + + +def test_save_widget_design_page_scope_no_role(client, users): + """ロールなしログインユーザーは拒否されること。""" + login_user_via_session(client=client, email=users[4]["email"]) + res = client.post( + "/admin/save_widget_design_page", + data=json.dumps({"repository_id": "Root Index", "page_id": 0}), + content_type="application/json") + assert res.status_code == 403 + + +def test_save_widget_design_page_scope_anonymous(client, users): + """未ログインの場合はログイン画面へリダイレクトされること(login_requiredが先に働く)。""" + res = client.post( + "/admin/save_widget_design_page", + data=json.dumps({"repository_id": "Root Index", "page_id": 0}), + content_type="application/json") + assert res.status_code == 302 + + +def test_save_widget_design_page_scope_both_params_required( + client, users, db_register): + """既存ページ更新時はDB側とbody側の双方が担当範囲内であることを要求すること。""" + login_user_via_session(client=client, email=users[3]["email"]) + with patch("weko_gridlayout.views.WidgetDesignPageServices.add_or_update_page", + return_value={}): + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=IN_SCOPE_COMMUNITY): + res = client.post( + "/admin/save_widget_design_page", + data=json.dumps({"repository_id": "OtherRepo", "page_id": 1}), + content_type="application/json") + assert res.status_code == 403 + + +# def delete_widget_design_page(): +def test_delete_widget_design_page_scope_community_admin_allowed( + client, users, db_register): + """担当コミュニティのCommunity Administratorは許可されること。""" + login_user_via_session(client=client, email=users[3]["email"]) + with patch("weko_gridlayout.views.WidgetDesignPageServices.delete_page", + return_value={}): + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=IN_SCOPE_COMMUNITY): + res = client.post( + "/admin/delete_widget_design_page", + data=json.dumps({"page_id": 1}), + content_type="application/json") + assert res.status_code == 200 + + +def test_delete_widget_design_page_scope_community_admin_denied( + client, users, db_register): + """担当外コミュニティのCommunity Administratorは拒否されること。""" + login_user_via_session(client=client, email=users[3]["email"]) + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=OUT_OF_SCOPE_COMMUNITY): + res = client.post( + "/admin/delete_widget_design_page", + data=json.dumps({"page_id": 1}), + content_type="application/json") + assert res.status_code == 403 + + +def test_delete_widget_design_page_scope_no_role(client, users, db_register): + """ロールなしログインユーザーは拒否されること。""" + login_user_via_session(client=client, email=users[4]["email"]) + res = client.post( + "/admin/delete_widget_design_page", + data=json.dumps({"page_id": 1}), + content_type="application/json") + assert res.status_code == 403 + + +def test_delete_widget_design_page_scope_anonymous(client, users, db_register): + """未ログインの場合はログイン画面へリダイレクトされること(login_requiredが先に働く)。""" + res = client.post( + "/admin/delete_widget_design_page", + data=json.dumps({"page_id": 1}), + content_type="application/json") + assert res.status_code == 302 + + +def test_delete_widget_design_page_scope_id_param_not_found(client, users): + """id_paramで指定したページが存在しない場合は404になること。""" + login_user_via_session(client=client, email=users[3]["email"]) + res = client.post( + "/admin/delete_widget_design_page", + data=json.dumps({"page_id": 999}), + content_type="application/json") + assert res.status_code == 404 + + +# def delete_widget_item(): +# +# WidgetItem's primary key column is widget_id, not id. If pk_attr were left +# at its default ('id'), WidgetItem.query.filter_by(id=...) would raise +# InvalidRequestError (no such column) and turn into a 500 for a Community +# Administrator (System/Repository Administrator would never hit this path +# since they return early). This is the specific regression this decorator +# application must avoid on this endpoint. +def test_delete_widget_item_scope_community_admin_allowed_pk_attr( + client, users, db_register): + """pk_attr='widget_id'が機能し、担当コミュニティ管理者が例外なく200になること。""" + login_user_via_session(client=client, email=users[3]["email"]) + with patch("weko_gridlayout.views.WidgetItemServices.delete_by_id", + return_value={}): + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=IN_SCOPE_COMMUNITY): + res = client.post( + "/admin/delete_widget_item", + data=json.dumps({"data_id": 1}), + content_type="application/json") + assert res.status_code == 200 + + +def test_delete_widget_item_scope_community_admin_denied( + client, users, db_register): + """担当外コミュニティのCommunity Administratorは拒否されること。""" + login_user_via_session(client=client, email=users[3]["email"]) + with patch("invenio_communities.models.Community.get_repositories_by_user", + return_value=OUT_OF_SCOPE_COMMUNITY): + res = client.post( + "/admin/delete_widget_item", + data=json.dumps({"data_id": 1}), + content_type="application/json") + assert res.status_code == 403 + + +def test_delete_widget_item_scope_no_role(client, users, db_register): + """ロールなしログインユーザーは拒否されること。""" + login_user_via_session(client=client, email=users[4]["email"]) + res = client.post( + "/admin/delete_widget_item", + data=json.dumps({"data_id": 1}), + content_type="application/json") + assert res.status_code == 403 + + +def test_delete_widget_item_scope_anonymous(client, users, db_register): + """未ログインの場合はログイン画面へリダイレクトされること(login_requiredが先に働く)。""" + res = client.post( + "/admin/delete_widget_item", + data=json.dumps({"data_id": 1}), + content_type="application/json") + assert res.status_code == 302 + + +def test_delete_widget_item_scope_id_param_not_found(client, users): + """id_paramで指定したウィジェットが存在しない場合は404になること。""" + login_user_via_session(client=client, email=users[3]["email"]) + res = client.post( + "/admin/delete_widget_item", + data=json.dumps({"data_id": 999}), + content_type="application/json") + assert res.status_code == 404 diff --git a/modules/weko-gridlayout/weko_gridlayout/views.py b/modules/weko-gridlayout/weko_gridlayout/views.py index e83d4fbbda..8782433997 100644 --- a/modules/weko-gridlayout/weko_gridlayout/views.py +++ b/modules/weko-gridlayout/weko_gridlayout/views.py @@ -22,9 +22,11 @@ from werkzeug.exceptions import NotFound from invenio_db import db +from weko_admin.permissions import repository_scope_required + from .api import WidgetItems from .config import WEKO_GRIDLAYOUT_ACCESS_COUNTER_TYPE -from .models import WidgetDesignPage +from .models import WidgetDesignPage, WidgetItem from .services import WidgetDataLoaderServices, WidgetDesignPageServices, \ WidgetDesignServices, WidgetItemServices from .utils import WidgetBucket, get_default_language, \ @@ -143,6 +145,7 @@ def load_widget_design_page_setting(page_id: str, current_language=''): @blueprint_api.route('/load_widget_list_design_setting', methods=['POST']) @login_required +@repository_scope_required(repository_id_param='repository_id') def load_widget_list_design_setting(): """Get Widget list, to display on the Widget List panel on UI. @@ -178,6 +181,8 @@ def load_widget_list_design_setting(): @blueprint_api.route('/save_widget_layout_setting', methods=['POST']) @login_required +@repository_scope_required(repository_id_param='repository_id', + id_param='page_id', id_model=WidgetDesignPage) # TODO: Allow this to be used for both or make a different path def save_widget_layout_setting(): """Save Widget design setting into DB. @@ -254,6 +259,8 @@ def load_widget_design_page(): @blueprint_api.route('/save_widget_design_page', methods=['POST']) @login_required +@repository_scope_required(repository_id_param='repository_id', + id_param='page_id', id_model=WidgetDesignPage) def save_widget_design_page(): """Save Widget design page into DB. @@ -275,6 +282,7 @@ def save_widget_design_page(): @blueprint_api.route('/delete_widget_design_page', methods=['POST']) @login_required +@repository_scope_required(id_param='page_id', id_model=WidgetDesignPage) def delete_widget_design_page(): """Delete Widget design page into DB. @@ -304,6 +312,9 @@ def load_widget_type(): @blueprint_api.route('/save_widget_item', methods=['POST']) @login_required +@repository_scope_required(repository_id_param='data.repository', + id_param='data_id', id_model=WidgetItem, + pk_attr='widget_id') def save_widget_item(): """Save Language List.""" if request.headers['Content-Type'] != 'application/json': @@ -315,6 +326,8 @@ def save_widget_item(): @blueprint_api.route('/delete_widget_item', methods=['POST']) @login_required +@repository_scope_required(id_param='data_id', id_model=WidgetItem, + pk_attr='widget_id') def delete_widget_item(): """Delete Language List.""" if request.headers['Content-Type'] != 'application/json': diff --git a/modules/weko-items-ui/tests/test_utils.py b/modules/weko-items-ui/tests/test_utils.py index ec20391119..4a9be3c7ef 100644 --- a/modules/weko-items-ui/tests/test_utils.py +++ b/modules/weko-items-ui/tests/test_utils.py @@ -8718,6 +8718,96 @@ def test__export_item(app, db, users, db_records, db_itemtype): assert _export_item(1,'JSON',True,'./tests/data/',records_data)[1] == {'1': {'weko_creator_id': '1', "weko_shared_ids": []}} +# issue62783: 非公開/他人所有アイテムのexportスキップ(所有権・公開状態チェック) +# .tox/c1/bin/pytest --cov=weko_items_ui tests/test_utils.py::test__export_item_permission_public_anonymous -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-items-ui/.tox/c1/tmp +def test__export_item_permission_public_anonymous(app, db, users, db_records, db_itemtype): + """未ログイン(匿名)ユーザーが公開アイテムをexport -> 成功すること.""" + depid, recid, parent, doi, record, item = db_records[0] + with app.test_request_context(headers=[("Accept-Language", "en")]): + # current_user未認証(匿名)を想定 + with patch("flask_login.utils._get_user", return_value=None), \ + patch("weko_items_ui.utils.get_user_roles", return_value=(False, None)), \ + patch("weko_items_ui.utils.check_created_id", return_value=False), \ + patch("weko_items_ui.utils.check_publish_status", return_value=True): + exported_item, list_item_role = _export_item( + 1, 'JSON', False, './tests/data/' + ) + assert exported_item != {} + assert exported_item['record_id'] == record.id + + +# .tox/c1/bin/pytest --cov=weko_items_ui tests/test_utils.py::test__export_item_permission_private_anonymous_skip -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-items-ui/.tox/c1/tmp +def test__export_item_permission_private_anonymous_skip(app, db, users, db_records, db_itemtype): + """未ログイン(匿名)ユーザーが非公開アイテムをexport -> スキップされること.""" + depid, recid, parent, doi, record, item = db_records[2] + with app.test_request_context(headers=[("Accept-Language", "en")]): + with patch("flask_login.utils._get_user", return_value=None), \ + patch("weko_items_ui.utils.get_user_roles", return_value=(False, None)), \ + patch("weko_items_ui.utils.check_created_id", return_value=False), \ + patch("weko_items_ui.utils.check_publish_status", return_value=False): + assert _export_item(2, 'JSON', False, './tests/data/') == ({}, {}) + + +# .tox/c1/bin/pytest --cov=weko_items_ui tests/test_utils.py::test__export_item_permission_private_owner -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-items-ui/.tox/c1/tmp +def test__export_item_permission_private_owner(app, db, users, db_records, db_itemtype): + """非公開アイテムをその所有者がexport -> 成功すること.""" + depid, recid, parent, doi, record, item = db_records[2] + with app.test_request_context(headers=[("Accept-Language", "en")]): + with patch("flask_login.utils._get_user", return_value=users[0]["obj"]), \ + patch("weko_items_ui.utils.get_user_roles", return_value=(False, [users[0]["id"]])), \ + patch("weko_items_ui.utils.check_created_id", return_value=True), \ + patch("weko_items_ui.utils.check_publish_status", return_value=False): + exported_item, list_item_role = _export_item( + 2, 'JSON', False, './tests/data/' + ) + assert exported_item != {} + assert exported_item['record_id'] == record.id + + +# .tox/c1/bin/pytest --cov=weko_items_ui tests/test_utils.py::test__export_item_permission_private_other_user_skip -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-items-ui/.tox/c1/tmp +def test__export_item_permission_private_other_user_skip(app, db, users, db_records, db_itemtype): + """非公開アイテムを所有者でない第三者(ログイン済)がexport -> スキップされること.""" + depid, recid, parent, doi, record, item = db_records[2] + with app.test_request_context(headers=[("Accept-Language", "en")]): + with patch("flask_login.utils._get_user", return_value=users[7]["obj"]), \ + patch("weko_items_ui.utils.get_user_roles", return_value=(False, [users[7]["id"]])), \ + patch("weko_items_ui.utils.check_created_id", return_value=False), \ + patch("weko_items_ui.utils.check_publish_status", return_value=False): + assert _export_item(2, 'JSON', False, './tests/data/') == ({}, {}) + + +# .tox/c1/bin/pytest --cov=weko_items_ui tests/test_utils.py::test_export_items_skip_non_permitted_records -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-items-ui/.tox/c1/tmp +def test_export_items_skip_non_permitted_records(app, db_itemtype, db_records, users): + """公開/非公開アイテムが混在する一括exportで、公開アイテムのみ結果に含まれること.""" + post_data = { + 'export_file_contents_radio': 'False', + 'export_format_radio': 'JSON', + 'record_ids': '[1,2]', + 'invalid_record_ids': '[]', + } + with app.test_request_context(headers=[("Accept-Language", "en")]): + with patch("flask_login.utils._get_user", return_value=users[1]["obj"]): + def fake_export_item(record_id, *args, **kwargs): + if str(record_id) == '1': + return ( + { + 'record_id': 1, + 'name': 'recid_1', + 'files': [], + 'path': 'recid_1', + 'item_type_id': '1', + 'researchmap_linkage': '', + }, + {}, + ) + # 非公開かつ権限なし -> スキップ扱い + return {}, {} + + with patch("weko_items_ui.utils._export_item", side_effect=fake_export_item): + res = export_items(post_data) + assert res.status_code == 200 + + # def _custom_export_metadata(record_metadata: dict, hide_item: bool = True, # .tox/c1/bin/pytest --cov=weko_items_ui tests/test_utils.py::test__custom_export_metadata -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-items-ui/.tox/c1/tmp def test__custom_export_metadata(app,db_itemtype,users): diff --git a/modules/weko-items-ui/weko_items_ui/utils.py b/modules/weko-items-ui/weko_items_ui/utils.py index 74e7f6fc76..e3ac780fff 100644 --- a/modules/weko-items-ui/weko_items_ui/utils.py +++ b/modules/weko-items-ui/weko_items_ui/utils.py @@ -2625,6 +2625,8 @@ def export_items(post_data): include_contents, record_path, ) + if not exported_item: + continue # 権限なしレコードはスキップ result['items'].append(exported_item) item_type_id = exported_item.get('item_type_id') @@ -2858,6 +2860,14 @@ def del_hide_sub_metadata(keys, metadata): record = WekoRecord.get_record_by_pid(record_id) list_item_role = {} if record: + roles = get_user_roles() + is_allowed = ( + roles[0] + or check_created_id(record) + or check_publish_status(record) + ) + if not is_allowed: + return {}, {} # 権限なしレコードはスキップ扱い exported_item['record_id'] = record.id exported_item['name'] = 'recid_{}'.format(record_id) exported_item['files'] = [] diff --git a/modules/weko-records-ui/tests/test_views.py b/modules/weko-records-ui/tests/test_views.py index af8ddd607a..c428874d7d 100644 --- a/modules/weko-records-ui/tests/test_views.py +++ b/modules/weko-records-ui/tests/test_views.py @@ -1640,27 +1640,53 @@ def test_preview_able(app): assert ret == False # def get_uri(): +# no.499: get_uri には record_edit_permission_required(param='pid_value') を +# 追加した。JSON body のキー名は pid_value(pid ではない)なので、記録の所有者 +# (records フィクスチャの owner=1 = users[7] "user@test.org")でのみ成功し、 +# 無関係な第三者(users[4] "generaluser@test.org")は拒否されることを確認する。 # .tox/c1/bin/pytest --cov=weko_records_ui tests/test_views.py::test_get_uri -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-records-ui/.tox/c1/tmp -def test_get_uri(app,client,db_sessionlifetime,records): +def test_get_uri(app,client,db_sessionlifetime,records,users): # 404を発生させるとwebassets.exceptions.FilterErrorが発生する対策 app.register_error_handler(404, None) url = url_for("weko_records_ui.get_uri", _external=True) + + # 匿名ユーザー -> 401(login_required) res = client.post(url,data=json.dumps({"uri":"https://localhost/record/1/files/001.jpg","pid_value":"1","accessrole":"1"}), content_type='application/json') - assert res.status_code == 200 - assert json.loads(res.data)=={'status': True} + assert res.status_code == 401 - res = client.post(url,data=json.dumps({"uri":"https://localhost/001.jpg","pid_value":"1","accessrole":"1"}), content_type='application/json') - assert res.status_code == 200 - assert json.loads(res.data)=={'status': True} + # レコード編集権限のないユーザー(第三者) -> 403 + with patch("flask_login.utils._get_user", return_value=users[4]["obj"]): + res = client.post(url,data=json.dumps({"uri":"https://localhost/record/1/files/001.jpg","pid_value":"1","accessrole":"1"}), content_type='application/json') + assert res.status_code == 403 - # Invalid request data - res = client.post("/get_uri") - assert res.status_code == 400 + # レコード編集権限のあるユーザー(所有者) -> 成功 + with patch("flask_login.utils._get_user", return_value=users[7]["obj"]): + res = client.post(url,data=json.dumps({"uri":"https://localhost/record/1/files/001.jpg","pid_value":"1","accessrole":"1"}), content_type='application/json') + assert res.status_code == 200 + assert json.loads(res.data)=={'status': True} - # Invalid pid_value - res = client.post(url,data=json.dumps({"uri":"https://localhost/001.jpg","pid_value":"test","accessrole":"1"}), content_type='application/json', follow_redirects=False) - assert res.status_code == 404 + res = client.post(url,data=json.dumps({"uri":"https://localhost/001.jpg","pid_value":"1","accessrole":"1"}), content_type='application/json') + assert res.status_code == 200 + assert json.loads(res.data)=={'status': True} + + # Invalid request data + res = client.post("/get_uri") + assert res.status_code == 400 + + # Invalid pid_value + # record_edit_permission_required が先に check_created_id_by_recid("test") + # を評価し、存在しない recid なので permitted=False として 403 を返す + # (view 本体の NoResultFound/PIDDoesNotExistError -> 404 処理まで到達しない) + res = client.post(url,data=json.dumps({"uri":"https://localhost/001.jpg","pid_value":"test","accessrole":"1"}), content_type='application/json', follow_redirects=False) + assert res.status_code == 403 + + # pid_value ではなく別のキー名(pid)で送るとパラメータが見つからず 400 + # (record_edit_permission_required(param='pid_value') が正しいキー名を + # 見ていることの確認) + with patch("flask_login.utils._get_user", return_value=users[7]["obj"]): + res = client.post(url,data=json.dumps({"uri":"https://localhost/record/1/files/001.jpg","pid":"1","accessrole":"1"}), content_type='application/json') + assert res.status_code == 400 # .tox/c1/bin/pytest --cov=weko_records_ui tests/test_views.py::test_default_view_method_fix35133 -v -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-records-ui/.tox/c1/tmp diff --git a/modules/weko-records-ui/weko_records_ui/rest.py b/modules/weko-records-ui/weko_records_ui/rest.py index 7e3fb0cd7f..bcf506d688 100644 --- a/modules/weko-records-ui/weko_records_ui/rest.py +++ b/modules/weko-records-ui/weko_records_ui/rest.py @@ -650,8 +650,8 @@ def __init__(self, serializers, ctx, *args, **kwargs): for key, value in ctx.items(): setattr(self, key, value) - # @pass_record - # @need_record_permission('read_permission_factory') + @require_api_auth(allow_anonymous=True) + @require_oauth_scopes(item_read_scope.id) def get(self, pid_value, **kwargs): """Render citation for record according to style and language.""" from weko_records.serializers import citeproc_v1 @@ -660,6 +660,8 @@ def get(self, pid_value, **kwargs): try: pid = PersistentIdentifier.get('depid', pid_value) record = WekoRecord.get_record(pid.object_uuid) + if not page_permission_factory(record).can(): + raise PermissionError() result = citeproc_v1.serialize(pid, record, style=style, locale=locale) result = escape_str(result) @@ -667,7 +669,7 @@ def get(self, pid_value, **kwargs): except Exception: current_app.logger.exception( 'Citation formatting for record {0} failed.'.format( - str(record.id))) + str(pid_value))) # record.id ではなく pid_value を参照(UnboundLocalError修正) return make_response(jsonify("Not found"), 404) diff --git a/modules/weko-records-ui/weko_records_ui/views.py b/modules/weko-records-ui/weko_records_ui/views.py index a5eb04f6c3..e118b35043 100644 --- a/modules/weko-records-ui/weko_records_ui/views.py +++ b/modules/weko-records-ui/weko_records_ui/views.py @@ -1459,6 +1459,8 @@ def preview_able(file_json): return True @blueprint.route("/get_uri", methods=['POST']) +@login_required +@record_edit_permission_required(param='pid_value') def get_uri(): """_summary_ --- diff --git a/modules/weko-theme/weko_theme/templates/weko_theme/macros/tabs_selector.html b/modules/weko-theme/weko_theme/templates/weko_theme/macros/tabs_selector.html index 93b2549dd5..071a82f245 100644 --- a/modules/weko-theme/weko_theme/templates/weko_theme/macros/tabs_selector.html +++ b/modules/weko-theme/weko_theme/templates/weko_theme/macros/tabs_selector.html @@ -21,12 +21,12 @@ {% macro tabs_selector(tab_value='top',community_id='') %} {%- if community_id %}
  • {{ _('Top') }}
    • - {%- if current_user.is_authenticated and current_user.roles %} + {%- if current_user.is_authenticated and current_user.roles | selectattr('name', 'in', ['System Administrator', 'Repository Administrator', 'Community Administrator', 'Contributor']) | list %}
  • {{ _('WorkFlow') }}
    • {%- endif %} {%- else %}
  • {{ _('Top') }}
    • - {%- if current_user.is_authenticated and current_user.roles %} + {%- if current_user.is_authenticated and current_user.roles | selectattr('name', 'in', ['System Administrator', 'Repository Administrator', 'Community Administrator', 'Contributor']) | list %}
  • {{ _('WorkFlow') }}
    • {%- endif %} {% if display_community %} diff --git a/modules/weko-workflow/tests/test_views.py b/modules/weko-workflow/tests/test_views.py index cc22744b2f..134176e250 100644 --- a/modules/weko-workflow/tests/test_views.py +++ b/modules/weko-workflow/tests/test_views.py @@ -11,7 +11,7 @@ from sqlalchemy.orm.attributes import flag_modified from sqlalchemy.orm import joinedload -from flask import json, jsonify, url_for, make_response, current_app +from flask import json, jsonify, url_for, make_response, current_app, g from flask_babelex import gettext as _ from invenio_db import db from sqlalchemy.exc import SQLAlchemyError, StatementError @@ -36,6 +36,7 @@ display_guest_activity, display_guest_activity_item_application, render_guest_workflow) +from weko_workflow.utils import generate_guest_activity_token_value from marshmallow.exceptions import ValidationError from weko_records_ui.models import FileOnetimeDownload, FilePermission from weko_records.models import ItemMetadata, ItemReference @@ -366,6 +367,47 @@ def test_new_activity(client, db, users, mocker): assert kwargs['community'] is None +# no.600/998: new_activity は item-access (Contributor以上) が無いと403になること +# .tox/c1/bin/pytest --cov=weko_workflow tests/test_views.py::test_new_activity_acl -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-workflow/.tox/c1/tmp +@pytest.mark.parametrize('users_index, status_code', [ + (0, 200), # contributor: has item-access + (1, 200), # repoadmin: has item-access + (2, 200), # sysadmin: superuser-access + (3, 200), # comadmin: has item-access + (4, 403), # generaluser: no item-access (no.600/998 fix) + (5, 403), # originalroleuser: no item-access (no.600/998 fix) + (6, 200), # originalroleuser2: also has repoadmin role -> item-access +]) +def test_new_activity_acl(client, db, users, mocker, users_index, status_code): + login(client=client, email=users[users_index]['email']) + + with patch("weko_workflow.views.WorkFlow.get_workflow_list", return_value=['wf1', 'wf2']), \ + patch("weko_workflow.views.WorkFlow.get_workflows_by_roles", return_value=['wf1']), \ + patch("weko_workflow.views.GetCommunity.get_community_by_id") as mock_get_community, \ + patch("weko_theme.utils.get_design_layout", return_value=('page_obj', 'widgets')), \ + patch("weko_workflow.utils.exclude_admin_workflow"), \ + patch("weko_workflow.views.render_template") as mock_render_template: + + mock_community = MagicMock() + mock_community.id = 'comm01' + mock_get_community.return_value = mock_community + mock_render_template.return_value = 'rendered_html' + + url = url_for('weko_workflow.new_activity', c='comm01') + res = client.get(url) + assert res.status_code == status_code + if status_code == 200: + mock_render_template.assert_called_once() + else: + mock_render_template.assert_not_called() + + +def test_new_activity_acl_nologin(client, db): + """no.600/998: 未ログインは@login_requiredでログイン画面へリダイレクトされること.""" + url = url_for('weko_workflow.new_activity') + res = client.get(url) + assert res.status_code == 302 + # .tox/c1/bin/pytest --cov=weko_workflow tests/test_views.py::test_init_activity_acl_nologin -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-workflow/.tox/c1/tmp def test_init_activity_acl_nologin(client,db_register2): @@ -689,13 +731,13 @@ def __init__(self): # .tox/c1/bin/pytest --cov=weko_workflow tests/test_views.py::test_list_activity_acl -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-workflow/.tox/c1/tmp @pytest.mark.parametrize('users_index, status_code', [ - (0, 200), - (1, 200), - (2, 200), - (3, 200), - (4, 200), - (5, 200), - (6, 200), + (0, 200), # contributor: has item-access + (1, 200), # repoadmin: has item-access + (2, 200), # sysadmin: superuser-access + (3, 200), # comadmin: has item-access + (4, 403), # generaluser: no item-access (no.602/1000 fix) + (5, 403), # originalroleuser: no item-access (no.602/1000 fix) + (6, 200), # originalroleuser2: also has repoadmin role -> item-access ]) def test_list_activity_acl(client, users, mocker, users_index, status_code): class MockActivity: @@ -715,20 +757,27 @@ def __init__(self): mock_render = mocker.patch('weko_workflow.views.render_template', return_value=jsonify({})) res = client.get(url) assert res.status_code == status_code - mock_condition.assert_called_with({}) - mock_activity.assert_called_with(conditions={'con1': 'val1'}) - mock_layout.assert_called_with('Root Index') - mock_render.assert_called_with( - 'weko_workflow/activity_list.html', - page=None, - pages=1, - name_param='pagesall', - size=1, - tab='todo', - maxpage=1, - render_widgets=True, - activities=[activity], - ) + if status_code == 200: + mock_condition.assert_called_with({}) + mock_activity.assert_called_with(conditions={'con1': 'val1'}) + mock_layout.assert_called_with('Root Index') + mock_render.assert_called_with( + 'weko_workflow/activity_list.html', + page=None, + pages=1, + name_param='pagesall', + size=1, + tab='todo', + maxpage=1, + render_widgets=True, + activities=[activity], + ) + else: + # no.602/1000: item-access が無いユーザーは403で拒否され、 + # 一覧取得ロジックには到達しない + mock_condition.assert_not_called() + mock_activity.assert_not_called() + mock_render.assert_not_called() # def init_activity_guest(): # .tox/c1/bin/pytest --cov=weko_workflow tests/test_views.py::test_init_activity_guest_nologin -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-workflow/.tox/c1/tmp @@ -807,6 +856,70 @@ def test_init_activity_guest_users(client, users, db_register_1, db_guestactivit res = client.post(url, json=input) assert res.status_code == status_code + +# no.603/1001: init_activity_guest は不正な形式の guest_mail を拒否すること +# .tox/c1/bin/pytest --cov=weko_workflow tests/test_views.py::test_init_activity_guest_invalid_email -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-workflow/.tox/c1/tmp +def test_init_activity_guest_invalid_email(client, db_register2): + """no.603/1001: guest_mail が不正な形式の場合は400を返すこと.""" + url = url_for('weko_workflow.init_activity_guest') + input = {'guest_mail': 'not-an-email', 'workflow_id': 1, 'flow_id': 1, + 'item_type_id': 1, 'record_id': '1', 'guest_item_title': 'test', + 'file_name': 'test_file'} + res = client.post(url, json=input) + assert res.status_code == 400 + data = json.loads(res.data) + assert data['msg'] == 'Invalid guest_mail' + + +# no.603/1001: init_activity_guest の workflow_id キー欠如でKeyErrorにならないこと +# (post_data["workflow_id"] -> post_data.get('workflow_id') 修正の確認) +# .tox/c1/bin/pytest --cov=weko_workflow tests/test_views.py::test_init_activity_guest_workflow_id_missing_key -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-workflow/.tox/c1/tmp +def test_init_activity_guest_workflow_id_missing_key(client, db_register2, mocker): + """no.603/1001: workflow_id キーが無くてもKeyError(500)にならないこと.""" + url = url_for('weko_workflow.init_activity_guest') + input = {'guest_mail': 'test@guest.com'} # workflow_id キーなし + mock_is_terms = mocker.patch( + 'weko_workflow.views.is_terms_of_use_only', return_value=False) + with patch('weko_workflow.views.init_activity_for_guest_user', + side_effect=Exception('boom')): + res = client.post(url, json=input) + assert res.status_code == 200 + mock_is_terms.assert_called_once_with(None) + data = json.loads(res.data) + assert data['msg'] == 'Cannot send mail' + + +# no.603/1001: init_activity_guest は同一IPからの短時間の連続リクエストを +# 5回/分に制限すること +# .tox/c1/bin/pytest --cov=weko_workflow tests/test_views.py::test_init_activity_guest_rate_limit -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-workflow/.tox/c1/tmp +def test_init_activity_guest_rate_limit(client, db_register2): + """no.603/1001: 6回目のリクエストは429(レート制限)となること.""" + url = url_for('weko_workflow.init_activity_guest') + input = {'guest_mail': 'test@guest.com', 'workflow_id': 1, 'flow_id': 1, + 'item_type_id': 1, 'record_id': '1', 'guest_item_title': 'test', + 'file_name': 'test_file'} + with patch("weko_workflow.views.is_terms_of_use_only", return_value=False), \ + patch('weko_workflow.views.init_activity_for_guest_user', + return_value=(WorkActivity(), 'url')), \ + patch('weko_workflow.views.db.session.commit'), \ + patch('weko_workflow.views.send_usage_application_mail_for_guest_user', + return_value=True): + for _ in range(5): + # NOTE: Flask-Limiter の判定は flask.g のフラグで1リクエストにつき + # 1回だけ行われる。本テストスイートの `app` フィクスチャは + # テスト全体を単一の app_context 内で実行するため、実際のHTTP + # サーバーと異なり client.post() をまたいで flask.g が引き継が + # れてしまう。本番のWSGIサーバーでは各リクエストで g がリセット + # されるため、ここではそれを模してリクエストごとに明示的にリセ + # ットする。 + g.pop('_rate_limiting_complete', None) + res = client.post(url, json=input) + assert res.status_code == 200 + g.pop('_rate_limiting_complete', None) + res = client.post(url, json=input) + assert res.status_code == 429 + + # def display_guest_activity(): # .tox/c1/bin/pytest --cov=weko_workflow tests/test_views.py::test_display_guest_activity -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-workflow/.tox/c1/tmp def test_display_guest_activity(client, users, db_register_full_action, db_guestactivity): @@ -4330,13 +4443,10 @@ def test_get_feedback_maillist(client, users, db_register_full_action, users_ind @pytest.mark.parametrize('users_index, status_code', [ - (0, 200), - #(1, 200), - #(2, 200), - #(3, 200), - #(4, 200), - #(5, 200), - #(6, 200), + (2, 200), # sysadmin: no.619/1017 で追加した @check_authority は + # 管理者を無条件で許可するため、この関数自体の分岐テストへの + # 影響はない。他ロールでのアクセス制御は + # test_get_request_maillist_authority で個別に検証する。 ]) # def get_request_maillist(activity_id='0') #.tox/c1/bin/pytest --cov=weko_workflow tests/test_views.py::test_get_request_maillist -vv -s --cov-branch --cov-report=term --cov-report=html --basetemp=/code/modules/weko-workflow/.tox/c1/tmp @@ -4437,6 +4547,62 @@ def test_get_request_maillist(client, users, users_index, status_code, mocker): {"email":"contributor@test.org","author_id":""}] ) + +# no.619/1017: get_request_maillist に付与した @check_authority(action_idを +# 取らないため新設したフォールバック分岐)のアクセス制御を検証する。 +# 申請者本人・管理者・共有編集者は許可(200)、無関係な第三者は拒否(403)。 +# .tox/c1/bin/pytest --cov=weko_workflow tests/test_views.py::test_get_request_maillist_authority -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-workflow/.tox/c1/tmp +# NOTE: このテストスイートの `app` フィクスチャはテスト全体を単一の +# app_context 内で実行するため、同一テスト内で login() (DBから都度Userを +# 取得する) を複数回呼び出すと2回目以降で +# sqlalchemy.orm.exc.DetachedInstanceError となる(weko_workflow.index等、 +# 本修正と無関係な既存エンドポイントでも同様に再現する、テスト環境固有の +# 制約)。そのため users_role をパラメータ化し、1テストにつきログインは +# 1回のみとする。 +@pytest.mark.parametrize('users_role, allowed', [ + ('owner', True), # 申請者本人 -> 許可 + ('admin', True), # 管理者(System Administrator) -> 許可 + ('shared', True), # 共有編集者(shared_user_ids に含まれる) -> 許可 + ('stranger', False), # 無関係な第三者 -> 拒否 +]) +def test_get_request_maillist_authority(client, db, users, db_register_full_action, + users_role, allowed): + flow_id = db_register_full_action["flow_define"].id + owner = users[0] # contributor: 申請者本人 + admin = users[2] # sysadmin: 管理者 + shared = users[5] # originalroleuser: 共有編集者 + stranger = users[4] # generaluser: 無関係な第三者 + login_user = {'owner': owner, 'admin': admin, 'shared': shared, + 'stranger': stranger}[users_role] + + activity = Activity( + activity_id="A-99999999-90001", workflow_id=1, flow_id=flow_id, + action_id=1, activity_login_user=owner["id"], + activity_update_user=owner["id"], action_order=1, + shared_user_ids=[{"user": shared["id"]}], + temp_data=json.dumps({"metainfo": {}}), + activity_start=datetime.strptime( + '2200/01/11 3:01:53.931', '%Y/%m/%d %H:%M:%S.%f'), + ) + db.session.add(activity) + db.session.commit() + + url = url_for('weko_workflow.get_request_maillist', + activity_id=activity.activity_id) + + with patch('weko_workflow.views.WorkActivity.get_activity_request_mail', + return_value=None): + login(client=client, email=login_user['email']) + res = client.get(url) + # check_authority自体はjsonifyのみでstatus未設定のため常に200 + assert res.status_code == 200 + data = response_data(res) + if allowed: + assert data != {"code": 403, "msg": "Authorization required"} + else: + assert data == {"code": 403, "msg": "Authorization required"} + + @pytest.mark.parametrize('users_index, status_code', [ (0, 200), #(1, 200), @@ -4643,6 +4809,87 @@ def prepare_activity(act_id, recid, with_item=False, is_deleted=False): assert res.status_code == 200 assert json.loads(res.data) == {"code": 200, 'for_delete': False, "is_deleted": True} + +# no.606/1004: verify_deletion の当事者検証(ログインユーザー)。 +# 申請者本人・管理者は許可(200)、無関係な第三者は拒否(403)。 +# NOTE: このテストスイートの `app` フィクスチャはテスト全体を単一の +# app_context 内で実行するため、同一テスト内で login() (DBから都度Userを +# 取得する) を複数回呼び出すと2回目以降で +# sqlalchemy.orm.exc.DetachedInstanceError となる(weko_workflow.index等、 +# 本修正と無関係な既存エンドポイントでも同様に再現する、テスト環境固有の +# 制約)。そのため users_index をパラメータ化し、1テストにつきログインは +# 1回のみとする。 +# .tox/c1/bin/pytest --cov=weko_workflow tests/test_views.py::test_verify_deletion_authority -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-workflow/.tox/c1/tmp +@pytest.mark.parametrize('users_index, status_code', [ + (0, 200), # contributor: 申請者本人 + (4, 403), # generaluser: 無関係な第三者 +]) +def test_verify_deletion_authority(client, db, users, db_register_full_action, + users_index, status_code): + flow_id = db_register_full_action["flow_define"].id + owner = users[0] # contributor: 申請者本人 + + activity = Activity( + activity_id="A-22000222-00001", workflow_id=1, flow_id=flow_id, + action_id=1, activity_login_user=owner["id"], + activity_update_user=owner["id"], action_order=1, + activity_start=datetime.strptime( + '2200/01/11 3:01:53.931', '%Y/%m/%d %H:%M:%S.%f'), + ) + db.session.add(activity) + db.session.commit() + + url = url_for("weko_workflow.verify_deletion", + activity_id=activity.activity_id) + + login(client=client, email=users[users_index]['email']) + res = client.get(url) + assert res.status_code == status_code + data = json.loads(res.data) + if status_code == 200: + assert data['code'] == 200 + else: + assert data == {"code": 403, "msg": "Authorization required"} + + +# no.606/1004: verify_deletion の当事者検証(ゲストユーザー)。 +# session['guest_token'] から復元した activity_id とリクエストの +# activity_id が一致する場合のみ許可すること。 +# .tox/c1/bin/pytest --cov=weko_workflow tests/test_views.py::test_verify_deletion_guest_authority -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-workflow/.tox/c1/tmp +def test_verify_deletion_guest_authority(client, db, db_register_full_action): + flow_id = db_register_full_action["flow_define"].id + activity_id1 = "A-22000222-00002" + activity_id2 = "A-22000222-00003" + for act_id in (activity_id1, activity_id2): + activity = Activity( + activity_id=act_id, workflow_id=1, flow_id=flow_id, + action_id=1, activity_login_user=1, + activity_update_user=1, action_order=1, + activity_start=datetime.strptime( + '2200/01/11 3:01:53.931', '%Y/%m/%d %H:%M:%S.%f'), + ) + db.session.add(activity) + db.session.commit() + + token = generate_guest_activity_token_value( + activity_id1, "test_file", datetime.utcnow(), "guest@test.org") + + # トークンに対応する activity_id へのアクセス -> 許可 + with client.session_transaction() as sess: + sess['guest_token'] = token + url = url_for("weko_workflow.verify_deletion", activity_id=activity_id1) + res = client.get(url) + assert res.status_code == 200 + + # トークンに対応しない他の activity_id へのアクセス -> 拒否 + with client.session_transaction() as sess: + sess['guest_token'] = token + url = url_for("weko_workflow.verify_deletion", activity_id=activity_id2) + res = client.get(url) + data = json.loads(res.data) + assert data == {"code": 403, "msg": "Authorization required"} + + # .tox/c1/bin/pytest --cov=weko_workflow tests/test_views.py::test_display_activity_nologin -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko_workflow/.tox/c1/tmp def test_display_activity_nologin(client,db_register2,mocker): """Test of display activity.""" diff --git a/modules/weko-workflow/weko_workflow/ext.py b/modules/weko-workflow/weko_workflow/ext.py index 21dd40855c..1fcbaf7b0a 100644 --- a/modules/weko-workflow/weko_workflow/ext.py +++ b/modules/weko-workflow/weko_workflow/ext.py @@ -43,11 +43,20 @@ def init_app(self, app): from .sessions import upt_activity_item from .views import depositactivity_blueprint, workflow_blueprint self.init_config(app) + self.init_limiter(app) item_created.connect(upt_activity_item, app) app.register_blueprint(workflow_blueprint) app.register_blueprint(depositactivity_blueprint) app.extensions['weko-workflow'] = self + def init_limiter(self, app): + """Initialize rate limiting. + + :param app: The flask application. + """ + from .utils import limiter + limiter.init_app(app) + def init_config(self, app): """Initialize configuration. diff --git a/modules/weko-workflow/weko_workflow/utils.py b/modules/weko-workflow/weko_workflow/utils.py index 1674860bb5..7ee3a88436 100644 --- a/modules/weko-workflow/weko_workflow/utils.py +++ b/modules/weko-workflow/weko_workflow/utils.py @@ -5249,6 +5249,16 @@ def create_limmiter(): return Limiter(app=Flask(__name__), key_func=get_remote_address, default_limits=WEKO_WORKFLOW_API_LIMIT_RATE_DEFAULT) +# NOTE: create_limmiter() above binds the Limiter to a throw-away Flask app +# (``Flask(__name__)``) instead of the real application, and is never +# init_app()'d against the running app, so it does not actually enforce any +# rate limit. ``limiter`` below is a module-level instance following the +# same pattern as ``weko_accounts.utils.limiter``: it is created unbound and +# then initialized against the real application in +# ``WekoWorkflow.init_limiter`` (see ``ext.py``). +limiter = Limiter(key_func=get_remote_address) + + def convert_to_timezone(dt, user_timezone=None): """ Convert a datetime object to the specified timezone. diff --git a/modules/weko-workflow/weko_workflow/views.py b/modules/weko-workflow/weko_workflow/views.py index cc52d2ee66..71ef993723 100644 --- a/modules/weko-workflow/weko_workflow/views.py +++ b/modules/weko-workflow/weko_workflow/views.py @@ -1,6 +1,7 @@ # -*- coding: utf-8 -*- """Blueprint for weko-workflow.""" +import base64 import json import re import shutil @@ -13,9 +14,12 @@ from typing import List from urllib.parse import urljoin +from email_validator import validate_email, EmailNotValidError +from flask_limiter.util import get_remote_address from weko_admin.models import AdminSettings from weko_items_ui.signals import cris_researchmap_linkage_request from weko_items_ui.models import CRIS_Institutions, CRISLinkageResult +from weko_items_ui.permissions import item_permission from weko_workflow.schema.marshmallow import ActionSchema, \ ActivitySchema, GetRequestMailListSchema, ResponseMessageSchema, CancelSchema, PasswdSchema, LockSchema,\ ResponseLockSchema, LockedValueSchema, GetFeedbackMailListSchema, SaveActivityResponseSchema,\ @@ -85,7 +89,7 @@ from .scopes import activity_scope from .utils import IdentifierHandle, auto_fill_title, \ check_authority_by_admin, check_continue, check_doi_validation_not_pass, \ - check_existed_doi, is_terms_of_use_only, \ + check_existed_doi, is_terms_of_use_only, limiter, \ delete_cache_data, delete_guest_activity, filter_all_condition, \ get_account_info, get_actionid, get_activity_display_info, \ get_application_and_approved_date, get_approval_keys, get_cache_data, \ @@ -420,6 +424,7 @@ def iframe_success(): @workflow_blueprint.route('/activity/new', methods=['GET']) @login_required +@item_permission.require(http_exception=403) def new_activity(): """New activity. Args: @@ -600,6 +605,7 @@ def _generate_download_url(record_id :str ,file_name :str) -> str: @workflow_blueprint.route('/activity/list', methods=['GET']) @login_required +@item_permission.require(http_exception=403) def list_activity(): """List activity.""" activity = WorkActivity() @@ -630,6 +636,7 @@ def list_activity(): @workflow_blueprint.route('/activity/init-guest', methods=['POST']) +@limiter.limit("5 per minute", key_func=get_remote_address) def init_activity_guest(): """Init workflow activity for guest user. Return URL of workflow activity made from the request body. @@ -639,12 +646,18 @@ def init_activity_guest(): """ post_data = request.get_json() + if post_data.get('guest_mail'): + try: + validate_email(post_data.get('guest_mail'), check_deliverability=False) + except EmailNotValidError: + return jsonify(msg='Invalid guest_mail'), 400 + password_for_download = "" if post_data.get('password_for_download'): pwd = post_data['password_for_download'] password_for_download = hash_password(pwd) - if is_terms_of_use_only(post_data["workflow_id"]): + if is_terms_of_use_only(post_data.get('workflow_id')): # if the workflow is terms_of_use_only(利用規約のみ) , # do not create activity. redirect file download. file_name = post_data["file_name"] @@ -780,6 +793,41 @@ def verify_deletion(activity_id="0"): Returns: dict: JSON response with code, is_deleted, and for_delete status. """ + # 当事者検証。login_required_customize は session["guest_token"] の + # 有無(ゲスト)またはログイン有無しか見ないため、activity_id との + # 整合性はここで検証する。 + if not current_user.is_authenticated: + # ゲストの場合: session["guest_token"] をデコードして得た + # activity_id とリクエストの activity_id が一致するかを検証する。 + # guest_token はサーバ側が display_guest_activity 等で発行時に + # 署名検証(oracle10.verify)済みの値をセッション(署名済みcookie)へ + # 格納したものであり、改ざんされていないことが前提のため、ここでは + # base64デコードのみで activity_id を取り出す。 + guest_token = session.get('guest_token') + token_activity_id = None + if guest_token: + try: + decoded = base64.b64decode(guest_token.encode()).decode() + params = decoded.split(" ") + if len(params) == 4: + token_activity_id = params[0] + except Exception as ex: + current_app.logger.debug( + 'failed to decode guest_token: {}'.format(ex)) + if not token_activity_id or token_activity_id != activity_id: + return jsonify(code=403, msg=_('Authorization required')), 403 + else: + # ログインユーザーの場合: System/Repository Administratorは無条件、 + # Community Administratorは担当コミュニティのアクティビティのみ、 + # それ以外は申請者本人のみ許可する。 + activity_for_auth = WorkActivity().get_activity_by_id(activity_id) + if activity_for_auth is None: + return jsonify(code=403, msg=_('Authorization required')), 403 + is_owner = str(activity_for_auth.activity_login_user) == \ + str(current_user.get_id()) + if not is_owner and not check_authority_by_admin(activity_for_auth): + return jsonify(code=403, msg=_('Authorization required')), 403 + is_deleted = False for_delete = False activity = WorkActivity().get_activity_by_id(activity_id) @@ -1270,6 +1318,26 @@ def display_activity(activity_id="0", community_id=None): ) +def _get_shared_user_ids_from_list(shared_user_ids_list): + """Get shared user ids from list. + + モジュールトップレベル関数(元は check_authority_action 内のネスト関数)。 + check_authority(action_idなし分岐)からも共用する。 + + Args: + shared_user_ids_list (list): List of shared user ids. + Returns: + list: List of shared user ids. + """ + shared_user_ids = [] + for shared_user in (shared_user_ids_list or []): + if isinstance(shared_user, dict): + shared_user_ids.append(shared_user.get('user')) + elif isinstance(shared_user, int): + shared_user_ids.append(shared_user) + return shared_user_ids + + def check_authority(func): """Check Authority.""" @wraps(func) @@ -1282,6 +1350,33 @@ def decorated_function(*args, **kwargs): if check_authority_by_admin(activity_detail): return func(*args, **kwargs) + action_id = kwargs.get('action_id') + if action_id is None: + # action_idを取らないエンドポイント向け: + # 申請者本人または共有編集者を許可する + if activity_detail is None: + return jsonify(code=403, msg=_('Authorization required')) + cur_user = int(current_user.get_id()) + if activity_detail.activity_login_user == cur_user: + return func(*args, **kwargs) + shared_ids = [] + im = ItemMetadata.query.filter_by( + id=activity_detail.item_id).one_or_none() + if im: + shared_ids += _get_shared_user_ids_from_list( + im.json.get('shared_user_ids', [])) + shared_ids += _get_shared_user_ids_from_list( + im.json.get('weko_shared_ids', [])) + elif activity_detail.temp_data: + temp_data = json.loads(activity_detail.temp_data) + shared_ids += _get_shared_user_ids_from_list( + activity_detail.shared_user_ids or []) + shared_ids += _get_shared_user_ids_from_list( + temp_data.get('metainfo', {}).get('shared_user_ids', [])) + if cur_user in shared_ids: + return func(*args, **kwargs) + return jsonify(code=403, msg=_('Authorization required')) + is_set, is_allow, is_deny = validate_action_role_user( activity_id=kwargs.get('activity_id'), action_id=kwargs.get('action_id'), @@ -1300,23 +1395,6 @@ def check_authority_action(activity_id='0', action_id=0, contain_login_item_application=False, action_order=0): """Check authority.""" - - def _get_shared_user_ids_from_list(shared_user_ids_list): - """Get shared user ids from list. - - Args: - shared_user_ids_list (list): List of shared user ids. - Returns: - list: List of shared user ids. - """ - shared_user_ids = [] - for shared_user in shared_user_ids_list: - if isinstance(shared_user, dict): - shared_user_ids.append(shared_user.get('user')) - elif isinstance(shared_user, int): - shared_user_ids.append(shared_user) - return shared_user_ids - if not current_user.is_authenticated: return 1 @@ -2879,6 +2957,7 @@ def get_feedback_maillist(activity_id='0'): @workflow_blueprint.route('/get_request_maillist/', methods=['GET']) @login_required +@check_authority def get_request_maillist(activity_id='0'): """アクティビティに設定されているリクエストメール送信先の情報を取得して返す