Compare commits

..
Author SHA1 Message Date
stumpylog 6c85e8877a Chore: type GlobalPermissionsByMethod.perms_map keys as HTTP methods
perms_map accepted any string key, so a misspelled method or a HEAD entry
would never match. Keys are now a Literal of the mappable methods, without
HEAD, and the map is typed read-only.
2026-10-14 13:17:26 -07:00
stumpylog 6b700393a0 Chore: share one base class for the method-mapped permission classes
ViewDocumentsPermissions, TrashPermissions, PaperlessNotePermissions and
AcknowledgeTasksPermissions each repeated the same has_permission body and
disagreed on HTTP methods missing from their perms_map.

GlobalPermissionsByMethod now holds that logic once. HEAD is checked
against the GET entry and a method missing from the map is denied. The
task acknowledge map gains OPTIONS, now requiring change_paperlesstask
like POST instead of being open to any authenticated user. The global
statistics and system status checks also share their active and
authenticated preamble.
2026-10-14 13:08:39 -07:00
11 changed files with 106 additions and 265 deletions

No files matched your search

+1 -1
View File
@@ -43,7 +43,7 @@ dependencies = [
"drf-spectacular~=0.30",
"drf-spectacular-sidecar>=2026.7.1,<2026.10",
"drf-writable-nested~=0.7.1",
"filelock>=3.32,<4.1",
"filelock~=3.32.0",
"flower>=2.0.1,<2.3",
"gotenberg-client[httpx]~=1.0",
"httpx-oauth~=0.17",
-2
View File
@@ -130,14 +130,12 @@ class DocumentMetadataOverrides:
get_users_with_perms(
doc,
only_with_perms_in=["view_document"],
with_group_users=False,
).values_list("id", flat=True),
)
overrides.change_users = list(
get_users_with_perms(
doc,
only_with_perms_in=["change_document"],
with_group_users=False,
).values_list("id", flat=True),
)
overrides.custom_fields = {
@@ -3,7 +3,6 @@ from itertools import islice
from pathlib import Path
from typing import TYPE_CHECKING
from typing import Any
from typing import Final
from allauth.mfa.models import Authenticator
from allauth.socialaccount.models import SocialAccount
@@ -71,10 +70,6 @@ from paperless.models import ApplicationConfiguration
from paperless_mail.models import MailAccount
from paperless_mail.models import MailRule
# Derived names replace the original's extension with these suffixes
THUMBNAIL_SUFFIX: Final[str] = "-thumbnail.webp"
ARCHIVE_SUFFIX: Final[str] = "-archive.pdf"
def serialize_queryset_batched(
queryset: "QuerySet[Any]",
@@ -517,12 +512,7 @@ class Command(CryptMixin, PaperlessCommand):
counter=filename_counter,
)
else:
# Leave room for the longer of the derived name suffixes
base_name = document.get_public_filename(
counter=filename_counter,
max_bytes=Document.MAX_PUBLIC_FILENAME_BYTES
- max(len(THUMBNAIL_SUFFIX), len(ARCHIVE_SUFFIX)),
)
base_name = document.get_public_filename(counter=filename_counter)
# The thumbnail and archive names are derived from the stem, so
# uniqueness must hold without the original's extension
@@ -551,7 +541,7 @@ class Command(CryptMixin, PaperlessCommand):
document_dict[EXPORTER_FILE_NAME] = original_arc
if not self.no_thumbnail:
thumbnail_name = base_name.parent / (base_name.stem + THUMBNAIL_SUFFIX)
thumbnail_name = base_name.parent / (base_name.stem + "-thumbnail.webp")
if self.use_folder_prefix:
thumbnail_name = Path("thumbnails") / thumbnail_name
thumbnail_arc = thumbnail_name.as_posix()
@@ -560,7 +550,7 @@ class Command(CryptMixin, PaperlessCommand):
thumbnail_arc = None
if not self.no_archive and document.has_archive_version:
archive_name = base_name.parent / (base_name.stem + ARCHIVE_SUFFIX)
archive_name = base_name.parent / (base_name.stem + "-archive.pdf")
if self.use_folder_prefix:
archive_name = Path("archive") / archive_name
archive_arc = archive_name.as_posix()
+7 -28
View File
@@ -157,8 +157,6 @@ class StoragePath(MatchingModel):
class Document(SoftDeleteModel, ModelWithOwner): # type: ignore[django-manager-missing]
MAX_STORED_FILENAME_LENGTH: Final[int] = 1024
# Per-component limit of common filesystems (ext4, XFS, btrfs, ZFS, APFS)
MAX_PUBLIC_FILENAME_BYTES: Final[int] = 255
correspondent = models.ForeignKey(
Correspondent,
@@ -486,47 +484,28 @@ class Document(SoftDeleteModel, ModelWithOwner): # type: ignore[django-manager-
def archive_file(self):
return Path(self.archive_path).open("rb")
def get_public_filename(
self,
*,
archive=False,
counter=0,
suffix=None,
max_bytes: int = MAX_PUBLIC_FILENAME_BYTES,
) -> str:
def get_public_filename(self, *, archive=False, counter=0, suffix=None) -> str:
"""
Returns a sanitized filename for the document, not including any paths.
The name fits in max_bytes (UTF-8). A title that is too long is
shortened, so the counter, suffix and extension are always kept.
"""
# Root owns metadata for all versions
context_document = (
self.root_document if self.root_document_id is not None else self
)
result = str(context_document)
tail = ""
if counter:
tail += f"_{counter:02}"
result += f"_{counter:02}"
if suffix:
tail += suffix
result += suffix
if archive:
tail += ".pdf"
result += ".pdf"
else:
tail += self.file_type
result += self.file_type
# sanitize_filename truncates from the end, which would cut off the
# counter and extension, so shorten the title before appending them
head_bytes = max_bytes - len(tail.encode())
head = str(context_document).encode()[:head_bytes].decode(errors="ignore")
return pathvalidate.sanitize_filename(
head + tail,
replacement_text="-",
max_len=max_bytes,
)
return pathvalidate.sanitize_filename(result, replacement_text="-")
@property
def file_type(self):
+48 -50
View File
@@ -1,5 +1,11 @@
from collections.abc import Mapping
from collections.abc import Sequence
from typing import Any
from typing import ClassVar
from typing import Literal
from typing import TypeGuard
from typing import TypeVar
from typing import cast
from django.contrib.auth.models import Group
from django.contrib.auth.models import Permission
@@ -62,28 +68,23 @@ class PaperlessAdminPermissions(BasePermission):
return request.user.is_active and request.user.is_staff
def has_global_statistics_permission(user: User | None) -> bool:
if (
user is None
or not getattr(user, "is_active", False)
or not getattr(user, "is_authenticated", False)
):
return False
def _is_active_and_authenticated(user: User | None) -> TypeGuard[User]:
return (
user is not None
and getattr(user, "is_active", False)
and getattr(user, "is_authenticated", False)
)
return getattr(user, "is_superuser", False) or user.has_perm(
"paperless.view_global_statistics",
def has_global_statistics_permission(user: User | None) -> bool:
return _is_active_and_authenticated(user) and (
getattr(user, "is_superuser", False)
or user.has_perm("paperless.view_global_statistics")
)
def has_system_status_permission(user: User | None) -> bool:
if (
user is None
or not getattr(user, "is_active", False)
or not getattr(user, "is_authenticated", False)
):
return False
return (
return _is_active_and_authenticated(user) and (
getattr(user, "is_superuser", False)
or getattr(user, "is_staff", False)
or user.has_perm("paperless.view_system_monitoring")
@@ -705,7 +706,32 @@ def has_perms_owner_aware(user, perms, obj):
return obj.owner is None or obj.owner == user or checker.has_perm(perms, obj)
class ViewDocumentsPermissions(BasePermission):
# HEAD is deliberately absent: GlobalPermissionsByMethod checks it against GET.
MappedMethod = Literal["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"]
class GlobalPermissionsByMethod(BasePermission):
"""
Requires the global (model-level) permissions listed in ``perms_map`` for
the request's HTTP method. HEAD is checked against the ``"GET"`` entry,
since DRF runs the GET handler for it. A method missing from
``perms_map`` is denied.
"""
perms_map: ClassVar[Mapping[MappedMethod, Sequence[str]]]
def has_permission(self, request: Any, view: Any) -> bool:
if not request.user or not request.user.is_authenticated:
return False
method = "GET" if request.method == "HEAD" else request.method
# An unknown method is not a MappedMethod, but .get() returns None for
# it, so it is still denied.
perms = self.perms_map.get(cast("MappedMethod", method))
return perms is not None and request.user.has_perms(perms)
class ViewDocumentsPermissions(GlobalPermissionsByMethod):
"""
Permissions class that checks for model permissions for only viewing Documents.
"""
@@ -716,31 +742,18 @@ class ViewDocumentsPermissions(BasePermission):
"POST": ["documents.view_document"],
}
def has_permission(self, request, view):
if not request.user or (not request.user.is_authenticated): # pragma: no cover
return False
return request.user.has_perms(self.perms_map.get(request.method, []))
class TrashPermissions(BasePermission):
class TrashPermissions(GlobalPermissionsByMethod):
"""Check the global document permission for each trash operation."""
perms_map = {
"OPTIONS": ["documents.view_document"],
"HEAD": ["documents.view_document"],
"GET": ["documents.view_document"],
"POST": ["documents.delete_document"],
}
def has_permission(self, request, view):
if not request.user or not request.user.is_authenticated: # pragma: no cover
return False
return request.user.has_perms(self.perms_map.get(request.method, []))
class PaperlessNotePermissions(BasePermission):
class PaperlessNotePermissions(GlobalPermissionsByMethod):
"""
Permissions class that checks for model permissions for Notes.
"""
@@ -760,28 +773,13 @@ class PaperlessNotePermissions(BasePermission):
],
}
def has_permission(self, request, view):
if not request.user or (not request.user.is_authenticated): # pragma: no cover
return False
perms = self.perms_map[request.method]
return request.user.has_perms(perms)
class AcknowledgeTasksPermissions(BasePermission):
class AcknowledgeTasksPermissions(GlobalPermissionsByMethod):
"""
Permissions class that checks for model permissions for acknowledging tasks.
"""
perms_map = {
"OPTIONS": ["documents.change_paperlesstask"],
"POST": ["documents.change_paperlesstask"],
}
def has_permission(self, request: Any, view: Any) -> bool:
if not request.user or not request.user.is_authenticated: # pragma: no cover
return False
perms = self.perms_map.get(request.method, [])
return request.user.has_perms(perms)
+28
View File
@@ -3653,6 +3653,34 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
def test_notes_head_and_unsupported_methods(self) -> None:
"""
GIVEN:
- A user allowed to view notes on their own document
WHEN:
- The notes endpoint is requested with HEAD, PUT and PATCH
THEN:
- HEAD is allowed like GET
- PUT and PATCH are rejected with 403 rather than a server error
"""
user = UserFactory()
grant_global(user, "view_note", "view_document")
doc = DocumentFactory(owner=user)
self.client.force_authenticate(user)
self.assertEqual(
self.client.head(f"/api/documents/{doc.pk}/notes/").status_code,
status.HTTP_200_OK,
)
self.assertEqual(
self.client.put(f"/api/documents/{doc.pk}/notes/").status_code,
status.HTTP_403_FORBIDDEN,
)
self.assertEqual(
self.client.patch(f"/api/documents/{doc.pk}/notes/").status_code,
status.HTTP_403_FORBIDDEN,
)
def test_delete_note(self) -> None:
"""
GIVEN:
+15
View File
@@ -1001,6 +1001,21 @@ class TestDocumentSearchApi(DirectoriesMixin, APITestCase):
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data), 10)
def test_search_autocomplete_head_requires_view_document(self) -> None:
"""
GIVEN:
- A user without the global view document permission
WHEN:
- A HEAD request is made for autocomplete
THEN:
- It is rejected like the equivalent GET request
"""
self.client.force_authenticate(user=UserFactory())
response = self.client.head("/api/search/autocomplete/?term=test")
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
def test_search_autocomplete_respect_permissions(self) -> None:
"""
GIVEN:
-28
View File
@@ -34,10 +34,8 @@ from documents.tests.utils import GetConsumerMixin
from paperless_mail.models import MailRule
from paperless_testing.assertions import FileSystemAssertsMixin
from paperless_testing.dirs import DirectoriesMixin
from paperless_testing.factories import DocumentFactory
from paperless_testing.factories import UserFactory
from paperless_testing.fakes.progress import FakeProgressManager
from paperless_testing.permissions import grant_object
class _BaseNewStyleParser:
@@ -1685,32 +1683,6 @@ class TestMetadataOverrides(TestCase):
self.assertEqual(base.actor_id, 2)
self.assertEqual(base.version_label, "v2")
def test_from_document_copies_user_and_group_grants_separately(self) -> None:
"""
GIVEN:
- A document with view and change granted directly to one user and
to a group with another user as a member
WHEN:
- Overrides are built from the document
THEN:
- The user lists hold only the directly granted user
- The group lists hold the group
"""
direct_user = UserFactory()
group_member = UserFactory()
group = Group.objects.create(name="group")
group_member.groups.add(group)
document = DocumentFactory()
grant_object(direct_user, document, "view_document", "change_document")
grant_object(group, document, "view_document", "change_document")
overrides = DocumentMetadataOverrides.from_document(document)
self.assertEqual(overrides.view_users, [direct_user.pk])
self.assertEqual(overrides.change_users, [direct_user.pk])
self.assertEqual(overrides.view_groups, [group.pk])
self.assertEqual(overrides.change_groups, [group.pk])
class TestBarcodeApplyDetectedASN(TestCase):
"""
@@ -4,7 +4,6 @@ from datetime import date
from pathlib import Path
from unittest import mock
import pytest
from django.test import TestCase
from django.test import override_settings
from faker import Faker
@@ -258,77 +257,3 @@ def test_suggestion_content() -> None:
)
doc.content = content_over_limit
assert doc.suggestion_content == expected_cropped_content
@pytest.mark.django_db
class TestPublicFilenameLength:
@pytest.mark.parametrize(
("title", "archive", "extension"),
[
pytest.param("a" * 300, False, ".jpg", id="ascii-original"),
pytest.param("a" * 300, True, ".pdf", id="ascii-archive"),
pytest.param("ä" * 200, False, ".jpg", id="multibyte-original"),
],
)
def test_long_title_keeps_extension_within_limit(
self,
title: str,
*,
archive: bool,
extension: str,
) -> None:
"""
GIVEN:
- A document whose title alone is longer than 255 bytes
WHEN:
- The public file name is generated
THEN:
- The name fits in 255 bytes and still ends with the extension
"""
doc = Document(mime_type="image/jpeg", title=title, created=date(2020, 12, 25))
name = doc.get_public_filename(archive=archive)
assert len(name.encode()) <= 255
assert name.endswith(extension)
def test_long_title_counters_give_distinct_names(self) -> None:
"""
GIVEN:
- A document whose title alone is longer than 255 bytes
WHEN:
- Public file names are generated with increasing counters
THEN:
- Each counter produces a distinct name that keeps the counter
"""
doc = Document(
mime_type="application/pdf",
title="ä" * 200,
created=date(2020, 12, 25),
)
names = [doc.get_public_filename(counter=counter) for counter in range(3)]
assert len(set(names)) == 3
assert names[1].endswith("_01.pdf")
assert all(len(name.encode()) <= 255 for name in names)
def test_max_bytes_bounds_name(self) -> None:
"""
GIVEN:
- A document with a long title
WHEN:
- The public file name is generated with a smaller byte limit
THEN:
- The name fits in that limit and keeps the counter and extension
"""
doc = Document(
mime_type="application/pdf",
title="a" * 300,
created=date(2020, 12, 25),
)
name = doc.get_public_filename(counter=1, max_bytes=100)
assert len(name.encode()) <= 100
assert name.endswith("_01.pdf")
@@ -1537,67 +1537,3 @@ class TestExportDerivedFileNameCollisions:
assert (
target / record[EXPORTER_ARCHIVE_NAME]
).read_text() == f"archive {pk}"
def test_long_titles_export_within_file_name_limit(
self,
paperless_dirs: PaperlessDirs,
tmp_path: Path,
) -> None:
"""
GIVEN:
- Two documents with the same created date and the same title
- The title alone is longer than 255 bytes and contains umlauts
WHEN:
- The exporter runs
THEN:
- The export finishes with distinct original, thumbnail and archive names
- Every exported file name fits in 255 bytes
- Each exported file holds its own document's content
"""
for pk in (1, 2):
(paperless_dirs.originals_dir / f"{pk:07}.pdf").write_bytes(
f"original {pk}".encode(),
)
(paperless_dirs.archive_dir / f"{pk:07}.pdf").write_bytes(
f"archive {pk}".encode(),
)
(paperless_dirs.thumbnail_dir / f"{pk:07}.webp").write_bytes(
f"thumbnail {pk}".encode(),
)
DocumentFactory(
pk=pk,
title="Bestellbestätigung " * 15,
created=date(2026, 10, 15),
filename=f"{pk:07}.pdf",
archive_filename=f"{pk:07}.pdf",
mime_type="application/pdf",
)
target = tmp_path / "export"
target.mkdir()
call_command(
"document_exporter",
str(target),
"--no-progress-bar",
skip_checks=True,
)
manifest = json.loads((target / "manifest.json").read_text())
records = {r["pk"]: r for r in manifest if r["model"] == "documents.document"}
for key in (
EXPORTER_FILE_NAME,
EXPORTER_THUMBNAIL_NAME,
EXPORTER_ARCHIVE_NAME,
):
assert records[1][key] != records[2][key]
for pk in (1, 2):
record = records[pk]
for key, content in (
(EXPORTER_FILE_NAME, f"original {pk}"),
(EXPORTER_THUMBNAIL_NAME, f"thumbnail {pk}"),
(EXPORTER_ARCHIVE_NAME, f"archive {pk}"),
):
assert len(Path(record[key]).name.encode()) <= 255
assert (target / record[key]).read_text() == content
Generated
+4 -4
View File
@@ -1272,11 +1272,11 @@ wheels = [
[[package]]
name = "filelock"
version = "4.0.12"
version = "3.32.6"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/53/e4/34efcb869715cf299e47d1ac7b2624d2bcb6f2d3dffc2f0abe8417f65ab2/filelock-4.0.12.tar.gz", hash = "sha256:cf42711a7ac791818b299fab0332a088c65aeeefa36290de98db92c434303b0c", size = 253880, upload-time = "2026-10-05T13:17:43.677Z" }
sdist = { url = "https://files.pythonhosted.org/packages/38/46/126b1831dca12060d4a8296bf9c4fe5c93c4f22197fa239cb0cc82042bba/filelock-3.32.6.tar.gz", hash = "sha256:a3f55a18af3652a94d8f47d6055df434f254ca1d02ef2524850c6d249ca2512c", size = 225172, upload-time = "2026-09-08T22:57:11.528Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/da/a1/b047e1c486a2c66283633b7dbb73872ca16ca1ce6a426f70d4d037fc8e1c/filelock-4.0.12-py3-none-any.whl", hash = "sha256:5f17ee83ecee8a6f3e389c75822fb70a1c2f0506b99438a6dffa1d56793588c8", size = 111732, upload-time = "2026-10-05T13:17:42.361Z" },
{ url = "https://files.pythonhosted.org/packages/cc/06/4f138f618dbea66803291274f228f01daf29f306fe8b96bc30dab765df75/filelock-3.32.6-py3-none-any.whl", hash = "sha256:3f16ecd0117feae0dfc147e8c62eb5daeccd8bd800378c3ddf416de9b4feb6b1", size = 100189, upload-time = "2026-09-08T22:57:10.182Z" },
]
[[package]]
@@ -3184,7 +3184,7 @@ requires-dist = [
{ name = "drf-spectacular", specifier = "~=0.30" },
{ name = "drf-spectacular-sidecar", specifier = ">=2026.7.1,<2026.10" },
{ name = "drf-writable-nested", specifier = "~=0.7.1" },
{ name = "filelock", specifier = ">=3.32,<4.1" },
{ name = "filelock", specifier = "~=3.32.0" },
{ name = "flower", specifier = ">=2.0.1,<2.3" },
{ name = "gotenberg-client", extras = ["httpx"], specifier = "~=1.0" },
{ name = "granian", extras = ["uvloop"], marker = "extra == 'webserver'", specifier = ">=2.7,<2.9" },