mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-10-16 04:57:13 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
252652b161 | ||
|
|
d4a57204eb | ||
|
|
c50c6f5517 |
No files matched your search
+1
-1
@@ -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.0",
|
||||
"filelock>=3.32,<4.1",
|
||||
"flower>=2.0.1,<2.3",
|
||||
"gotenberg-client[httpx]~=1.0",
|
||||
"httpx-oauth~=0.17",
|
||||
|
||||
@@ -34,7 +34,6 @@ from documents.models import DocumentType
|
||||
from documents.models import PaperlessTask
|
||||
from documents.models import StoragePath
|
||||
from documents.models import Tag
|
||||
from documents.permissions import PermissionsSpec
|
||||
from documents.permissions import set_permissions_for_objects
|
||||
from documents.plugins.helpers import DocumentsStatusManager
|
||||
from documents.tasks import bulk_update_documents
|
||||
@@ -446,7 +445,7 @@ def reprocess(doc_ids: list[int], *, remote_ocr: bool = False) -> Literal["OK"]:
|
||||
|
||||
def set_permissions(
|
||||
doc_ids: list[int],
|
||||
set_permissions: PermissionsSpec,
|
||||
set_permissions: dict,
|
||||
*,
|
||||
owner: User | None = None,
|
||||
merge: bool = False,
|
||||
|
||||
@@ -38,8 +38,7 @@ from documents.models import StoragePath
|
||||
from documents.models import Tag
|
||||
from documents.models import WorkflowTrigger
|
||||
from documents.parsers import ParseError
|
||||
from documents.permissions import PermissionsSpec
|
||||
from documents.permissions import set_permissions_for_objects
|
||||
from documents.permissions import set_permissions_for_object
|
||||
from documents.plugins.base import AlwaysRunPluginMixin
|
||||
from documents.plugins.base import ConsumeTaskPlugin
|
||||
from documents.plugins.base import NoCleanupPluginMixin
|
||||
@@ -949,7 +948,7 @@ class ConsumerPlugin(
|
||||
or self.metadata.change_users is not None
|
||||
or self.metadata.change_groups is not None
|
||||
):
|
||||
permissions: PermissionsSpec = {
|
||||
permissions = {
|
||||
"view": {
|
||||
"users": self.metadata.view_users or [],
|
||||
"groups": self.metadata.view_groups or [],
|
||||
@@ -959,7 +958,7 @@ class ConsumerPlugin(
|
||||
"groups": self.metadata.change_groups or [],
|
||||
},
|
||||
}
|
||||
set_permissions_for_objects(permissions, Document, [document.pk])
|
||||
set_permissions_for_object(permissions=permissions, object=document)
|
||||
|
||||
if self.metadata.custom_fields:
|
||||
for field in CustomField.objects.filter(
|
||||
|
||||
@@ -130,12 +130,14 @@ 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,6 +3,7 @@ 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
|
||||
@@ -70,6 +71,10 @@ 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]",
|
||||
@@ -512,7 +517,12 @@ class Command(CryptMixin, PaperlessCommand):
|
||||
counter=filename_counter,
|
||||
)
|
||||
else:
|
||||
base_name = document.get_public_filename(counter=filename_counter)
|
||||
# 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)),
|
||||
)
|
||||
|
||||
# The thumbnail and archive names are derived from the stem, so
|
||||
# uniqueness must hold without the original's extension
|
||||
@@ -541,7 +551,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.webp")
|
||||
thumbnail_name = base_name.parent / (base_name.stem + THUMBNAIL_SUFFIX)
|
||||
if self.use_folder_prefix:
|
||||
thumbnail_name = Path("thumbnails") / thumbnail_name
|
||||
thumbnail_arc = thumbnail_name.as_posix()
|
||||
@@ -550,7 +560,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.pdf")
|
||||
archive_name = base_name.parent / (base_name.stem + ARCHIVE_SUFFIX)
|
||||
if self.use_folder_prefix:
|
||||
archive_name = Path("archive") / archive_name
|
||||
archive_arc = archive_name.as_posix()
|
||||
|
||||
+28
-7
@@ -157,6 +157,8 @@ 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,
|
||||
@@ -484,28 +486,47 @@ 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) -> str:
|
||||
def get_public_filename(
|
||||
self,
|
||||
*,
|
||||
archive=False,
|
||||
counter=0,
|
||||
suffix=None,
|
||||
max_bytes: int = MAX_PUBLIC_FILENAME_BYTES,
|
||||
) -> 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:
|
||||
result += f"_{counter:02}"
|
||||
tail += f"_{counter:02}"
|
||||
|
||||
if suffix:
|
||||
result += suffix
|
||||
tail += suffix
|
||||
|
||||
if archive:
|
||||
result += ".pdf"
|
||||
tail += ".pdf"
|
||||
else:
|
||||
result += self.file_type
|
||||
tail += self.file_type
|
||||
|
||||
return pathvalidate.sanitize_filename(result, replacement_text="-")
|
||||
# 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,
|
||||
)
|
||||
|
||||
@property
|
||||
def file_type(self):
|
||||
|
||||
+120
-37
@@ -1,10 +1,4 @@
|
||||
from collections.abc import Iterable
|
||||
from collections.abc import Mapping
|
||||
from typing import Any
|
||||
from typing import Final
|
||||
from typing import Literal
|
||||
from typing import TypeAlias
|
||||
from typing import TypedDict
|
||||
from typing import TypeVar
|
||||
|
||||
from django.contrib.auth.models import Group
|
||||
@@ -26,29 +20,16 @@ from django.db.models.functions import Coalesce
|
||||
from guardian.core import ObjectPermissionChecker
|
||||
from guardian.models import GroupObjectPermission
|
||||
from guardian.models import UserObjectPermission
|
||||
from guardian.shortcuts import assign_perm
|
||||
from guardian.shortcuts import get_objects_for_user
|
||||
from guardian.shortcuts import get_users_with_perms
|
||||
from guardian.shortcuts import remove_perm
|
||||
from rest_framework.permissions import BasePermission
|
||||
from rest_framework.permissions import DjangoObjectPermissions
|
||||
|
||||
from documents.models import Document
|
||||
from documents.versioning import get_root_document
|
||||
|
||||
PermissionAction = Literal["view", "change"]
|
||||
PERMISSION_ACTIONS: Final[tuple[PermissionAction, ...]] = ("view", "change")
|
||||
|
||||
|
||||
class PermissionGrantees(TypedDict, total=False):
|
||||
"""
|
||||
The user and group ids to grant one permission action to. A missing key
|
||||
leaves that kind of grant untouched.
|
||||
"""
|
||||
|
||||
users: Iterable[int]
|
||||
groups: Iterable[int]
|
||||
|
||||
|
||||
PermissionsSpec: TypeAlias = Mapping[PermissionAction, PermissionGrantees]
|
||||
|
||||
|
||||
class PaperlessObjectPermissions(DjangoObjectPermissions):
|
||||
"""
|
||||
@@ -123,10 +104,87 @@ def get_groups_with_only_permission(obj, codename):
|
||||
return Group.objects.filter(id__in=group_object_perm_group_ids).distinct()
|
||||
|
||||
|
||||
def set_permissions_for_object(
|
||||
permissions: dict,
|
||||
object,
|
||||
*,
|
||||
merge: bool = False,
|
||||
) -> None:
|
||||
"""
|
||||
Set permissions for an object. The permissions are given as a mapping of actions
|
||||
to a dict of user / group id lists, e.g.
|
||||
{"view": {"users": [1], "groups": [2]}, "change": {"users": [], "groups": []}}.
|
||||
|
||||
If merge is True, the permissions are merged with the existing permissions and
|
||||
no users or groups are removed. If False, the permissions are set to exactly
|
||||
the given list of users and groups.
|
||||
"""
|
||||
|
||||
for action, entry in permissions.items():
|
||||
permission = f"{action}_{object.__class__.__name__.lower()}"
|
||||
if "users" in entry:
|
||||
# users
|
||||
users_to_add = User.objects.filter(id__in=entry["users"])
|
||||
users_to_remove = (
|
||||
get_users_with_perms(
|
||||
object,
|
||||
only_with_perms_in=[permission],
|
||||
with_group_users=False,
|
||||
)
|
||||
if not merge
|
||||
else User.objects.none()
|
||||
)
|
||||
if users_to_add.exists() and users_to_remove.exists():
|
||||
users_to_remove = users_to_remove.exclude(id__in=users_to_add)
|
||||
if users_to_remove.exists():
|
||||
for user in users_to_remove:
|
||||
remove_perm(permission, user, object)
|
||||
if users_to_add.exists():
|
||||
for user in users_to_add:
|
||||
assign_perm(permission, user, object)
|
||||
if action == "change":
|
||||
# change gives view too
|
||||
assign_perm(
|
||||
f"view_{object.__class__.__name__.lower()}",
|
||||
user,
|
||||
object,
|
||||
)
|
||||
if "groups" in entry:
|
||||
# groups
|
||||
groups_to_add = Group.objects.filter(id__in=entry["groups"])
|
||||
groups_to_remove = (
|
||||
get_groups_with_only_permission(
|
||||
object,
|
||||
permission,
|
||||
)
|
||||
if not merge
|
||||
else Group.objects.none()
|
||||
)
|
||||
if groups_to_add.exists() and groups_to_remove.exists():
|
||||
groups_to_remove = groups_to_remove.exclude(id__in=groups_to_add)
|
||||
if groups_to_remove.exists():
|
||||
for group in groups_to_remove:
|
||||
remove_perm(permission, group, object)
|
||||
if groups_to_add.exists():
|
||||
for group in groups_to_add:
|
||||
assign_perm(permission, group, object)
|
||||
if action == "change":
|
||||
# change gives view too
|
||||
assign_perm(
|
||||
f"view_{object.__class__.__name__.lower()}",
|
||||
group,
|
||||
object,
|
||||
)
|
||||
|
||||
|
||||
def _resolve_permissions(codenames: set[str], ctype: ContentType) -> list[Permission]:
|
||||
"""
|
||||
Resolve ``codenames`` to Permission rows, raising for any that don't exist
|
||||
so a bad action name fails loudly instead of granting nothing.
|
||||
Resolves `codenames` to Permission rows, raising like the single-object
|
||||
assign_perm() this bulk path replaces does (via a `.get()` internally)
|
||||
if any codename doesn't exist. SetPermissionsSerializer rejects unknown
|
||||
action names at the API, but a caller passing one directly would
|
||||
otherwise get a plain `.filter()` that silently builds zero rows and
|
||||
no-ops instead of reporting the bad input.
|
||||
"""
|
||||
permission_objs = list(
|
||||
Permission.objects.filter(content_type=ctype, codename__in=codenames),
|
||||
@@ -145,13 +203,16 @@ def _apply_bulk_permission_entry(
|
||||
perm_model: type[UserObjectPermission] | type[GroupObjectPermission],
|
||||
identity_model: type[User] | type[Group],
|
||||
identity_field: str,
|
||||
ids: Iterable[int],
|
||||
ids: list[int],
|
||||
codename: str,
|
||||
permission_objs: list[Permission],
|
||||
ctype: ContentType,
|
||||
object_pks: list[str],
|
||||
merge: bool,
|
||||
) -> None:
|
||||
# Only the ids are needed to build permission rows (via `<field>_id=`),
|
||||
# so avoid fetching full User/Group rows for identities that may not
|
||||
# even end up being granted anything new.
|
||||
add_ids = set(
|
||||
identity_model.objects.filter(id__in=ids).values_list("id", flat=True),
|
||||
)
|
||||
@@ -189,26 +250,45 @@ def _apply_bulk_permission_entry(
|
||||
for pk in object_pks
|
||||
for identity_id in add_ids
|
||||
]
|
||||
# ignore_conflicts skips grants that already exist (unique constraint).
|
||||
# ignore_conflicts skips only rows that already exist as an exact
|
||||
# (identity, permission, object) match -- the same de-dup the
|
||||
# underlying (user|group, permission, object_pk) unique constraint
|
||||
# already enforces for the single-object assign_perm() this replaces,
|
||||
# so it doesn't change what counts as "already granted". batch_size
|
||||
# caps how many rows go into a single INSERT statement.
|
||||
perm_model.objects.bulk_create(rows, ignore_conflicts=True, batch_size=1000)
|
||||
|
||||
|
||||
def set_permissions_for_objects(
|
||||
permissions: PermissionsSpec,
|
||||
permissions: dict,
|
||||
model: type[Model],
|
||||
pks: QuerySet | list,
|
||||
*,
|
||||
merge: bool = False,
|
||||
) -> None:
|
||||
"""
|
||||
Set the view/change grants for every object in ``pks``. With ``merge``,
|
||||
the given users and groups are added; without it, each given action's
|
||||
direct grants are replaced. Granting change also grants view, and an
|
||||
action entry with neither key is left untouched.
|
||||
Bulk equivalent of set_permissions_for_object: applies the same
|
||||
permission changes to every object identified by `pks` at once.
|
||||
|
||||
guardian's bulk assign_perm is avoided because it skips the direct row
|
||||
for anyone who already has access through a group, so revoking the group
|
||||
later would silently remove access that was granted explicitly.
|
||||
Takes a model + pks (rather than model instances) deliberately -- the
|
||||
permission rows built below only ever need `pk`, `content_type`, and
|
||||
identity ids, so callers shouldn't have to fetch full rows (with every
|
||||
other field) just to hand them to this function.
|
||||
|
||||
Deliberately does not use guardian's queryset/list-aware assign_perm:
|
||||
passing a list as the object routes to bulk_assign_perm, which skips
|
||||
creating a direct permission row for anyone who already has the
|
||||
permission via ANY group membership (it checks
|
||||
ObjectPermissionChecker.has_perm, which is group-inheritance-aware) --
|
||||
unlike the single-object assign_perm this replaces, which always
|
||||
ensures a direct row via get_or_create regardless of group-derived
|
||||
access. Losing that guarantee would mean a later revocation of the
|
||||
group's grant silently strips access an admin explicitly asked to be
|
||||
direct. Bulk-creating rows straight against the permission models
|
||||
instead (see _apply_bulk_permission_entry) preserves the original
|
||||
always-create-a-direct-row semantics while still batching every object
|
||||
and every identity into one query per action, rather than one query per
|
||||
(object, user) pair.
|
||||
"""
|
||||
object_pks = [str(pk) for pk in pks]
|
||||
if not object_pks: # pragma: no cover
|
||||
@@ -217,9 +297,12 @@ def set_permissions_for_objects(
|
||||
model_name = model.__name__.lower()
|
||||
ctype = ContentType.objects.get_for_model(model)
|
||||
|
||||
# Resolve every action before writing, so a bad one can't leave the
|
||||
# actions ahead of it applied.
|
||||
permissions_by_action: dict[PermissionAction, list[Permission]] = {}
|
||||
# Every action is resolved up front, before anything is written, so an
|
||||
# unrecognized action name (see _resolve_permissions) aborts the whole
|
||||
# call instead of leaving the actions ahead of it already applied.
|
||||
# SetPermissionsSerializer rejects unknown actions at the API, so this
|
||||
# guards any other caller.
|
||||
permissions_by_action: dict[str, list[Permission]] = {}
|
||||
for action, entry in permissions.items():
|
||||
if "users" not in entry and "groups" not in entry:
|
||||
continue
|
||||
|
||||
@@ -78,16 +78,12 @@ from documents.models import WorkflowActionEmail
|
||||
from documents.models import WorkflowActionWebhook
|
||||
from documents.models import WorkflowTrigger
|
||||
from documents.parsers import is_mime_type_supported
|
||||
from documents.permissions import PERMISSION_ACTIONS
|
||||
from documents.permissions import PermissionAction
|
||||
from documents.permissions import PermissionGrantees
|
||||
from documents.permissions import PermissionsSpec
|
||||
from documents.permissions import get_document_count_filter_for_user
|
||||
from documents.permissions import get_groups_with_only_permission
|
||||
from documents.permissions import has_perms_owner_aware
|
||||
from documents.permissions import permitted_document_ids
|
||||
from documents.permissions import restrict_queryset_to_visible
|
||||
from documents.permissions import set_permissions_for_objects
|
||||
from documents.permissions import set_permissions_for_object
|
||||
from documents.regex import validate_regex_pattern
|
||||
from documents.templating.filepath import validate_filepath_template_and_render
|
||||
from documents.templating.utils import convert_format_str_to_template_format
|
||||
@@ -215,6 +211,9 @@ class MatchingModelSerializer(serializers.ModelSerializer[Any]):
|
||||
return match
|
||||
|
||||
|
||||
PERMISSION_ACTIONS = ("view", "change")
|
||||
|
||||
|
||||
class SetPermissionsMixin:
|
||||
def _validate_user_ids(self, user_ids):
|
||||
users = User.objects.none()
|
||||
@@ -236,10 +235,8 @@ class SetPermissionsMixin:
|
||||
)
|
||||
return groups
|
||||
|
||||
def validate_set_permissions(self, set_permissions=None) -> PermissionsSpec:
|
||||
permissions_dict: dict[PermissionAction, PermissionGrantees] = {
|
||||
action: {} for action in PERMISSION_ACTIONS
|
||||
}
|
||||
def validate_set_permissions(self, set_permissions=None):
|
||||
permissions_dict = {action: {} for action in PERMISSION_ACTIONS}
|
||||
if set_permissions is not None:
|
||||
for action in PERMISSION_ACTIONS:
|
||||
if action in set_permissions:
|
||||
@@ -247,18 +244,18 @@ class SetPermissionsMixin:
|
||||
users = set_permissions[action]["users"]
|
||||
permissions_dict[action]["users"] = self._validate_user_ids(
|
||||
users,
|
||||
).values_list("id", flat=True)
|
||||
)
|
||||
if "groups" in set_permissions[action]:
|
||||
groups = set_permissions[action]["groups"]
|
||||
permissions_dict[action]["groups"] = self._validate_group_ids(
|
||||
groups,
|
||||
).values_list("id", flat=True)
|
||||
)
|
||||
else:
|
||||
del permissions_dict[action]
|
||||
return permissions_dict
|
||||
|
||||
def _set_permissions(self, permissions: PermissionsSpec, object) -> None:
|
||||
set_permissions_for_objects(permissions, type(object), [object.pk])
|
||||
def _set_permissions(self, permissions, object) -> None:
|
||||
set_permissions_for_object(permissions, object)
|
||||
|
||||
|
||||
class SerializerWithPerms(serializers.Serializer[dict[str, Any]]):
|
||||
@@ -3089,7 +3086,7 @@ class BulkEditObjectsSerializer(SerializerWithPerms, SetPermissionsMixin):
|
||||
)
|
||||
return objects
|
||||
|
||||
def _validate_permissions(self, permissions) -> PermissionsSpec:
|
||||
def _validate_permissions(self, permissions) -> dict:
|
||||
return self.validate_set_permissions(
|
||||
permissions,
|
||||
)
|
||||
|
||||
@@ -23,7 +23,6 @@ from documents.models import Document
|
||||
from documents.models import DocumentType
|
||||
from documents.models import StoragePath
|
||||
from documents.models import Tag
|
||||
from documents.permissions import PermissionsSpec
|
||||
from documents.permissions import set_permissions_for_objects
|
||||
from paperless_testing.dirs import DirectoriesMixin
|
||||
from paperless_testing.factories import DocumentFactory
|
||||
@@ -446,7 +445,7 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
|
||||
|
||||
grant_object(self.group1, self.doc1, "view_document")
|
||||
|
||||
permissions: PermissionsSpec = {
|
||||
permissions = {
|
||||
"view": {
|
||||
"users": [self.user1.id, self.user2.id],
|
||||
"groups": [self.group2.id],
|
||||
@@ -489,7 +488,7 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
|
||||
grant_object(self.user1, self.doc1, "view_document")
|
||||
grant_object(self.group1, self.doc1, "view_document")
|
||||
|
||||
permissions: PermissionsSpec = {
|
||||
permissions = {
|
||||
"view": {
|
||||
"users": [self.user2.id],
|
||||
"groups": [self.group2.id],
|
||||
@@ -540,7 +539,7 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
|
||||
batched call rather than one call per (document, identity)
|
||||
pair
|
||||
"""
|
||||
permissions: PermissionsSpec = {
|
||||
permissions = {
|
||||
"view": {
|
||||
"users": [self.user1.id, self.user2.id],
|
||||
"groups": [self.group2.id],
|
||||
@@ -647,7 +646,7 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
|
||||
"""
|
||||
with self.assertRaises(Permission.DoesNotExist):
|
||||
set_permissions_for_objects(
|
||||
{"not_a_real_action": {"users": [self.user1.id], "groups": []}}, # type: ignore[bad-assignment]
|
||||
{"not_a_real_action": {"users": [self.user1.id], "groups": []}},
|
||||
Document,
|
||||
[self.doc1.pk],
|
||||
)
|
||||
@@ -675,7 +674,7 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
|
||||
set_permissions_for_objects(
|
||||
{
|
||||
"view": {"users": [self.user1.id], "groups": []},
|
||||
"not_a_real_action": {"users": [self.user1.id], "groups": []}, # type: ignore[bad-assignment]
|
||||
"not_a_real_action": {"users": [self.user1.id], "groups": []},
|
||||
},
|
||||
Document,
|
||||
[self.doc1.pk],
|
||||
|
||||
@@ -34,8 +34,10 @@ 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:
|
||||
@@ -1683,6 +1685,32 @@ 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,6 +4,7 @@ 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
|
||||
@@ -257,3 +258,77 @@ 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,3 +1537,67 @@ 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
|
||||
@@ -7,8 +7,7 @@ from documents.data_models import DocumentMetadataOverrides
|
||||
from documents.models import CustomFieldInstance
|
||||
from documents.models import Document
|
||||
from documents.models import WorkflowAction
|
||||
from documents.permissions import PermissionsSpec
|
||||
from documents.permissions import set_permissions_for_objects
|
||||
from documents.permissions import set_permissions_for_object
|
||||
from documents.templating.workflows import parse_w_workflow_placeholders
|
||||
|
||||
logger = logging.getLogger("paperless.workflows.mutations")
|
||||
@@ -74,7 +73,7 @@ def apply_assignment_to_document(
|
||||
action.has_assign_change_groups,
|
||||
],
|
||||
):
|
||||
permissions: PermissionsSpec = {
|
||||
permissions = {
|
||||
"view": {
|
||||
"users": action.assign_view_users.values_list("id", flat=True),
|
||||
"groups": action.assign_view_groups.values_list("id", flat=True),
|
||||
@@ -84,10 +83,9 @@ def apply_assignment_to_document(
|
||||
"groups": action.assign_change_groups.values_list("id", flat=True),
|
||||
},
|
||||
}
|
||||
set_permissions_for_objects(
|
||||
permissions,
|
||||
Document,
|
||||
[document.pk],
|
||||
set_permissions_for_object(
|
||||
permissions=permissions,
|
||||
object=document,
|
||||
merge=True,
|
||||
)
|
||||
|
||||
@@ -245,14 +243,13 @@ def apply_removal_to_document(
|
||||
document.owner = None
|
||||
|
||||
if action.remove_all_permissions:
|
||||
permissions: PermissionsSpec = {
|
||||
permissions = {
|
||||
"view": {"users": [], "groups": []},
|
||||
"change": {"users": [], "groups": []},
|
||||
}
|
||||
set_permissions_for_objects(
|
||||
permissions,
|
||||
Document,
|
||||
[document.pk],
|
||||
set_permissions_for_object(
|
||||
permissions=permissions,
|
||||
object=document,
|
||||
merge=False,
|
||||
)
|
||||
|
||||
|
||||
@@ -1272,11 +1272,11 @@ wheels = [
|
||||
|
||||
[[package]]
|
||||
name = "filelock"
|
||||
version = "3.32.6"
|
||||
version = "4.0.12"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
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" }
|
||||
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" }
|
||||
wheels = [
|
||||
{ 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" },
|
||||
{ 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" },
|
||||
]
|
||||
|
||||
[[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.0" },
|
||||
{ name = "filelock", specifier = ">=3.32,<4.1" },
|
||||
{ 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" },
|
||||
|
||||
Reference in new issue
Block a user