diff --git a/modules/weko-records-ui/tests/test_utils.py b/modules/weko-records-ui/tests/test_utils.py index 680831e434..f0148fc4ab 100644 --- a/modules/weko-records-ui/tests/test_utils.py +++ b/modules/weko-records-ui/tests/test_utils.py @@ -60,6 +60,7 @@ can_manage_secret_url, validate_token, validate_url_download, + ensure_url_record_matches, generate_one_time_download_url, parse_one_time_download_token, ) @@ -762,27 +763,50 @@ def test_parse_one_time_download_token(app): def test_is_private_index(app,records): indexer, results = records record = results[0]["record"] + # Regression: the fixture record belongs to a single public index. assert is_private_index(record)==False - data1 = [ - [0, 1, 2, 3, 4, 5, 6], - [0, 1, 2, 3, 4, 5, 6], - [0, 1, 2, 3, 4, 5, 6], - [0, 1, 2, 3, 4, 5, 6], - [0, 1, 2, 3, 4, 5, 6], - [0, 1, 2, 3, 4, 5, 6], - [0, 1, 2, 3, 4, 5, 6], - ] - with patch("weko_index_tree.api.Indexes.get_path_list", return_value=data1): - assert is_private_index(record) == False +# def is_private_index(record): +# .tox/c1/bin/pytest --cov=weko_records_ui tests/test_utils.py::test_is_private_index_decision_table -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-records-ui/.tox/c1/tmp +@pytest.mark.parametrize( + "delegate_return, expected", + [ + pytest.param( + True, False, + id="rule1_at_least_one_fully_public_chain"), + pytest.param( + False, True, + id="rule2_no_fully_public_chain_eg_future_dated_or_nonpublic_ancestor"), + pytest.param( + None, True, + id="rule3_delegate_returns_none_eg_all_indexes_deleted"), + ], +) +def test_is_private_index_decision_table(delegate_return, expected): + with patch( + "weko_index_tree.api.Indexes.is_public_state_and_not_in_future", + return_value=delegate_return, + ) as mock_delegate: + assert is_private_index({"path": ["1", "2"]}) is expected + mock_delegate.assert_called_once_with(["1", "2"]) - data1 = [ - [0, 1, 2, 3, 4, 5, False], - ] - with patch("weko_index_tree.api.Indexes.get_path_list", return_value=data1): - assert is_private_index(record) == True +@pytest.mark.parametrize( + "record", + [ + pytest.param({"path": []}, id="rule4_empty_path_list"), + pytest.param({}, id="rule5_no_path_key_at_all"), + ], +) +def test_is_private_index_no_index_short_circuits(record): + # An item that belongs to no index at all must be treated as private + # without even querying Indexes.is_public_state_and_not_in_future(). + with patch( + "weko_index_tree.api.Indexes.is_public_state_and_not_in_future" + ) as mock_delegate: + assert is_private_index(record) is True + mock_delegate.assert_not_called() # def validate_download_record(record: dict): @@ -1480,6 +1504,13 @@ def test_validate_token(app, users): match = re.search(r'[?&]token=([^&]+)', url) onetime_token = match.group(1) assert validate_token(onetime_token, is_secret_url=False) is True + + # Cross-type reuse: a validly-signed secret-URL token must not + # validate via the onetime-URL path, and vice versa. + with app.test_request_context(): + assert validate_token(secret_token, is_secret_url=False) is False + assert validate_token(onetime_token, is_secret_url=True) is False + invalid_bytes = b'\xb2q\xff\x19\xaf\xfc\xc6T\x8bt\xd6\xf6\xc6 \ \x08D\xe7\xf3G;cN\x1bn|\xa2\x88\x01v\xed\x1cA_1' with app.test_request_context(): @@ -1564,6 +1595,8 @@ def test_convert_token_into_obj(vldt_token, app, users): @patch('weko_records_ui.utils.validate_download_record') def test_validate_url_download(vldt_record, vldt_file, is_enabled, vldt_token, app, db, users): + matching_record = {'recid': 1} + matching_filename = 'test.txt' with app.test_request_context(): secret_obj = FileSecretDownload.create( creator_id=1, @@ -1579,7 +1612,8 @@ def test_validate_url_download(vldt_record, vldt_file, is_enabled, vldt_token, is_enabled.return_value = True vldt_file.return_value = True vldt_record.return_value = True - assert validate_url_download('', '', secret_token, True) == (True, '') + assert validate_url_download( + matching_record, matching_filename, secret_token, True) == (True, '') with app.test_request_context(): onetime_obj = FileOnetimeDownload.create( @@ -1594,58 +1628,177 @@ def test_validate_url_download(vldt_record, vldt_file, is_enabled, vldt_token, match = re.search(r'[?&]token=([^&]+)', create_download_url(onetime_obj)) onetime_token = match.group(1) db.session.flush() - assert validate_url_download('', '', onetime_token, False) == (True, '') + assert validate_url_download( + matching_record, matching_filename, onetime_token, False) == (True, '') with patch('weko_records_ui.utils.validate_token', return_value=False): - assert validate_url_download('', '', secret_token, True) == ( + assert validate_url_download( + matching_record, matching_filename, secret_token, True) == ( False, 'The provided token is invalid.') with patch('weko_records_ui.utils.is_secret_url_feature_enabled', return_value=False): - assert validate_url_download('', '', secret_token, True) == ( + assert validate_url_download( + matching_record, matching_filename, secret_token, True) == ( False, 'This feature is currently disabled.') with patch('weko_records_ui.utils.validate_file_access', return_value=False): - assert validate_url_download('', '', secret_token, True) == ( + assert validate_url_download( + matching_record, matching_filename, secret_token, True) == ( False, 'This file is currently not available for this feature.') with patch('weko_records_ui.utils.validate_download_record', return_value=False): - assert validate_url_download('', '', secret_token, True) == ( + assert validate_url_download( + matching_record, matching_filename, secret_token, True) == ( False, 'This file is currently not available for this feature.') + # record_id/file_name mismatch: Reject + assert validate_url_download( + {'recid': 999}, matching_filename, secret_token, True) == ( + False, 'This file is currently not available for this feature.') + + # record_id match, file_name mismatch: not a real file on the item: Reject + assert validate_url_download( + matching_record, 'nonexistent.txt', secret_token, True) == ( + False, 'This file is currently not available for this feature.') + + # record_id match, file_name mismatch: a different, legitimately + # existing file on the *same* item: Reject + assert validate_url_download( + matching_record, 'other_legit_file.txt', secret_token, True) == ( + False, 'This file is currently not available for this feature.') + + # record_id mismatch, file_name mismatch: Reject + assert validate_url_download( + {'recid': 999}, 'other.txt', secret_token, True) == ( + False, 'This file is currently not available for this feature.') + + # Same four mismatch cases as above (record_id only / file_name only + # with a nonexistent name / file_name only with a different real + # file / both), repeated for the onetime-URL path with onetime_token, + # to confirm the match check applies identically to both URL types. + assert validate_url_download( + {'recid': 999}, matching_filename, onetime_token, False) == ( + False, 'This file is currently not available for this feature.') + assert validate_url_download( + matching_record, 'nonexistent.txt', onetime_token, False) == ( + False, 'This file is currently not available for this feature.') + assert validate_url_download( + matching_record, 'other_legit_file.txt', onetime_token, False) == ( + False, 'This file is currently not available for this feature.') + assert validate_url_download( + {'recid': 999}, 'other.txt', onetime_token, False) == ( + False, 'This file is currently not available for this feature.') + secret_obj.is_deleted = True db.session.commit() - assert validate_url_download('', '', secret_token, True) == ( + assert validate_url_download( + matching_record, matching_filename, secret_token, True) == ( False, 'This URL has been deactivated.') secret_obj.is_deleted = False secret_obj.download_count = 10 db.session.commit() - assert validate_url_download('', '', secret_token, True) == ( + assert validate_url_download( + matching_record, matching_filename, secret_token, True) == ( False, 'The download limit has been exceeded.') secret_obj.download_count = 0 db.session.commit() with patch('weko_records_ui.utils.dt') as mock_dt: mock_dt.now.return_value = dt.now(timezone.utc) + timedelta(days=31) - assert validate_url_download('', '', secret_token, True) == ( + assert validate_url_download( + matching_record, matching_filename, secret_token, True) == ( False, 'The expiration date for download has been exceeded.') onetime_obj.is_deleted = True db.session.commit() - assert validate_url_download('', '', onetime_token, False) == ( + assert validate_url_download( + matching_record, matching_filename, onetime_token, False) == ( False, 'This URL has been deactivated.') onetime_obj.is_deleted = False onetime_obj.download_count = 10 db.session.commit() - assert validate_url_download('', '', onetime_token, False) == ( + assert validate_url_download( + matching_record, matching_filename, onetime_token, False) == ( False, 'The download limit has been exceeded.') onetime_obj.download_count = 0 db.session.commit() with patch('weko_records_ui.utils.dt') as mock_dt: mock_dt.now.return_value = dt.now(timezone.utc) + timedelta(days=31) - assert validate_url_download('', '', onetime_token, False) == ( + assert validate_url_download( + matching_record, matching_filename, onetime_token, False) == ( False, 'The expiration date for download has been exceeded.') +# def ensure_url_record_matches(url_obj, record_id, file_name): +# .tox/c1/bin/pytest --cov=weko_records_ui tests/test_utils.py::test_ensure_url_record_matches -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-records-ui/.tox/c1/tmp +def test_ensure_url_record_matches(): + + # Build real FileSecretDownload instances. + def make_secret(record_id, file_name): + return FileSecretDownload( + creator_id=1, record_id=record_id, file_name=file_name, + label_name='test', expiration_date=dt.now(timezone.utc), + download_limit=1) + + # Build real FileOnetimeDownload instances. + def make_onetime(record_id, file_name): + return FileOnetimeDownload( + approver_id=1, record_id=record_id, file_name=file_name, + expiration_date=dt.now(timezone.utc), download_limit=1, + user_mail='test@example.org', is_guest=False, extra_info={}) + + # Decision table. + url_obj = make_secret(record_id='1', file_name='test.txt') + # rule 1: record_id match, file_name match -> allow (True) + assert ensure_url_record_matches(url_obj, '1', 'test.txt') is True + # rule 2: record_id match, file_name mismatch -> reject (False) + assert ensure_url_record_matches(url_obj, '1', 'other.txt') is False + # rule 3: record_id mismatch, file_name match -> reject (False) + assert ensure_url_record_matches(url_obj, '2', 'test.txt') is False + # rule 4: record_id mismatch, file_name mismatch -> reject (False) + assert ensure_url_record_matches(url_obj, '2', 'other.txt') is False + + # Both accepted url_obj types must be recognized. + onetime_obj = make_onetime(record_id='1', file_name='test.txt') + assert ensure_url_record_matches(onetime_obj, '1', 'test.txt') is True + + # url_obj is None: rejected without raising. + assert ensure_url_record_matches(None, '1', 'test.txt') is False + + # url_obj is of a type other than FileSecretDownload/FileOnetimeDownload. + class UnexpectedType: + def __init__(self, record_id, file_name): + self.record_id = record_id + self.file_name = file_name + assert ensure_url_record_matches( + UnexpectedType(record_id='1', file_name='test.txt'), + '1', 'test.txt') is False + + # Boundary/abnormal inputs must not raise, and must be rejected when + # they do not match a non-None url_obj. + assert ensure_url_record_matches(url_obj, None, 'test.txt') is False + assert ensure_url_record_matches(url_obj, '1', None) is False + assert ensure_url_record_matches(url_obj, '', 'test.txt') is False + + # url_obj.record_id and record_id have different types: compared + # after str() conversion, so equal values should still match. + int_id_obj = make_secret(record_id=1, file_name='test.txt') + assert ensure_url_record_matches(int_id_obj, '1', 'test.txt') is True + assert ensure_url_record_matches(int_id_obj, 1, 'test.txt') is True + assert ensure_url_record_matches(int_id_obj, '2', 'test.txt') is False + + # file_name containing path traversal / script-injection-like + # characters must not raise, and must be rejected unless it is an + # exact match. + malicious_names = ['../../etc/passwd', '', + "' OR '1'='1", '--', 'a' * 5000] + for name in malicious_names: + assert ensure_url_record_matches(url_obj, '1', name) is False + # An exact match (even of an unusual file name) is still allowed. + exact_obj = make_secret(record_id='1', file_name=name) + assert ensure_url_record_matches(exact_obj, '1', name) is True + + # .tox/c1/bin/pytest --cov=weko_records_ui tests/test_utils.py::test_validate_file_access -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-records-ui/.tox/c1/tmp def test_validate_file_access(): with patch('weko_records_ui.utils.is_secret_file', return_value=True): diff --git a/modules/weko-records-ui/tests/test_views.py b/modules/weko-records-ui/tests/test_views.py index 3384266728..1373ad6459 100644 --- a/modules/weko-records-ui/tests/test_views.py +++ b/modules/weko-records-ui/tests/test_views.py @@ -846,21 +846,47 @@ def test_copy_secret_url(app, client, records): assert res.json['url'] == expected_secret_url with patch('weko_records_ui.views.can_manage_secret_url', return_value=False): - with pytest.raises(Exception): - res = client.get(url) - assert res.status_code == 403 - with patch('weko_records_ui.views.create_download_url', + # abort(403) must propagate as an actual 403 response. + res = client.get(url) + assert res.status_code == 403 + with patch('weko_records_ui.views.can_manage_secret_url', + return_value=True), \ + patch('weko_records_ui.views.create_download_url', side_effect=Exception('Test Error')): + # can_manage_secret_url must stay True here, otherwise the real + # permission check would reject with 403 before ever reaching + # create_download_url(), and this would not test a genuine 500. res = client.get(url) assert res.status_code == 500 with patch('weko_records_ui.views.can_manage_secret_url', return_value=True): + # secret_url_id does not exist url = url_for('invenio_records_ui.recid_copy_secret_url', pid_value=records[1]['recid'].pid_value, filename=records[1]['filename'], secret_url_id=99) # invalid secret_url_id - res = client.get(url) - assert res.json['url'] is None + res_not_exist = client.get(url) + assert res_not_exist.status_code == 404 + + # secret_url_id exists, but belongs to a *different* item. + # Its file_name is deliberately set to match the *requested* filename. + other_secret_obj = FileSecretDownload.create( + creator_id=1, + record_id=records[0]['recid'].pid_value, + file_name=records[1]['filename'], + label_name='other item link', + expiration_date=datetime.now(timezone.utc) + timedelta(days=1), + download_limit=1, + ) + mismatched_url = url_for( + 'invenio_records_ui.recid_copy_secret_url', + pid_value=records[1]['recid'].pid_value, + filename=records[1]['filename'], + secret_url_id=other_secret_obj.id) + res_mismatch = client.get(mismatched_url) + assert res_mismatch.status_code == 404 + # The mismatch/not exist response must be indistinguishable to the caller. + assert res_mismatch.get_data() == res_not_exist.get_data() # .tox/c1/bin/pytest --cov=weko_records_ui tests/test_views.py::test_copy_onetime_url -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-records-ui/.tox/c1/tmp @@ -894,21 +920,49 @@ def test_copy_onetime_url(app, client, records): assert res.json['url'] == expected_onetime_url with patch('weko_records_ui.views.can_manage_onetime_url', return_value=False): - with pytest.raises(Exception): - res = client.get(url) - assert res.status_code == 403 - with patch('weko_records_ui.views.create_download_url', + # abort(403) must propagate as an actual 403 response + res = client.get(url) + assert res.status_code == 403 + with patch('weko_records_ui.views.can_manage_onetime_url', + return_value=True), \ + patch('weko_records_ui.views.create_download_url', side_effect=Exception('Test Error')): + # can_manage_onetime_url must stay True here, otherwise the real + # permission check would reject with 403 before ever reaching + # create_download_url(), and this would not test a genuine 500. res = client.get(url) assert res.status_code == 500 with patch('weko_records_ui.views.can_manage_onetime_url', return_value=True): + # onetime_url_id does not exist url = url_for('invenio_records_ui.recid_copy_onetime_url', pid_value=records[1]['recid'].pid_value, filename=records[1]['filename'], onetime_url_id=99) # invalid onetime_url_id - res = client.get(url) - assert res.json['url'] is None + res_not_exist = client.get(url) + assert res_not_exist.status_code == 404 + + # onetime_url_id exists, but belongs to a *different* item. + # Its file_name is deliberately set to match the *requested* filename. + other_onetime_obj = FileOnetimeDownload.create( + approver_id=1, + record_id=records[0]['recid'].pid_value, + file_name=records[1]['filename'], + expiration_date=datetime.now(timezone.utc) + timedelta(days=1), + download_limit=1, + user_mail='test@example.org', + is_guest=False, + extra_info={} + ) + mismatched_url = url_for( + 'invenio_records_ui.recid_copy_onetime_url', + pid_value=records[1]['recid'].pid_value, + filename=records[1]['filename'], + onetime_url_id=other_onetime_obj.id) + res_mismatch = client.get(mismatched_url) + assert res_mismatch.status_code == 404 + # The mismatch/not exist response must be indistinguishable to the caller. + assert res_mismatch.get_data() == res_not_exist.get_data() # .tox/c1/bin/pytest --cov=weko_records_ui tests/test_views.py::test_delete_secret_url -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-records-ui/.tox/c1/tmp @@ -936,11 +990,16 @@ def test_delete_secret_url(client, records): assert secret_obj.is_deleted == True with patch('weko_records_ui.views.can_manage_secret_url', return_value=False): - with pytest.raises(Exception): - res = client.delete(url) - assert res.status_code == 403 - with patch('weko_records_ui.models.FileSecretDownload.delete_logically', + # abort(403) must propagate as an actual 403 response. + res = client.delete(url) + assert res.status_code == 403 + with patch('weko_records_ui.views.can_manage_secret_url', + return_value=True), \ + patch('weko_records_ui.models.FileSecretDownload.delete_logically', side_effect=Exception('Test Error')): + # can_manage_secret_url must stay True here, otherwise the real + # permission check would reject with 403 before ever reaching + # delete_logically(), and this would not test a genuine 500. res = client.delete(url) assert res.status_code == 500 with patch('weko_records_ui.views.can_manage_secret_url', @@ -949,9 +1008,32 @@ def test_delete_secret_url(client, records): pid_value=records[1]['recid'].pid_value, filename=records[1]['filename'], secret_url_id=99) # invalid secret_url_id - with pytest.raises(Exception): - res = client.delete(url) - assert res.status_code == 404 + # abort(404) must propagate as an actual 404 response, not be + # swallowed into a 500. + res_not_exist = client.delete(url) + assert res_not_exist.status_code == 404 + + # secret_url_id exists, but belongs to a *different* item. + # Its file_name is deliberately set to match the *requested* filename. + other_secret_obj = FileSecretDownload.create( + creator_id=1, + record_id=records[0]['recid'].pid_value, + file_name=records[1]['filename'], + label_name='other item link', + expiration_date=datetime.now(timezone.utc) + timedelta(days=1), + download_limit=1, + ) + mismatched_url = url_for( + 'invenio_records_ui.recid_delete_secret_url', + pid_value=records[1]['recid'].pid_value, + filename=records[1]['filename'], + secret_url_id=other_secret_obj.id) + res_mismatch = client.delete(mismatched_url) + assert res_mismatch.status_code == 404 + # The mismatch/not exist response must be indistinguishable to the caller. + assert res_mismatch.get_data() == res_not_exist.get_data() + # The other item's secret URL must remain untouched. + assert other_secret_obj.is_deleted == False # .tox/c1/bin/pytest --cov=weko_records_ui tests/test_views.py::test_delete_onetime_url -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-records-ui/.tox/c1/tmp @@ -981,11 +1063,17 @@ def test_delete_onetime_url(client, records): assert onetime_obj.is_deleted == True with patch('weko_records_ui.views.can_manage_onetime_url', return_value=False): - with pytest.raises(Exception): - res = client.delete(url) - assert res.status_code == 403 - with patch('weko_records_ui.models.FileOnetimeDownload.delete_logically', + # abort(403) must propagate as an actual 403 response. + res = client.delete(url) + assert res.status_code == 403 + with patch('weko_records_ui.views.can_manage_onetime_url', + return_value=True), \ + patch('weko_records_ui.models.FileOnetimeDownload.delete_logically', side_effect=Exception('Test Error')): + # can_manage_onetime_url must stay True here, otherwise the + # real permission check would reject with 403 before ever + # reaching delete_logically(), and this would not test a + # genuine 500. res = client.delete(url) assert res.status_code == 500 with patch('weko_records_ui.views.can_manage_onetime_url', @@ -994,9 +1082,34 @@ def test_delete_onetime_url(client, records): pid_value=records[1]['recid'].pid_value, filename=records[1]['filename'], onetime_url_id=99) - with pytest.raises(Exception): - res = client.delete(url) - assert res.status_code == 404 + # abort(404) must propagate as an actual 404 response, not be + # swallowed into a 500. + res_not_exist = client.delete(url) + assert res_not_exist.status_code == 404 + + # onetime_url_id exists, but belongs to a *different* item. + # Its file_name is deliberately set to match the *requested* filename. + other_onetime_obj = FileOnetimeDownload.create( + approver_id=1, + record_id=records[0]['recid'].pid_value, + file_name=records[1]['filename'], + expiration_date=datetime.now(timezone.utc) + timedelta(days=1), + download_limit=1, + user_mail='test@example.org', + is_guest=False, + extra_info={} + ) + mismatched_url = url_for( + 'invenio_records_ui.recid_delete_onetime_url', + pid_value=records[1]['recid'].pid_value, + filename=records[1]['filename'], + onetime_url_id=other_onetime_obj.id) + res_mismatch = client.delete(mismatched_url) + assert res_mismatch.status_code == 404 + # The mismatch/not exist response must be indistinguishable to the caller. + assert res_mismatch.get_data() == res_not_exist.get_data() + # The other item's onetime URL must remain untouched. + assert other_onetime_obj.is_deleted == False # def default_view_method(pid, record, filename=None, template=None, **kwargs): diff --git a/modules/weko-records-ui/weko_records_ui/utils.py b/modules/weko-records-ui/weko_records_ui/utils.py index 31d433ed95..f1b920c51a 100644 --- a/modules/weko-records-ui/weko_records_ui/utils.py +++ b/modules/weko-records-ui/weko_records_ui/utils.py @@ -1328,23 +1328,31 @@ def validate_onetime_download_token( def is_private_index(record): - """Check index of workflow is private. + """Check index of the record is private. - :param record:Record data. - :return: + Args: + record (dict): Record data. + Returns: + bool: + True if all indexes the item belongs to are non-public, + False if at least one index is public. + Note: + - An item that belongs to no index at all is treated as private. + - An index is only treated as public if its own public_state is + True and its public_date (if set) is today or in the past, AND + every ancestor index above it in the tree satisfies the same + condition. If any ancestor index is non-public (or has a + public_date in the future), the index is treated as non-public + as well, regardless of the index's own public_state. + - If the item belongs to at least one such public index, it is not + private. """ from weko_index_tree.api import Indexes list_index = record.get("path") - indexes = Indexes.get_path_list(list_index) - publish_state = 6 - for index in indexes: - if len(indexes) == 1: - if not index[publish_state]: - return True - else: - if index[publish_state]: - return False - return False + if not list_index: + return True + # Delegate the public/private determination to the Indexes API. + return not Indexes.is_public_state_and_not_in_future(list_index) def validate_download_record(record): @@ -2312,6 +2320,32 @@ def convert_token_into_obj(token, is_secret_url): return url_obj +def ensure_url_record_matches(url_obj, record_id, file_name): + """Check that the issued URL record matches the requested target. + + Args: + url_obj: The download URL record fetched from the database. + record_id: The record (item) ID actually being requested. + file_name: The file name actually being requested. + + Returns: + bool: True if 'url_obj' was issued for the given + 'record_id'/'file_name', False otherwise (including when the + values do not match, or are None/empty). + """ + # Ensure that the URL object is valid and of the correct type before proceeding. + if ( + url_obj is None or + type(url_obj) not in (FileSecretDownload, FileOnetimeDownload) + ): + return False + + return ( + str(url_obj.record_id) == str(record_id) + and url_obj.file_name == file_name + ) + + def validate_url_download(record, filename, token, is_secret_url=None): """Validate the request for URL download. @@ -2344,6 +2378,12 @@ def validate_url_download(record, filename, token, is_secret_url=None): # Check if the URL is still valid url_obj = convert_token_into_obj(token, is_secret_url) + + # Check that the token was actually issued for the requested item/file. + # The same generic message as the file-access check above is used. + if not ensure_url_record_matches(url_obj, record.get('recid'), filename): + return False, _('This file is currently not available for this feature.') + if url_obj.is_deleted is True: return False, _('This URL has been deactivated.') if url_obj.download_count >= url_obj.download_limit: diff --git a/modules/weko-records-ui/weko_records_ui/views.py b/modules/weko-records-ui/weko_records_ui/views.py index 2bd15555ec..d4564a39a5 100644 --- a/modules/weko-records-ui/weko_records_ui/views.py +++ b/modules/weko-records-ui/weko_records_ui/views.py @@ -37,6 +37,7 @@ from flask_login import login_required from flask_security import current_user from sqlalchemy.orm.exc import NoResultFound +from werkzeug.exceptions import HTTPException from invenio_db import db from invenio_files_rest.models import ObjectVersion, FileInstance from invenio_files_rest.permissions import has_update_version_role @@ -95,6 +96,7 @@ from weko_records_ui.utils import ( check_items_settings, can_manage_onetime_url, can_manage_secret_url, create_download_url, create_secret_url_record, delete_version, + ensure_url_record_matches, export_preprocess, get_billing_file_download_permission, get_file_info_list, get_google_detaset_meta, get_google_scholar_meta, get_groups_price, get_min_price_billing_file_download, get_record_permalink, hide_by_email, @@ -953,13 +955,23 @@ def copy_secret_url(pid, record, **kwargs): Raises: flask.abort: - 403 if the user does not have enough permissions. + - 404 if the URL does not exist or does not belong to the requested pid/filename. - 500 if an error occurs while preparing the URL. """ try: if not can_manage_secret_url(record, kwargs['filename']): abort(403) url_record = FileSecretDownload.get_by_id(kwargs['secret_url_id']) + # Reject (as 404) when the secret_url_id does not exist, + # or when it exists but does not match the requested pid/filename. + if not url_record or not ensure_url_record_matches( + url_record, pid.pid_value, kwargs['filename']): + abort(404) url = create_download_url(url_record) + except HTTPException: + # Let abort()'s intended status code (403/404) propagate as-is, + # so we simply re-raise the HTTPException. + raise except Exception as e: current_app.logger.error(e) abort(500) @@ -986,13 +998,23 @@ def copy_onetime_url(pid, record, **kwargs): Raises: flask.abort: - 403 if the user does not have enough permissions. + - 404 if the URL does not exist or does not belong to the requested pid/filename. - 500 if an error occurs while preparing the URL. """ try: if not can_manage_onetime_url(record, kwargs['filename']): abort(403) url_record = FileOnetimeDownload.get_by_id(kwargs['onetime_url_id']) + # Reject (as 404) when the onetime_url_id does not exist, + # or when it exists but does not actually belong to the requested pid/filename. + if not url_record or not ensure_url_record_matches( + url_record, pid.pid_value, kwargs['filename']): + abort(404) url = create_download_url(url_record) + except HTTPException: + # Let abort()'s intended status code (403/404) propagate as-is, + # so we simply re-raise the HTTPException. + raise except Exception as e: current_app.logger.error(e) abort(500) @@ -1019,15 +1041,23 @@ def delete_secret_url(pid, record, **kwargs): Raises: flask.abort: - 403 if the user does not have enough permissions. + - 404 if the URL does not exist or does not belong to the requested pid/filename. - 500 if an error occurs while deleting the URL. """ try: if not can_manage_secret_url(record, kwargs['filename']): abort(403) url_record = FileSecretDownload.get_by_id(kwargs['secret_url_id']) - if not url_record: + # Reject (as 404) when the secret_url_id does not exist, + # or when it exists but does not actually belong to the requested pid/filename. + if not url_record or not ensure_url_record_matches( + url_record, pid.pid_value, kwargs['filename']): abort(404) url_record.delete_logically() + except HTTPException: + # Let abort()'s intended status code (403/404) propagate as-is, + # so we simply re-raise the HTTPException. + raise except Exception as e: current_app.logger.error(e) abort(500) @@ -1054,15 +1084,23 @@ def delete_onetime_url(pid, record, **kwargs): Raises: flask.abort: - 403 if the user does not have enough permissions. + - 404 if the URL does not exist or does not belong to the requested pid/filename. - 500 if an error occurs while deleting the URL. """ try: if not can_manage_onetime_url(record, kwargs['filename']): abort(403) url_record = FileOnetimeDownload.get_by_id(kwargs['onetime_url_id']) - if not url_record: + # Reject (as 404) when the onetime_url_id does not exist, + # or when it exists but does not actually belong to the requested pid/filename. + if not url_record or not ensure_url_record_matches( + url_record, pid.pid_value, kwargs['filename']): abort(404) url_record.delete_logically() + except HTTPException: + # Let abort()'s intended status code (403/404) propagate as-is, + # so we simply re-raise the HTTPException. + raise except Exception as e: current_app.logger.error(e) abort(500)