Hotfix/v2.0.4

modules/invenio-deposit/invenio_deposit/views/rest.py

@@ -28,7 +28,7 @@
 
 import json
 from copy import deepcopy
-from functools import partial
+from functools import partial, wraps
 
 from flask import Blueprint, abort, current_app, make_response, request, \
     url_for
@@ -63,6 +63,26 @@ def create_error_handlers(blueprint):
     records_rest_error_handlers(blueprint)
 
 
+def decorate_view(view_func, decorators, methods):
+    """Apply ``decorators`` to the requests of ``view_func`` in ``methods``.
+
+    :param view_func: The view function.
+    :param decorators: List of decorators or their import strings.
+    :param methods: HTTP methods the decorators are applied to.
+    :returns: The wrapped view function.
+    """
+    decorated = view_func
+    for decorator in reversed(decorators):
+        decorated = obj_or_import_string(decorator)(decorated)
+
+    @wraps(view_func)
+    def dispatch(*args, **kwargs):
+        if request.method in methods:
+            return decorated(*args, **kwargs)
+        return view_func(*args, **kwargs)
+    return dispatch
+
+
 def create_blueprint(endpoints):
     """Create Invenio-Deposit-REST blueprint.
 
@@ -116,6 +136,13 @@ def dbsession_clean(exception):
             '{0}/files/<path:key>'.format(options['item_route'])
         )
 
+        route_decorators = {
+            options['list_route']: (
+                options.pop('create_decorators', []), ('POST',)),
+            options['item_route']: (
+                options.pop('update_decorators', []), ('PUT', 'PATCH')),
+        }
+
         options.setdefault('search_class', DepositSearch)
         search_class = obj_or_import_string(options['search_class'])
 
@@ -127,6 +154,11 @@ def dbsession_clean(exception):
         options.setdefault('indexer_class', None)
 
         for rule in records_rest_url_rules(endpoint, **options):
+            decorators, methods = route_decorators.get(
+                rule['rule'], ([], ()))
+            if decorators:
+                rule['view_func'] = decorate_view(
+                    rule['view_func'], decorators, methods)
             blueprint.add_url_rule(**rule)
 
         search_class_kwargs = {}

modules/invenio-deposit/tests/test_views_rest.py

@@ -623,3 +623,80 @@ def test_simple_rest_flow(app, test_client, api, es, location, fake_schemas, use
             # assert res.status_code == 200
             # data = json.loads(res.data.decode('utf-8'))
             # assert 'Revision 2' == data['metadata']['title']
+
+
+def test_decorate_view():
+    """Test decorators are applied to the given methods only."""
+    from flask import Flask, abort
+
+    from invenio_deposit.views.rest import decorate_view
+
+    def deny(f):
+        def decorated(*args, **kwargs):
+            abort(418)
+        return decorated
+
+    def view():
+        return 'ok'
+
+    app = Flask('testapp')
+    app.add_url_rule(
+        '/items', view_func=decorate_view(view, [deny], ('POST', 'PUT')),
+        methods=['GET', 'POST', 'PUT', 'PATCH'])
+    with app.test_client() as client:
+        assert client.get('/items').status_code == 200
+        assert client.patch('/items').status_code == 200
+        assert client.post('/items').status_code == 418
+        assert client.put('/items').status_code == 418
+
+
+def test_create_blueprint_route_decorators():
+    """Test create_decorators and update_decorators options."""
+    from flask import Flask
+    from invenio_db import InvenioDB
+
+    from invenio_deposit.views.rest import create_blueprint
+
+    endpoints = dict(
+        depid=dict(
+            pid_type='depid',
+            pid_minter='deposit',
+            pid_fetcher='deposit',
+            record_serializers={
+                'application/json': ('invenio_records_rest.serializers'
+                                     ':json_v1_response'),
+            },
+            search_serializers={
+                'application/json': ('invenio_records_rest.serializers'
+                                     ':json_v1_search'),
+            },
+            list_route='/deposits/',
+            item_route='/deposits/<pid_value>',
+            default_media_type='application/json',
+            create_decorators=[_abort_factory(418)],
+            update_decorators=[_abort_factory(409)],
+        )
+    )
+    app = Flask('testapp')
+    app.config.update(
+        SQLALCHEMY_DATABASE_URI='sqlite://',
+        SQLALCHEMY_TRACK_MODIFICATIONS=False,
+    )
+    InvenioDB(app)
+    app.register_blueprint(create_blueprint(endpoints))
+    with app.test_client() as client:
+        assert client.post('/deposits/').status_code == 418
+        assert client.put('/deposits/1').status_code == 409
+        assert client.patch('/deposits/1').status_code == 409
+    assert 'create_decorators' in endpoints['depid']
+    assert 'update_decorators' in endpoints['depid']
+
+
+def _abort_factory(code):
+    from flask import abort
+
+    def decorator(f):
+        def decorated(*args, **kwargs):
+            abort(code)
+        return decorated
+    return decorator

modules/weko-admin/tests/conftest.py

@@ -132,6 +132,8 @@ def base_app(instance_path, cache_config,request ,search_class):
     os.environ['INVENIO_ELASTICSEARCH_HOST']='elasticsearch_test'
     app_ = Flask('test_weko_admin_app', instance_path=instance_path)
     app_.config.update(
+        WEKO_WORKFLOW_GUEST_TOKEN_ISSUE_ENABLED=True,
+        WEKO_WORKFLOW_GUEST_TOKEN_ACCEPT_ENABLED=True,
         SERVER_NAME='test_server',
         ACCOUNTS_USE_CELERY=False,
         SECRET_KEY='SECRET_KEY',

modules/weko-admin/tests/test_utils.py

@@ -2328,3 +2328,33 @@ def test_get_detail_search_list(i18n_app, users):
     with patch("flask_login.utils._get_user", return_value=users[3]['obj']):
         result =  get_detail_search_list()
         assert result
+
+
+# class UsageReport: send_reminder_mail
+def test_usage_report_send_reminder_mail_guest_token_disabled():
+    from flask import Flask
+
+    from weko_admin.utils import UsageReport
+
+    app = Flask('testapp')
+    app.config.update(
+        THEME_SITEURL='https://example.org',
+        THEME_INSTITUTION_NAME={'ja': '', 'en': ''},
+        WEKO_WORKFLOW_GUEST_TOKEN_ISSUE_ENABLED=False,
+    )
+    from mock import MagicMock
+
+    activity = MagicMock()
+    activity.extra_info = {'is_guest': True, 'guest_mail': 'a@example.org'}
+    with app.test_request_context():
+        with patch('weko_admin.utils.UsageReport._UsageReport__get_site_info',
+                   return_value=('', '')), \
+                patch('weko_admin.utils.UsageReport.'
+                      '_UsageReport__get_default_mail_sender',
+                      return_value=''):
+            usage_report = UsageReport()
+            with patch.object(usage_report,
+                              '_UsageReport__process_send_mail') as send:
+                assert usage_report.send_reminder_mail(
+                    [], 'mail-id', [activity]) is True
+                send.assert_not_called()

modules/weko-admin/weko_admin/utils.py

@@ -1875,12 +1875,13 @@ def __init__(self):
         from weko_workflow.api import WorkActivity
         from weko_workflow.models import ActionStatusPolicy
         from weko_workflow.utils import generate_guest_activity_token_value, \
-            process_send_mail
+            is_guest_token_issue_enabled, process_send_mail
         self.__activities_id = []
         self.__activities_number = 0
         self.__work_activity = WorkActivity()
         self.__action_status_policy = ActionStatusPolicy()
         self.__activity_token_value = generate_guest_activity_token_value
+        self.__is_guest_token_issue_enabled = is_guest_token_issue_enabled
         self.__process_send_mail = process_send_mail
         self.__page_number = 1
         self.__usage_report_activities_data = []
@@ -1991,6 +1992,9 @@ def send_reminder_mail(self, activities_id: list,
                 .get("WEKO_WORKFLOW_REQUEST_FOR_REGISTER_USAGE_REPORT")
 
         for activity in activities:
+            if (activity.extra_info or {}).get('is_guest') \
+                    and not self.__is_guest_token_issue_enabled():
+                continue
             if activity.item_id:
                 records_id.append(activity.item_id)
 

modules/weko-deposit/tests/test_weko_deposit.py

@@ -81,3 +81,19 @@ def test_ItemResource_put(app, db):
             res = client.put("/deposits/redirect/1",
                             data=json.dumps(data), headers=headers)
             assert res.status_code == 200
+
+
+def test_deposit_rest_endpoints_route_decorators():
+    """Test route decorators of the deposit REST endpoints."""
+    from weko_deposit.config import (DEPOSIT_REST_ENDPOINTS,
+                                     WEKO_DEPOSIT_REST_ENDPOINTS)
+
+    expected = ['weko_accounts.utils:login_required_customize']
+    assert DEPOSIT_REST_ENDPOINTS['depid']['create_decorators'] == expected
+    assert WEKO_DEPOSIT_REST_ENDPOINTS['depid']['create_decorators'] \
+        == expected
+
+    expected = ['weko_deposit.rest:require_item_edit_permission']
+    assert DEPOSIT_REST_ENDPOINTS['depid']['update_decorators'] == expected
+    assert WEKO_DEPOSIT_REST_ENDPOINTS['depid']['update_decorators'] \
+        == expected

modules/weko-deposit/weko_deposit/config.py

@@ -128,6 +128,8 @@
         # read_permission_factory_imp='',
         # update_permission_factory_imp='',
         delete_permission_factory_imp=deny_all,
+        create_decorators=['weko_accounts.utils:login_required_customize'],
+        update_decorators=['weko_deposit.rest:require_item_edit_permission'],
     )
 )
 

modules/weko-records-ui/tests/conftest.py

@@ -190,6 +190,8 @@ def base_app(instance_path):
 
     # os.environ['FPDF_FONT_DIR'] = "/code/modules/weko-records-ui/weko_records_ui/fonts/"
     app_.config.update(
+        WEKO_WORKFLOW_GUEST_TOKEN_ISSUE_ENABLED=True,
+        WEKO_WORKFLOW_GUEST_TOKEN_ACCEPT_ENABLED=True,
         CELERY_ALWAYS_EAGER=True,
         CELERY_CACHE_BACKEND="memory",
         CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,

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

@@ -1879,3 +1879,19 @@ def test_delete_version(app, records):
                     "weko_records_ui.utils.call_external_system") as mock_external:
                     delete_version("2.0")
                     mock_external.assert_not_called()
+
+
+# def create_usage_report_for_user
+def test_create_usage_report_for_user_guest_token_disabled():
+    from flask import Flask
+
+    from weko_records_ui.utils import create_usage_report_for_user
+
+    app = Flask('testapp')
+    app.config['WEKO_WORKFLOW_GUEST_TOKEN_ISSUE_ENABLED'] = False
+    with app.app_context():
+        with patch('weko_records_ui.utils.WorkActivity') as work_activity:
+            assert create_usage_report_for_user(
+                {'usage_application_activity_id': 'A-1',
+                 'is_guest': True}) is None
+            work_activity.assert_not_called()

modules/weko-records-ui/weko_records_ui/rest.py

@@ -61,6 +61,7 @@
 from weko_workflow.models import GuestActivity
 from weko_workflow.scopes import activity_scope
 from weko_workflow.utils import check_etag, check_pretty, init_activity_for_guest_user
+from weko_workflow.utils import is_guest_token_issue_enabled
 
 from .errors import AvailableFilesNotFoundRESTError, ContentsNotFoundError, DateFormatRESTError, \
     FilesNotFoundRESTError, InternalServerError, InvalidEmailError, InvalidRequestError, \
@@ -502,6 +503,8 @@ def post_v1(self, **kwargs):
             except Exception as ex:
                 # invalid email
                 raise InvalidEmailError() # 400 Error
+            if not is_guest_token_issue_enabled():
+                raise PermissionError()  # 403 Error
 
         # Check pretty
         check_pretty(param_pretty)

modules/weko-records-ui/weko_records_ui/utils.py

@@ -1094,7 +1094,10 @@ def create_usage_report_for_user(onetime_download_extra_info: dict):
     is_guest = onetime_download_extra_info.get('is_guest', False)
 
     # Get Usage Application Activity.
-    from weko_workflow.utils import create_record_metadata_for_user
+    from weko_workflow.utils import create_record_metadata_for_user, \
+        is_guest_token_issue_enabled
+    if is_guest and not is_guest_token_issue_enabled():
+        return None
     usage_application_activity = WorkActivity().get_activity_by_id(
         activity_id)
 

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

@@ -43,3 +43,17 @@ def test_init(app):
     # assert 'weko-search-ui' not in app.extensions
     # ext.init_app(app)
     # assert 'weko-search-ui' in app.extensions
+
+
+def test_records_rest_endpoints_permission_factories():
+    """Test permission factories of the records REST endpoints."""
+    from weko_search_ui.config import RECORDS_REST_ENDPOINTS
+
+    for endpoint in ("recid", "opensearch", "worksapce"):
+        options = RECORDS_REST_ENDPOINTS[endpoint]
+        for key in (
+            "create_permission_factory_imp",
+            "update_permission_factory_imp",
+            "delete_permission_factory_imp",
+        ):
+            assert options[key] == "invenio_records_rest.utils:deny_all"

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

@@ -138,6 +138,17 @@
     "application/json": ("weko_records.serializers:opensearch_v1_search"),
 }
 
+for _endpoint in ("recid", "opensearch", "worksapce"):
+    RECORDS_REST_ENDPOINTS[_endpoint][
+        "create_permission_factory_imp"
+    ] = "invenio_records_rest.utils:deny_all"
+    RECORDS_REST_ENDPOINTS[_endpoint][
+        "update_permission_factory_imp"
+    ] = "invenio_records_rest.utils:deny_all"
+    RECORDS_REST_ENDPOINTS[_endpoint][
+        "delete_permission_factory_imp"
+    ] = "invenio_records_rest.utils:deny_all"
+
 # RECORDS_REST_ENDPOINTS['recid']['read_permission_factory_imp'] = allow_all
 
 INDEXER_DEFAULT_INDEX = "{}-weko-item-v1.0.0".format(index_prefix)  # Use direct index

modules/weko-workflow/tests/conftest.py

@@ -218,6 +218,8 @@ def base_app(instance_path, search_class, cache_config):
     """Flask application fixture."""
     app_ = Flask('testapp', instance_path=instance_path)
     app_.config.update(
+        WEKO_WORKFLOW_GUEST_TOKEN_ISSUE_ENABLED=True,
+        WEKO_WORKFLOW_GUEST_TOKEN_ACCEPT_ENABLED=True,
         SECRET_KEY='SECRET_KEY',
         TESTING=True,
         SERVER_NAME='TEST_SERVER.localdomain',