Coverage for paperless_mail/views.py: 65%
161 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 09:07 +0000
1import datetime
2import logging
3from datetime import timedelta
4from http import HTTPStatus
5from typing import Any
7from django.http import HttpResponseBadRequest
8from django.http import HttpResponseForbidden
9from django.http import HttpResponseRedirect
10from django.utils import timezone
11from django_filters.rest_framework import DjangoFilterBackend
12from drf_spectacular.types import OpenApiTypes
13from drf_spectacular.utils import extend_schema
14from drf_spectacular.utils import extend_schema_view
15from drf_spectacular.utils import inline_serializer
16from httpx_oauth.oauth2 import GetAccessTokenError
17from rest_framework import serializers
18from rest_framework.decorators import action
19from rest_framework.filters import OrderingFilter
20from rest_framework.generics import GenericAPIView
21from rest_framework.permissions import BasePermission
22from rest_framework.permissions import IsAuthenticated
23from rest_framework.response import Response
24from rest_framework.viewsets import ModelViewSet
25from rest_framework.viewsets import ReadOnlyModelViewSet
27from documents.filters import PermittedObjectsFilter
28from documents.models import PaperlessTask
29from documents.permissions import PaperlessObjectPermissions
30from documents.permissions import has_perms_owner_aware
31from documents.permissions import permitted_object_ids
32from documents.views import PassUserMixin
33from paperless.views import StandardPagination
34from paperless_mail.filters import ProcessedMailFilterSet
35from paperless_mail.mail import MailError
36from paperless_mail.mail import get_mailbox
37from paperless_mail.mail import mailbox_login
38from paperless_mail.models import MailAccount
39from paperless_mail.models import MailRule
40from paperless_mail.models import ProcessedMail
41from paperless_mail.oauth import PaperlessMailOAuth2Manager
42from paperless_mail.serialisers import MailAccountSerializer
43from paperless_mail.serialisers import MailRuleSerializer
44from paperless_mail.serialisers import ProcessedMailSerializer
45from paperless_mail.tasks import process_mail_accounts
48class DeleteProcessedMailPermissions(BasePermission):
49 def has_permission(self, request, view):
50 return bool(
51 request.user
52 and request.user.is_authenticated
53 and request.user.has_perm("paperless_mail.delete_processedmail"),
54 )
57@extend_schema_view(
58 test=extend_schema(
59 operation_id="mail_account_test",
60 request=MailAccountSerializer,
61 description="Test a mail account",
62 responses={
63 200: inline_serializer(
64 name="MailAccountTestResponse",
65 fields={"success": serializers.BooleanField()},
66 ),
67 400: OpenApiTypes.STR,
68 },
69 ),
70 process=extend_schema(
71 operation_id="mail_account_process",
72 description="Manually process the selected mail account for new messages.",
73 responses={
74 200: inline_serializer(
75 name="MailAccountProcessResponse",
76 fields={"result": serializers.CharField(default="OK")},
77 ),
78 404: None,
79 },
80 ),
81)
82class MailAccountViewSet(PassUserMixin, ModelViewSet[MailAccount]):
83 model = MailAccount
85 queryset = MailAccount.objects.all().order_by("pk")
86 serializer_class = MailAccountSerializer
87 pagination_class = StandardPagination
88 permission_classes = (IsAuthenticated, PaperlessObjectPermissions)
89 filter_backends = (PermittedObjectsFilter,)
91 def get_permissions(self):
92 if self.action == "test":
93 # Test action does not require object level permissions
94 self.permission_classes = (IsAuthenticated,)
95 return super().get_permissions()
97 @action(methods=["post"], detail=False)
98 def test(self, request):
99 logger = logging.getLogger("paperless_mail")
100 request.data["name"] = datetime.datetime.now().isoformat()
101 serializer = self.get_serializer(data=request.data)
102 serializer.is_valid(raise_exception=True)
103 existing_account = None
104 account_id = request.data.get("id")
106 # testing a new connection requires add permission
107 if account_id is None and not request.user.has_perms( 107 ↛ 110line 107 didn't jump to line 110 because the condition on line 107 was never true
108 ["paperless_mail.add_mailaccount"],
109 ):
110 return HttpResponseForbidden("Insufficient permissions")
112 # testing an existing account requires change permission on that account
113 if account_id is not None: 113 ↛ 114line 113 didn't jump to line 114 because the condition on line 113 was never true
114 try:
115 existing_account = MailAccount.objects.get(pk=account_id)
116 except (TypeError, ValueError, MailAccount.DoesNotExist):
117 return HttpResponseForbidden("Insufficient permissions")
119 if not request.user.has_perms(
120 ["paperless_mail.change_mailaccount"],
121 ) or not has_perms_owner_aware(
122 request.user,
123 "change_mailaccount",
124 existing_account,
125 ):
126 return HttpResponseForbidden("Insufficient permissions")
128 # account exists, use the password from there instead of ***
129 if ( 129 ↛ 133line 129 didn't jump to line 133 because the condition on line 129 was never true
130 len(serializer.validated_data.get("password").replace("*", "")) == 0
131 and existing_account is not None
132 ):
133 serializer.validated_data["password"] = existing_account.password
134 serializer.validated_data["account_type"] = existing_account.account_type
135 serializer.validated_data["refresh_token"] = existing_account.refresh_token
136 serializer.validated_data["expiration"] = existing_account.expiration
138 account = MailAccount(**serializer.validated_data)
139 try:
140 with get_mailbox(
141 account.imap_server,
142 account.imap_port,
143 account.imap_security,
144 ) as M:
145 if (
146 existing_account is not None
147 and account.is_token
148 and account.expiration is not None
149 and account.expiration < timezone.now()
150 ):
151 oauth_manager = PaperlessMailOAuth2Manager()
152 if oauth_manager.refresh_account_oauth_token(existing_account):
153 # User is not changing password and token needs to be refreshed
154 existing_account.refresh_from_db()
155 account.password = existing_account.password
156 else:
157 logger.error(
158 "Mail account connectivity test failed: Unable to refresh oauth token",
159 )
160 raise MailError("Unable to refresh oauth token")
162 mailbox_login(M, account)
163 return Response({"success": True})
164 except MailError:
165 logger.error(
166 "Mail account connectivity test failed",
167 )
168 return HttpResponseBadRequest("Unable to connect to server")
170 @action(methods=["post"], detail=True)
171 def process(self, request, pk=None):
172 account = self.get_object()
173 process_mail_accounts.apply_async(
174 kwargs={"account_ids": [account.pk]},
175 headers={"trigger_source": PaperlessTask.TriggerSource.MANUAL},
176 )
178 return Response({"result": "OK"})
181@extend_schema_view(
182 bulk_delete=extend_schema(
183 operation_id="processed_mail_bulk_delete",
184 description="Delete multiple processed mail records by ID.",
185 request=inline_serializer(
186 name="BulkDeleteMailRequest",
187 fields={
188 "mail_ids": serializers.ListField(child=serializers.IntegerField()),
189 },
190 ),
191 responses={
192 (HTTPStatus.OK, "application/json"): inline_serializer(
193 name="BulkDeleteMailResponse",
194 fields={
195 "result": serializers.CharField(),
196 "deleted_mail_ids": serializers.ListField(
197 child=serializers.IntegerField(),
198 ),
199 },
200 ),
201 HTTPStatus.BAD_REQUEST: None,
202 HTTPStatus.FORBIDDEN: None,
203 },
204 ),
205)
206class ProcessedMailViewSet(PassUserMixin, ReadOnlyModelViewSet[ProcessedMail]):
207 permission_classes = (IsAuthenticated, PaperlessObjectPermissions)
208 serializer_class = ProcessedMailSerializer
209 pagination_class = StandardPagination
210 filter_backends = (
211 DjangoFilterBackend,
212 OrderingFilter,
213 PermittedObjectsFilter,
214 )
215 filterset_class = ProcessedMailFilterSet
217 queryset = ProcessedMail.objects.all().order_by("-processed")
219 @action(
220 methods=["post"],
221 detail=False,
222 permission_classes=[IsAuthenticated, DeleteProcessedMailPermissions],
223 )
224 def bulk_delete(self, request):
225 mail_ids = request.data.get("mail_ids", [])
226 if not isinstance(mail_ids, list) or not all(
227 isinstance(i, int) for i in mail_ids
228 ):
229 return HttpResponseBadRequest("mail_ids must be a list of integers")
230 mails = ProcessedMail.objects.filter(id__in=mail_ids)
231 # Check every id up front so an unpermitted one rejects the whole
232 # request rather than deleting the mails ahead of it first.
233 if mails.exclude( 233 ↛ 240line 233 didn't jump to line 240 because the condition on line 233 was never true
234 pk__in=permitted_object_ids(
235 request.user,
236 ProcessedMail,
237 "delete_processedmail",
238 ),
239 ).exists():
240 return HttpResponseForbidden("Insufficient permissions")
241 mails.delete()
242 return Response({"result": "OK", "deleted_mail_ids": mail_ids})
245class MailRuleViewSet(PassUserMixin, ModelViewSet[MailRule]):
246 model = MailRule
248 queryset = MailRule.objects.all().order_by("order")
249 serializer_class = MailRuleSerializer
250 pagination_class = StandardPagination
251 permission_classes = (IsAuthenticated, PaperlessObjectPermissions)
252 filter_backends = (PermittedObjectsFilter,)
255@extend_schema_view(
256 get=extend_schema(
257 description="Callback view for OAuth2 authentication",
258 responses={200: None},
259 ),
260)
261class OauthCallbackView(GenericAPIView[Any]):
262 permission_classes = (IsAuthenticated,)
264 def get(self, request, format=None):
265 if not ( 265 ↛ 268line 265 didn't jump to line 268 because the condition on line 265 was never true
266 request.user and request.user.has_perms(["paperless_mail.add_mailaccount"])
267 ):
268 return HttpResponseBadRequest(
269 "You do not have permission to add mail accounts",
270 )
272 logger = logging.getLogger("paperless_mail")
273 code = request.query_params.get("code")
274 # Gmail passes scope as a query param, Outlook does not
275 scope = request.query_params.get("scope")
277 if code is None: 277 ↛ 283line 277 didn't jump to line 283 because the condition on line 277 was always true
278 logger.error(
279 "Invalid oauth callback request: missing code",
280 )
281 return HttpResponseBadRequest("Invalid request, see logs for more detail")
283 oauth_manager = PaperlessMailOAuth2Manager(
284 state=request.session.get("oauth_state"),
285 )
287 state = request.query_params.get("state", "")
288 if not oauth_manager.validate_state(state):
289 logger.error(
290 "Invalid oauth callback request: state validation failed",
291 )
292 return HttpResponseBadRequest("Invalid request, see logs for more detail")
294 try:
295 if scope is not None and "google" in scope:
296 # Google
297 account_type = MailAccount.MailAccountType.GMAIL_OAUTH
298 imap_server = "imap.gmail.com"
299 defaults = {
300 "name": f"Gmail OAuth {timezone.now()}",
301 "username": "",
302 "imap_security": MailAccount.ImapSecurity.SSL,
303 "imap_port": 993,
304 "account_type": account_type,
305 }
306 result = oauth_manager.get_gmail_access_token(code)
308 elif scope is None:
309 # Outlook
310 account_type = MailAccount.MailAccountType.OUTLOOK_OAUTH
311 imap_server = "outlook.office365.com"
312 defaults = {
313 "name": f"Outlook OAuth {timezone.now()}",
314 "username": "",
315 "imap_security": MailAccount.ImapSecurity.SSL,
316 "imap_port": 993,
317 "account_type": account_type,
318 }
320 result = oauth_manager.get_outlook_access_token(code)
322 access_token = result["access_token"]
323 refresh_token = result["refresh_token"]
324 expires_in = result["expires_in"]
325 account, _ = MailAccount.objects.update_or_create(
326 password=access_token,
327 is_token=True,
328 imap_server=imap_server,
329 refresh_token=refresh_token,
330 expiration=timezone.now() + timedelta(seconds=expires_in),
331 owner=request.user,
332 defaults=defaults,
333 )
334 return HttpResponseRedirect(
335 f"{oauth_manager.oauth_redirect_url}?oauth_success=1&account_id={account.pk}",
336 )
337 except GetAccessTokenError:
338 logger.error("Error getting access token from OAuth provider")
339 return HttpResponseRedirect(
340 f"{oauth_manager.oauth_redirect_url}?oauth_success=0",
341 )