17425918180

Committed 03 Sep 2025 07:11AM UTC coverage: 95.796% (-0.6%) from 96.378%

Build # 17425918180

Build Type

push

github

Committed by

PascalRepond

Commit Message

translations: extract messages

Co-Authored-by: Pascal Repond <pascal.repond@rero.ch>

Run Details

7816 of 8159 relevant lines covered (95.8%)

0.96 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.85

/sonar/modules/documents/permissions.py

# Swiss Open Access Repository
# Copyright (C) 2021 RERO
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, version 3 of the License.
#
# This program 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""Permissions for documents."""

from flask import request
from invenio_files_rest.models import Bucket, ObjectVersion
from invenio_pidstore.errors import PIDDoesNotExistError
from invenio_pidstore.models import PersistentIdentifier

from sonar.modules.documents.api import DocumentRecord
from sonar.modules.organisations.api import current_organisation
from sonar.modules.permissions import FilesPermission, RecordPermission

from .utils import get_file_restriction, get_organisations


class DocumentPermission(RecordPermission):
    """Documents permissions."""

    @classmethod
    def list(cls, user, record=None):
        """List permission check.

        :param user: Current user record.
        :param record: Record to check.
        :returns: True is action can be done.
        """

        # Documents are accessible in public view, but eventually filtered
        # later by organisation
        if request.args.get("view"):
            return True

        # Only for moderators users.
        return bool(user and user.is_moderator and current_organisation)

    @classmethod
    def create(cls, user, record=None):
        """Create permission check.

        :param user: Current user record.
        :param record: Record to check.
        :returns: True is action can be done.
        """
        # Only for moderators users
        return user and user.is_moderator

    @classmethod
    def read(cls, user, record):
        """Read permission check.

        :param user: Current user record.
        :param record: Record to check.
        :returns: True is action can be done.
        """
        # Superuser is allowed.
        if user and user.is_superuser:
            return True

        document = DocumentRecord.get_record_by_pid(record["pid"])
        document = document.resolve()
        # Moderator can read their own documents.
        if user and user.is_moderator and document.has_organisation(current_organisation["pid"]):
            return True
        return not document.is_masked

    @classmethod
    def update(cls, user, record):
        """Update permission check.

        :param user: Current user record.
        :param record: Record to check.
        :returns: True is action can be done.
        """
        if not user or not user.is_moderator:
            return False

        if user.is_superuser:
            return True

        document = DocumentRecord.get_record_by_pid(record["pid"])
        document = document.resolve()

        # Moderator can update their own documents.
        if not document.has_organisation(current_organisation["pid"]):
            return False

        user = user.replace_refs()
        return document.has_subdivision(user.get("subdivision", {}).get("pid"))

    @classmethod
    def delete(cls, user, record):
        """Delete permission check.

        :param user: Current user record.
        :param record: Record to check.
        :returns: True is action can be done.
        """
        # Delete is only for admins.
        if not user or not user.is_admin:
            return False

        if user.is_superuser:
            return True

        # Check delete conditions and consider same rules as update
        return not cls.has_urn(record) and cls.update(user, record)

    @classmethod
    def has_urn(cls, record):
        """The record has an SONAR URN.

        :param record: Record to check.
        :returns: True if action can be done.
        """
        # Delete only documents with no URN or no registred URN
        document = DocumentRecord.get_record_by_pid(record["pid"])
        if document:
            # check if document has urn
            try:
                urn_identifier = PersistentIdentifier.get_by_object("urn", "rec", document.id)
            except PIDDoesNotExistError:
                return False

            return True
        return False


class DocumentFilesPermission(FilesPermission):
    """Documents files permissions.

    Write operations are limited to admin users, read depends if the
    corresponding document is masked or if the file is restricted.
    """

    @classmethod
    def get_document(cls, parent_record):
        """Get the parent document."""
        return DocumentRecord.get_record_by_pid(parent_record.get("pid"))

    @classmethod
    def read(cls, user, record, pid, parent_record):
        """Read permission check.

        :param user: current user record.
        :param record: Record to check.
        :param pid: The :class:`invenio_pidstore.models.PersistentIdentifier`
        instance.
        :param parent_record: the record related to the bucket.
        :returns: True is action can be done.
        """
        # Superuser is allowed.
        if user and user.is_superuser:
            return True
        document = cls.get_document(parent_record)
        if document and not DocumentPermission.read(user, document):
            return False

        # read the bucket metadata
        # TODO: filter the list of files based on embargo
        if isinstance(record, Bucket):
            return True
        file = document.files[record.key] if hasattr(record, "key") else document.files[record["key"]]
        file_type = file["type"]
        if file_type == "fulltext" and (not user or not user.is_admin):
            return False
        file_restriction = get_file_restriction(file, get_organisations(document), True)
        return not file_restriction.get("restricted", True)

    @classmethod
    def update(cls, user, record, pid, parent_record):
        """Update permission check.

        :param user: Current user record.
        :param record: Record to check.
        :param pid: The :class:`invenio_pidstore.models.PersistentIdentifier`
        instance.
        :param parent_record: the record related to the bucket.
        :returns: True is action can be done.
        """
        if user and user.is_superuser:
            return True
        document = cls.get_document(parent_record)
        if isinstance(record, ObjectVersion):
            file_type = document.files[record.key]["type"]
            if file_type == "file" and record.mimetype == "application/pdf":
                return not DocumentPermission.has_urn(parent_record) and DocumentPermission.update(user, parent_record)
        if isinstance(record, dict):
            file_type = record["type"]
            if file_type == "file" and record["mimetype"] == "application/pdf":
                return not DocumentPermission.has_urn(parent_record) and DocumentPermission.update(user, parent_record)
        if document := cls.get_document(parent_record):
            return DocumentPermission.update(user, document)
        return False

    @classmethod
    def delete(cls, user, record, pid, parent_record):
        """Delete permission check.

        :param user: Current user record.
        :param record: Record to check.
        :param pid: The :class:`invenio_pidstore.models.PersistentIdentifier`
        instance.
        :param parent_record: the record related to the bucket.
        :returns: True if action can be done.
        """
        return cls.update(user, record, pid, parent_record)

1	# Swiss Open Access Repository
2	# Copyright (C) 2021 RERO
3	#
4	# This program is free software: you can redistribute it and/or modify
5	# it under the terms of the GNU Affero General Public License as published by
6	# the Free Software Foundation, version 3 of the License.
7	#
8	# This program is distributed in the hope that it will be useful,
9	# but WITHOUT ANY WARRANTY; without even the implied warranty of
10	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11	# GNU Affero General Public License for more details.
12	#
13	# You should have received a copy of the GNU Affero General Public License
14	# along with this program. If not, see <http://www.gnu.org/licenses/>.
15
16	"""Permissions for documents."""
17
18	from flask import request	1✔
19	from invenio_files_rest.models import Bucket, ObjectVersion	1✔
20	from invenio_pidstore.errors import PIDDoesNotExistError	1✔
21	from invenio_pidstore.models import PersistentIdentifier	1✔
22
23	from sonar.modules.documents.api import DocumentRecord	1✔
24	from sonar.modules.organisations.api import current_organisation	1✔
25	from sonar.modules.permissions import FilesPermission, RecordPermission	1✔
26
27	from .utils import get_file_restriction, get_organisations	1✔
28
29
30	class DocumentPermission(RecordPermission):	1✔
31	"""Documents permissions."""
32
33	@classmethod	1✔
34	def list(cls, user, record=None):	1✔
35	"""List permission check.
36
37	:param user: Current user record.
38	:param record: Record to check.
39	:returns: True is action can be done.
40	"""
41
42	# Documents are accessible in public view, but eventually filtered
43	# later by organisation
44	if request.args.get("view"):	1✔
45	return True	1✔
46
47	# Only for moderators users.
48	return bool(user and user.is_moderator and current_organisation)	1✔
49
50	@classmethod	1✔
51	def create(cls, user, record=None):	1✔
52	"""Create permission check.
53
54	:param user: Current user record.
55	:param record: Record to check.
56	:returns: True is action can be done.
57	"""
58	# Only for moderators users
59	return user and user.is_moderator	1✔
60
61	@classmethod	1✔
62	def read(cls, user, record):	1✔
63	"""Read permission check.
64
65	:param user: Current user record.
66	:param record: Record to check.
67	:returns: True is action can be done.
68	"""
69	# Superuser is allowed.
70	if user and user.is_superuser:	1✔
71	return True	1✔
72
73	document = DocumentRecord.get_record_by_pid(record["pid"])	1✔
74	document = document.resolve()	1✔
75	# Moderator can read their own documents.
76	if user and user.is_moderator and document.has_organisation(current_organisation["pid"]):	1✔
77	return True	1✔
78	return not document.is_masked	1✔
79
80	@classmethod	1✔
81	def update(cls, user, record):	1✔
82	"""Update permission check.
83
84	:param user: Current user record.
85	:param record: Record to check.
86	:returns: True is action can be done.
87	"""
88	if not user or not user.is_moderator:	1✔
89	return False	1✔
90
91	if user.is_superuser:	1✔
92	return True	1✔
93
94	document = DocumentRecord.get_record_by_pid(record["pid"])	1✔
95	document = document.resolve()	1✔
96
97	# Moderator can update their own documents.
98	if not document.has_organisation(current_organisation["pid"]):	1✔
99	return False	1✔
100
101	user = user.replace_refs()	1✔
102	return document.has_subdivision(user.get("subdivision", {}).get("pid"))	1✔
103
104	@classmethod	1✔
105	def delete(cls, user, record):	1✔
106	"""Delete permission check.
107
108	:param user: Current user record.
109	:param record: Record to check.
110	:returns: True is action can be done.
111	"""
112	# Delete is only for admins.
113	if not user or not user.is_admin:	1✔
114	return False	1✔
115
116	if user.is_superuser:	1✔
117	return True	1✔
118
119	# Check delete conditions and consider same rules as update
120	return not cls.has_urn(record) and cls.update(user, record)	1✔
121
122	@classmethod	1✔
123	def has_urn(cls, record):	1✔
124	"""The record has an SONAR URN.
125
126	:param record: Record to check.
127	:returns: True if action can be done.
128	"""
129	# Delete only documents with no URN or no registred URN
130	document = DocumentRecord.get_record_by_pid(record["pid"])	1✔
131	if document:	1✔
132	# check if document has urn
133	try:	1✔
134	urn_identifier = PersistentIdentifier.get_by_object("urn", "rec", document.id)	1✔
135	except PIDDoesNotExistError:	1✔
136	return False	1✔
137
138	return True	1✔
139	return False	×
140
141
142	class DocumentFilesPermission(FilesPermission):	1✔
143	"""Documents files permissions.
144
145	Write operations are limited to admin users, read depends if the
146	corresponding document is masked or if the file is restricted.
147	"""
148
149	@classmethod	1✔
150	def get_document(cls, parent_record):	1✔
151	"""Get the parent document."""
152	return DocumentRecord.get_record_by_pid(parent_record.get("pid"))	1✔
153
154	@classmethod	1✔
155	def read(cls, user, record, pid, parent_record):	1✔
156	"""Read permission check.
157
158	:param user: current user record.
159	:param record: Record to check.
160	:param pid: The :class:`invenio_pidstore.models.PersistentIdentifier`
161	instance.
162	:param parent_record: the record related to the bucket.
163	:returns: True is action can be done.
164	"""
165	# Superuser is allowed.
166	if user and user.is_superuser:	1✔
167	return True	1✔
168	document = cls.get_document(parent_record)	1✔
169	if document and not DocumentPermission.read(user, document):	1✔
170	return False	1✔
171
172	# read the bucket metadata
173	# TODO: filter the list of files based on embargo
174	if isinstance(record, Bucket):	1✔
175	return True	1✔
176	file = document.files[record.key] if hasattr(record, "key") else document.files[record["key"]]	1✔
177	file_type = file["type"]	1✔
178	if file_type == "fulltext" and (not user or not user.is_admin):	1✔
179	return False	1✔
180	file_restriction = get_file_restriction(file, get_organisations(document), True)	1✔
181	return not file_restriction.get("restricted", True)	1✔
182
183	@classmethod	1✔
184	def update(cls, user, record, pid, parent_record):	1✔
185	"""Update permission check.
186
187	:param user: Current user record.
188	:param record: Record to check.
189	:param pid: The :class:`invenio_pidstore.models.PersistentIdentifier`
190	instance.
191	:param parent_record: the record related to the bucket.
192	:returns: True is action can be done.
193	"""
194	if user and user.is_superuser:	1✔
195	return True	1✔
196	document = cls.get_document(parent_record)	1✔
197	if isinstance(record, ObjectVersion):	1✔
198	file_type = document.files[record.key]["type"]	1✔
199	if file_type == "file" and record.mimetype == "application/pdf":	1✔
200	return not DocumentPermission.has_urn(parent_record) and DocumentPermission.update(user, parent_record)	1✔
201	if isinstance(record, dict):	1✔
202	file_type = record["type"]	1✔
203	if file_type == "file" and record["mimetype"] == "application/pdf":	1✔
204	return not DocumentPermission.has_urn(parent_record) and DocumentPermission.update(user, parent_record)	1✔
205	if document := cls.get_document(parent_record):	1✔
206	return DocumentPermission.update(user, document)	1✔
207	return False	×
208
209	@classmethod	1✔
210	def delete(cls, user, record, pid, parent_record):	1✔
211	"""Delete permission check.
212
213	:param user: Current user record.
214	:param record: Record to check.
215	:param pid: The :class:`invenio_pidstore.models.PersistentIdentifier`
216	instance.
217	:param parent_record: the record related to the bucket.
218	:returns: True if action can be done.
219	"""
220	return cls.update(user, record, pid, parent_record)	1✔

rero / sonar / 17425918180

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous