fix: validate PDF output doc indexes in bulk edit

pyproject.toml

@@ -77,7 +77,6 @@ dependencies = [
   "torch~=2.13.0",
   "watchfiles>=1.2",
   "whitenoise~=6.11",
-  "whoosh-compat[tantivy]==0.1",
   "zxing-cpp~=3.1.0",
 ]
 [project.optional-dependencies]

src/documents/bulk_edit.py

@@ -899,17 +899,26 @@ def edit_pdf(
     pdf_docs: list[pikepdf.Pdf] = []
 
     try:
+        if not operations:
+            raise ValueError("Output document index is out of bounds")
+
+        max_idx = max(op.get("doc", 0) for op in operations)
+        if update_document and max_idx > 0:
+            logger.error(
+                "Update requested but multiple output documents specified",
+            )
+            raise ValueError("Multiple output documents specified")
+
+        if any(
+            op.get("doc", 0) < 0 or op.get("doc", 0) >= len(operations)
+            for op in operations
+        ):
+            raise ValueError("Output document index is out of bounds")
+
         with pikepdf.open(pair.source_doc.source_path) as src:
             # prepare output documents
-            max_idx = max(op.get("doc", 0) for op in operations)
             pdf_docs = [pikepdf.new() for _ in range(max_idx + 1)]
 
-            if update_document and len(pdf_docs) > 1:
-                logger.error(
-                    "Update requested but multiple output documents specified",
-                )
-                raise ValueError("Multiple output documents specified")
-
             for op in operations:
                 dst = pdf_docs[op.get("doc", 0)]
                 page = src.pages[op["page"] - 1]

src/documents/permissions.py

@@ -657,16 +657,41 @@ class ViewDocumentsPermissions(BasePermission):
         return request.user.has_perms(self.perms_map.get(request.method, []))
 
 
+class TrashPermissions(BasePermission):
+    """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):
     """
     Permissions class that checks for model permissions for Notes.
     """
 
     perms_map = {
-        "OPTIONS": ["documents.view_note"],
-        "GET": ["documents.view_note"],
-        "POST": ["documents.add_note"],
-        "DELETE": ["documents.delete_note"],
+        "OPTIONS": ["documents.view_note", "documents.view_document"],
+        "GET": ["documents.view_note", "documents.view_document"],
+        "POST": [
+            "documents.add_note",
+            "documents.view_document",
+            "documents.change_document",
+        ],
+        "DELETE": [
+            "documents.delete_note",
+            "documents.view_document",
+            "documents.change_document",
+        ],
     }
 
     def has_permission(self, request, view):

src/documents/search/_fields.py

@@ -1,42 +0,0 @@
-from __future__ import annotations
-
-from whoosh_compat import FieldKind
-from whoosh_compat import FieldSpec
-from whoosh_compat import SubpathSpec
-
-# Internal-only schema fields with no query-syntax meaning of their own
-# (sort shadow fields, bigram CJK fields, simple_title/simple_content,
-# autocomplete_word, notes_text) are NOT represented here, they are
-# declared in _schema.py's field_descriptors().
-#
-# analyzer/pattern_normalizer are deliberately left at FieldSpec's default
-# (None): they're language-specific and only meaningful to whoosh-compat's
-# parser, so _registry.py attaches them per-language via dataclasses.replace()
-# rather than PUBLIC_FIELDS declaring them itself. _schema.py only reads
-# name/kind/fast and never sees the analyzer at all.
-PUBLIC_FIELDS: tuple[FieldSpec, ...] = (
-    FieldSpec("title", FieldKind.TEXT),
-    FieldSpec("content", FieldKind.TEXT),
-    FieldSpec("correspondent", FieldKind.TEXT),
-    FieldSpec("document_type", FieldKind.TEXT, aliases=("type",)),
-    FieldSpec("storage_path", FieldKind.TEXT, aliases=("path",)),
-    FieldSpec("original_filename", FieldKind.TEXT),
-    FieldSpec("tag", FieldKind.TEXT, comma_values=True),
-    FieldSpec("checksum", FieldKind.KEYWORD),
-    FieldSpec("asn", FieldKind.U64, fast=True),
-    FieldSpec("page_count", FieldKind.U64, fast=True),
-    FieldSpec("num_notes", FieldKind.U64, fast=True),
-    FieldSpec("created", FieldKind.DATE, date_only=True, fast=True),
-    FieldSpec("modified", FieldKind.DATETIME, fast=True),
-    FieldSpec("added", FieldKind.DATETIME, fast=True),
-    FieldSpec(
-        "notes",
-        FieldKind.JSON,
-        subpaths={"user": SubpathSpec(), "note": SubpathSpec(default=True)},
-    ),
-    FieldSpec(
-        "custom_fields",
-        FieldKind.JSON,
-        subpaths={"name": SubpathSpec(), "value": SubpathSpec(default=True)},
-    ),
-)

src/documents/search/_registry.py

@@ -1,91 +0,0 @@
-from __future__ import annotations
-
-import dataclasses
-from typing import TYPE_CHECKING
-
-from whoosh_compat import FieldKind
-from whoosh_compat import FieldRegistry
-
-from documents.search._fields import PUBLIC_FIELDS
-from documents.search._tokenizer import ascii_fold
-from documents.search._tokenizer import paperless_text_analyzer
-from documents.search._tokenizer import stem_pattern_text
-
-if TYPE_CHECKING:
-    from whoosh_compat import PatternNormalizer
-
-_registry_cache: dict[str | None, FieldRegistry] = {}
-
-
-def _identity_analyzer(text: str) -> list[str]:
-    """Analyzer for KEYWORD fields indexed with the raw tokenizer (no splitting)."""
-    return [text]
-
-
-def _fold_normalizer(text: str) -> str:
-    """Wildcard/regex literal-run normalizer for fields indexed without stemming."""
-    return ascii_fold(text.lower())
-
-
-def _make_pattern_normalizer(language: str | None) -> PatternNormalizer:
-    """Build the wildcard/regex literal-run normalizer for a search language."""
-
-    def _pattern_normalizer(text: str) -> tuple[str, ...]:
-        """Normalize a literal run into the forms a term may match.
-
-        TEXT index terms go through lowercase -> ascii_fold -> stem, so a
-        pattern that skips stemming can never match one: "invoice*" would look
-        for a term starting with "invoice" while the index holds "invoic". The
-        run is therefore offered stemmed as well. KEYWORD fields are indexed
-        raw and get _fold_normalizer instead, so their patterns stay literal.
-
-        Both forms are returned, as alternatives, because neither is a prefix
-        of the other in general: English stemming substitutes as well as
-        truncates ("copy" -> "copi"), so the stem alone loses the compounds
-        the typed run reaches ("copyright") while the typed run alone loses
-        the inflections the stem reaches ("copies"). whoosh-compat ORs the
-        alternatives per literal run and deduplicates them, so a run the
-        stemmer leaves alone costs exactly the one branch it did before.
-
-        Inside a bracket class the emitter calls this once per character and
-        uses the answer only if it is a single one-character form; two forms
-        there leave the character as typed. A stemmer does not change a lone
-        character, so the two forms deduplicate to one and the class body is
-        folded as before.
-        """
-        folded = ascii_fold(text.lower())
-        stemmed = stem_pattern_text(folded, language)
-        return (folded, stemmed)
-
-    return _pattern_normalizer
-
-
-def get_field_registry(language: str | None) -> FieldRegistry:
-    """Build (or return the cached) FieldRegistry for the given search language.
-
-    Cached keyed by language, rebuilt on the same trigger register_tokenizers()
-    uses (settings.SEARCH_LANGUAGE change). A fresh call with a new language
-    builds and caches a new registry rather than mutating the old one.
-    """
-    if language in _registry_cache:
-        return _registry_cache[language]
-
-    text_analyzer = paperless_text_analyzer(language).analyze
-    pattern_normalizer = _make_pattern_normalizer(language)
-
-    specs = [
-        dataclasses.replace(
-            field,
-            analyzer=_identity_analyzer
-            if field.kind is FieldKind.KEYWORD
-            else text_analyzer,
-            pattern_normalizer=_fold_normalizer
-            if field.kind is FieldKind.KEYWORD
-            else pattern_normalizer,
-        )
-        for field in PUBLIC_FIELDS
-    ]
-
-    registry = FieldRegistry(specs)
-    _registry_cache[language] = registry
-    return registry

src/documents/search/_schema.py

@@ -1,19 +1,14 @@
 from __future__ import annotations
 
-import hashlib
 import json
 import logging
 import shutil
 from typing import TYPE_CHECKING
 from typing import Final
-from typing import NamedTuple
 from typing import cast
 
 import tantivy
 from django.conf import settings
-from whoosh_compat import FieldKind
-
-from documents.search._fields import PUBLIC_FIELDS
 
 if TYPE_CHECKING:
     from pathlib import Path
@@ -21,185 +16,7 @@ if TYPE_CHECKING:
 logger = logging.getLogger("paperless.search")
 
 # v1 - Initial tantivy schema format
-# v2 - build_schema() derived from PUBLIC_FIELDS, changing the field declaration
-#      order, and the write-only correspondent/document_type/storage_path/tag id
-#      columns dropped. tantivy compares schemas by ordered field list, so an
-#      index built by v1 rejects every write against the v2 schema.
-SCHEMA_VERSION: Final[int] = 2
-
-
-class FieldDescriptor(NamedTuple):
-    """One tantivy field, in declaration order.
-
-    The descriptor vocabulary is paperless', not tantivy-py's: it is both the
-    input to the SchemaBuilder and the input to schema_fingerprint(), so the
-    persisted fingerprint cannot move under a tantivy-py upgrade.
-    """
-
-    name: str
-    kind: str
-    stored: bool
-    indexed: bool
-    fast: bool
-    tokenizer: str | None
-
-
-# (schema kind, tokenizer) for the FieldKind -> FieldDescriptor mapping that
-# doesn't need special-casing. JSON is handled separately below since it can
-# emit a second, synthetic descriptor.
-_KIND_TABLE: Final[dict[FieldKind, tuple[str, str | None]]] = {
-    FieldKind.TEXT: ("text", "paperless_text"),
-    FieldKind.KEYWORD: ("text", "raw"),
-    FieldKind.U64: ("u64", None),
-    FieldKind.DATE: ("date", None),
-    FieldKind.DATETIME: ("date", None),
-}
-# Kinds whose fast-field flag follows FieldSpec.fast rather than always False.
-_FAST_FROM_FIELD: Final[frozenset[FieldKind]] = frozenset(
-    {FieldKind.U64, FieldKind.DATE, FieldKind.DATETIME},
-)
-
-
-def _public_field_descriptors() -> list[FieldDescriptor]:
-    """Descriptors for the query-visible fields declared in PUBLIC_FIELDS."""
-    descriptors: list[FieldDescriptor] = []
-    for field in PUBLIC_FIELDS:
-        if field.kind is FieldKind.JSON:
-            descriptors.append(
-                FieldDescriptor(
-                    field.name,
-                    "json",
-                    stored=True,
-                    indexed=True,
-                    fast=False,
-                    tokenizer="paperless_text",
-                ),
-            )
-            if field.name == "notes":
-                # Plain-text companion for snippet generation: tantivy's
-                # SnippetGenerator does not support JSON fields. Schema-only,
-                # no query-syntax meaning, not in PUBLIC_FIELDS.
-                descriptors.append(
-                    FieldDescriptor(
-                        "notes_text",
-                        "text",
-                        stored=True,
-                        indexed=True,
-                        fast=False,
-                        tokenizer="paperless_text",
-                    ),
-                )
-            continue
-        schema_kind, tokenizer = _KIND_TABLE[field.kind]
-        descriptors.append(
-            FieldDescriptor(
-                field.name,
-                schema_kind,
-                stored=True,
-                indexed=True,
-                fast=field.fast if field.kind in _FAST_FROM_FIELD else False,
-                tokenizer=tokenizer,
-            ),
-        )
-    return descriptors
-
-
-def field_descriptors() -> list[FieldDescriptor]:
-    """Every field of the document index, in the order tantivy declares them.
-
-    tantivy compares schemas by *ordered* field list, so the order here is
-    part of the on-disk contract: schema_fingerprint() hashes it and
-    needs_rebuild() acts on the result.
-    """
-    return [
-        FieldDescriptor(
-            "id",
-            "u64",
-            stored=True,
-            indexed=True,
-            fast=True,
-            tokenizer=None,
-        ),
-        *_public_field_descriptors(),
-        # Shadow sort fields - fast, not stored
-        *(
-            FieldDescriptor(
-                name,
-                "text",
-                stored=False,
-                indexed=True,
-                fast=True,
-                tokenizer="simple_analyzer",
-            )
-            for name in ("title_sort", "correspondent_sort", "type_sort")
-        ),
-        # CJK support - not stored, indexed only
-        *(
-            FieldDescriptor(
-                name,
-                "text",
-                stored=False,
-                indexed=True,
-                fast=False,
-                tokenizer="bigram_analyzer",
-            )
-            for name in (
-                "bigram_content",
-                "bigram_title",
-                "bigram_correspondent",
-                "bigram_document_type",
-                "bigram_tag",
-            )
-        ),
-        # Simple substring search support for title/content - not stored,
-        # indexed only
-        *(
-            FieldDescriptor(
-                name,
-                "text",
-                stored=False,
-                indexed=True,
-                fast=False,
-                tokenizer="simple_search_analyzer",
-            )
-            for name in ("simple_title", "simple_content")
-        ),
-        # Autocomplete prefix scan via terms_with_prefix, which walks the
-        # field's term dictionary - so the field must be indexed (term dict),
-        # not stored. The stored value is never read back, so storing it only
-        # wastes space.
-        FieldDescriptor(
-            "autocomplete_word",
-            "text",
-            stored=False,
-            indexed=True,
-            fast=False,
-            tokenizer="raw",
-        ),
-        # Permission filter columns, read by build_permission_filter.
-        *(
-            FieldDescriptor(
-                name,
-                "u64",
-                stored=False,
-                indexed=True,
-                fast=True,
-                tokenizer=None,
-            )
-            for name in ("owner_id", "viewer_id", "viewer_group_id")
-        ),
-    ]
-
-
-def schema_fingerprint() -> str:
-    """Hash of the field descriptors, stamped into .index_settings.json.
-
-    Changes whenever a field is added, removed, retyped, re-optioned or
-    reordered, so an index built from a different schema shape is detected
-    even when SCHEMA_VERSION was not bumped.
-    """
-    payload = json.dumps([list(descriptor) for descriptor in field_descriptors()])
-    return hashlib.blake2b(payload.encode()).hexdigest()
+SCHEMA_VERSION: Final[int] = 1
 
 
 def build_schema() -> tantivy.Schema:
@@ -215,37 +32,85 @@ def build_schema() -> tantivy.Schema:
     """
     sb = tantivy.SchemaBuilder()
 
-    for descriptor in field_descriptors():
-        if descriptor.kind == "text":
-            sb.add_text_field(
-                descriptor.name,
-                stored=descriptor.stored,
-                fast=descriptor.fast,
-                tokenizer_name=cast("str", descriptor.tokenizer),
-            )
-        elif descriptor.kind == "json":
-            sb.add_json_field(
-                descriptor.name,
-                stored=descriptor.stored,
-                fast=descriptor.fast,
-                tokenizer_name=cast("str", descriptor.tokenizer),
-            )
-        elif descriptor.kind == "u64":
-            sb.add_unsigned_field(
-                descriptor.name,
-                stored=descriptor.stored,
-                indexed=descriptor.indexed,
-                fast=descriptor.fast,
-            )
-        elif descriptor.kind == "date":
-            sb.add_date_field(
-                descriptor.name,
-                stored=descriptor.stored,
-                indexed=descriptor.indexed,
-                fast=descriptor.fast,
-            )
-        else:
-            raise ValueError(f"Unknown schema field kind: {descriptor.kind}")
+    sb.add_unsigned_field("id", stored=True, indexed=True, fast=True)
+    sb.add_text_field("checksum", stored=True, tokenizer_name="raw")
+
+    for field in (
+        "title",
+        "correspondent",
+        "document_type",
+        "storage_path",
+        "original_filename",
+        "content",
+    ):
+        sb.add_text_field(field, stored=True, tokenizer_name="paperless_text")
+
+    # Shadow sort fields - fast, not stored/indexed
+    for field in ("title_sort", "correspondent_sort", "type_sort"):
+        sb.add_text_field(
+            field,
+            stored=False,
+            tokenizer_name="simple_analyzer",
+            fast=True,
+        )
+
+    # CJK support - not stored, indexed only
+    sb.add_text_field("bigram_content", stored=False, tokenizer_name="bigram_analyzer")
+    sb.add_text_field("bigram_title", stored=False, tokenizer_name="bigram_analyzer")
+    sb.add_text_field(
+        "bigram_correspondent",
+        stored=False,
+        tokenizer_name="bigram_analyzer",
+    )
+    sb.add_text_field(
+        "bigram_document_type",
+        stored=False,
+        tokenizer_name="bigram_analyzer",
+    )
+    sb.add_text_field("bigram_tag", stored=False, tokenizer_name="bigram_analyzer")
+
+    # Simple substring search support for title/content - not stored, indexed only
+    sb.add_text_field(
+        "simple_title",
+        stored=False,
+        tokenizer_name="simple_search_analyzer",
+    )
+    sb.add_text_field(
+        "simple_content",
+        stored=False,
+        tokenizer_name="simple_search_analyzer",
+    )
+
+    # Autocomplete prefix scan via terms_with_prefix, which walks the field's
+    # term dictionary - so the field must be indexed (term dict), not stored.
+    # The stored value is never read back, so storing it only wastes space.
+    sb.add_text_field("autocomplete_word", stored=False, tokenizer_name="raw")
+
+    sb.add_text_field("tag", stored=True, tokenizer_name="paperless_text")
+
+    # JSON fields — structured queries: notes.user:alice, custom_fields.name:invoice
+    sb.add_json_field("notes", stored=True, tokenizer_name="paperless_text")
+    # Plain-text companion for notes — tantivy's SnippetGenerator does not support
+    # JSON fields, so highlights require a text field with the same content.
+    sb.add_text_field("notes_text", stored=True, tokenizer_name="paperless_text")
+    sb.add_json_field("custom_fields", stored=True, tokenizer_name="paperless_text")
+
+    for field in (
+        "correspondent_id",
+        "document_type_id",
+        "storage_path_id",
+        "tag_id",
+        "owner_id",
+        "viewer_id",
+        "viewer_group_id",
+    ):
+        sb.add_unsigned_field(field, stored=False, indexed=True, fast=True)
+
+    for field in ("created", "modified", "added"):
+        sb.add_date_field(field, stored=True, indexed=True, fast=True)
+
+    for field in ("asn", "page_count", "num_notes"):
+        sb.add_unsigned_field(field, stored=True, indexed=True, fast=True)
 
     return sb.build()
 
@@ -254,9 +119,9 @@ def needs_rebuild(index_dir: Path) -> bool:
     """
     Check if the search index needs rebuilding.
 
-    Reads .index_settings.json to compare the stored schema version, search
-    language and schema fingerprint against the current configuration. Returns
-    True if the file is missing, unparsable, or any value mismatches.
+    Reads .index_settings.json to compare the stored schema version and
+    search language against the current configuration. Returns True if the
+    file is missing, unparsable, or either value mismatches.
 
     Args:
         index_dir: Path to the search index directory
@@ -275,9 +140,6 @@ def needs_rebuild(index_dir: Path) -> bool:
         if "language" not in data or data["language"] != settings.SEARCH_LANGUAGE:
             logger.info("Search index language changed - rebuilding.")
             return True
-        if data.get("schema_fingerprint") != schema_fingerprint():
-            logger.info("Search index schema fingerprint mismatch - rebuilding.")
-            return True
     except ValueError:
         return True
     return False
@@ -308,7 +170,6 @@ def _write_sentinels(index_dir: Path) -> None:
             {
                 "schema_version": SCHEMA_VERSION,
                 "language": settings.SEARCH_LANGUAGE,
-                "schema_fingerprint": schema_fingerprint(),
             },
         ),
     )

src/documents/search/_tokenizer.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import logging
-from functools import cache
 from typing import Final
 
 import tantivy
@@ -72,7 +71,7 @@ def register_tokenizers(index: tantivy.Index, language: str | None) -> None:
         use fast=True and Tantivy requires fast-field tokenizers to exist
         even for documents that omit those fields.
     """
-    index.register_tokenizer("paperless_text", paperless_text_analyzer(language))
+    index.register_tokenizer("paperless_text", _paperless_text(language))
     index.register_tokenizer("simple_analyzer", _simple_analyzer())
     index.register_tokenizer("bigram_analyzer", _bigram_analyzer())
     index.register_tokenizer("simple_search_analyzer", _simple_search_analyzer())
@@ -80,7 +79,7 @@ def register_tokenizers(index: tantivy.Index, language: str | None) -> None:
     index.register_fast_field_tokenizer("simple_analyzer", _simple_analyzer())
 
 
-def paperless_text_analyzer(language: str | None) -> tantivy.TextAnalyzer:
+def _paperless_text(language: str | None) -> tantivy.TextAnalyzer:
     """Main full-text tokenizer for content, title, etc: simple -> remove_long(129) -> lowercase -> ascii_fold [-> stemmer]"""
     builder = (
         tantivy.TextAnalyzerBuilder(tantivy.Tokenizer.simple())
@@ -101,54 +100,6 @@ def paperless_text_analyzer(language: str | None) -> tantivy.TextAnalyzer:
     return builder.build()
 
 
-@cache
-def _pattern_stemmer(language: str | None) -> tantivy.TextAnalyzer | None:
-    """The stemming tail of paperless_text_analyzer, over a whole literal run.
-
-    Same language gate and same Snowball stemmer paperless_text_analyzer
-    applies at index time, so query patterns follow SEARCH_LANGUAGE. Returns
-    None when that gate disables stemming; paperless_text_analyzer already
-    warns about an unsupported language, so this stays quiet.
-
-    The raw tokenizer keeps the run whole (a wildcard literal is a fragment,
-    not necessarily a word), and remove_long is kept so an over-long run is
-    treated the same way the index treats it.
-    """
-    if not language:
-        return None
-    tantivy_lang = _LANGUAGE_MAP.get(language.lower())
-    if tantivy_lang is None:
-        return None
-    return (
-        tantivy.TextAnalyzerBuilder(tantivy.Tokenizer.raw())
-        .filter(tantivy.Filter.remove_long(_TOKEN_REMOVE_LONG_LIMIT))
-        .filter(tantivy.Filter.stemmer(tantivy_lang))
-        .build()
-    )
-
-
-def stem_pattern_text(text: str, language: str | None) -> str:
-    """Stem an already lowercased/ascii-folded run the way index terms are.
-
-    Returns text unchanged when stemming is disabled for language, and also
-    when the stem step does not yield exactly one token: remove_long drops a run
-    past the length limit, leaving no stem to substitute. Falling back to the
-    text as typed is the safe direction for a pattern prefix, since it can only
-    be as narrow as it was before stemming was considered.
-
-    The raw tokenizer emits one token whatever the input and the stemmer is
-    1-to-1, so only the zero-token case can fire today; the guard covers both
-    counts so a tokenizer change cannot turn this into an IndexError.
-    """
-    analyzer = _pattern_stemmer(language)
-    if analyzer is None:
-        return text
-    tokens = analyzer.analyze(text)
-    if len(tokens) != 1:
-        return text
-    return tokens[0]
-
-
 def _simple_analyzer() -> tantivy.TextAnalyzer:
     """Tokenizer for shadow sort fields (title_sort, correspondent_sort, type_sort): simple -> lowercase -> ascii_fold."""
     return (

src/documents/serialisers.py

@@ -1335,6 +1335,7 @@ class DocumentSerializer(
             "root_document",
             "versions",
         )
+        read_only_fields = ("deleted_at",)
         list_serializer_class = OwnedObjectListSerializer
 
 
@@ -1787,6 +1788,12 @@ class EditPdfDocumentsSerializer(DocumentListSerializer, SourceModeValidationMix
                     "update_document only allowed with a single output document",
                 )
 
+        if any(
+            op.get("doc", 0) < 0 or op.get("doc", 0) >= len(operations)
+            for op in operations
+        ):
+            raise serializers.ValidationError("doc index is out of bounds")
+
         doc = Document.objects.get(id=documents[0])
         if doc.page_count:
             for op in operations:
@@ -2150,6 +2157,12 @@ class BulkEditSerializer(
                     "update_document only allowed with a single output document",
                 )
 
+        if any(
+            op.get("doc", 0) < 0 or op.get("doc", 0) >= len(parameters["operations"])
+            for op in parameters["operations"]
+        ):
+            raise serializers.ValidationError("doc index is out of bounds")
+
         doc = Document.objects.get(id=document_id)
         # doc existence is already validated
         if doc.page_count:
@@ -2839,10 +2852,14 @@ class ShareLinkSerializer(OwnedObjectSerializer):
         return super().create(validated_data)
 
     def validate_document(self, document):
-        if self.user is not None and has_perms_owner_aware(
-            self.user,
-            "view_document",
-            document,
+        if (
+            self.user is not None
+            and self.user.has_perm("documents.view_document")
+            and has_perms_owner_aware(
+                self.user,
+                "view_document",
+                document,
+            )
         ):
             return document
         raise PermissionDenied(
@@ -3603,6 +3620,8 @@ class WorkflowSerializer(serializers.ModelSerializer[Workflow]):
 
         if "actions" in validated_data:
             actions = validated_data.pop("actions")
+            for action in actions:
+                action.pop("id", None)
 
         instance = super().create(validated_data)
 

src/documents/tests/search/test_json_subpath_completeness.py

@@ -1,92 +0,0 @@
-"""Every declared JSON subpath must actually be written to the index.
-
-PUBLIC_FIELDS declares each JSON field's subpaths (e.g. ``notes`` ->
-{"user", "note"}), but nothing coupled that declaration to what
-``_backend.py``'s document builder actually writes into the JSON blob at
-index time. A subpath declared but never written would be
-queryable-but-always-empty -- syntactically valid, silently matching
-nothing -- with no test failure anywhere.
-
-This indexes one real document carrying values for every JSON field
-(a Note, a CustomFieldInstance) and inspects the document's own stored
-JSON payload, rather than running field-specific queries: that way a
-future JSON field's subpaths are covered automatically, without a new
-per-subpath query having to be added by hand each time.
-"""
-
-from __future__ import annotations
-
-from typing import TYPE_CHECKING
-
-import pytest
-import tantivy
-from django.contrib.auth.models import User
-from whoosh_compat import FieldKind
-
-from documents.models import CustomField
-from documents.models import CustomFieldInstance
-from documents.models import Document
-from documents.models import Note
-from documents.search._fields import PUBLIC_FIELDS
-
-if TYPE_CHECKING:
-    from documents.search._backend import TantivyBackend
-
-pytestmark = [pytest.mark.search, pytest.mark.django_db]
-
-
-class TestJsonSubpathsAreWrittenAtIndexTime:
-    def test_every_declared_json_subpath_appears_in_the_stored_document(
-        self,
-        backend: TantivyBackend,
-    ) -> None:
-        """
-        GIVEN:
-            - A document with a Note and a CustomFieldInstance attached
-        WHEN:
-            - The document is indexed via TantivyBackend.add_or_update
-        THEN:
-            - Every subpath PUBLIC_FIELDS declares for notes/custom_fields
-              is present as a key in the document's stored JSON payload
-        """
-        user = User.objects.create_user(username="completeness-user")
-        field = CustomField.objects.create(
-            name="Completeness Field",
-            data_type=CustomField.FieldDataType.STRING,
-        )
-        doc = Document.objects.create(
-            title="Completeness doc",
-            content="x",
-            checksum="json-subpath-completeness",
-        )
-        Note.objects.create(document=doc, user=user, note="a note")
-        CustomFieldInstance.objects.create(
-            document=doc,
-            field=field,
-            value_text="a value",
-        )
-        backend.add_or_update(doc)
-
-        index = backend._index
-        searcher = index.searcher()
-        hits = searcher.search(
-            tantivy.Query.term_query(index.schema, "id", doc.pk),
-            limit=1,
-        ).hits
-        assert hits, "the document was not indexed"
-        stored = searcher.doc(hits[0][1]).to_dict()
-
-        json_fields = [f for f in PUBLIC_FIELDS if f.kind is FieldKind.JSON]
-        assert json_fields, "no JSON fields declared - fixture is stale"
-        for field_spec in json_fields:
-            stored_values = stored.get(field_spec.name)
-            assert stored_values, (
-                f"{field_spec.name} was not written to the index at all"
-            )
-            written_keys = stored_values[0].keys()
-            for subpath in field_spec.subpaths:
-                assert subpath in written_keys, (
-                    f"{field_spec.name}.{subpath} is declared in PUBLIC_FIELDS "
-                    "but _backend.py's document builder never writes it - it "
-                    "would be queryable but always empty"
-                )

src/documents/tests/search/test_keyword_pattern_literal.py

@@ -1,62 +0,0 @@
-"""Wildcard patterns on KEYWORD fields must stay literal.
-
-``checksum`` is the only KEYWORD field: it is indexed with the raw tokenizer,
-so its terms are never lowercased, folded or stemmed. Running its wildcard
-patterns through the stemming normalizer rewrote hex prefixes ("ceded" ->
-"cede") and returned documents whose checksum did not start with what the user
-typed, which for an identity field is a wrong answer.
-
-This covers only the registry-level normalizer, which is all that exists to
-prove at this point in the stack: user queries are not yet routed through
-whoosh-compat (that lands with the query-layer PR), so the same fact proven
-end to end against real indexed documents lives in
-``test_checksum_prefix_queries.py``.
-"""
-
-from __future__ import annotations
-
-from typing import TYPE_CHECKING
-
-import pytest
-
-from documents.search._registry import get_field_registry
-
-if TYPE_CHECKING:
-    from whoosh_compat import FieldRegistry
-    from whoosh_compat import PatternNormalizer
-
-pytestmark = [pytest.mark.search, pytest.mark.django_db]
-
-
-def _normalizer(registry: FieldRegistry, name: str) -> PatternNormalizer:
-    ref = registry.make_ref(name)
-    assert ref is not None
-    resolved = registry.resolve(ref)
-    assert resolved is not None
-    assert resolved.spec.pattern_normalizer is not None
-    return resolved.spec.pattern_normalizer
-
-
-class TestKeywordPatternNormalizer:
-    @pytest.mark.parametrize(
-        "run",
-        [
-            pytest.param("ceded", id="stems_to_cede"),
-            pytest.param("added", id="stems_to_ad"),
-            pytest.param("cafed", id="stems_to_cafe"),
-        ],
-    )
-    def test_keyword_runs_are_folded_not_stemmed(self, run: str) -> None:
-        """
-        GIVEN:
-            - The "checksum" field's registered pattern normalizer
-              (KEYWORD kind, "en" registry)
-        WHEN:
-            - A wildcard pattern run is normalized
-        THEN:
-            - The run is returned unchanged, never widened to a stem (which
-              would return checksums that do not start with what the user
-              typed)
-        """
-        normalize = _normalizer(get_field_registry("en"), "checksum")
-        assert normalize(run) == run

src/documents/tests/search/test_pattern_normalizer.py

@@ -1,156 +0,0 @@
-"""The pattern normalizer's stem-alternates contract, and its consistency
-with the index-side analyzer.
-
-Query patterns are normalized but were not stemmed, while index terms are
-stemmed, so the natural spelling of a prefix search matched nothing:
-``invoice*`` found no document although ``invoic*`` did. v2's index was
-UNSTEMMED (whoosh ``TEXT()`` defaults to ``StandardAnalyzer``), so this
-regressed against both baselines.
-
-These are pure unit tests against ``_make_pattern_normalizer`` and
-``stem_pattern_text`` directly, no query routing involved. The end-to-end
-proof that a real wildcard query actually reaches a stemmed index term
-lives in ``test_pattern_stemming.py``.
-"""
-
-from __future__ import annotations
-
-from typing import TYPE_CHECKING
-
-import pytest
-
-from documents.search._registry import _make_pattern_normalizer
-from documents.search._tokenizer import ascii_fold
-from documents.search._tokenizer import paperless_text_analyzer
-from documents.search._tokenizer import stem_pattern_text
-
-if TYPE_CHECKING:
-    from whoosh_compat import PatternNormalizer
-
-
-class TestStemsMatchTheIndexAnalyzer:
-    """stem_pattern_text rebuilds paperless_text_analyzer's stemming tail rather
-    than sharing it, so a filter added to the index analyzer alone would silently
-    stop patterns from reaching the terms it produces.
-    """
-
-    @pytest.mark.parametrize(
-        "language",
-        ["en", "de", "fr", "es", "sv", None, "klingon"],
-    )
-    @pytest.mark.parametrize(
-        "word",
-        ["Copies", "copyright", "Companies", "Invoices", "laufen", "casas", "Straße"],
-    )
-    def test_stem_equals_the_index_term(self, word: str, language: str | None) -> None:
-        """
-        GIVEN:
-            - A word, across several representative index languages
-              ("en", "de", "fr", "es", "sv"), no language, and an
-              unsupported language ("klingon")
-        WHEN:
-            - `stem_pattern_text` (the pattern-side stemmer) processes the
-              folded word, and `paperless_text_analyzer` (the index-side
-              analyzer) independently processes the same word
-        THEN:
-            - The two produce the identical term. `stem_pattern_text`
-              rebuilds `paperless_text_analyzer`'s stemming tail rather
-              than sharing it, so a filter added to the index analyzer
-              alone would silently stop patterns from reaching the terms
-              it produces; this pins the two staying in sync
-        """
-        indexed = paperless_text_analyzer(language).analyze(word)[0]
-        assert stem_pattern_text(ascii_fold(word.lower()), language) == indexed
-
-
-def _forms(normalize: PatternNormalizer, text: str) -> tuple[str, ...]:
-    """The distinct forms a term may match, in order, the way the emitter reads
-    the normalizer's answer (see whoosh_compat.PatternNormalizer)."""
-    result = normalize(text)
-    if isinstance(result, str):
-        return (result,)
-    return tuple(dict.fromkeys(result))
-
-
-class TestPatternNormalizer:
-    @pytest.mark.parametrize(
-        ("text", "expected"),
-        [
-            ("Invoice", ("invoice", "invoic")),
-            ("companies", ("companies", "compani")),
-            # y -> i is a substitution, so both forms are needed: the index
-            # holds "librari" for "library" and "library" for "librarian".
-            ("library", ("library", "librari")),
-            # A run the stemmer leaves alone collapses back to one form, so it
-            # costs exactly the one regex branch it did before.
-            ("invoic", ("invoic",)),
-            ("Universit", ("universit",)),
-            ("Café", ("cafe",)),
-        ],
-    )
-    def test_offers_the_typed_run_and_its_stem(
-        self,
-        text: str,
-        expected: tuple[str, ...],
-    ) -> None:
-        """
-        GIVEN:
-            - The "en" pattern normalizer
-        WHEN:
-            - It processes a literal run (e.g. "Invoice", "library",
-              "Café")
-        THEN:
-            - It returns the folded run and, where it differs, the
-              stemmed form, as distinct alternatives; a run the stemmer
-              leaves alone (e.g. "invoic") collapses back to the single
-              folded form. "library" needs both forms since y -> i is a
-              substitution: the index holds "librari" for "library" and
-              "library" for "librarian"
-        """
-        assert _forms(_make_pattern_normalizer("en"), text) == expected
-
-    def test_run_that_yields_no_token_falls_back_to_the_typed_run(self) -> None:
-        """
-        GIVEN:
-            - The "en" pattern normalizer
-        WHEN:
-            - It processes a run past the analyzer's remove_long limit
-        THEN:
-            - The run analyzes to zero tokens, so there is no stem to
-              offer, and only the folded run remains
-        """
-        over_long = "invoices" * 20
-        assert _forms(_make_pattern_normalizer("en"), over_long) == (over_long,)
-
-    @pytest.mark.parametrize("language", [None, "klingon"])
-    def test_unstemmed_language_folds_only(self, language: str | None) -> None:
-        """
-        GIVEN:
-            - A pattern normalizer with no language configured, or one
-              this build has no stemmer for ("klingon")
-        WHEN:
-            - It processes "Invoices"
-        THEN:
-            - Only the folded form ("invoices") is offered, since with no
-              stemmer configured the index holds surface forms and the
-              pattern must keep them too
-        """
-        assert _forms(_make_pattern_normalizer(language), "Invoices") == ("invoices",)
-
-    @pytest.mark.parametrize("char", ["a", "Z", "é"])
-    def test_a_single_character_collapses_to_one_folded_form(self, char: str) -> None:
-        """
-        GIVEN:
-            - The "en" pattern normalizer
-        WHEN:
-            - It processes a single character
-        THEN:
-            - Exactly one, one-character form is returned. A bracket
-              class body is normalized one character at a time and the
-              answer is used only when it is a single one-character
-              form, so a stemmer that changed a lone character would
-              silently disable folding inside classes
-        """
-        forms = _forms(_make_pattern_normalizer("en"), char)
-        assert len(forms) == 1
-        assert len(forms[0]) == 1

src/documents/tests/search/test_registry.py

@@ -1,224 +0,0 @@
-from collections.abc import Sequence
-
-import pytest
-from whoosh_compat import FieldKind
-from whoosh_compat import FieldRegistry
-from whoosh_compat.fields import ResolvedField
-
-from documents.search._fields import PUBLIC_FIELDS
-from documents.search._registry import get_field_registry
-
-
-@pytest.fixture
-def registry() -> FieldRegistry:
-    return get_field_registry(None)
-
-
-def _resolve(registry: FieldRegistry, name: str) -> ResolvedField:
-    ref = registry.make_ref(name)
-    assert ref is not None, f"{name} is not a valid field ref"
-    resolved = registry.resolve(ref)
-    assert resolved is not None, f"{name} did not resolve"
-    return resolved
-
-
-def _distinct_forms(result: str | Sequence[str]) -> tuple[str, ...]:
-    """The forms a term may match, in order, the way whoosh-compat's emitter
-    reads a pattern_normalizer's answer: a bare str is one form, a sequence is
-    several, deduplicated."""
-    if isinstance(result, str):
-        return (result,)
-    return tuple(dict.fromkeys(result))
-
-
-class TestFieldRegistry:
-    def test_no_queryable_field_name_ends_in_id(self) -> None:
-        """
-        GIVEN:
-            - PUBLIC_FIELDS, the canonical query-syntax field table
-        WHEN:
-            - Every declared field name is inspected
-        THEN:
-            - None of them end in "_id" (internal id columns, written for
-              permission filtering and joins, must never reach the query
-              surface; checked against PUBLIC_FIELDS rather than the
-              registry so a leak is caught where it is declared)
-        """
-        leaked = [f.name for f in PUBLIC_FIELDS if f.name.endswith("_id")]
-        assert not leaked, f"internal id fields reached the query surface: {leaked}"
-
-    def test_type_alias_resolves_to_document_type(
-        self,
-        registry: FieldRegistry,
-    ) -> None:
-        """
-        GIVEN:
-            - The field registry
-        WHEN:
-            - The alias "type" is resolved
-        THEN:
-            - It resolves to the canonical "document_type" field
-        """
-        assert _resolve(registry, "type").spec.name == "document_type"
-
-    def test_path_alias_resolves_to_storage_path(self, registry: FieldRegistry) -> None:
-        """
-        GIVEN:
-            - The field registry
-        WHEN:
-            - The alias "path" is resolved
-        THEN:
-            - It resolves to the canonical "storage_path" field
-        """
-        assert _resolve(registry, "path").spec.name == "storage_path"
-
-    def test_notes_json_subpaths_resolve(self, registry: FieldRegistry) -> None:
-        """
-        GIVEN:
-            - The field registry
-        WHEN:
-            - "notes.user" is resolved
-        THEN:
-            - It resolves to the "notes" field with json_path "user"
-        """
-        resolved = _resolve(registry, "notes.user")
-        assert resolved.spec.name == "notes"
-        assert resolved.json_path == "user"
-        assert resolved.is_subpath is True
-
-    def test_custom_fields_json_subpaths_resolve(self, registry: FieldRegistry) -> None:
-        """
-        GIVEN:
-            - The field registry
-        WHEN:
-            - "custom_fields.name" and "custom_fields.value" are resolved
-        THEN:
-            - Both resolve without error
-        """
-        for raw in ("custom_fields.name", "custom_fields.value"):
-            _resolve(registry, raw)
-
-    def test_tag_is_comma_values(self, registry: FieldRegistry) -> None:
-        """
-        GIVEN:
-            - The field registry
-        WHEN:
-            - The "tag" field is resolved
-        THEN:
-            - It is marked comma_values=True
-        """
-        assert _resolve(registry, "tag").spec.comma_values is True
-
-    def test_correspondent_is_not_comma_values(self, registry: FieldRegistry) -> None:
-        """
-        GIVEN:
-            - The field registry
-        WHEN:
-            - The "correspondent" field is resolved
-        THEN:
-            - It is not marked comma_values ("tag" is the only field that
-              opts in; end to end the two readings of
-              "correspondent:foo,bar" agree anyway, since the analyzer
-              splits the literal value on the comma regardless, so this is
-              only observable at the registry level)
-        """
-        assert _resolve(registry, "correspondent").spec.comma_values is False
-
-    def test_created_is_date_kind(self, registry: FieldRegistry) -> None:
-        """
-        GIVEN:
-            - The field registry
-        WHEN:
-            - The "created" field is resolved
-        THEN:
-            - Its kind is DATE and date_only is True
-        """
-        resolved = _resolve(registry, "created")
-        assert resolved.spec.kind is FieldKind.DATE
-        assert resolved.spec.date_only is True
-
-    def test_analyzer_lowercases_and_ascii_folds(self, registry: FieldRegistry) -> None:
-        """
-        GIVEN:
-            - The field registry with no language configured (no stemmer
-              in the analyzer chain)
-        WHEN:
-            - The "title" field's analyzer processes "Café"
-        THEN:
-            - It is lowercased and ASCII-folded to the single token "cafe"
-        """
-        resolved = _resolve(registry, "title")
-        assert resolved.spec.analyzer is not None
-        assert resolved.spec.analyzer("Café") == ["cafe"]
-
-    def test_checksum_analyzer_is_identity_single_token(
-        self,
-        registry: FieldRegistry,
-    ) -> None:
-        """
-        GIVEN:
-            - The field registry
-        WHEN:
-            - The "checksum" field's analyzer (raw tokenizer, no
-              splitting) processes "ABC-123"
-        THEN:
-            - It is returned unchanged as a single token
-        """
-        resolved = _resolve(registry, "checksum")
-        assert resolved.spec.analyzer is not None
-        assert resolved.spec.analyzer("ABC-123") == ["ABC-123"]
-
-    def test_pattern_normalizer_follows_the_registry_language(
-        self,
-        registry: FieldRegistry,
-    ) -> None:
-        """
-        GIVEN:
-            - A registry with no language, and a registry built for "en"
-        WHEN:
-            - The "title" field's pattern normalizer processes "Running"
-        THEN:
-            - With no language, only the folded run is offered
-              ("running"), since the index holds surface forms
-            - With "en", the stem is offered too ("run"), since indexed
-              terms are stemmed and the pattern has to reach them
-        """
-        resolved = _resolve(registry, "title")
-        assert resolved.spec.pattern_normalizer is not None
-        assert _distinct_forms(resolved.spec.pattern_normalizer("Running")) == (
-            "running",
-        )
-
-        resolved_en = _resolve(get_field_registry("en"), "title")
-        assert resolved_en.spec.pattern_normalizer is not None
-        assert _distinct_forms(resolved_en.spec.pattern_normalizer("Running")) == (
-            "running",
-            "run",
-        )
-
-    def test_registry_is_cached_per_language(self) -> None:
-        """
-        GIVEN:
-            - Two calls to get_field_registry("en")
-        WHEN:
-            - Both calls are made
-        THEN:
-            - They return the same registry instance
-        """
-        a = get_field_registry("en")
-        b = get_field_registry("en")
-        assert a is b
-
-    def test_registry_rebuilds_on_language_change(self) -> None:
-        """
-        GIVEN:
-            - A call to get_field_registry("en") and a call to
-              get_field_registry("de")
-        WHEN:
-            - Both calls are made
-        THEN:
-            - They return different registry instances
-        """
-        a = get_field_registry("en")
-        b = get_field_registry("de")
-        assert a is not b

src/documents/tests/search/test_schema.py

@@ -5,17 +5,12 @@ from typing import TYPE_CHECKING
 
 import pytest
 
-from documents.search._fields import PUBLIC_FIELDS
 from documents.search._schema import SCHEMA_VERSION
-from documents.search._schema import build_schema
-from documents.search._schema import field_descriptors
 from documents.search._schema import needs_rebuild
-from documents.search._schema import schema_fingerprint
 
 if TYPE_CHECKING:
     from pathlib import Path
 
-    import tantivy
     from pytest_django.fixtures import Settings
 
 
@@ -35,13 +30,7 @@ class TestNeedsRebuild:
     ) -> None:
         settings.SEARCH_LANGUAGE = "en"
         (index_dir / ".index_settings.json").write_text(
-            json.dumps(
-                {
-                    "schema_version": SCHEMA_VERSION,
-                    "language": "en",
-                    "schema_fingerprint": schema_fingerprint(),
-                },
-            ),
+            json.dumps({"schema_version": SCHEMA_VERSION, "language": "en"}),
         )
         assert needs_rebuild(index_dir) is False
 
@@ -88,61 +77,3 @@ class TestNeedsRebuild:
             json.dumps({"schema_version": SCHEMA_VERSION, "language": "en"}),
         )
         assert needs_rebuild(index_dir) is True
-
-
-def _schema_fields(schema: tantivy.Schema) -> dict[str, dict]:
-    """{name: field-state} for every field declared on a tantivy Schema.
-
-    tantivy-py 0.26 exposes no public introspection API on Schema (no
-    __iter__, get_field, to_json, etc.) -- __reduce__() (used internally for
-    pickling) is the only way to recover the field list, so we lean on it
-    here for test assertions only.
-    """
-    state = schema.__reduce__()[1][0]
-    return {field["name"]: field for field in state["inner"]}
-
-
-class TestSchemaMatchesPublicFields:
-    def test_every_public_field_is_in_the_schema(self) -> None:
-        """
-        GIVEN:
-            - PUBLIC_FIELDS and the tantivy schema built by build_schema()
-        WHEN:
-            - Every field declared in PUBLIC_FIELDS is checked against the
-              schema
-        THEN:
-            - Each one is present as a field in the built schema
-        """
-        schema = build_schema()
-        schema_field_names = set(_schema_fields(schema))
-        for field in PUBLIC_FIELDS:
-            assert field.name in schema_field_names, (
-                f"{field.name} is in PUBLIC_FIELDS but missing from build_schema()"
-            )
-
-
-class TestFastFlagAgreement:
-    def test_every_public_field_fast_flag_matches_the_built_schema(self) -> None:
-        """
-        GIVEN:
-            - PUBLIC_FIELDS and field_descriptors() (the latter is exactly
-              the input build_schema()'s SchemaBuilder consumes for the
-              `fast` kwarg on every field kind, so it pins the agreement
-              without depending on a private tantivy-py pickled
-              representation)
-        WHEN:
-            - Every PUBLIC_FIELDS entry's fast flag is compared against
-              field_descriptors()' fast flag for the same field
-        THEN:
-            - They agree for every field, catching a fast=True
-              PUBLIC_FIELDS entry the builder silently ignores here
-              instead of at a user's field:* existence query, which
-              whoosh-compat's registry trusts PUBLIC_FIELDS' fast flag to
-              resolve
-        """
-        descriptor_fast = {d.name: d.fast for d in field_descriptors()}
-        for public_field in PUBLIC_FIELDS:
-            assert descriptor_fast[public_field.name] == public_field.fast, (
-                f"{public_field.name}: PUBLIC_FIELDS says fast={public_field.fast} but"
-                f" field_descriptors() says fast={descriptor_fast[public_field.name]}"
-            )

src/documents/tests/search/test_schema_fingerprint.py

@@ -1,587 +0,0 @@
-"""The schema fingerprint stamped into .index_settings.json.
-
-tantivy compares schemas by *ordered* field list, and `tantivy.Index(schema,
-path=...)` (what every write path does) raises on any difference. SCHEMA_VERSION
-is the manual guard against that, but build_schema() is edited for *parser*
-reasons - adding an alias, flipping fast=True, adding a subpath - by people not
-thinking about the on-disk index, and forgetting the bump is exactly how this
-branch's bug happened.
-
-The fingerprint is the automatic guard: it hashes the field descriptor list that
-build_schema() itself iterates, so any change to a field's name, kind, options
-or *position* forces a rebuild on its own.
-"""
-
-from __future__ import annotations
-
-import hashlib
-import json
-from typing import TYPE_CHECKING
-
-import pytest
-import tantivy
-
-from documents.search import _schema
-from documents.search._schema import SCHEMA_VERSION
-from documents.search._schema import FieldDescriptor
-from documents.search._schema import _write_sentinels
-from documents.search._schema import build_schema
-from documents.search._schema import field_descriptors
-from documents.search._schema import needs_rebuild
-from documents.search._schema import schema_fingerprint
-
-if TYPE_CHECKING:
-    from pathlib import Path
-
-    from pytest_django.fixtures import SettingsWrapper
-
-pytestmark = pytest.mark.search
-
-# The on-disk field layout of a v2 index, pinned as data. Any edit here is an
-# index-format change: it must come with a rebuild, which the fingerprint now
-# forces automatically. Reproduced from build_schema()'s output as it stood
-# before the descriptor refactor, so it also pins that the refactor changed
-# nothing.
-PINNED_DESCRIPTORS: tuple[FieldDescriptor, ...] = (
-    FieldDescriptor("id", "u64", stored=True, indexed=True, fast=True, tokenizer=None),
-    FieldDescriptor(
-        "title",
-        "text",
-        stored=True,
-        indexed=True,
-        fast=False,
-        tokenizer="paperless_text",
-    ),
-    FieldDescriptor(
-        "content",
-        "text",
-        stored=True,
-        indexed=True,
-        fast=False,
-        tokenizer="paperless_text",
-    ),
-    FieldDescriptor(
-        "correspondent",
-        "text",
-        stored=True,
-        indexed=True,
-        fast=False,
-        tokenizer="paperless_text",
-    ),
-    FieldDescriptor(
-        "document_type",
-        "text",
-        stored=True,
-        indexed=True,
-        fast=False,
-        tokenizer="paperless_text",
-    ),
-    FieldDescriptor(
-        "storage_path",
-        "text",
-        stored=True,
-        indexed=True,
-        fast=False,
-        tokenizer="paperless_text",
-    ),
-    FieldDescriptor(
-        "original_filename",
-        "text",
-        stored=True,
-        indexed=True,
-        fast=False,
-        tokenizer="paperless_text",
-    ),
-    FieldDescriptor(
-        "tag",
-        "text",
-        stored=True,
-        indexed=True,
-        fast=False,
-        tokenizer="paperless_text",
-    ),
-    FieldDescriptor(
-        "checksum",
-        "text",
-        stored=True,
-        indexed=True,
-        fast=False,
-        tokenizer="raw",
-    ),
-    FieldDescriptor("asn", "u64", stored=True, indexed=True, fast=True, tokenizer=None),
-    FieldDescriptor(
-        "page_count",
-        "u64",
-        stored=True,
-        indexed=True,
-        fast=True,
-        tokenizer=None,
-    ),
-    FieldDescriptor(
-        "num_notes",
-        "u64",
-        stored=True,
-        indexed=True,
-        fast=True,
-        tokenizer=None,
-    ),
-    FieldDescriptor(
-        "created",
-        "date",
-        stored=True,
-        indexed=True,
-        fast=True,
-        tokenizer=None,
-    ),
-    FieldDescriptor(
-        "modified",
-        "date",
-        stored=True,
-        indexed=True,
-        fast=True,
-        tokenizer=None,
-    ),
-    FieldDescriptor(
-        "added",
-        "date",
-        stored=True,
-        indexed=True,
-        fast=True,
-        tokenizer=None,
-    ),
-    FieldDescriptor(
-        "notes",
-        "json",
-        stored=True,
-        indexed=True,
-        fast=False,
-        tokenizer="paperless_text",
-    ),
-    FieldDescriptor(
-        "notes_text",
-        "text",
-        stored=True,
-        indexed=True,
-        fast=False,
-        tokenizer="paperless_text",
-    ),
-    FieldDescriptor(
-        "custom_fields",
-        "json",
-        stored=True,
-        indexed=True,
-        fast=False,
-        tokenizer="paperless_text",
-    ),
-    FieldDescriptor(
-        "title_sort",
-        "text",
-        stored=False,
-        indexed=True,
-        fast=True,
-        tokenizer="simple_analyzer",
-    ),
-    FieldDescriptor(
-        "correspondent_sort",
-        "text",
-        stored=False,
-        indexed=True,
-        fast=True,
-        tokenizer="simple_analyzer",
-    ),
-    FieldDescriptor(
-        "type_sort",
-        "text",
-        stored=False,
-        indexed=True,
-        fast=True,
-        tokenizer="simple_analyzer",
-    ),
-    FieldDescriptor(
-        "bigram_content",
-        "text",
-        stored=False,
-        indexed=True,
-        fast=False,
-        tokenizer="bigram_analyzer",
-    ),
-    FieldDescriptor(
-        "bigram_title",
-        "text",
-        stored=False,
-        indexed=True,
-        fast=False,
-        tokenizer="bigram_analyzer",
-    ),
-    FieldDescriptor(
-        "bigram_correspondent",
-        "text",
-        stored=False,
-        indexed=True,
-        fast=False,
-        tokenizer="bigram_analyzer",
-    ),
-    FieldDescriptor(
-        "bigram_document_type",
-        "text",
-        stored=False,
-        indexed=True,
-        fast=False,
-        tokenizer="bigram_analyzer",
-    ),
-    FieldDescriptor(
-        "bigram_tag",
-        "text",
-        stored=False,
-        indexed=True,
-        fast=False,
-        tokenizer="bigram_analyzer",
-    ),
-    FieldDescriptor(
-        "simple_title",
-        "text",
-        stored=False,
-        indexed=True,
-        fast=False,
-        tokenizer="simple_search_analyzer",
-    ),
-    FieldDescriptor(
-        "simple_content",
-        "text",
-        stored=False,
-        indexed=True,
-        fast=False,
-        tokenizer="simple_search_analyzer",
-    ),
-    FieldDescriptor(
-        "autocomplete_word",
-        "text",
-        stored=False,
-        indexed=True,
-        fast=False,
-        tokenizer="raw",
-    ),
-    FieldDescriptor(
-        "owner_id",
-        "u64",
-        stored=False,
-        indexed=True,
-        fast=True,
-        tokenizer=None,
-    ),
-    FieldDescriptor(
-        "viewer_id",
-        "u64",
-        stored=False,
-        indexed=True,
-        fast=True,
-        tokenizer=None,
-    ),
-    FieldDescriptor(
-        "viewer_group_id",
-        "u64",
-        stored=False,
-        indexed=True,
-        fast=True,
-        tokenizer=None,
-    ),
-)
-
-
-def _schema_fields(schema: tantivy.Schema) -> list[dict]:
-    """The tantivy-level field list, in declaration order.
-
-    tantivy-py 0.26 exposes no public introspection API on Schema, so
-    __reduce__() (its pickling hook) is the only way to recover the field list.
-    It is used here, in a test, precisely because it is the representation the
-    persisted fingerprint must NOT depend on.
-    """
-    return schema.__reduce__()[1][0]["inner"]
-
-
-def _sentinels(index_dir: Path, **overrides: object) -> None:
-    data = {
-        "schema_version": SCHEMA_VERSION,
-        "language": None,
-        "schema_fingerprint": schema_fingerprint(),
-    }
-    data.update(overrides)
-    (index_dir / ".index_settings.json").write_text(json.dumps(data))
-
-
-class TestDescriptorsDescribeTheBuiltSchema:
-    def test_descriptors_match_the_pinned_field_layout(self) -> None:
-        """
-        GIVEN:
-            - PINNED_DESCRIPTORS, a frozen snapshot of the v2 on-disk field
-              layout, reproduced from build_schema()'s output as it stood
-              before the descriptor refactor
-        WHEN:
-            - field_descriptors() is called
-        THEN:
-            - It matches the pinned layout exactly, in the same order,
-              pinning that the refactor changed nothing
-        """
-        assert tuple(field_descriptors()) == PINNED_DESCRIPTORS
-
-    def test_built_schema_matches_the_descriptors(self) -> None:
-        """
-        GIVEN:
-            - The schema built by build_schema()
-        WHEN:
-            - Its fields are read back via __reduce__() (schema.__reduce__(),
-              tantivy-py's pickling hook)
-        THEN:
-            - Every field's name, kind, stored/fast flags and tokenizer
-              match what field_descriptors() declared as input; the
-              descriptors are not a parallel description, they are the
-              input, so a descriptor edit cannot claim a shape the
-              SchemaBuilder did not actually build
-        """
-        kinds = {"text": "text", "json": "json_object", "u64": "u64", "date": "date"}
-        built = [
-            (
-                field["name"],
-                field["type"],
-                field["options"]["stored"],
-                bool(field["options"].get("fast")),
-                (field["options"].get("indexing") or {}).get("tokenizer"),
-            )
-            for field in _schema_fields(build_schema())
-        ]
-        expected = [
-            (
-                descriptor.name,
-                kinds[descriptor.kind],
-                descriptor.stored,
-                descriptor.fast,
-                descriptor.tokenizer,
-            )
-            for descriptor in field_descriptors()
-        ]
-        assert built == expected
-
-
-class TestFingerprintSensitivity:
-    def test_a_field_option_change_moves_the_fingerprint(
-        self,
-        monkeypatch: pytest.MonkeyPatch,
-    ) -> None:
-        """
-        GIVEN:
-            - The current schema fingerprint
-        WHEN:
-            - A single field descriptor's "fast" option is changed, with
-              no other change
-        THEN:
-            - The fingerprint changes
-        """
-        before = schema_fingerprint()
-        changed = field_descriptors()
-        changed[1] = changed[1]._replace(fast=True)
-        monkeypatch.setattr(_schema, "field_descriptors", lambda: changed)
-
-        assert schema_fingerprint() != before
-
-    def test_reordering_alone_moves_the_fingerprint(
-        self,
-        monkeypatch: pytest.MonkeyPatch,
-    ) -> None:
-        """
-        GIVEN:
-            - The current schema fingerprint
-        WHEN:
-            - Two field descriptors are swapped, with no other change (the
-              original bug: same fields, different declaration order)
-        THEN:
-            - The fingerprint changes; a set- or dict-based fingerprint
-              would be blind to this, and tantivy would reject every write
-              against the existing index
-        """
-        before = schema_fingerprint()
-        swapped = field_descriptors()
-        swapped[1], swapped[2] = swapped[2], swapped[1]
-        monkeypatch.setattr(_schema, "field_descriptors", lambda: swapped)
-
-        assert schema_fingerprint() != before
-
-
-class TestFingerprintIsIndependentOfTantivy:
-    def test_a_tantivy_option_key_addition_would_not_move_it(self) -> None:
-        """
-        GIVEN:
-            - The built schema's raw field list, and the same list with a
-              new tantivy-internal option key added (simulating a
-              tantivy-py upgrade)
-        WHEN:
-            - Both raw lists are hashed directly, and schema_fingerprint()
-              is compared against a hash of field_descriptors()
-        THEN:
-            - The raw hashes differ (hashing schema.__reduce__() would
-              force a global reindex on every tantivy-py upgrade), but
-              schema_fingerprint() is unaffected, since it hashes
-              field_descriptors(), never tantivy's own representation
-        """
-        fields = _schema_fields(build_schema())
-        upgraded = [
-            {**field, "options": {**field["options"], "coerce": True}}
-            for field in fields
-        ]
-        assert _hash(upgraded) != _hash(fields)
-        assert schema_fingerprint() == _fingerprint_of(field_descriptors())
-
-    def test_fingerprint_never_touches_the_schema_builder(
-        self,
-        monkeypatch: pytest.MonkeyPatch,
-    ) -> None:
-        """
-        GIVEN:
-            - tantivy.SchemaBuilder replaced with a stand-in that raises if
-              constructed
-        WHEN:
-            - build_schema() is called (and raises), then
-              schema_fingerprint() is called again
-        THEN:
-            - schema_fingerprint() still matches its earlier value,
-              proving it never consults SchemaBuilder
-        """
-        before = schema_fingerprint()
-
-        class _RemovedSchemaBuilder:
-            def __init__(self) -> None:
-                raise AssertionError("tantivy.SchemaBuilder was consulted")
-
-        monkeypatch.setattr(tantivy, "SchemaBuilder", _RemovedSchemaBuilder)
-        with pytest.raises(AssertionError):
-            build_schema()
-
-        assert schema_fingerprint() == before
-
-
-def _hash(payload: object) -> str:
-    return hashlib.blake2b(json.dumps(payload).encode()).hexdigest()
-
-
-def _fingerprint_of(descriptors: list[FieldDescriptor]) -> str:
-    return _hash([list(descriptor) for descriptor in descriptors])
-
-
-class TestNeedsRebuildOnFingerprint:
-    def test_matching_fingerprint_does_not_rebuild(
-        self,
-        index_dir: Path,
-        settings: SettingsWrapper,
-    ) -> None:
-        """
-        GIVEN:
-            - An index directory whose sentinel file records the current
-              schema_fingerprint()
-        WHEN:
-            - needs_rebuild() is called
-        THEN:
-            - It returns False
-        """
-        settings.SEARCH_LANGUAGE = None
-        _sentinels(index_dir)
-
-        assert needs_rebuild(index_dir) is False
-
-    def test_stale_fingerprint_rebuilds_despite_a_matching_version(
-        self,
-        index_dir: Path,
-        settings: SettingsWrapper,
-        monkeypatch: pytest.MonkeyPatch,
-    ) -> None:
-        """
-        GIVEN:
-            - An index directory whose sentinel matches SCHEMA_VERSION,
-              but field_descriptors() is patched to add a field the
-              fingerprint never saw (schema edited, version not bumped)
-        WHEN:
-            - needs_rebuild() is called
-        THEN:
-            - It returns True; without the fingerprint check,
-              `reindex --if-needed` would report the index up to date and
-              every subsequent write would raise
-        """
-        settings.SEARCH_LANGUAGE = None
-        _sentinels(index_dir)
-        extended = [
-            *field_descriptors(),
-            FieldDescriptor(
-                "new_field",
-                "u64",
-                stored=False,
-                indexed=True,
-                fast=True,
-                tokenizer=None,
-            ),
-        ]
-        monkeypatch.setattr(_schema, "field_descriptors", lambda: extended)
-
-        assert needs_rebuild(index_dir) is True
-
-    def test_reordered_schema_rebuilds(
-        self,
-        index_dir: Path,
-        settings: SettingsWrapper,
-        monkeypatch: pytest.MonkeyPatch,
-    ) -> None:
-        """
-        GIVEN:
-            - An index directory whose sentinel matches the current
-              fingerprint, but field_descriptors() is patched to swap two
-              fields' order
-        WHEN:
-            - needs_rebuild() is called
-        THEN:
-            - It returns True
-        """
-        settings.SEARCH_LANGUAGE = None
-        _sentinels(index_dir)
-        reordered = field_descriptors()
-        reordered[1], reordered[2] = reordered[2], reordered[1]
-        monkeypatch.setattr(_schema, "field_descriptors", lambda: reordered)
-
-        assert needs_rebuild(index_dir) is True
-
-    def test_missing_fingerprint_rebuilds(
-        self,
-        index_dir: Path,
-        settings: SettingsWrapper,
-    ) -> None:
-        """
-        GIVEN:
-            - An index directory whose sentinel has no "schema_fingerprint"
-              key at all
-        WHEN:
-            - needs_rebuild() is called
-        THEN:
-            - It returns True; an index whose schema shape nobody recorded
-              is rebuilt rather than trusted
-        """
-        settings.SEARCH_LANGUAGE = None
-        (index_dir / ".index_settings.json").write_text(
-            json.dumps({"schema_version": SCHEMA_VERSION, "language": None}),
-        )
-
-        assert needs_rebuild(index_dir) is True
-
-    def test_written_sentinels_satisfy_the_check(
-        self,
-        index_dir: Path,
-        settings: SettingsWrapper,
-    ) -> None:
-        """
-        GIVEN:
-            - An index directory whose sentinels are written by
-              _write_sentinels() itself
-        WHEN:
-            - needs_rebuild() is called
-        THEN:
-            - It returns False
-        """
-        settings.SEARCH_LANGUAGE = "en"
-        _write_sentinels(index_dir)
-
-        assert needs_rebuild(index_dir) is False

src/documents/tests/search/test_schema_version.py

@@ -1,178 +0,0 @@
-"""SCHEMA_VERSION must change whenever build_schema()'s field list or order does.
-
-tantivy compares schemas by *ordered* field list. ``Index.open()`` loads the
-schema from the index's own ``meta.json``, so reads against an index built by an
-older release keep working after a field reorder. Writes do not:
-``WriteBatch.__enter__`` calls ``tantivy.Index(build_schema(), path=...)``, an
-open-or-create that raises ``ValueError`` on any schema difference. Nothing
-catches that ValueError, so consumption, index_document and bulk edit all
-hard-fail while ``/api/status/`` still reports the index healthy.
-
-The only thing that saves such an install is ``needs_rebuild()`` noticing the
-version stamped in ``.index_settings.json`` is stale.
-"""
-
-from __future__ import annotations
-
-import json
-from typing import TYPE_CHECKING
-
-import pytest
-import tantivy
-from django.conf import settings as django_settings
-
-from documents.search._schema import build_schema
-from documents.search._schema import needs_rebuild
-from documents.search._schema import open_or_rebuild_index
-
-if TYPE_CHECKING:
-    from pathlib import Path
-
-pytestmark = [pytest.mark.search]
-
-RELEASED_V1_SCHEMA_VERSION = 1
-
-
-def _build_released_v1_schema() -> tantivy.Schema:
-    """Frozen copy of build_schema() as shipped in v3.0.x (schema version 1).
-
-    Deliberately duplicated rather than imported: it must keep describing the
-    on-disk layout of already-deployed indexes even as build_schema() evolves.
-    """
-    sb = tantivy.SchemaBuilder()
-
-    sb.add_unsigned_field("id", stored=True, indexed=True, fast=True)
-    sb.add_text_field("checksum", stored=True, tokenizer_name="raw")
-
-    for field in (
-        "title",
-        "correspondent",
-        "document_type",
-        "storage_path",
-        "original_filename",
-        "content",
-    ):
-        sb.add_text_field(field, stored=True, tokenizer_name="paperless_text")
-
-    for field in ("title_sort", "correspondent_sort", "type_sort"):
-        sb.add_text_field(
-            field,
-            stored=False,
-            tokenizer_name="simple_analyzer",
-            fast=True,
-        )
-
-    for field in (
-        "bigram_content",
-        "bigram_title",
-        "bigram_correspondent",
-        "bigram_document_type",
-        "bigram_tag",
-    ):
-        sb.add_text_field(field, stored=False, tokenizer_name="bigram_analyzer")
-
-    for field in ("simple_title", "simple_content"):
-        sb.add_text_field(field, stored=False, tokenizer_name="simple_search_analyzer")
-
-    sb.add_text_field("autocomplete_word", stored=False, tokenizer_name="raw")
-    sb.add_text_field("tag", stored=True, tokenizer_name="paperless_text")
-
-    sb.add_json_field("notes", stored=True, tokenizer_name="paperless_text")
-    sb.add_text_field("notes_text", stored=True, tokenizer_name="paperless_text")
-    sb.add_json_field("custom_fields", stored=True, tokenizer_name="paperless_text")
-
-    for field in (
-        "correspondent_id",
-        "document_type_id",
-        "storage_path_id",
-        "tag_id",
-        "owner_id",
-        "viewer_id",
-        "viewer_group_id",
-    ):
-        sb.add_unsigned_field(field, stored=False, indexed=True, fast=True)
-
-    for field in ("created", "modified", "added"):
-        sb.add_date_field(field, stored=True, indexed=True, fast=True)
-
-    for field in ("asn", "page_count", "num_notes"):
-        sb.add_unsigned_field(field, stored=True, indexed=True, fast=True)
-
-    return sb.build()
-
-
-@pytest.fixture
-def released_v1_index(tmp_path: Path) -> Path:
-    """An index directory as a v3.0.x install would leave it on disk."""
-    index_dir = tmp_path / "index"
-    index_dir.mkdir()
-    tantivy.Index(_build_released_v1_schema(), path=str(index_dir))
-    (index_dir / ".index_settings.json").write_text(
-        json.dumps(
-            {
-                "schema_version": RELEASED_V1_SCHEMA_VERSION,
-                "language": django_settings.SEARCH_LANGUAGE,
-            },
-        ),
-    )
-    return index_dir
-
-
-class TestUpgradeFromReleasedV1Index:
-    def test_released_v1_index_is_flagged_for_rebuild(
-        self,
-        released_v1_index: Path,
-    ) -> None:
-        """
-        GIVEN:
-            - An index directory laid out exactly as a v3.0.x (schema
-              version 1) install would leave it
-        WHEN:
-            - needs_rebuild() is called
-        THEN:
-            - It returns True; if this fails,
-              `document_index reindex --if-needed` prints "Search index is
-              up to date" and skips, leaving the mismatched index in place
-        """
-        assert needs_rebuild(released_v1_index) is True
-
-    def test_opening_a_v1_index_leaves_it_writable(
-        self,
-        released_v1_index: Path,
-    ) -> None:
-        """
-        GIVEN:
-            - A v1 index directory
-        WHEN:
-            - open_or_rebuild_index() is called against it
-        THEN:
-            - The directory can be reopened with the current schema
-              without raising; end to end, open_or_rebuild_index must
-              hand back an index the write path can reopen. Before the
-              version bump, needs_rebuild() returned False here, and the
-              stale directory survived untouched, so every subsequent
-              write against it raised tantivy's own schema-mismatch
-              ValueError
-        """
-        open_or_rebuild_index(released_v1_index)
-
-        tantivy.Index(build_schema(), path=str(released_v1_index))
-
-    def test_rebuilt_index_is_not_rebuilt_again(
-        self,
-        released_v1_index: Path,
-    ) -> None:
-        """
-        GIVEN:
-            - A v1 index directory that has just been rebuilt by
-              open_or_rebuild_index()
-        WHEN:
-            - needs_rebuild() is called again
-        THEN:
-            - It returns False; the rebuild must stamp the version it
-              actually wrote, otherwise every startup wipes and reindexes
-              the whole corpus
-        """
-        open_or_rebuild_index(released_v1_index)
-
-        assert needs_rebuild(released_v1_index) is False

src/documents/tests/search/test_stem_pattern_text.py

@@ -1,37 +0,0 @@
-from __future__ import annotations
-
-import pytest
-
-from documents.search._tokenizer import stem_pattern_text
-
-pytestmark = pytest.mark.search
-
-
-class TestStemPatternText:
-    def test_unsupported_language_returns_text_unchanged(self) -> None:
-        """
-        GIVEN:
-            - A language code with no Snowball stemmer mapping
-        WHEN:
-            - A pattern run is stemmed for that language
-        THEN:
-            - The run is returned unchanged, since the stemming gate that
-              disables stemming for an unsupported language also disables
-              the pattern-side stemmer
-        """
-        assert stem_pattern_text("running", "klingon") == "running"
-
-    def test_run_past_remove_long_limit_returns_text_unchanged(self) -> None:
-        """
-        GIVEN:
-            - A supported language and a run longer than the remove_long
-              filter's limit (129 characters, matching Document.title's
-              max_length)
-        WHEN:
-            - The over-long run is stemmed
-        THEN:
-            - The remove_long filter drops the token entirely, leaving no
-              stem to substitute, so the run is returned unchanged
-        """
-        long_run = "a" * 130
-        assert stem_pattern_text(long_run, "en") == long_run

src/documents/tests/search/test_tokenizer.py

@@ -7,8 +7,8 @@ import pytest
 import tantivy
 
 from documents.search._tokenizer import _bigram_analyzer
+from documents.search._tokenizer import _paperless_text
 from documents.search._tokenizer import _simple_search_analyzer
-from documents.search._tokenizer import paperless_text_analyzer
 from documents.search._tokenizer import register_tokenizers
 
 if TYPE_CHECKING:
@@ -25,7 +25,7 @@ class TestTokenizers:
         sb.add_text_field("content", stored=True, tokenizer_name="paperless_text")
         schema = sb.build()
         idx = tantivy.Index(schema, path=None)
-        idx.register_tokenizer("paperless_text", paperless_text_analyzer(""))
+        idx.register_tokenizer("paperless_text", _paperless_text(""))
         return idx
 
     @pytest.fixture

src/documents/tests/test_api_bulk_download.py

@@ -4,6 +4,7 @@ import json
 import shutil
 import zipfile
 
+from django.contrib.auth.models import Permission
 from django.contrib.auth.models import User
 from django.test import override_settings
 from django.utils import timezone
@@ -326,6 +327,9 @@ class TestBulkDownload(DirectoriesMixin, SampleDirMixin, APITestCase):
 
     def test_download_insufficient_permissions(self) -> None:
         user = User.objects.create_user(username="temp_user")
+        user.user_permissions.add(
+            Permission.objects.get(codename="view_document"),
+        )
         self.client.force_authenticate(user=user)
 
         self.doc2.owner = self.user

src/documents/tests/test_api_bulk_edit.py

@@ -1084,6 +1084,8 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         user1 = User.objects.create(username="user1")
         self.client.force_authenticate(user=user1)
 
+        assign_perm("view_document", user1, self.doc2)
+
         response = self.client.post(
             "/api/documents/selection_data/",
             json.dumps({"documents": [self.doc2.id]}),
@@ -1091,7 +1093,18 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         )
 
         self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
-        self.assertEqual(response.content, b"Insufficient permissions")
+
+        user1.user_permissions.add(
+            Permission.objects.get(codename="view_document"),
+        )
+        user1 = User.objects.get(pk=user1.pk)
+        self.client.force_authenticate(user=user1)
+        response = self.client.post(
+            "/api/documents/selection_data/",
+            json.dumps({"documents": [self.doc2.id]}),
+            content_type="application/json",
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
 
     @mock.patch("documents.serialisers.bulk_edit.set_permissions")
     def test_set_permissions(self, m) -> None:
@@ -1636,6 +1649,24 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
 
         self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
 
+    def test_legacy_bulk_edit_rejects_out_of_bounds_pdf_doc_index(self) -> None:
+        response = self.client.post(
+            "/api/documents/bulk_edit/",
+            json.dumps(
+                {
+                    "documents": [self.doc2.id],
+                    "method": "edit_pdf",
+                    "parameters": {
+                        "operations": [{"page": 1, "doc": 2**32}],
+                    },
+                },
+            ),
+            content_type="application/json",
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertIn(b"doc index is out of bounds", response.content)
+
     @mock.patch("documents.views.bulk_edit.edit_pdf")
     def test_edit_pdf(self, m) -> None:
         self.setup_mock(m, "edit_pdf")
@@ -1738,6 +1769,21 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
         self.assertIn(b"doc must be an integer", response.content)
 
+        for doc_index in (-1, 2**32):
+            with self.subTest(doc_index=doc_index):
+                response = self.client.post(
+                    "/api/documents/edit_pdf/",
+                    json.dumps(
+                        {
+                            "documents": [self.doc2.id],
+                            "operations": [{"page": 1, "doc": doc_index}],
+                        },
+                    ),
+                    content_type="application/json",
+                )
+                self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+                self.assertIn(b"doc index is out of bounds", response.content)
+
         response = self.client.post(
             "/api/documents/edit_pdf/",
             json.dumps(

src/documents/tests/test_api_documents.py

@@ -3615,6 +3615,55 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
         self.assertEqual(response.content, b"Insufficient permissions to delete notes")
         self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
 
+    def test_notes_require_global_document_permissions(self) -> None:
+        user = User.objects.create_user(username="note_editor")
+        user.user_permissions.add(
+            *Permission.objects.filter(
+                codename__in=["view_note", "add_note", "delete_note"],
+            ),
+        )
+        doc = Document.objects.create(
+            title="test",
+            mime_type="application/pdf",
+            content="notes",
+            owner=user,
+        )
+        note = Note.objects.create(note="Existing", document=doc, user=user)
+        self.client.force_authenticate(user)
+
+        response = self.client.get(f"/api/documents/{doc.pk}/notes/")
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
+        user.user_permissions.add(
+            Permission.objects.get(codename="view_document"),
+        )
+        user = User.objects.get(pk=user.pk)
+        self.client.force_authenticate(user)
+        response = self.client.get(f"/api/documents/{doc.pk}/notes/")
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+        response = self.client.post(
+            f"/api/documents/{doc.pk}/notes/",
+            data={"note": "New"},
+        )
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
+        user.user_permissions.add(
+            Permission.objects.get(codename="change_document"),
+        )
+        user = User.objects.get(pk=user.pk)
+        self.client.force_authenticate(user)
+        response = self.client.post(
+            f"/api/documents/{doc.pk}/notes/",
+            data={"note": "New"},
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+        response = self.client.delete(
+            f"/api/documents/{doc.pk}/notes/?id={note.pk}",
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
     def test_delete_note(self) -> None:
         """
         GIVEN:
@@ -3981,6 +4030,21 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
 
         assign_perm("view_document", user1, doc)
 
+        create_resp = self.client.post(
+            "/api/share_links/",
+            data={
+                "document": doc.pk,
+                "file_version": "original",
+            },
+            format="json",
+        )
+        self.assertEqual(create_resp.status_code, status.HTTP_403_FORBIDDEN)
+
+        user1.user_permissions.add(
+            Permission.objects.get(codename="view_document"),
+        )
+        user1 = User.objects.get(pk=user1.pk)
+        self.client.force_authenticate(user1)
         create_resp = self.client.post(
             "/api/share_links/",
             data={

src/documents/tests/test_api_objects.py

@@ -457,6 +457,9 @@ class TestApiStoragePaths(DirectoriesMixin, APITestCase):
     def test_test_storage_path_requires_document_view_permission(self) -> None:
         owner = User.objects.create_user(username="owner")
         unprivileged = User.objects.create_user(username="unprivileged")
+        unprivileged.user_permissions.add(
+            Permission.objects.get(codename="view_document"),
+        )
         document = Document.objects.create(
             mime_type="application/pdf",
             owner=owner,
@@ -488,6 +491,23 @@ class TestApiStoragePaths(DirectoriesMixin, APITestCase):
         )
         assign_perm("view_document", viewer, document)
 
+        self.client.force_authenticate(user=viewer)
+        response = self.client.post(
+            f"{self.ENDPOINT}test/",
+            json.dumps(
+                {
+                    "document": document.id,
+                    "path": "path/{{ title }}",
+                },
+            ),
+            content_type="application/json",
+        )
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
+        viewer.user_permissions.add(
+            Permission.objects.get(codename="view_document"),
+        )
+        viewer = User.objects.get(pk=viewer.pk)
         self.client.force_authenticate(user=viewer)
         response = self.client.post(
             f"{self.ENDPOINT}test/",
@@ -530,6 +550,9 @@ class TestApiStoragePaths(DirectoriesMixin, APITestCase):
             password="password",
             email="owner@example.com",
         )
+        owner.user_permissions.add(
+            Permission.objects.get(codename="view_document"),
+        )
         document = Document.objects.create(
             mime_type="application/pdf",
             owner=owner,
@@ -605,6 +628,9 @@ class TestApiStoragePaths(DirectoriesMixin, APITestCase):
             checksum="123",
         )
         assign_perm("view_document", viewer, document)
+        viewer.user_permissions.add(
+            Permission.objects.get(codename="view_document"),
+        )
 
         self.client.force_authenticate(user=viewer)
         response = self.client.post(
@@ -692,6 +718,9 @@ class TestApiStoragePaths(DirectoriesMixin, APITestCase):
         )
         document.tags.add(private_tag)
         assign_perm("view_document", viewer, document)
+        viewer.user_permissions.add(
+            Permission.objects.get(codename="view_document"),
+        )
 
         self.client.force_authenticate(user=viewer)
         response = self.client.post(
@@ -745,6 +774,9 @@ class TestApiStoragePaths(DirectoriesMixin, APITestCase):
             value_int=42,
         )
         assign_perm("view_document", viewer, document)
+        viewer.user_permissions.add(
+            Permission.objects.get(codename="view_document"),
+        )
 
         self.client.force_authenticate(user=viewer)
         response = self.client.post(

src/documents/tests/test_api_trash.py

@@ -69,6 +69,16 @@ class TestTrashAPI(DirectoriesMixin, APITestCase):
         self.assertEqual(resp.status_code, status.HTTP_200_OK)
         self.assertEqual(Document.global_objects.count(), 0)
 
+    def test_trash_list_requires_global_document_view_permission(self) -> None:
+        user = User.objects.create_user(username="trash_owner")
+        document = Document.objects.create(title="Owned", owner=user)
+        document.delete()
+        self.client.force_authenticate(user)
+
+        response = self.client.get("/api/trash/")
+
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
     def test_trash_api_empty_all(self) -> None:
         """
         GIVEN:

src/documents/tests/test_api_workflows.py

@@ -194,6 +194,48 @@ class TestApiWorkflows(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_201_CREATED)
         self.assertEqual(Workflow.objects.count(), 2)
 
+    def test_api_create_workflow_ignores_nested_action_id(self) -> None:
+        """
+        GIVEN:
+            - An existing workflow action
+        WHEN:
+            - API request to create a workflow includes that action's ID
+        THEN:
+            - A new action is created without changing the existing action
+        """
+        original_title = self.action.assign_title
+
+        response = self.client.post(
+            self.ENDPOINT,
+            json.dumps(
+                {
+                    "name": "Workflow 2",
+                    "order": 1,
+                    "triggers": [
+                        {
+                            "sources": [DocumentSource.ApiUpload],
+                            "type": WorkflowTrigger.WorkflowTriggerType.CONSUMPTION,
+                            "filter_filename": "*",
+                        },
+                    ],
+                    "actions": [
+                        {
+                            "id": self.action.id,
+                            "assign_title": "New Action Title",
+                        },
+                    ],
+                },
+            ),
+            content_type="application/json",
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+        self.action.refresh_from_db()
+        self.assertEqual(self.action.assign_title, original_title)
+        new_action = Workflow.objects.get(name="Workflow 2").actions.get()
+        self.assertNotEqual(new_action.id, self.action.id)
+        self.assertEqual(new_action.assign_title, "New Action Title")
+
     def test_api_create_workflow_nested(self) -> None:
         """
         GIVEN:

src/documents/tests/test_bulk_edit.py

@@ -1642,6 +1642,17 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         mock_group.assert_not_called()
         mock_consume_file.assert_not_called()
 
+    @mock.patch("pikepdf.open")
+    def test_edit_pdf_rejects_out_of_bounds_output_index(self, mock_open) -> None:
+        with self.assertLogs("paperless.bulk_edit", level="ERROR"):
+            with self.assertRaisesRegex(ValueError, "index is out of bounds"):
+                bulk_edit.edit_pdf(
+                    [self.doc2.id],
+                    [{"page": 1, "doc": 2**32}],
+                )
+
+        mock_open.assert_not_called()
+
     @mock.patch("documents.bulk_edit.update_document_content_maybe_archive_file.delay")
     @mock.patch("documents.tasks.consume_file.apply_async")
     @mock.patch("documents.bulk_edit.tempfile.mkdtemp")

src/documents/tests/test_permission_filtering_security.py

@@ -309,6 +309,9 @@ class TestEmailDocumentPermissionBoundary:
     ):
         owner = User.objects.create_user(username="owner")
         requester = User.objects.create_user(username="requester")
+        requester.user_permissions.add(
+            Permission.objects.get(codename="view_document"),
+        )
         rest_api_client.force_authenticate(user=requester)
         hidden = DocumentFactory(owner=owner)
 
@@ -364,6 +367,27 @@ class TestBulkEditChangePermissionBoundary:
 
 @pytest.mark.django_db
 class TestBulkDownloadPermissionChecksRootDocument:
+    def test_download_requires_global_view_permission(
+        self,
+        rest_api_client,
+        paperless_dirs,
+        _media_settings,
+    ):
+        owner = User.objects.create_user(username="owner")
+        requester = User.objects.create_user(username="requester")
+        root = DocumentFactory(owner=owner)
+        root.source_path.write_bytes(b"%PDF-1.4 test")
+        assign_perm("view_document", requester, root)
+        rest_api_client.force_authenticate(user=requester)
+
+        response = rest_api_client.post(
+            "/api/documents/bulk_download/",
+            {"documents": [root.pk]},
+            format="json",
+        )
+
+        assert response.status_code == HTTPStatus.FORBIDDEN
+
     def test_permission_checked_on_root_not_on_version(
         self,
         rest_api_client,
@@ -372,6 +396,9 @@ class TestBulkDownloadPermissionChecksRootDocument:
     ):
         owner = User.objects.create_user(username="owner")
         requester = User.objects.create_user(username="requester")
+        requester.user_permissions.add(
+            Permission.objects.get(codename="view_document"),
+        )
         rest_api_client.force_authenticate(user=requester)
         root = DocumentFactory(owner=owner)
         # a version of root that the requester has NOT been individually granted
@@ -396,6 +423,9 @@ class TestBulkDownloadPermissionChecksRootDocument:
         # `stranger` case) can't tell the two apart, since they're denied
         # either way.
         version_only_grantee = User.objects.create_user(username="version_only_grantee")
+        version_only_grantee.user_permissions.add(
+            Permission.objects.get(codename="view_document"),
+        )
         assign_perm("view_document", version_only_grantee, version)
         rest_api_client.force_authenticate(user=version_only_grantee)
         response = rest_api_client.post(
@@ -417,6 +447,9 @@ class TestTrashRestorePermissionBoundary:
     ):
         owner = User.objects.create_user(username="owner")
         requester = User.objects.create_user(username="requester")
+        requester.user_permissions.add(
+            Permission.objects.get(codename="delete_document"),
+        )
         rest_api_client.force_authenticate(user=requester)
         doc = DocumentFactory(owner=owner)
         assign_perm("view_document", requester, doc)  # view only, NOT delete
@@ -435,6 +468,9 @@ class TestTrashRestorePermissionBoundary:
     ):
         owner = User.objects.create_user(username="owner")
         requester = User.objects.create_user(username="requester")
+        requester.user_permissions.add(
+            Permission.objects.get(codename="delete_document"),
+        )
         rest_api_client.force_authenticate(user=requester)
         doc = DocumentFactory(owner=owner)
         assign_perm("delete_document", requester, doc)
@@ -447,6 +483,22 @@ class TestTrashRestorePermissionBoundary:
         )
         assert response.status_code == HTTPStatus.OK
 
+    def test_restore_requires_global_delete_permission(self, rest_api_client):
+        owner = User.objects.create_user(username="owner")
+        requester = User.objects.create_user(username="requester")
+        rest_api_client.force_authenticate(user=requester)
+        doc = DocumentFactory(owner=owner)
+        assign_perm("delete_document", requester, doc)
+        doc.delete()
+
+        response = rest_api_client.post(
+            "/api/trash/",
+            {"documents": [doc.pk], "action": "restore"},
+            format="json",
+        )
+
+        assert response.status_code == HTTPStatus.FORBIDDEN
+
 
 @pytest.mark.django_db
 class TestTrashViewExcludesExplicitlyGrantedDocuments:
@@ -463,6 +515,9 @@ class TestTrashViewExcludesExplicitlyGrantedDocuments:
     def test_explicit_grant_does_not_leak_trashed_document(self, rest_api_client):
         owner = User.objects.create_user(username="trash_owner")
         grantee = User.objects.create_user(username="trash_grantee")
+        grantee.user_permissions.add(
+            Permission.objects.get(codename="view_document"),
+        )
         doc = DocumentFactory(owner=owner)
         doc.delete()  # soft delete
         assign_perm("view_document", grantee, doc)

src/documents/tests/test_share_link_bundles.py

@@ -6,8 +6,10 @@ from pathlib import Path
 from unittest import mock
 
 from django.conf import settings
+from django.contrib.auth.models import Permission
 from django.contrib.auth.models import User
 from django.utils import timezone
+from guardian.shortcuts import assign_perm
 from rest_framework import serializers
 from rest_framework import status
 from rest_framework.test import APITestCase
@@ -48,6 +50,37 @@ class ShareLinkBundleAPITests(DirectoriesMixin, APITestCase):
         delay_mock.assert_called_once()
         self.assertEqual(delay_mock.call_args.kwargs["kwargs"]["bundle_id"], bundle.pk)
 
+    @mock.patch("documents.views.build_share_link_bundle.apply_async")
+    def test_create_bundle_requires_global_document_view_permission(
+        self,
+        delay_mock,
+    ) -> None:
+        owner = User.objects.create_user(username="document_owner")
+        requester = User.objects.create_user(username="bundle_creator")
+        requester.user_permissions.add(
+            Permission.objects.get(codename="add_sharelinkbundle"),
+        )
+        document = DocumentFactory.create(owner=owner)
+        assign_perm("view_document", requester, document)
+        self.client.force_authenticate(requester)
+        payload = {
+            "document_ids": [document.pk],
+            "file_version": ShareLink.FileVersion.ARCHIVE,
+            "expiration_days": 7,
+        }
+
+        response = self.client.post(self.ENDPOINT, payload, format="json")
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
+        requester.user_permissions.add(
+            Permission.objects.get(codename="view_document"),
+        )
+        requester = User.objects.get(pk=requester.pk)
+        self.client.force_authenticate(requester)
+        response = self.client.post(self.ENDPOINT, payload, format="json")
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+        delay_mock.assert_called_once()
+
     def test_create_bundle_rejects_missing_documents(self) -> None:
         payload = {
             "document_ids": [9999],

src/documents/tests/test_views.py

@@ -141,6 +141,9 @@ class TestViews(DirectoriesMixin, TestCase):
             codename__contains="sharelink",
         )
         self.user.user_permissions.add(*sharelink_permissions)
+        self.user.user_permissions.add(
+            Permission.objects.get(codename="view_document"),
+        )
         self.user.save()
 
         self.client.force_login(self.user)
@@ -202,6 +205,9 @@ class TestViews(DirectoriesMixin, TestCase):
             codename__contains="sharelink",
         )
         self.user.user_permissions.add(*sharelink_permissions)
+        self.user.user_permissions.add(
+            Permission.objects.get(codename="view_document"),
+        )
         self.client.force_login(self.user)
 
         create_response = self.client.post(

src/documents/views.py

@@ -170,6 +170,7 @@ from documents.permissions import AcknowledgeTasksPermissions
 from documents.permissions import PaperlessAdminPermissions
 from documents.permissions import PaperlessNotePermissions
 from documents.permissions import PaperlessObjectPermissions
+from documents.permissions import TrashPermissions
 from documents.permissions import ViewDocumentsPermissions
 from documents.permissions import annotate_document_count_by_ids
 from documents.permissions import annotate_document_count_for_related_queryset
@@ -3519,7 +3520,7 @@ class PostDocumentView(GenericAPIView[Any]):
     ),
 )
 class SelectionDataView(GenericAPIView[Any]):
-    permission_classes = (IsAuthenticated,)
+    permission_classes = (IsAuthenticated, ViewDocumentsPermissions)
     serializer_class = DocumentListSerializer
     parser_classes = (parsers.MultiPartParser, parsers.JSONParser)
 
@@ -4010,7 +4011,7 @@ class StatisticsView(GenericAPIView[Any]):
     ),
 )
 class BulkDownloadView(DocumentSelectionMixin, GenericAPIView[Any]):
-    permission_classes = (IsAuthenticated,)
+    permission_classes = (IsAuthenticated, ViewDocumentsPermissions)
     serializer_class = BulkDownloadSerializer
     parser_classes = (parsers.JSONParser,)
 
@@ -4109,7 +4110,7 @@ class StoragePathViewSet(PermissionsAwareDocumentCountMixin, ModelViewSet[Storag
     def get_permissions(self):
         if self.action == "test":
             # Test action does not require object level permissions
-            self.permission_classes = (IsAuthenticated,)
+            self.permission_classes = (IsAuthenticated, ViewDocumentsPermissions)
         return super().get_permissions()
 
     def destroy(self, request, *args, **kwargs):
@@ -4676,6 +4677,12 @@ class ShareLinkBundleViewSet(PassUserMixin, ModelViewSet[ShareLinkBundle]):
     filterset_class = ShareLinkBundleFilterSet
     ordering_fields = ("created", "expiration", "status")
 
+    def get_permissions(self):
+        permissions = super().get_permissions()
+        if self.action == "create":
+            permissions.append(ViewDocumentsPermissions())
+        return permissions
+
     def get_queryset(self):
         return (
             super()
@@ -5494,7 +5501,7 @@ class SystemStatusView(PassUserMixin):
 
 
 class TrashView(ListModelMixin, PassUserMixin):
-    permission_classes = (IsAuthenticated,)
+    permission_classes = (IsAuthenticated, TrashPermissions)
     serializer_class = TrashSerializer
 
     class _TrashPermittedObjectsFilter(PermittedObjectsFilter):

src/locale/en_US/LC_MESSAGES/django.po

@@ -2,7 +2,7 @@ msgid ""
 msgstr ""
 "Project-Id-Version: paperless-ngx\n"
 "Report-Msgid-Bugs-To: \n"
-"POT-Creation-Date: 2026-09-11 15:47+0000\n"
+"POT-Creation-Date: 2026-09-12 23:18+0000\n"
 "PO-Revision-Date: 2022-02-17 04:17\n"
 "Last-Translator: \n"
 "Language-Team: English\n"
@@ -1632,7 +1632,7 @@ msgid "workflow runs"
 msgstr ""
 
 #: documents/serialisers.py:525 documents/serialisers.py:882
-#: documents/serialisers.py:2849 documents/views.py:318 documents/views.py:2693
+#: documents/serialisers.py:2854 documents/views.py:319 documents/views.py:2694
 #: paperless_mail/serialisers.py:156
 msgid "Insufficient permissions."
 msgstr ""
@@ -1641,39 +1641,39 @@ msgstr ""
 msgid "Invalid color."
 msgstr ""
 
-#: documents/serialisers.py:2326
+#: documents/serialisers.py:2327
 #, python-format
 msgid "File type %(type)s not supported"
 msgstr ""
 
-#: documents/serialisers.py:2370
+#: documents/serialisers.py:2371
 #, python-format
 msgid "Custom field id must be an integer: %(id)s"
 msgstr ""
 
-#: documents/serialisers.py:2377
+#: documents/serialisers.py:2378
 #, python-format
 msgid "Custom field with id %(id)s does not exist"
 msgstr ""
 
-#: documents/serialisers.py:2394 documents/serialisers.py:2404
+#: documents/serialisers.py:2395 documents/serialisers.py:2405
 msgid ""
 "Custom fields must be a list of integers or an object mapping ids to values."
 msgstr ""
 
-#: documents/serialisers.py:2399
+#: documents/serialisers.py:2400
 msgid "Some custom fields don't exist or were specified twice."
 msgstr ""
 
-#: documents/serialisers.py:2546
+#: documents/serialisers.py:2547
 msgid "Invalid variable detected."
 msgstr ""
 
-#: documents/serialisers.py:2905
+#: documents/serialisers.py:2910
 msgid "Duplicate document identifiers are not allowed."
 msgstr ""
 
-#: documents/serialisers.py:2935 documents/views.py:4700
+#: documents/serialisers.py:2940 documents/views.py:4707
 #, python-format
 msgid "Documents not found: %(ids)s"
 msgstr ""
@@ -1941,40 +1941,40 @@ msgstr ""
 msgid "Unable to parse URI {value}"
 msgstr ""
 
-#: documents/views.py:311 documents/views.py:2690
+#: documents/views.py:312 documents/views.py:2691
 msgid "Invalid more_like_id"
 msgstr ""
 
-#: documents/views.py:1645
+#: documents/views.py:1646
 msgid "Invalid AI configuration."
 msgstr ""
 
-#: documents/views.py:1656
+#: documents/views.py:1657
 msgid "AI backend request timed out."
 msgstr ""
 
-#: documents/views.py:1668
+#: documents/views.py:1669
 msgid "AI backend rejected the request. Check logs for details."
 msgstr ""
 
-#: documents/views.py:2515 documents/views.py:2836
+#: documents/views.py:2516 documents/views.py:2837
 msgid "Specify only one of text, title_search, query, or more_like_id."
 msgstr ""
 
-#: documents/views.py:4713
+#: documents/views.py:4720
 #, python-format
 msgid "Insufficient permissions to share document %(id)s."
 msgstr ""
 
-#: documents/views.py:4759
+#: documents/views.py:4766
 msgid "Bundle is already being processed."
 msgstr ""
 
-#: documents/views.py:4823
+#: documents/views.py:4830
 msgid "The share link bundle is still being prepared. Please try again later."
 msgstr ""
 
-#: documents/views.py:4837
+#: documents/views.py:4844
 msgid "The share link bundle is unavailable."
 msgstr ""
 

src/paperless_mail/tests/test_api.py

@@ -854,6 +854,36 @@ class TestAPIProcessedMails(DirectoriesMixin, APITestCase):
         )
         self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
 
+    def test_bulk_delete_requires_global_delete_permission(self) -> None:
+        owner = User.objects.create_user(username="mail_owner")
+        requester = User.objects.create_user(username="mail_deleter")
+        requester.user_permissions.add(
+            Permission.objects.get(codename="add_processedmail"),
+        )
+        mail = ProcessedMailFactory(owner=owner)
+        assign_perm("delete_processedmail", requester, mail)
+        self.client.force_authenticate(requester)
+
+        response = self.client.post(
+            f"{self.ENDPOINT}bulk_delete/",
+            data={"mail_ids": [mail.pk]},
+            format="json",
+        )
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
+        requester.user_permissions.add(
+            Permission.objects.get(codename="delete_processedmail"),
+        )
+        requester = User.objects.get(pk=requester.pk)
+        self.client.force_authenticate(requester)
+        response = self.client.post(
+            f"{self.ENDPOINT}bulk_delete/",
+            data={"mail_ids": [mail.pk]},
+            format="json",
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertFalse(ProcessedMail.objects.filter(pk=mail.pk).exists())
+
     def test_bulk_delete_processed_mails_rejects_mixed_batch_atomically(self) -> None:
         """
         GIVEN:

src/paperless_mail/views.py

@@ -18,6 +18,7 @@ from rest_framework import serializers
 from rest_framework.decorators import action
 from rest_framework.filters import OrderingFilter
 from rest_framework.generics import GenericAPIView
+from rest_framework.permissions import BasePermission
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.response import Response
 from rest_framework.viewsets import ModelViewSet
@@ -44,6 +45,15 @@ from paperless_mail.serialisers import ProcessedMailSerializer
 from paperless_mail.tasks import process_mail_accounts
 
 
+class DeleteProcessedMailPermissions(BasePermission):
+    def has_permission(self, request, view):
+        return bool(
+            request.user
+            and request.user.is_authenticated
+            and request.user.has_perm("paperless_mail.delete_processedmail"),
+        )
+
+
 @extend_schema_view(
     test=extend_schema(
         operation_id="mail_account_test",
@@ -206,7 +216,11 @@ class ProcessedMailViewSet(PassUserMixin, ReadOnlyModelViewSet[ProcessedMail]):
 
     queryset = ProcessedMail.objects.all().order_by("-processed")
 
-    @action(methods=["post"], detail=False)
+    @action(
+        methods=["post"],
+        detail=False,
+        permission_classes=[IsAuthenticated, DeleteProcessedMailPermissions],
+    )
     def bulk_delete(self, request):
         mail_ids = request.data.get("mail_ids", [])
         if not isinstance(mail_ids, list) or not all(

uv.lock

@@ -2932,7 +2932,6 @@ dependencies = [
     { name = "torch", version = "2.13.0+cpu", source = { registry = "https://download.pytorch.org/whl/cpu" }, marker = "sys_platform == 'linux'" },
     { name = "watchfiles" },
     { name = "whitenoise" },
-    { name = "whoosh-compat", extra = ["tantivy"] },
     { name = "zxing-cpp" },
 ]
 
@@ -3091,7 +3090,6 @@ requires-dist = [
     { name = "torch", specifier = "~=2.13.0", index = "https://download.pytorch.org/whl/cpu" },
     { name = "watchfiles", specifier = ">=1.2" },
     { name = "whitenoise", specifier = "~=6.11" },
-    { name = "whoosh-compat", extras = ["tantivy"], specifier = "==0.1.0" },
     { name = "zxing-cpp", specifier = "~=3.1.0" },
 ]
 provides-extras = ["mariadb", "postgres", "webserver"]
@@ -5640,23 +5638,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/db/eb/d5583a11486211f3ebd4b385545ae787f32363d453c19fffd81106c9c138/whitenoise-6.12.0-py3-none-any.whl", hash = "sha256:fc5e8c572e33ebf24795b47b6a7da8da3c00cff2349f5b04c02f28d0cc5a3cc2", size = 20302, upload-time = "2026-02-27T00:05:40.086Z" },
 ]
 
-[[package]]
-name = "whoosh-compat"
-version = "0.1.0"
-source = { registry = "https://pypi.org/simple" }
-dependencies = [
-    { name = "python-dateutil" },
-]
-sdist = { url = "https://files.pythonhosted.org/packages/5d/f7/3e45f4a484afa174cd42e424ce2b8c514ae54f564830122656ad17b765e2/whoosh_compat-0.1.0.tar.gz", hash = "sha256:86935bdc159ed9b0a06a4661d17f1251d8280340b84e218cc73d915a2edaddf7", size = 577543, upload-time = "2026-08-25T15:22:42.316Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/3f/51/9a8399d0f472814e136a2884bf910c6c59843a5a2c66afc4b5133eb531ea/whoosh_compat-0.1.0-py3-none-any.whl", hash = "sha256:3e7c5f519b4d397dbf4f8d7bbe4ca1eb6004da24c8892bc701844ed393bc97e7", size = 153875, upload-time = "2026-08-25T15:22:40.787Z" },
-]
-
-[package.optional-dependencies]
-tantivy = [
-    { name = "tantivy" },
-]
-
 [[package]]
 name = "wrapt"
 version = "2.0.1"