Coverage for paperless/views.py: 79%
259 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 09:07 +0000
1from collections import OrderedDict
2from pathlib import Path
3from typing import Any
5from allauth.mfa import signals
6from allauth.mfa.adapter import get_adapter as get_mfa_adapter
7from allauth.mfa.base.internal.flows import delete_and_cleanup
8from allauth.mfa.models import Authenticator
9from allauth.mfa.recovery_codes.internal.flows import auto_generate_recovery_codes
10from allauth.mfa.totp.internal import auth as totp_auth
11from allauth.socialaccount.adapter import get_adapter
12from allauth.socialaccount.models import SocialAccount
13from django.conf import settings
14from django.contrib.auth.models import Group
15from django.contrib.auth.models import User
16from django.contrib.staticfiles.storage import staticfiles_storage
17from django.db.models.functions import Lower
18from django.http import FileResponse
19from django.http import HttpResponseBadRequest
20from django.http import HttpResponseForbidden
21from django.http import HttpResponseNotFound
22from django.views.generic import View
23from django_filters.rest_framework import DjangoFilterBackend
24from drf_spectacular.types import OpenApiTypes
25from drf_spectacular.utils import extend_schema
26from drf_spectacular.utils import extend_schema_view
27from rest_framework.authtoken.models import Token
28from rest_framework.authtoken.views import ObtainAuthToken
29from rest_framework.decorators import action
30from rest_framework.exceptions import ValidationError
31from rest_framework.fields import BooleanField
32from rest_framework.filters import OrderingFilter
33from rest_framework.generics import GenericAPIView
34from rest_framework.pagination import PageNumberPagination
35from rest_framework.permissions import DjangoModelPermissions
36from rest_framework.permissions import IsAuthenticated
37from rest_framework.response import Response
38from rest_framework.throttling import ScopedRateThrottle
39from rest_framework.viewsets import ModelViewSet
41from documents.models import PaperlessTask
42from documents.permissions import PaperlessObjectPermissions
43from documents.tasks import llmindex_index
44from paperless.filters import GroupFilterSet
45from paperless.filters import UserFilterSet
46from paperless.models import ApplicationConfiguration
47from paperless.serialisers import ApplicationConfigurationSerializer
48from paperless.serialisers import GroupSerializer
49from paperless.serialisers import PaperlessAuthTokenSerializer
50from paperless.serialisers import ProfileSerializer
51from paperless.serialisers import UserSerializer
52from paperless_ai.indexing import llm_index_exists
55class PaperlessObtainAuthTokenView(ObtainAuthToken):
56 serializer_class = PaperlessAuthTokenSerializer
57 throttle_classes = [ScopedRateThrottle]
58 throttle_scope = "login"
61class StandardPagination(PageNumberPagination):
62 page_size = 25
63 page_size_query_param = "page_size"
64 max_page_size = 100000
66 def _get_api_version(self) -> int:
67 request = getattr(self, "request", None)
68 default_version = settings.REST_FRAMEWORK["DEFAULT_VERSION"]
69 return int(request.version if request else default_version)
71 def _should_include_all(self) -> bool:
72 # TODO: remove legacy `all` support when API v9 is dropped.
73 return self._get_api_version() < 10
75 def get_paginated_response(self, data):
76 response_data = [
77 ("count", self.page.paginator.count),
78 ("next", self.get_next_link()),
79 ("previous", self.get_previous_link()),
80 ]
81 if self._should_include_all(): 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true
82 response_data.append(("all", self.get_all_result_ids()))
83 response_data.append(("results", data))
85 return Response(
86 OrderedDict(response_data),
87 )
89 def get_all_result_ids(self):
90 from documents.search import TantivyRelevanceList
92 query = self.page.paginator.object_list
93 if isinstance(query, TantivyRelevanceList):
94 return query.get_all_ids()
95 return self.page.paginator.object_list.values_list("pk", flat=True)
97 def get_paginated_response_schema(self, schema):
98 response_schema = super().get_paginated_response_schema(schema)
99 if self._should_include_all(): 99 ↛ 100line 99 didn't jump to line 100 because the condition on line 99 was never true
100 response_schema["properties"]["all"] = {
101 "type": "array",
102 "example": "[1, 2, 3]",
103 "items": {"type": "integer"},
104 }
105 else:
106 response_schema["properties"].pop("all", None)
107 return response_schema
110class FaviconView(View):
111 def get(self, request, *args, **kwargs):
112 try:
113 path = Path(staticfiles_storage.path("paperless/img/favicon.ico"))
114 return FileResponse(path.open("rb"), content_type="image/x-icon")
115 except FileNotFoundError:
116 return HttpResponseNotFound("favicon.ico not found")
119class UserViewSet(ModelViewSet[User]):
120 _BOOL_NOT_PROVIDED = object()
121 model = User
123 queryset = User.objects.exclude(
124 username__in=["consumer", "AnonymousUser"],
125 ).order_by(Lower("username"))
127 serializer_class = UserSerializer
128 pagination_class = StandardPagination
129 permission_classes = (IsAuthenticated, PaperlessObjectPermissions)
130 filter_backends = (DjangoFilterBackend, OrderingFilter)
131 filterset_class = UserFilterSet
132 ordering_fields = ("username",)
134 @staticmethod
135 def _parse_requested_bool(data, key: str):
136 if key not in data:
137 return UserViewSet._BOOL_NOT_PROVIDED
138 try:
139 return BooleanField().to_internal_value(data.get(key))
140 except ValidationError:
141 # Let serializer validation report invalid values as 400 responses
142 return UserViewSet._BOOL_NOT_PROVIDED
144 def create(self, request, *args, **kwargs):
145 requested_is_superuser = self._parse_requested_bool(
146 request.data,
147 "is_superuser",
148 )
149 requested_is_staff = self._parse_requested_bool(request.data, "is_staff")
151 if not request.user.is_superuser: 151 ↛ 152line 151 didn't jump to line 152 because the condition on line 151 was never true
152 if requested_is_superuser is True:
153 return HttpResponseForbidden(
154 "Superuser status can only be granted by a superuser",
155 )
156 if requested_is_staff is True:
157 return HttpResponseForbidden(
158 "Staff status can only be granted by a superuser",
159 )
161 return super().create(request, *args, **kwargs)
163 def update(self, request, *args, **kwargs):
164 user_to_update: User = self.get_object()
166 if not request.user.is_superuser and user_to_update.is_superuser: 166 ↛ 167line 166 didn't jump to line 167 because the condition on line 166 was never true
167 return HttpResponseForbidden(
168 "Superusers can only be modified by other superusers",
169 )
171 requested_is_superuser = self._parse_requested_bool(
172 request.data,
173 "is_superuser",
174 )
175 requested_is_staff = self._parse_requested_bool(request.data, "is_staff")
177 if ( 177 ↛ 182line 177 didn't jump to line 182 because the condition on line 177 was never true
178 not request.user.is_superuser
179 and requested_is_superuser is not self._BOOL_NOT_PROVIDED
180 and requested_is_superuser != user_to_update.is_superuser
181 ):
182 return HttpResponseForbidden(
183 "Superuser status can only be changed by a superuser",
184 )
185 if ( 185 ↛ 190line 185 didn't jump to line 190 because the condition on line 185 was never true
186 not request.user.is_superuser
187 and requested_is_staff is not self._BOOL_NOT_PROVIDED
188 and requested_is_staff != user_to_update.is_staff
189 ):
190 return HttpResponseForbidden(
191 "Staff status can only be changed by a superuser",
192 )
193 return super().update(request, *args, **kwargs)
195 def destroy(self, request, *args, **kwargs):
196 user_to_delete: User = self.get_object()
198 if not request.user.is_superuser and user_to_delete.is_superuser: 198 ↛ 199line 198 didn't jump to line 199 because the condition on line 198 was never true
199 return HttpResponseForbidden(
200 "Superusers can only be deleted by other superusers",
201 )
203 return super().destroy(request, *args, **kwargs)
205 @extend_schema(
206 request=None,
207 responses={
208 200: OpenApiTypes.BOOL,
209 404: OpenApiTypes.STR,
210 },
211 )
212 @action(detail=True, methods=["post"])
213 def deactivate_totp(self, request, pk=None):
214 request_user = request.user
215 user = User.objects.get(pk=pk)
216 if not request_user.is_superuser and request_user != user: 216 ↛ 217line 216 didn't jump to line 217 because the condition on line 216 was never true
217 return HttpResponseForbidden(
218 "You do not have permission to deactivate TOTP for this user",
219 )
220 authenticator = Authenticator.objects.filter(
221 user=user,
222 type=Authenticator.Type.TOTP,
223 ).first()
224 if authenticator is not None: 224 ↛ 225line 224 didn't jump to line 225 because the condition on line 224 was never true
225 delete_and_cleanup(request, authenticator)
226 return Response(data=True)
227 else:
228 return HttpResponseNotFound("TOTP not found")
231class GroupViewSet(ModelViewSet[Group]):
232 model = Group
234 queryset = Group.objects.order_by(Lower("name"))
236 serializer_class = GroupSerializer
237 pagination_class = StandardPagination
238 permission_classes = (IsAuthenticated, PaperlessObjectPermissions)
239 filter_backends = (DjangoFilterBackend, OrderingFilter)
240 filterset_class = GroupFilterSet
241 ordering_fields = ("name",)
244class ProfileView(GenericAPIView[Any]):
245 """
246 User profile view, only available when logged in
247 """
249 permission_classes = [IsAuthenticated]
250 serializer_class = ProfileSerializer
252 def get(self, request, *args, **kwargs):
253 user = self.request.user
255 serializer = self.get_serializer(data=request.data)
256 return Response(serializer.to_representation(user))
258 def patch(self, request, *args, **kwargs):
259 serializer = self.get_serializer(data=request.data)
260 serializer.is_valid(raise_exception=True)
261 user = self.request.user if hasattr(self.request, "user") else None
263 password = serializer.validated_data.pop("password", None)
264 if password and password.replace("*", ""):
265 user.set_password(password)
266 user.save()
268 for key, value in serializer.validated_data.items():
269 setattr(user, key, value)
270 user.save()
272 return Response(serializer.to_representation(user))
275@extend_schema_view(
276 get=extend_schema(
277 responses={
278 (200, "application/json"): OpenApiTypes.OBJECT,
279 },
280 ),
281 post=extend_schema(
282 request={
283 "application/json": {
284 "type": "object",
285 "properties": {
286 "secret": {"type": "string"},
287 "code": {"type": "string"},
288 },
289 "required": ["secret", "code"],
290 },
291 },
292 responses={
293 (200, "application/json"): OpenApiTypes.OBJECT,
294 },
295 ),
296 delete=extend_schema(
297 responses={
298 (200, "application/json"): OpenApiTypes.BOOL,
299 404: OpenApiTypes.STR,
300 },
301 ),
302)
303class TOTPView(GenericAPIView[Any]):
304 """
305 TOTP views
306 """
308 permission_classes = [IsAuthenticated]
310 def get(self, request, *args, **kwargs):
311 """
312 Generates a new TOTP secret and returns the URL and SVG
313 """
314 user = self.request.user
315 mfa_adapter = get_mfa_adapter()
316 secret = totp_auth.get_totp_secret(regenerate=True)
317 url = mfa_adapter.build_totp_url(user, secret)
318 svg = mfa_adapter.build_totp_svg(url)
319 return Response(
320 {
321 "url": url,
322 "qr_svg": svg,
323 "secret": secret,
324 },
325 )
327 def post(self, request, *args, **kwargs):
328 """
329 Validates a TOTP code and activates the TOTP authenticator
330 """
331 valid = totp_auth.validate_totp_code(
332 request.data["secret"],
333 request.data["code"],
334 )
335 recovery_codes = None
336 if valid: 336 ↛ 337line 336 didn't jump to line 337 because the condition on line 336 was never true
337 auth = totp_auth.TOTP.activate(
338 request.user,
339 request.data["secret"],
340 ).instance
341 signals.authenticator_added.send(
342 sender=Authenticator,
343 request=request,
344 user=request.user,
345 authenticator=auth,
346 )
347 rc_auth: Authenticator = auto_generate_recovery_codes(request)
348 if rc_auth:
349 recovery_codes = rc_auth.wrap().get_unused_codes()
350 return Response(
351 {
352 "success": valid,
353 "recovery_codes": recovery_codes,
354 },
355 )
357 def delete(self, request, *args, **kwargs):
358 """
359 Deactivates the TOTP authenticator
360 """
361 user = self.request.user
362 authenticator = Authenticator.objects.filter(
363 user=user,
364 type=Authenticator.Type.TOTP,
365 ).first()
366 if authenticator is not None: 366 ↛ 367line 366 didn't jump to line 367 because the condition on line 366 was never true
367 delete_and_cleanup(request, authenticator)
368 return Response(data=True)
369 else:
370 return HttpResponseNotFound("TOTP not found")
373@extend_schema_view(
374 post=extend_schema(
375 request={
376 "application/json": None,
377 },
378 responses={
379 (200, "application/json"): OpenApiTypes.STR,
380 },
381 ),
382)
383class GenerateAuthTokenView(GenericAPIView[Any]):
384 """
385 Generates (or re-generates) an auth token, requires a logged in user
386 unlike the default DRF endpoint
387 """
389 permission_classes = [IsAuthenticated]
391 def post(self, request, *args, **kwargs):
392 user = self.request.user
394 existing_token = Token.objects.filter(user=user).first()
395 if existing_token is not None:
396 existing_token.delete()
397 token = Token.objects.create(user=user)
398 return Response(
399 token.key,
400 )
403@extend_schema_view(
404 list=extend_schema(
405 description="Get the application configuration",
406 external_docs={
407 "description": "Application Configuration",
408 "url": "https://docs.paperless-ngx.com/configuration/",
409 },
410 ),
411)
412class ApplicationConfigurationViewSet(ModelViewSet[ApplicationConfiguration]):
413 model = ApplicationConfiguration
415 queryset = ApplicationConfiguration.objects
417 serializer_class = ApplicationConfigurationSerializer
418 permission_classes = (IsAuthenticated, DjangoModelPermissions)
420 @extend_schema(exclude=True)
421 def create(self, request, *args, **kwargs):
422 return Response(status=405) # Not Allowed
424 def perform_update(self, serializer):
425 old_instance = ApplicationConfiguration.objects.all().first()
426 old_llm_embedding_backend = (
427 old_instance.llm_embedding_backend or settings.LLM_EMBEDDING_BACKEND
428 )
429 old_llm_embedding_chunk_size = (
430 old_instance.llm_embedding_chunk_size or settings.LLM_EMBEDDING_CHUNK_SIZE
431 )
432 old_llm_embedding_endpoint = (
433 old_instance.llm_embedding_endpoint or settings.LLM_EMBEDDING_ENDPOINT
434 )
435 old_llm_embedding_model = (
436 old_instance.llm_embedding_model or settings.LLM_EMBEDDING_MODEL
437 )
438 old_llm_context_size = (
439 old_instance.llm_context_size or settings.LLM_CONTEXT_SIZE
440 )
442 new_instance: ApplicationConfiguration = serializer.save()
443 new_llm_embedding_backend = (
444 new_instance.llm_embedding_backend or settings.LLM_EMBEDDING_BACKEND
445 )
446 new_ai_enabled = (
447 new_instance.ai_enabled
448 if new_instance.ai_enabled is not None
449 else settings.AI_ENABLED
450 )
451 new_ai_index_enabled = bool(
452 new_ai_enabled and new_llm_embedding_backend,
453 )
454 new_llm_embedding_chunk_size = (
455 new_instance.llm_embedding_chunk_size or settings.LLM_EMBEDDING_CHUNK_SIZE
456 )
457 new_llm_embedding_endpoint = (
458 new_instance.llm_embedding_endpoint or settings.LLM_EMBEDDING_ENDPOINT
459 )
460 new_llm_embedding_model = (
461 new_instance.llm_embedding_model or settings.LLM_EMBEDDING_MODEL
462 )
463 new_llm_context_size = (
464 new_instance.llm_context_size or settings.LLM_CONTEXT_SIZE
465 )
467 embedding_config_changed = (
468 old_llm_embedding_backend != new_llm_embedding_backend
469 or old_llm_embedding_chunk_size != new_llm_embedding_chunk_size
470 or old_llm_embedding_endpoint != new_llm_embedding_endpoint
471 or old_llm_embedding_model != new_llm_embedding_model
472 or old_llm_context_size != new_llm_context_size
473 )
474 rebuild_needed = new_ai_index_enabled and (
475 not llm_index_exists() or embedding_config_changed
476 )
478 if rebuild_needed: 478 ↛ 479line 478 didn't jump to line 479 because the condition on line 478 was never true
479 llmindex_index.apply_async(
480 kwargs={"rebuild": True},
481 headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
482 )
485@extend_schema_view(
486 post=extend_schema(
487 request={
488 "application/json": {
489 "type": "object",
490 "properties": {
491 "id": {"type": "integer"},
492 },
493 "required": ["id"],
494 },
495 },
496 responses={
497 (200, "application/json"): OpenApiTypes.INT,
498 400: OpenApiTypes.STR,
499 },
500 ),
501)
502class DisconnectSocialAccountView(GenericAPIView[Any]):
503 """
504 Disconnects a social account provider from the user account
505 """
507 permission_classes = [IsAuthenticated]
509 def post(self, request, *args, **kwargs):
510 user = self.request.user
512 try:
513 account = user.socialaccount_set.get(pk=request.data["id"])
514 account_id = account.id
515 account.delete()
516 return Response(account_id)
517 except SocialAccount.DoesNotExist:
518 return HttpResponseBadRequest("Social account not found")
521@extend_schema_view(
522 get=extend_schema(
523 responses={
524 (200, "application/json"): OpenApiTypes.OBJECT,
525 },
526 ),
527)
528class SocialAccountProvidersView(GenericAPIView[Any]):
529 """
530 List of social account providers
531 """
533 permission_classes = [IsAuthenticated]
535 def get(self, request, *args, **kwargs):
536 adapter = get_adapter()
537 providers = adapter.list_providers(request)
538 resp = [
539 {"name": p.name, "login_url": p.get_login_url(request, process="connect")}
540 for p in providers
541 if p.id != "openid"
542 ]
544 for openid_provider in filter(lambda p: p.id == "openid", providers): 544 ↛ 545line 544 didn't jump to line 545 because the loop on line 544 never started
545 resp += [
546 {
547 "name": b["name"],
548 "login_url": openid_provider.get_login_url(
549 request,
550 process="connect",
551 openid=b["openid_url"],
552 ),
553 }
554 for b in openid_provider.get_brands()
555 ]
557 return Response(sorted(resp, key=lambda p: p["name"]))