diff --git a/modules/invenio-resourcesyncserver/invenio_resourcesyncserver/api.py b/modules/invenio-resourcesyncserver/invenio_resourcesyncserver/api.py
index 3bd2018f2c..61619251e6 100644
--- a/modules/invenio-resourcesyncserver/invenio_resourcesyncserver/api.py
+++ b/modules/invenio-resourcesyncserver/invenio_resourcesyncserver/api.py
@@ -49,6 +49,7 @@
 
 from .config import INVENIO_CAPABILITY_URL, VALIDATE_MESSAGE, WEKO_ROOT_INDEX
 from .models import ChangeListIndexes, ResourceListIndexes
+from .permissions import can_download_file
 from .query import get_items_by_index_tree
 
 import urllib.parse
@@ -475,6 +476,8 @@ def get_resource_dump_manifest(self, record_id):
             record = WekoRecord.get_record_by_pid(record_id)
             if record:
                 for file in record.files:
+                    if not can_download_file(record, file):
+                        continue
                     current_app.logger.debug(file.info())
                     file_info = file.info()
                     path = 'recid_{}/{}'.format(
@@ -969,18 +972,24 @@ def get_change_dump_manifest_xml(self, record_id):
             else:
                 prev_record = None
             if current_record:
-                list_file = [file for file in current_record.files]
+                list_file = [
+                    (current_record, file) for file in current_record.files
+                ]
                 current_checksum = [
                     file.info().get('checksum') for file in current_record.files
                 ]
                 prev_checksum = []
                 if prev_record:
-                    list_file.extend([file for file in prev_record.files])
+                    list_file.extend(
+                        [(prev_record, file) for file in prev_record.files]
+                    )
                     prev_checksum = [
                         file.info().get('checksum') for file in
                         prev_record.files
                     ]
-                for file in list_file:
+                for record, file in list_file:
+                    if not can_download_file(record, file):
+                        continue
                     file_info = file.info()
                     change = None
                     if file_info.get('checksum') in prev_checksum:
diff --git a/modules/invenio-resourcesyncserver/invenio_resourcesyncserver/permissions.py b/modules/invenio-resourcesyncserver/invenio_resourcesyncserver/permissions.py
new file mode 100644
index 0000000000..f06602e64e
--- /dev/null
+++ b/modules/invenio-resourcesyncserver/invenio_resourcesyncserver/permissions.py
@@ -0,0 +1,87 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of WEKO3.
+# Copyright (C) 2017 National Institute of Informatics.
+#
+# WEKO3 is free software; you can redistribute it
+# and/or modify it under the terms of the GNU General Public License as
+# published by the Free Software Foundation; either version 2 of the
+# License, or (at your option) any later version.
+#
+# WEKO3 is distributed in the hope that it will be
+# useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with WEKO3; if not, write to the
+# Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston,
+# MA 02111-1307, USA.
+
+"""Permissions for ResourceSync Server."""
+
+from functools import wraps
+
+from flask import abort, current_app
+
+
+def is_public_record(record_id):
+    """Check that the record is available to the public.
+
+    The record must be published, its publication date must have come and
+    it must belong to a public index. For a versioned identifier
+    ("<recid>.<version>") the parent record must satisfy the same
+    conditions.
+
+    :param record_id: Identifier of the record.
+    :return: True if the record can be distributed.
+    """
+    from invenio_oaiserver.response import is_private_index
+    from weko_deposit.api import WekoRecord
+    from weko_records_ui.permissions import check_publish_status
+
+    record_ids = [str(record_id)]
+    if '.' in record_ids[0]:
+        record_ids.append(record_ids[0].split('.')[0])
+
+    for _id in record_ids:
+        try:
+            record = WekoRecord.get_record_by_pid(_id)
+        except Exception as ex:
+            current_app.logger.debug(ex)
+            return False
+        if not record or not check_publish_status(record) \
+                or is_private_index(record):
+            return False
+    return True
+
+
+def can_download_file(record, file):
+    """Check that the current user can download the file of the record.
+
+    :param record: Record that owns the file.
+    :param file: File object of the record.
+    :return: True if the file can be downloaded.
+    """
+    from weko_records_ui.permissions import check_file_download_permission
+    try:
+        return bool(check_file_download_permission(record, file.info()))
+    except Exception as ex:
+        current_app.logger.debug(ex)
+        return False
+
+
+def public_record_required(param='record_id'):
+    """Abort with 404 unless the record in the URL is public.
+
+    :param param: name of the view argument holding the record identifier.
+    """
+    def decorator(f):
+        @wraps(f)
+        def decorated(*args, **kwargs):
+            record_id = kwargs.get(param)
+            if record_id is None or not is_public_record(record_id):
+                abort(404)
+            return f(*args, **kwargs)
+        return decorated
+    return decorator
diff --git a/modules/invenio-resourcesyncserver/invenio_resourcesyncserver/query.py b/modules/invenio-resourcesyncserver/invenio_resourcesyncserver/query.py
index c7ee054d33..1a70a13c3e 100644
--- a/modules/invenio-resourcesyncserver/invenio_resourcesyncserver/query.py
+++ b/modules/invenio-resourcesyncserver/invenio_resourcesyncserver/query.py
@@ -234,6 +234,16 @@ def _get_index_search_query(_date_from: str, _date_until: str) -> dict:
             "post_filter": {
                 "bool": {
                     "must": [
+                        {
+                            # Deleted items are kept so that the change
+                            # list can report their deletion.
+                            "terms": {
+                                "publish_status": [
+                                    PublishStatus.PUBLIC.value,
+                                    PublishStatus.DELETE.value
+                                ]
+                            }
+                        },
                         {
                             "range": {
                                 "publish_date": {
diff --git a/modules/invenio-resourcesyncserver/invenio_resourcesyncserver/views.py b/modules/invenio-resourcesyncserver/invenio_resourcesyncserver/views.py
index c3c8246699..b31337db9f 100644
--- a/modules/invenio-resourcesyncserver/invenio_resourcesyncserver/views.py
+++ b/modules/invenio-resourcesyncserver/invenio_resourcesyncserver/views.py
@@ -21,6 +21,7 @@
 from weko_index_tree.models import Index
 
 from .api import ChangeListHandler, ResourceListHandler
+from .permissions import public_record_required
 from .utils import render_capability_xml, render_well_know_resourcesync
 
 blueprint = Blueprint(
@@ -60,6 +61,7 @@ def resource_dump(index_id):
 
 
 @blueprint.route("/resync/<index_id>/<record_id>/file_content.zip")
+@public_record_required()
 def file_content(index_id, record_id):
     """Download file content."""
     resource = ResourceListHandler.get_resource_by_repository_id(index_id)
@@ -80,6 +82,7 @@ def capability():
 
 
 @blueprint.route("/resync/<index_id>/<record_id>/resourcedump_manifest.xml")
+@public_record_required()
 def resource_dump_manifest(index_id, record_id):
     """Render resource dump manifest."""
     resource = ResourceListHandler.get_resource_by_repository_id(index_id)
@@ -138,6 +141,7 @@ def change_dump(index_id, from_date):
 
 
 @blueprint.route("/resync/<index_id>/<record_id>/changedump_manifest.xml")
+@public_record_required()
 def change_dump_manifest(index_id, record_id):
     """Render change dump manifest."""
     cl = ChangeListHandler.get_change_list_by_repo_id(index_id)
@@ -149,6 +153,7 @@ def change_dump_manifest(index_id, record_id):
 
 
 @blueprint.route("/resync/<index_id>/<record_id>/change_dump_content.zip")
+@public_record_required()
 def change_dump_content(index_id, record_id):
     """Render change dump content."""
     cl = ChangeListHandler.get_change_list_by_repo_id(index_id)
diff --git a/modules/invenio-resourcesyncserver/tests/test_api.py b/modules/invenio-resourcesyncserver/tests/test_api.py
index 454f1840f9..beece032f7 100644
--- a/modules/invenio-resourcesyncserver/tests/test_api.py
+++ b/modules/invenio-resourcesyncserver/tests/test_api.py
@@ -305,11 +305,42 @@ def as_xml_sample():
 
     with patch("invenio_resourcesyncserver.api.ResourceListHandler._validation", return_value=True):
         with patch("weko_deposit.api.WekoRecord.get_record_by_pid", return_value=return_data):
-            # try and except is for bypassing ResourceDumpManifest.as_xml()
-            try:
-                assert test.get_resource_dump_manifest(record_id)
-            except:
-                pass
+            with patch("invenio_resourcesyncserver.api.can_download_file", return_value=True):
+                # try and except is for bypassing ResourceDumpManifest.as_xml()
+                try:
+                    assert test.get_resource_dump_manifest(record_id)
+                except:
+                    pass
+
+
+def _sample_file(key, checksum):
+    file = MagicMock()
+    file.info.return_value = {
+        "key": key,
+        "checksum": "sha256:{}".format(checksum),
+        "size": 10,
+    }
+    return file
+
+
+def test_get_resource_dump_manifest_download_permission_ResourceListHandler(i18n_app):
+    test = sample_ResourceListHandler()
+    test.resource_dump_manifest = True
+    allowed = _sample_file("allowed.txt", "aaaa")
+    denied = _sample_file("denied.txt", "bbbb")
+    record = MagicMock()
+    record.files = [allowed, denied]
+    record.get.return_value = "1"
+
+    with patch("invenio_resourcesyncserver.api.ResourceListHandler._validation", return_value=True):
+        with patch("weko_deposit.api.WekoRecord.get_record_by_pid", return_value=record):
+            with patch("invenio_resourcesyncserver.api.can_download_file",
+                       side_effect=lambda r, f: f is allowed) as m:
+                xml = test.get_resource_dump_manifest("1")
+    assert "allowed.txt" in xml
+    assert "denied.txt" not in xml
+    assert "bbbb" not in xml
+    assert m.call_count == 2
 
 
 #     def get_record_content_file(self, record_id):
@@ -545,7 +576,48 @@ def _is_record_in_index(key):
     with patch("weko_deposit.api.WekoRecord.get_record_by_pid", return_value=return_data):
         with patch("invenio_resourcesyncserver.utils.get_pid", return_value=return_data):
             with patch("weko_deposit.api.WekoRecord.get_record", return_value=return_data):
-                assert test_str.get_change_dump_manifest_xml(record_id)
+                with patch("invenio_resourcesyncserver.api.can_download_file", return_value=True):
+                    assert test_str.get_change_dump_manifest_xml(record_id)
+
+
+def test_get_change_dump_manifest_xml_download_permission_ChangeListHandler(i18n_app):
+    test_str = sample_ChangeListHandler("str")
+    test_str._validation = lambda: True
+    test_str._is_record_in_index = lambda key: True
+
+    kept = _sample_file("kept.txt", "0000")
+    created_ok = _sample_file("created_ok.txt", "1111")
+    created_ng = _sample_file("created_ng.txt", "2222")
+    deleted_ok = _sample_file("deleted_ok.txt", "3333")
+    deleted_ng = _sample_file("deleted_ng.txt", "4444")
+    current_record = MagicMock()
+    current_record.files = [kept, created_ok, created_ng]
+    current_record.get.return_value = "8"
+    prev_record = MagicMock()
+    prev_record.files = [kept, deleted_ok, deleted_ng]
+
+    denied = (created_ng, deleted_ng)
+    checked = []
+
+    def _can_download_file(record, file):
+        checked.append((record, file))
+        return file not in denied
+
+    with patch("weko_deposit.api.WekoRecord.get_record_by_pid", return_value=current_record):
+        with patch("invenio_resourcesyncserver.utils.get_pid", return_value=MagicMock()):
+            with patch("weko_deposit.api.WekoRecord.get_record", return_value=prev_record):
+                with patch("invenio_resourcesyncserver.api.can_download_file",
+                           side_effect=_can_download_file):
+                    xml = test_str.get_change_dump_manifest_xml("8.2")
+
+    assert "created_ok.txt" in xml
+    assert "deleted_ok.txt" in xml
+    assert "created_ng.txt" not in xml
+    assert "deleted_ng.txt" not in xml
+    assert "kept.txt" not in xml
+    # each file is checked against the record that owns it
+    assert (current_record, created_ng) in checked
+    assert (prev_record, deleted_ng) in checked
 
 
 #     def delete(cls, change_list_id):
diff --git a/modules/invenio-resourcesyncserver/tests/test_permissions.py b/modules/invenio-resourcesyncserver/tests/test_permissions.py
new file mode 100644
index 0000000000..ff4c7abb49
--- /dev/null
+++ b/modules/invenio-resourcesyncserver/tests/test_permissions.py
@@ -0,0 +1,119 @@
+import pytest
+from mock import MagicMock, patch
+from werkzeug.exceptions import NotFound
+
+from invenio_resourcesyncserver.permissions import (
+    can_download_file,
+    is_public_record,
+    public_record_required
+)
+
+
+PUBLIC = {
+    "publish_status": "0",
+    "pubdate": {"attribute_value": "2022-08-20"},
+    "path": ["1"],
+}
+PRIVATE = {
+    "publish_status": "1",
+    "pubdate": {"attribute_value": "2022-08-20"},
+    "path": ["1"],
+}
+DELETED = {
+    "publish_status": "-1",
+    "pubdate": {"attribute_value": "2022-08-20"},
+    "path": ["1"],
+}
+FUTURE = {
+    "publish_status": "0",
+    "pubdate": {"attribute_value": "2999-01-01"},
+    "path": ["1"],
+}
+
+
+def _records(mapping):
+    def get_record_by_pid(pid):
+        if pid not in mapping:
+            raise Exception("pid does not exist")
+        return mapping[pid]
+    return get_record_by_pid
+
+
+# def is_public_record(record_id):
+@pytest.mark.parametrize("records, record_id, private_index, expected", [
+    ({"1": PUBLIC}, "1", False, True),
+    ({"1": PUBLIC}, 1, False, True),
+    ({"1": PRIVATE}, "1", False, False),
+    ({"1": DELETED}, "1", False, False),
+    ({"1": FUTURE}, "1", False, False),
+    ({"1": PUBLIC}, "1", True, False),
+    ({}, "1", False, False),
+    # versioned identifier: the parent record is checked as well
+    ({"1": PUBLIC, "1.1": PUBLIC}, "1.1", False, True),
+    ({"1": PRIVATE, "1.1": PUBLIC}, "1.1", False, False),
+    ({"1": PUBLIC, "1.1": PRIVATE}, "1.1", False, False),
+    ({"1.1": PUBLIC}, "1.1", False, False),
+])
+def test_is_public_record(i18n_app, records, record_id, private_index, expected):
+    with patch("weko_deposit.api.WekoRecord.get_record_by_pid", side_effect=_records(records)):
+        with patch("invenio_oaiserver.response.is_private_index", return_value=private_index):
+            assert is_public_record(record_id) == expected
+
+
+# def public_record_required(param='record_id'):
+def test_public_record_required(i18n_app):
+    @public_record_required()
+    def view(index_id, record_id):
+        return "ok"
+
+    with patch("invenio_resourcesyncserver.permissions.is_public_record", return_value=True) as m:
+        assert view(index_id=1, record_id="2") == "ok"
+        m.assert_called_once_with("2")
+
+    with patch("invenio_resourcesyncserver.permissions.is_public_record", return_value=False):
+        with pytest.raises(NotFound):
+            view(index_id=1, record_id="2")
+
+    @public_record_required(param="recid")
+    def view2(recid=None):
+        return "ok"
+
+    with patch("invenio_resourcesyncserver.permissions.is_public_record", return_value=True):
+        assert view2(recid="3") == "ok"
+        with pytest.raises(NotFound):
+            view2()
+
+
+# def can_download_file(record, file):
+def test_can_download_file(i18n_app):
+    record = {"recid": "1"}
+    file = MagicMock()
+    file.info.return_value = {"filename": "a.txt", "accessrole": "open_access"}
+    target = "weko_records_ui.permissions.check_file_download_permission"
+
+    with patch(target, return_value=True) as m:
+        assert can_download_file(record, file) is True
+        m.assert_called_once_with(record, file.info.return_value)
+    with patch(target, return_value=False):
+        assert can_download_file(record, file) is False
+    with patch(target, return_value=None):
+        assert can_download_file(record, file) is False
+    with patch(target, side_effect=Exception("error")):
+        assert can_download_file(record, file) is False
+
+
+# check_file_download_permission is not mocked: a file whose access role
+# does not allow anonymous download is excluded.
+@pytest.mark.parametrize("fjson, expected", [
+    ({"filename": "a.txt", "accessrole": "open_access",
+      "date": [{"dateType": "Available", "dateValue": "2000-01-01"}]}, True),
+    ({"filename": "a.txt", "accessrole": "open_access",
+      "date": [{"dateType": "Available", "dateValue": "2999-01-01"}]}, False),
+    ({"filename": "a.txt", "accessrole": "open_no"}, False),
+])
+def test_can_download_file_guest(i18n_app, users, fjson, expected):
+    record = {"recid": "1", "item_type_id": "1", "owner": "1",
+              "_deposit": {"created_by": 1}}
+    file = MagicMock()
+    file.info.return_value = fjson
+    assert can_download_file(record, file) == expected
diff --git a/modules/invenio-resourcesyncserver/tests/test_query.py b/modules/invenio-resourcesyncserver/tests/test_query.py
index a9cfd0dbd8..a6808b3e04 100644
--- a/modules/invenio-resourcesyncserver/tests/test_query.py
+++ b/modules/invenio-resourcesyncserver/tests/test_query.py
@@ -194,3 +194,42 @@ def update_from_dict(self, q):
         search = DummySearch()
         with pytest.raises(query_mod.InvalidQueryRESTError):
             query_mod.item_changes_search_factory(search, index_id=1)
+
+
+@pytest.mark.parametrize("is_root", [True, False])
+def test_item_changes_search_factory_publish_status(monkeypatch, is_root):
+    from invenio_resourcesyncserver import query as query_mod
+    from weko_schema_ui.models import PublishStatus
+
+    class DummySearch:
+        def __init__(self):
+            self.query = None
+        def update_from_dict(self, q):
+            self.query = q
+
+    class DummyIndexes:
+        @staticmethod
+        def get_list_path_publish(index_id):
+            return ["1"]
+        @staticmethod
+        def get_child_list(q):
+            return []
+
+    monkeypatch.setattr(query_mod, "Indexes", DummyIndexes)
+    app = Flask(__name__)
+    app.config["WEKO_ROOT_INDEX"] = 0
+    with app.app_context():
+        search = DummySearch()
+        result = query_mod.item_changes_search_factory(
+            search, index_id=0 if is_root else 1,
+            date_from="2020-01-01", date_until="2020-12-31")
+        must = result.query["post_filter"]["bool"]["must"]
+        status_filters = [
+            m["terms"]["publish_status"] for m in must
+            if isinstance(m, dict) and "publish_status" in m.get("terms", {})
+        ]
+        assert len(status_filters) == 1
+        assert PublishStatus.PUBLIC.value in status_filters[0]
+        assert PublishStatus.DELETE.value in status_filters[0]
+        assert PublishStatus.PRIVATE.value not in status_filters[0]
+        assert PublishStatus.NEW.value not in status_filters[0]
diff --git a/modules/invenio-resourcesyncserver/tests/test_views.py b/modules/invenio-resourcesyncserver/tests/test_views.py
index 47c1d81221..f863d7048d 100644
--- a/modules/invenio-resourcesyncserver/tests/test_views.py
+++ b/modules/invenio-resourcesyncserver/tests/test_views.py
@@ -5,6 +5,7 @@
 from flask import current_app, make_response, request
 from flask_login import current_user
 from mock import patch, MagicMock
+from werkzeug.exceptions import NotFound
 from werkzeug.local import LocalProxy
 
 from invenio_resourcesyncserver.api import ResourceListHandler, ChangeListHandler
@@ -184,8 +185,9 @@ def get_record_content_file(item):
         return item
     data1.get_record_content_file = get_record_content_file
 
-    with patch("invenio_resourcesyncserver.api.ResourceListHandler.get_resource_by_repository_id", return_value=data1):
-        assert file_content(index_id=index_id, record_id=record_id) is not None
+    with patch("invenio_resourcesyncserver.permissions.is_public_record", return_value=True):
+        with patch("invenio_resourcesyncserver.api.ResourceListHandler.get_resource_by_repository_id", return_value=data1):
+            assert file_content(index_id=index_id, record_id=record_id) is not None
     
     # Exception coverage
     # file_content(index_id=index_id, record_id=record_id) is not None
@@ -209,8 +211,9 @@ def get_record_content_file(item):
         return item
     data1.get_record_content_file = get_record_content_file
 
-    with patch("invenio_resourcesyncserver.api.ResourceListHandler.get_resource_by_repository_id", return_value=data1):
-        assert "Response" in str(type(resource_dump_manifest(index_id=index_id, record_id=record_id)))
+    with patch("invenio_resourcesyncserver.permissions.is_public_record", return_value=True):
+        with patch("invenio_resourcesyncserver.api.ResourceListHandler.get_resource_by_repository_id", return_value=data1):
+            assert "Response" in str(type(resource_dump_manifest(index_id=index_id, record_id=record_id)))
 
     # Exception coverage
     # resource_dump_manifest(index_id=index_id, record_id=record_id)
@@ -323,8 +326,9 @@ def get_change_list_by_repo_id(item):
         
     data1.get_change_list_by_repo_id = get_change_list_by_repo_id
 
-    with patch("invenio_resourcesyncserver.api.ChangeListHandler.get_change_list_by_repo_id", return_value=data1):
-        assert "Response" in str(type(change_dump_manifest(index_id=index_id, record_id=record_id)))
+    with patch("invenio_resourcesyncserver.permissions.is_public_record", return_value=True):
+        with patch("invenio_resourcesyncserver.api.ChangeListHandler.get_change_list_by_repo_id", return_value=data1):
+            assert "Response" in str(type(change_dump_manifest(index_id=index_id, record_id=record_id)))
 
     # Exception coverage
     # change_dump_manifest(index_id=index_id)
@@ -346,8 +350,9 @@ def get_change_list_by_repo_id(item):
         
     data1.get_change_list_by_repo_id = get_change_list_by_repo_id
 
-    with patch("invenio_resourcesyncserver.api.ChangeListHandler.get_change_list_by_repo_id", return_value=data1):
-        assert str(type(change_dump_content(index_id=index_id, record_id=record_id))) == str(type(get_change_list_by_repo_id(index_id)))
+    with patch("invenio_resourcesyncserver.permissions.is_public_record", return_value=True):
+        with patch("invenio_resourcesyncserver.api.ChangeListHandler.get_change_list_by_repo_id", return_value=data1):
+            assert str(type(change_dump_content(index_id=index_id, record_id=record_id))) == str(type(get_change_list_by_repo_id(index_id)))
 
     # Exception coverage
     # change_dump_content(index_id=index_id)
@@ -375,4 +380,22 @@ def zfill(item):
     test.append( etree.Element("test1") )
 
     with patch("invenio_resourcesyncserver.views.getrecord", return_value=test):
-        assert "Response" in str(type(record_detail_in_index(index_id=index_id, record_id=record_id)))
\ No newline at end of file
+        assert "Response" in str(type(record_detail_in_index(index_id=index_id, record_id=record_id)))
+
+
+@pytest.mark.parametrize("view", [
+    file_content,
+    resource_dump_manifest,
+    change_dump_manifest,
+    change_dump_content,
+])
+def test_record_views_not_public(i18n_app, view):
+    handler = MagicMock()
+    with patch("invenio_resourcesyncserver.permissions.is_public_record", return_value=False):
+        with patch("invenio_resourcesyncserver.api.ResourceListHandler.get_resource_by_repository_id", return_value=handler):
+            with patch("invenio_resourcesyncserver.api.ChangeListHandler.get_change_list_by_repo_id", return_value=handler):
+                with pytest.raises(NotFound):
+                    view(index_id=33, record_id="1")
+    handler.get_record_content_file.assert_not_called()
+    handler.get_resource_dump_manifest.assert_not_called()
+    handler.get_change_dump_manifest_xml.assert_not_called()
diff --git a/modules/invenio-stats/invenio_stats/permissions.py b/modules/invenio-stats/invenio_stats/permissions.py
index 949414645a..d026156c5b 100644
--- a/modules/invenio-stats/invenio_stats/permissions.py
+++ b/modules/invenio-stats/invenio_stats/permissions.py
@@ -20,7 +20,73 @@
 
 """Permissions for invenio-stats."""
 
+import uuid
+from functools import wraps
+
+from flask import abort
 from invenio_access import Permission, action_factory
+from sqlalchemy.orm.exc import NoResultFound
 
 stats_api_access = action_factory('stats-api-access')
 stats_api_permission = Permission(stats_api_access)
+
+
+def _can_view_record(record):
+    """Check the record detail page permission of the current user."""
+    from weko_records_ui.permissions import page_permission_factory
+
+    return page_permission_factory(record).can()
+
+
+def record_view_permission_required(f):
+    """Require the permission to view the record given by ``record_id``.
+
+    Uses the same permission as the record detail page.
+    Aborts with 400 for a malformed id, 404 for an unknown record and
+    403 when the current user cannot view the record.
+    """
+    @wraps(f)
+    def decorated(*args, **kwargs):
+        from weko_deposit.api import WekoRecord
+
+        try:
+            record_uuid = uuid.UUID(str(kwargs.get('record_id')))
+        except ValueError:
+            abort(400)
+        try:
+            record = WekoRecord.get_record(record_uuid)
+        except NoResultFound:
+            abort(404)
+        if not _can_view_record(record):
+            abort(403)
+        return f(*args, **kwargs)
+    return decorated
+
+
+def bucket_view_permission_required(f):
+    """Require the permission to view a record owning ``bucket_id``.
+
+    Uses the same permission as the record detail page.
+    Aborts with 400 for a malformed id, 404 when no record owns the
+    bucket and 403 when the current user cannot view the record.
+    """
+    @wraps(f)
+    def decorated(*args, **kwargs):
+        from invenio_records_files.models import RecordsBuckets
+        from weko_deposit.api import WekoRecord
+
+        try:
+            bucket_uuid = uuid.UUID(str(kwargs.get('bucket_id')))
+        except ValueError:
+            abort(400)
+        found = False
+        for rb in RecordsBuckets.query.filter_by(bucket_id=bucket_uuid).all():
+            try:
+                record = WekoRecord.get_record(rb.record_id)
+            except NoResultFound:
+                continue
+            found = True
+            if _can_view_record(record):
+                return f(*args, **kwargs)
+        abort(403 if found else 404)
+    return decorated
diff --git a/modules/invenio-stats/invenio_stats/views.py b/modules/invenio-stats/invenio_stats/views.py
index ecb0ce4581..ae5a59eced 100644
--- a/modules/invenio-stats/invenio_stats/views.py
+++ b/modules/invenio-stats/invenio_stats/views.py
@@ -23,6 +23,8 @@
 
 from . import config
 from .errors import InvalidRequestInputError, UnknownQueryError
+from .permissions import bucket_view_permission_required, \
+    record_view_permission_required
 from .proxies import current_stats
 from .utils import QueryCommonReportsHelper, QueryFileReportsHelper, \
     QueryItemRegReportHelper, QueryRecordViewPerIndexReportHelper, \
@@ -46,6 +48,24 @@ def wrapper(*args, **kwargs):
     return wrapper
 
 
+def get_query_date(data):
+    """Get the target month from the request body.
+
+    :param data: Request body, ``{'date': 'total'}`` or
+        ``{'date': 'YYYY-MM'}``.
+    :returns: None for 'total', otherwise the 'YYYY-MM' string.
+    """
+    try:
+        date = data['date']
+        if date == 'total':
+            return None
+        datetime.strptime(date, '%Y-%m')
+        return date
+    except (TypeError, KeyError, ValueError):
+        current_app.logger.error(traceback.format_exc())
+        abort(400)
+
+
 class WekoQuery(ContentNegotiatedMethodView):
     """Docstring for WekoQuery."""
 
@@ -241,6 +261,7 @@ def get_data_by_pid_value(self, pid_value, query_date=None, get_period=False):
 
         return result
 
+    @record_view_permission_required
     def get(self, **kwargs):
         """Get total record view count."""
         record_id = kwargs.get('record_id')
@@ -251,19 +272,16 @@ def get(self, **kwargs):
             abort(400)
         return self.make_response(self.get_data(record_uuid, get_period=True))
 
+    @record_view_permission_required
     def post(self, **kwargs):
         """Get record view count with date."""
         record_id = kwargs.get('record_id')
-        d = request.get_json(force=False)
         try:
             record_uuid = uuid.UUID(record_id)
-            if d['date'] == 'total':
-                date = None
-            else:
-                date = d['date']
-        except (TypeError, ValueError):
+        except ValueError:
             current_app.logger.error(traceback.format_exc())
             abort(400)
+        date = get_query_date(request.get_json(force=False, silent=True))
         return self.make_response(self.get_data(record_uuid, date))
 
 
@@ -384,6 +402,7 @@ def get_data(self, bucket_id, file_key, query_date=None,
 
         return result
 
+    @bucket_view_permission_required
     def get(self, **kwargs):
         """Get total file download/preview count."""
         bucket_id = kwargs.get('bucket_id')
@@ -394,15 +413,12 @@ def get(self, **kwargs):
                 file_key,
                 get_period=True))
 
+    @bucket_view_permission_required
     def post(self, **kwargs):
         """Get file download/preview count with date."""
         bucket_id = kwargs.get('bucket_id')
         file_key = kwargs.get('file_key')
-        d = request.get_json(force=False)
-        if d['date'] == 'total':
-            date = None
-        else:
-            date = d['date']
+        date = get_query_date(request.get_json(force=False, silent=True))
         return self.make_response(self.get_data(bucket_id, file_key, date))
 
 
diff --git a/modules/invenio-stats/tests/conftest.py b/modules/invenio-stats/tests/conftest.py
index 4a7181aee4..30cac98345 100644
--- a/modules/invenio-stats/tests/conftest.py
+++ b/modules/invenio-stats/tests/conftest.py
@@ -491,6 +491,8 @@ def db(app):
     # invenio_stats.utils reads Community, and the model has to be imported
     # before create_all() for its table to be part of the metadata.
     from invenio_communities.models import Community  # noqa: F401
+    # The stats views look up the record of a bucket.
+    from invenio_records_files.models import RecordsBuckets  # noqa: F401
 
     if not database_exists(str(db_.engine.url)):
         create_database(str(db_.engine.url))
diff --git a/modules/invenio-stats/tests/test_views.py b/modules/invenio-stats/tests/test_views.py
index 9887559d33..474c0a4673 100644
--- a/modules/invenio-stats/tests/test_views.py
+++ b/modules/invenio-stats/tests/test_views.py
@@ -131,9 +131,29 @@ def __init__(self, child):
         self.children = self.mockChild(child)
         pass
 
+
+class mockPermissionChecker:
+    def __init__(self, result):
+        self.result = result
+
+    def can(self):
+        return self.result
+
+
+def patch_page_permission(result):
+    """Patch the record detail page permission used by the stats views."""
+    return patch(
+        "weko_records_ui.permissions.page_permission_factory",
+        return_value=mockPermissionChecker(result))
+
 # class QueryRecordViewCount(WekoQuery):
 # .tox/c1/bin/pytest --cov=invenio_stats tests/test_views.py::test_query_record_view_count -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/invenio-stats/.tox/c1/tmp
 def test_query_record_view_count(client, db, es, records):
+    with patch_page_permission(True):
+        _test_query_record_view_count(client, records)
+
+
+def _test_query_record_view_count(client, records):
     _uuid = str(records[0][0].object_uuid)
 
     # get
@@ -176,38 +196,95 @@ def test_query_record_view_count(client, db, es, records):
 
 # .tox/c1/bin/pytest --cov=invenio_stats tests/test_views.py::test_query_record_view_count_error -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/invenio-stats/.tox/c1/tmp
 def test_query_record_view_count_error(client, db, records):
-    _uuid = uuid.uuid4()
-    res = client.get(
-        url_for('invenio_stats.get_record_view_count', record_id=_uuid))
-    assert res.status_code==200
+    # リクエストの後はフィクスチャのオブジェクトがセッションから外れるので先に控える
+    record_uuid = str(records[0][0].object_uuid)
+    with patch_page_permission(True):
+        # record does not exist
+        _uuid = uuid.uuid4()
+        res = client.get(
+            url_for('invenio_stats.get_record_view_count', record_id=_uuid))
+        assert res.status_code==404
+        res = client.post(
+            url_for('invenio_stats.get_record_view_count', record_id=_uuid),
+            data=json.dumps({'date': 'total'}),
+            content_type='application/json',
+        )
+        assert res.status_code==404
 
+        res = client.get(
+            url_for('invenio_stats.get_record_view_count', record_id=record_uuid))
+        assert res.status_code==200
+
+        # GET:Invalid uuid
+        res = client.get(
+            url_for('invenio_stats.get_record_view_count', record_id='test'))
+        assert res.status_code==400
+
+        # POST:Invalid uuid
+        res = client.post(
+            url_for('invenio_stats.get_record_view_count', record_id='test'),
+            data=json.dumps({'date': 'total'}),
+            content_type='application/json',
+        )
+        assert res.status_code==400
+
+        # POST:Invalid request data
+        res = client.post('/api/stats/{}'.format(record_uuid))
+        assert res.status_code==400
+        for _data in [{}, {'date': 'test'}, {'date': 202209}, []]:
+            res = client.post(
+                url_for('invenio_stats.get_record_view_count',
+                        record_id=record_uuid),
+                data=json.dumps(_data),
+                content_type='application/json',
+            )
+            assert res.status_code==400
+
+
+# .tox/c1/bin/pytest --cov=invenio_stats tests/test_views.py::test_query_record_view_count_permission -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/invenio-stats/.tox/c1/tmp
+def test_query_record_view_count_permission(client, db, records):
     _uuid = str(records[0][0].object_uuid)
-    res = client.get(
-        url_for('invenio_stats.get_record_view_count', record_id=_uuid))
-    assert res.status_code==200
+    url = url_for('invenio_stats.get_record_view_count', record_id=_uuid)
+    headers = [('Content-Type', 'application/json'),
+               ('Accept', 'application/json')]
 
-    # GET:Invalid uuid
-    res = client.get(
-        url_for('invenio_stats.get_record_view_count', record_id='test'))
-    assert res.status_code==400
+    # the permission of the record detail page is checked
+    with patch_page_permission(True) as mock_factory:
+        res = client.get(url)
+        assert res.status_code==200
+        assert str(mock_factory.call_args[0][0].id) == _uuid
 
-    # POST:Invalid uuid
-    res = client.post(
-        url_for('invenio_stats.get_record_view_count', record_id='test'),
-        data=json.dumps({'date': 'total'}),
-        content_type='application/json',
-    )
-    assert res.status_code==400
+        res = client.post(url, headers=headers,
+                          data=json.dumps({'date': 'total'}))
+        assert res.status_code==200
+        assert str(mock_factory.call_args[0][0].id) == _uuid
 
-    # POST:Invalid request data
-    res = client.post('/api/stats/{}'.format(_uuid))
-    assert res.status_code==400
+    # users who cannot view the record are rejected
+    with patch_page_permission(False):
+        res = client.get(url)
+        assert res.status_code==403
+
+        res = client.post(url, headers=headers,
+                          data=json.dumps({'date': 'total'}))
+        assert res.status_code==403
 
 
 # class QueryFileStatsCount(WekoQuery):
 # .tox/c1/bin/pytest --cov=invenio_stats tests/test_views.py::test_query_file_stats_count -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/invenio-stats/.tox/c1/tmp
-def test_query_file_stats_count(client, db):
-    _uuid = uuid.uuid4()
+def test_query_file_stats_count(client, db, records, bucket):
+    with patch_page_permission(True):
+        _test_query_file_stats_count(client, db, records, bucket)
+
+
+def _link_bucket(db, record, bucket):
+    from invenio_records_files.models import RecordsBuckets
+    RecordsBuckets.create(record=record.model, bucket=bucket)
+    db.session.commit()
+
+
+def _test_query_file_stats_count(client, db, records, bucket):
+    _link_bucket(db, records[0][1], bucket)
+    _uuid = bucket.id
 
     # get_data
     res = QueryFileStatsCount.get_data(QueryFileStatsCount, bucket_id=_uuid, file_key='test.pdf', root_file_id=uuid.uuid4())
@@ -266,6 +343,72 @@ def test_query_file_stats_count(client, db):
     assert res.status_code==200
 
 
+# .tox/c1/bin/pytest --cov=invenio_stats tests/test_views.py::test_query_file_stats_count_permission -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/invenio-stats/.tox/c1/tmp
+def test_query_file_stats_count_permission(client, db, records, bucket):
+    _link_bucket(db, records[0][1], bucket)
+    # リクエストの後はフィクスチャのオブジェクトがセッションから外れるので先に控える
+    record_id = records[0][1].id
+    url = url_for('invenio_stats.get_file_stats_count',
+                  bucket_id=bucket.id, file_key='test.pdf')
+    headers = [('Content-Type', 'application/json'),
+               ('Accept', 'application/json')]
+
+    # the permission of the record detail page is checked
+    with patch_page_permission(True) as mock_factory:
+        res = client.get(url)
+        assert res.status_code==200
+        assert mock_factory.call_args[0][0].id == record_id
+
+        res = client.post(url, headers=headers,
+                          data=json.dumps({'date': 'total'}))
+        assert res.status_code==200
+        assert mock_factory.call_args[0][0].id == record_id
+
+    # users who cannot view the record are rejected
+    with patch_page_permission(False):
+        res = client.get(url)
+        assert res.status_code==403
+
+        res = client.post(url, headers=headers,
+                          data=json.dumps({'date': 'total'}))
+        assert res.status_code==403
+
+
+# .tox/c1/bin/pytest --cov=invenio_stats tests/test_views.py::test_query_file_stats_count_error -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/invenio-stats/.tox/c1/tmp
+def test_query_file_stats_count_error(client, db, records, bucket):
+    headers = [('Content-Type', 'application/json'),
+               ('Accept', 'application/json')]
+    with patch_page_permission(True):
+        # bucket without record
+        url = url_for('invenio_stats.get_file_stats_count',
+                      bucket_id=uuid.uuid4(), file_key='test.pdf')
+        res = client.get(url)
+        assert res.status_code==404
+        res = client.post(url, headers=headers,
+                          data=json.dumps({'date': 'total'}))
+        assert res.status_code==404
+
+        # invalid bucket id
+        url = url_for('invenio_stats.get_file_stats_count',
+                      bucket_id='test', file_key='test.pdf')
+        res = client.get(url)
+        assert res.status_code==400
+        res = client.post(url, headers=headers,
+                          data=json.dumps({'date': 'total'}))
+        assert res.status_code==400
+
+        # invalid request data
+        _link_bucket(db, records[0][1], bucket)
+        url = url_for('invenio_stats.get_file_stats_count',
+                      bucket_id=bucket.id, file_key='test.pdf')
+        res = client.post(url)
+        assert res.status_code==400
+        for _data in [{}, {'date': 'test'}, {'date': 202209}, []]:
+            res = client.post(url, headers=headers,
+                              data=json.dumps(_data))
+            assert res.status_code==400
+
+
 # class QueryItemRegReport(WekoQuery):
 # .tox/c1/bin/pytest --cov=invenio_stats tests/test_views.py::test_query_item_reg_report -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/invenio-stats/.tox/c1/tmp
 @pytest.mark.parametrize(
diff --git a/modules/weko-items-ui/tests/test_utils.py b/modules/weko-items-ui/tests/test_utils.py
index ec20391119..697efb8b82 100644
--- a/modules/weko-items-ui/tests/test_utils.py
+++ b/modules/weko-items-ui/tests/test_utils.py
@@ -9336,7 +9336,7 @@ def test_get_workflow_by_item_type_id(app, db_workflow, db_itemtype):
 
 # def validate_bibtex(record_ids):
 # .tox/c1/bin/pytest --cov=weko_items_ui tests/test_utils.py::test_validate_bibtex -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-items-ui/.tox/c1/tmp
-def test_validate_bibtex(app, db, db_records, db_itemtype, db_oaischema):
+def test_validate_bibtex(app, db, users, db_records, db_itemtype, db_oaischema):
     app.config.update(OAISERVER_XSL_URL=None)
     schema = {}
     schema['root_name'] = db_oaischema.form_data.get('root_name')
@@ -9346,7 +9346,39 @@ def test_validate_bibtex(app, db, db_records, db_itemtype, db_oaischema):
         db_oaischema.xsd, object_pairs_hook=OrderedDict)
     with patch('weko_schema_ui.schema.cache_schema', return_value=schema):
         with patch('weko_schema_ui.serializers.WekoBibTexSerializer.serialize', return_value='test_data'):
-            assert validate_bibtex([1]) == []
+            with app.test_request_context():
+                with patch("flask_login.utils._get_user", return_value=users[2]["obj"]):
+                    assert validate_bibtex([1]) == []
+
+
+# .tox/c1/bin/pytest --cov=weko_items_ui tests/test_utils.py::test_validate_bibtex_viewable_only -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-items-ui/.tox/c1/tmp
+def test_validate_bibtex_viewable_only(app, db, users, db_records, db_itemtype, db_oaischema):
+    app.config.update(OAISERVER_XSL_URL=None)
+    schema = {}
+    schema['root_name'] = db_oaischema.form_data.get('root_name')
+    schema['schema_location'] = db_oaischema.schema_location
+    schema['namespaces'] = db_oaischema.namespaces
+    schema['schema'] = json.loads(
+        db_oaischema.xsd, object_pairs_hook=OrderedDict)
+    with patch('weko_schema_ui.schema.cache_schema', return_value=schema):
+        with patch('weko_schema_ui.serializers.WekoBibTexSerializer.serialize', return_value='test_data') as mock_serialize:
+            with app.test_request_context():
+                # guest: a private record (recid 2) is reported in the same
+                # way as a record that does not exist, and is not serialized
+                assert validate_bibtex([2, 9999]) == [2, 9999]
+                mock_serialize.assert_not_called()
+
+                # the detail page permission decides whether a record is used
+                with patch('weko_items_ui.utils.page_permission_factory') as mock_perm:
+                    mock_perm.return_value.can.return_value = True
+                    assert validate_bibtex([1]) == []
+                    mock_perm.return_value.can.return_value = False
+                    assert validate_bibtex([1]) == [1]
+
+                # sysadmin can validate the private record
+                with patch("flask_login.utils._get_user", return_value=users[2]["obj"]):
+                    assert validate_bibtex([2]) == []
+                    assert validate_bibtex([9999]) == [9999]
 
 
 # def make_bibtex_data(record_ids):
diff --git a/modules/weko-items-ui/tests/test_views.py b/modules/weko-items-ui/tests/test_views.py
index b8b9dbaad8..0624b2763f 100644
--- a/modules/weko-items-ui/tests/test_views.py
+++ b/modules/weko-items-ui/tests/test_views.py
@@ -21642,6 +21642,51 @@ def test_validate_bibtex_export_acl_nologin(
             assert res.status_code == 200
 
 
+# .tox/c1/bin/pytest --cov=weko_items_ui tests/test_views.py::test_validate_bibtex_export_invalid_input -v --cov-branch --cov-report=term --basetemp=/code/modules/weko-items-ui/.tox/c1/tmp
+@pytest.mark.parametrize(
+    "data, content_type",
+    [
+        ("", "application/json"),
+        ("not json", "application/json"),
+        (json.dumps({}), "application/json"),
+        (json.dumps([1]), "application/json"),
+        (json.dumps({"record_ids": 1}), "application/json"),
+        (json.dumps({"record_ids": [{"id": 1}]}), "application/json"),
+        (json.dumps({"record_ids": [True]}), "application/json"),
+        (json.dumps({"record_ids": [1]}), "text/plain"),
+    ],
+)
+def test_validate_bibtex_export_invalid_input(app, client, users, data, content_type):
+    url = url_for("weko_items_ui.validate_bibtex_export", _external=True)
+    with patch("weko_items_ui.utils.validate_bibtex", return_value=[]) as mock_validate:
+        res = client.post(url, data=data, content_type=content_type)
+        assert res.status_code == 400
+        mock_validate.assert_not_called()
+
+
+# .tox/c1/bin/pytest --cov=weko_items_ui tests/test_views.py::test_validate_bibtex_export_not_viewable -v --cov-branch --cov-report=term --basetemp=/code/modules/weko-items-ui/.tox/c1/tmp
+def test_validate_bibtex_export_not_viewable(
+    app, client, users, db_records, db_itemtype, db_oaischema
+):
+    app.config.update(OAISERVER_XSL_URL=None)
+    schema = {}
+    schema['root_name'] = db_oaischema.form_data.get('root_name')
+    schema['schema_location'] = db_oaischema.schema_location
+    schema['namespaces'] = db_oaischema.namespaces
+    schema['schema'] = json.loads(
+        db_oaischema.xsd, object_pairs_hook=OrderedDict)
+    url = url_for("weko_items_ui.validate_bibtex_export", _external=True)
+    with patch('weko_schema_ui.schema.cache_schema', return_value=schema):
+        with patch('weko_schema_ui.serializers.WekoBibTexSerializer.serialize', return_value='test_data'):
+            # guest: a private record and a missing record get the same answer
+            res = client.post(
+                url, data=json.dumps({"record_ids": [2, 9999]}),
+                content_type="application/json"
+            )
+            assert res.status_code == 200
+            assert json.loads(res.data) == {"invalid_record_ids": [2, 9999]}
+
+
 # def export():
 # .tox/c1/bin/pytest --cov=weko_items_ui tests/test_views.py::test_export_acl_nologin -v --cov-branch --cov-report=term --basetemp=/code/modules/weko-items-ui/.tox/c1/tmp
 def test_export_acl_nologin(client, users, db_oaischema):
diff --git a/modules/weko-items-ui/weko_items_ui/utils.py b/modules/weko-items-ui/weko_items_ui/utils.py
index 74e7f6fc76..841425482b 100644
--- a/modules/weko-items-ui/weko_items_ui/utils.py
+++ b/modules/weko-items-ui/weko_items_ui/utils.py
@@ -38,6 +38,7 @@
 
 import bagit
 from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.orm.exc import NoResultFound
 from elasticsearch.exceptions import NotFoundError
 from elasticsearch import exceptions as es_exceptions
 from flask import abort, current_app, flash, redirect, request, send_file, url_for
@@ -75,7 +76,8 @@
 from weko_records.utils import replace_fqdn_of_file_metadata
 from weko_records_ui.errors import AvailableFilesNotFoundRESTError
 from weko_records_ui.permissions import (
-    check_created_id, check_file_download_permission, check_publish_status
+    check_created_id, check_file_download_permission, check_publish_status,
+    page_permission_factory
 )
 from weko_redis.redis import RedisConnection
 from weko_search_ui.config import ROCRATE_METADATA_FILE, WEKO_IMPORT_DOI_TYPE
@@ -3742,14 +3744,24 @@ def get_workflow_by_item_type_id(
 def validate_bibtex(record_ids):
     """Validate data of records for Bibtex exporting.
 
+    A record that does not exist or that the current user may not view
+    (as judged by the detail page permission) is reported as invalid,
+    in the same way as a record lacking required items.
+
     @param record_ids:
-    @return:
+    @return: list of record ids that cannot be exported.
     """
     lst_invalid_ids = []
     err_msg = _('Please input all required item.')
     from weko_schema_ui.serializers import WekoBibTexSerializer
     for record_id in record_ids:
-        record = WekoRecord.get_record_by_pid(record_id)
+        try:
+            record = WekoRecord.get_record_by_pid(record_id)
+        except (PIDDoesNotExistError, NoResultFound):
+            record = None
+        if record is None or not page_permission_factory(record).can():
+            lst_invalid_ids.append(record_id)
+            continue
         pid = record.pid_recid
         serializer = WekoBibTexSerializer()
         result = serializer.serialize(pid, record, True)
diff --git a/modules/weko-items-ui/weko_items_ui/views.py b/modules/weko-items-ui/weko_items_ui/views.py
index 6f605097c1..2ff3f5ce49 100644
--- a/modules/weko-items-ui/weko_items_ui/views.py
+++ b/modules/weko-items-ui/weko_items_ui/views.py
@@ -1529,8 +1529,13 @@ def validate_bibtex_export():
     @return:
     """
     from .utils import validate_bibtex
-    post_data = request.get_json()
-    record_ids = post_data['record_ids']
+    post_data = request.get_json(silent=True)
+    record_ids = post_data.get('record_ids') \
+        if isinstance(post_data, dict) else None
+    if not isinstance(record_ids, list) or not all(
+            isinstance(i, (int, str)) and not isinstance(i, bool)
+            for i in record_ids):
+        abort(400)
     invalid_record_ids = validate_bibtex(record_ids)
     return jsonify(invalid_record_ids=invalid_record_ids)
 
diff --git a/modules/weko-records-ui/weko_records_ui/config.py b/modules/weko-records-ui/weko_records_ui/config.py
index 05b1eb5aa8..d24f193c5d 100644
--- a/modules/weko-records-ui/weko_records_ui/config.py
+++ b/modules/weko-records-ui/weko_records_ui/config.py
@@ -99,6 +99,8 @@
         pid_type='recid',
         route='/records/<pid_value>',
         view_imp='weko_signposting.api.requested_signposting',
+        permission_factory_imp='weko_records_ui.permissions'
+                               ':page_permission_factory',
         methods=['HEAD']
     ),
     recid=dict(
diff --git a/modules/weko-search-ui/tests/test_views.py b/modules/weko-search-ui/tests/test_views.py
index ea2482a644..a0240ea72e 100644
--- a/modules/weko-search-ui/tests/test_views.py
+++ b/modules/weko-search-ui/tests/test_views.py
@@ -166,6 +166,61 @@ def test_get_path_name_dict(i18n_app, users, indices):
         assert get_path_name_dict('33_44')
 
 
+def _add_unpublished_index(db, index_id):
+    from weko_index_tree.models import Index
+    # (parent, position) に一意制約があるので、フィクスチャのインデックスと重ならない位置に置く
+    used = [i.position for i in Index.query.filter_by(parent=0).all()]
+    with db.session.begin_nested():
+        db.session.add(Index(
+            index_name="unpublished",
+            index_name_english="unpublished",
+            public_state=False,
+            id=index_id,
+            parent=0,
+            position=(max(used) + 1) if used else 0,
+        ))
+    db.session.commit()
+
+
+# .tox/c1/bin/pytest --cov=weko_search_ui tests/test_views.py::test_get_path_name_dict_admin_sees_unpublished -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-search-ui/.tox/c1/tmp
+def test_get_path_name_dict_admin_sees_unpublished(i18n_app, db, users, indices):
+    _add_unpublished_index(db, 77)
+    with patch("flask_login.utils._get_user", return_value=users[3]['obj']):
+        res = get_path_name_dict('33_44_77')
+        data = json.loads(res.data)
+        assert set(data.keys()) == {"33", "44", "77"}
+
+
+# .tox/c1/bin/pytest --cov=weko_search_ui tests/test_views.py::test_get_path_name_dict_filters_unbrowsable -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-search-ui/.tox/c1/tmp
+def test_get_path_name_dict_filters_unbrowsable(i18n_app, db, users, indices):
+    _add_unpublished_index(db, 77)
+    # contributor
+    with patch("flask_login.utils._get_user", return_value=users[1]['obj']):
+        res = get_path_name_dict('77')
+        assert json.loads(res.data) == {}
+    # non-existent index is skipped
+    with patch("flask_login.utils._get_user", return_value=users[3]['obj']):
+        res = get_path_name_dict('33_99999')
+        assert set(json.loads(res.data).keys()) == {"33"}
+
+
+# .tox/c1/bin/pytest --cov=weko_search_ui tests/test_views.py::test_get_path_name_dict_guest -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-search-ui/.tox/c1/tmp
+def test_get_path_name_dict_guest(app, client, db, indices):
+    _add_unpublished_index(db, 77)
+    url = url_for("weko_search_ui.get_path_name_dict", path_str="33_77", _external=True)
+    res = client.get(url)
+    assert res.status_code == 200
+    assert "77" not in json.loads(res.data)
+
+
+# .tox/c1/bin/pytest --cov=weko_search_ui tests/test_views.py::test_get_path_name_dict_invalid -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-search-ui/.tox/c1/tmp
+@pytest.mark.parametrize("path_str", ["abc", "33_x", "33__44", "-1", "1" * 19])
+def test_get_path_name_dict_invalid(app, client, db, indices, path_str):
+    url = url_for("weko_search_ui.get_path_name_dict", path_str=path_str, _external=True)
+    res = client.get(url)
+    assert res.status_code == 400
+
+
 # def gettitlefacet():
 def test_gettitlefacet(i18n_app, users, client, facet_search_setting):
     with patch("flask_login.utils._get_user", return_value=users[3]['obj']):
diff --git a/modules/weko-search-ui/weko_search_ui/views.py b/modules/weko-search-ui/weko_search_ui/views.py
index 1b83f85761..c492604135 100644
--- a/modules/weko-search-ui/weko_search_ui/views.py
+++ b/modules/weko-search-ui/weko_search_ui/views.py
@@ -20,12 +20,13 @@
 
 """Blueprint for weko-search-ui."""
 
+import re
 import time
 import traceback
 from xml.etree import ElementTree
 
 from blinker import Namespace
-from flask import Blueprint, current_app, flash, jsonify, render_template, request
+from flask import Blueprint, abort, current_app, flash, jsonify, render_template, request
 from flask_babelex import gettext as _
 from flask_login import login_required
 from flask_security import current_user
@@ -371,13 +372,21 @@ def get_child_list(index_id=0):
 
 @blueprint.route("/get_path_name_dict/<string:path_str>", methods=["GET"])
 def get_path_name_dict(path_str=""):
-    """Get path and name."""
+    """Get path and name.
+
+    Only indexes that the current user can browse are included.
+    """
+    from weko_index_tree.utils import check_index_permissions
     path_name_dict = {}
     path_arr = path_str.split("_")
+    if not all(re.match(r"^[0-9]{1,18}$", path) for path in path_arr):
+        abort(400)
     for path in path_arr:
-        index = Indexes.get_index(index_id=path)
+        index = Indexes.get_index(index_id=int(path))
+        if index is None or not check_index_permissions(index_id=index.id):
+            continue
         idx_name = index.index_name
-        idx_name_en = index.index_name_english
+        idx_name_en = index.index_name_english or ""
         if current_i18n.language == "ja" and idx_name:
             path_name_dict[path] = idx_name.replace("\n", r"<br\>").replace(
                 "&EMPTY&", ""
diff --git a/modules/weko-signposting/tests/conftest.py b/modules/weko-signposting/tests/conftest.py
index d5678564b8..bd5e7e6c2c 100644
--- a/modules/weko-signposting/tests/conftest.py
+++ b/modules/weko-signposting/tests/conftest.py
@@ -20,6 +20,7 @@
 
 import pytest
 from flask import Flask
+from flask_babelex import Babel
 from sqlalchemy_utils.functions import create_database, database_exists
 
 from invenio_access import InvenioAccess
@@ -39,6 +40,7 @@
 from weko_index_tree import WekoIndexTree
 from weko_index_tree.api import Indexes
 from weko_logging.audit import WekoLoggingUserActivity
+from weko_records_ui.config import RECORDS_UI_ENDPOINTS
 from weko_signposting import WekoSignposting
 from weko_signposting.views import blueprint
 from weko_workflow import WekoWorkflow
@@ -76,13 +78,14 @@ def base_app(instance_path):
         INDEXER_FILE_DOC_TYPE="content",
         TESTING=True,
         OAISERVER_METADATA_FORMATS=oaiserver,
+        WEKO_PERMISSION_SUPER_ROLE_USER=[
+            'System Administrator',
+            'Repository Administrator',
+        ],
+        WEKO_PERMISSION_ROLE_COMMUNITY=['Community Administrator'],
         RECORDS_UI_ENDPOINTS=dict(
-            recid_signposting=dict(
-                pid_type='recid',
-                route='/records/<pid_value>',
-                view_imp='weko_signposting.api.requested_signposting',
-                methods=['HEAD']
-            ),
+            # Use the definition shipped by weko-records-ui as is.
+            recid_signposting=RECORDS_UI_ENDPOINTS['recid_signposting'],
             recid=dict(
                 pid_type='recid',
                 route='/records/<pid_value>',
@@ -94,6 +97,9 @@ def base_app(instance_path):
             ),
         ),
     )
+    # The detail page permission compares the publish date in the user's
+    # timezone.
+    Babel(app_)
     InvenioAccess(app_)
     InvenioAccounts(app_)
     InvenioDB(app_)
diff --git a/modules/weko-signposting/tests/test_api.py b/modules/weko-signposting/tests/test_api.py
index ed7bec9de8..5de96e6c10 100644
--- a/modules/weko-signposting/tests/test_api.py
+++ b/modules/weko-signposting/tests/test_api.py
@@ -23,6 +23,10 @@ def test_requested_signposting(app, client, db_records, mocker):
     depid, recid, parent, doi, record, item = db_records[0]
     expected = json_data("data/link_str.json")
 
+    # the record is public and its index is viewable
+    mocker.patch(
+        "weko_records_ui.permissions.check_index_permissions",
+        return_value=True)
     mock_permalink = mocker.patch("weko_signposting.api.get_record_doi")
     mock_permalink.return_value = None
 
@@ -44,6 +48,34 @@ def test_requested_signposting(app, client, db_records, mocker):
     assert res.status_code == 200
 
 
+# .tox/c1/bin/pytest --cov=weko_signposting tests/test_api.py::test_requested_signposting_permission -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-signposting/.tox/c1/tmp
+def test_requested_signposting_permission(app, client, db_records, users,
+                                          mocker):
+    # the record is not public
+    depid, recid, parent, doi, record, item = db_records[2]
+    assert record["publish_status"] == "1"
+    mocker.patch("weko_signposting.api.get_record_doi", return_value=None)
+    url = url_for("invenio_records_ui.recid_signposting",
+                  pid_value=recid.pid_value)
+
+    # guest: redirected to the login page, no links
+    res = client.head(url)
+    assert res.status_code == 302
+    assert "Link" not in res.headers
+
+    # user without permission to view the record
+    with patch("flask_login.utils._get_user", return_value=users[0]["obj"]):
+        res = client.head(url)
+    assert res.status_code == 403
+    assert "Link" not in res.headers
+
+    # administrator can view the record
+    with patch("flask_login.utils._get_user", return_value=users[3]["obj"]):
+        res = client.head(url)
+    assert res.status_code == 200
+    assert "Link" in res.headers
+
+
 # def get_record_doi(recid):
 # .tox/c1/bin/pytest --cov=weko_signposting tests/test_api.py::test_get_record_doi -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-signposting/.tox/c1/tmp
 def test_get_record_doi(app, client, db, db_records, mocker):
diff --git a/modules/weko-sitemap/tests/test_ext.py b/modules/weko-sitemap/tests/test_ext.py
index eaa3955590..01af3a4ab7 100644
--- a/modules/weko-sitemap/tests/test_ext.py
+++ b/modules/weko-sitemap/tests/test_ext.py
@@ -134,12 +134,46 @@ def test_gzip_response(create_app):
 # .tox/c1/bin/pytest --cov=weko_sitemap tests/test_ext.py::test_generate_all_item_urls -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-sitemap/.tox/c1/tmp
 def test_generate_all_item_urls(app,records):
     with app.test_request_context():
-        result = iter(current_app.extensions["weko-sitemap"]._generate_all_item_urls())
-        for i,r in enumerate(result):
-            if i==0:
-                assert r["loc"] == "http://test_server/records/1"
-            if i==1:
-                assert r["loc"] == "http://test_server/records/2"
+        with patch("weko_index_tree.utils.check_index_permissions", return_value=True):
+            result = list(current_app.extensions["weko-sitemap"]._generate_all_item_urls())
+        assert [r["loc"] for r in result] == [
+            "http://test_server/records/1",
+            "http://test_server/records/2",
+        ]
+
+
+def _sitemap_record(publish_status="0", pubdate="2022-08-20"):
+    return {
+        "publish_status": publish_status,
+        "pubdate": {"attribute_name": "PubDate", "attribute_value": pubdate},
+        "path": ["1"],
+    }
+
+
+# .tox/c1/bin/pytest --cov=weko_sitemap tests/test_ext.py::test_generate_all_item_urls_not_public -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-sitemap/.tox/c1/tmp
+@pytest.mark.parametrize("record_2, index_permission, expected", [
+    (_sitemap_record(), True, ["1", "2"]),
+    (_sitemap_record(publish_status="1"), True, ["1"]),
+    (_sitemap_record(publish_status="-1"), True, ["1"]),
+    (_sitemap_record(pubdate="2999-01-01"), True, ["1"]),
+    (None, True, ["1"]),
+    (_sitemap_record(), False, []),
+])
+def test_generate_all_item_urls_not_public(app, records, record_2, index_permission, expected):
+    def get_record_by_pid(pid):
+        if pid == "1":
+            return _sitemap_record()
+        if record_2 is None:
+            raise Exception("pid does not exist")
+        return record_2
+
+    with app.test_request_context():
+        with patch("weko_deposit.api.WekoRecord.get_record_by_pid", side_effect=get_record_by_pid):
+            with patch("weko_index_tree.utils.check_index_permissions", return_value=index_permission):
+                result = list(current_app.extensions["weko-sitemap"]._generate_all_item_urls())
+    assert [r["loc"] for r in result] == [
+        "http://test_server/records/{}".format(i) for i in expected
+    ]
     
 # .tox/c1/bin/pytest --cov=weko_sitemap tests/test_ext.py::test_load_cache_pages -vv -s --cov-branch --cov-report=term --cov-report=html --basetemp=/code/modules/weko-sitemap/.tox/c1/tmp
 def test_load_cache_pages(create_app):
diff --git a/modules/weko-sitemap/weko_sitemap/ext.py b/modules/weko-sitemap/weko_sitemap/ext.py
index b667c0788e..c337028be0 100644
--- a/modules/weko-sitemap/weko_sitemap/ext.py
+++ b/modules/weko-sitemap/weko_sitemap/ext.py
@@ -118,16 +118,40 @@ def _generate_all_item_urls(self):
              .limit(current_app.config['WEKO_SITEMAP_TOTAL_MAX_URL_COUNT']))
 
         for recid, rm in q.yield_per(1000):
+            pid_value = (recid.pid_value).replace('.1', '')
+            if not self._is_public_item(pid_value):
+                continue
             yield {
                 'loc': url_for('invenio_records_ui.recid',
-                               pid_value=(recid.pid_value).replace(
-                                   '.1', ''),
+                               pid_value=pid_value,
                                _external=True),
                 # W3C Datetime format YYYY-MM-DDThh:mmTZD
                 'lastmod': format_datetime(
                     rm.updated, 'yyyy-MM-ddTHH:mm:ssz', 'full')
             }
 
+    @staticmethod
+    def _is_public_item(pid_value):
+        """Check that the item detail page is open to the public.
+
+        Uses the same checks as the detail page for a guest user: the item
+        is published, its publication date has come and it belongs to an
+        index that guests can browse.
+
+        :param pid_value: Identifier of the item (without version).
+        :return: True if the item can be listed in the sitemap.
+        """
+        from weko_deposit.api import WekoRecord
+        from weko_index_tree.utils import check_index_permissions
+        from weko_records_ui.permissions import check_publish_status
+        try:
+            record = WekoRecord.get_record_by_pid(pid_value)
+            return bool(check_publish_status(record)) \
+                and bool(check_index_permissions(record))
+        except Exception as ex:
+            current_app.logger.debug(ex)
+            return False
+
     def _load_cache_pages(self):
         """Get pages from cache instead of re-creating them."""
         kwargs = dict(