import datetime import logging from datetime import timedelta from http import HTTPStatus from typing import Any from django.http import HttpResponseBadRequest from django.http import HttpResponseForbidden from django.http import HttpResponseRedirect from django.utils import timezone from django_filters.rest_framework import DjangoFilterBackend from drf_spectacular.types import OpenApiTypes from drf_spectacular.utils import extend_schema from drf_spectacular.utils import extend_schema_view from drf_spectacular.utils import inline_serializer from httpx_oauth.oauth2 import GetAccessTokenError 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 IsAuthenticated from rest_framework.response import Response from rest_framework.viewsets import ModelViewSet from rest_framework.viewsets import ReadOnlyModelViewSet from documents.filters import ObjectOwnedOrGrantedPermissionsFilter from documents.models import PaperlessTask from documents.permissions import PaperlessObjectPermissions from documents.permissions import has_perms_owner_aware from documents.views import PassUserMixin from paperless.views import StandardPagination from paperless_mail.filters import ProcessedMailFilterSet from paperless_mail.mail import MailError from paperless_mail.mail import get_mailbox from paperless_mail.mail import mailbox_login from paperless_mail.models import MailAccount from paperless_mail.models import MailRule from paperless_mail.models import ProcessedMail from paperless_mail.oauth import PaperlessMailOAuth2Manager from paperless_mail.serialisers import MailAccountSerializer from paperless_mail.serialisers import MailRuleSerializer from paperless_mail.serialisers import ProcessedMailSerializer from paperless_mail.tasks import process_mail_accounts @extend_schema_view( test=extend_schema( operation_id="mail_account_test", request=MailAccountSerializer, description="Test a mail account", responses={ 200: inline_serializer( name="MailAccountTestResponse", fields={"success": serializers.BooleanField()}, ), 400: OpenApiTypes.STR, }, ), process=extend_schema( operation_id="mail_account_process", description="Manually process the selected mail account for new messages.", responses={ 200: inline_serializer( name="MailAccountProcessResponse", fields={"result": serializers.CharField(default="OK")}, ), 404: None, }, ), ) class MailAccountViewSet(PassUserMixin, ModelViewSet[MailAccount]): model = MailAccount queryset = MailAccount.objects.all().order_by("pk") serializer_class = MailAccountSerializer pagination_class = StandardPagination permission_classes = (IsAuthenticated, PaperlessObjectPermissions) filter_backends = (ObjectOwnedOrGrantedPermissionsFilter,) def get_permissions(self): if self.action == "test": # Test action does not require object level permissions self.permission_classes = (IsAuthenticated,) return super().get_permissions() @action(methods=["post"], detail=False) def test(self, request): logger = logging.getLogger("paperless_mail") request.data["name"] = datetime.datetime.now().isoformat() serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) existing_account = None account_id = request.data.get("id") # testing a new connection requires add permission if account_id is None and not request.user.has_perms( ["paperless_mail.add_mailaccount"], ): return HttpResponseForbidden("Insufficient permissions") # testing an existing account requires change permission on that account if account_id is not None: try: existing_account = MailAccount.objects.get(pk=account_id) except (TypeError, ValueError, MailAccount.DoesNotExist): return HttpResponseForbidden("Insufficient permissions") if not has_perms_owner_aware( request.user, "change_mailaccount", existing_account, ): return HttpResponseForbidden("Insufficient permissions") # account exists, use the password from there instead of *** if ( len(serializer.validated_data.get("password").replace("*", "")) == 0 and existing_account is not None ): serializer.validated_data["password"] = existing_account.password serializer.validated_data["account_type"] = existing_account.account_type serializer.validated_data["refresh_token"] = existing_account.refresh_token serializer.validated_data["expiration"] = existing_account.expiration account = MailAccount(**serializer.validated_data) try: with get_mailbox( account.imap_server, account.imap_port, account.imap_security, ) as M: if ( existing_account is not None and account.is_token and account.expiration is not None and account.expiration < timezone.now() ): oauth_manager = PaperlessMailOAuth2Manager() if oauth_manager.refresh_account_oauth_token(existing_account): # User is not changing password and token needs to be refreshed existing_account.refresh_from_db() account.password = existing_account.password else: logger.error( "Mail account connectivity test failed: Unable to refresh oauth token", ) raise MailError("Unable to refresh oauth token") mailbox_login(M, account) return Response({"success": True}) except MailError: logger.error( "Mail account connectivity test failed", ) return HttpResponseBadRequest("Unable to connect to server") @action(methods=["post"], detail=True) def process(self, request, pk=None): account = self.get_object() process_mail_accounts.apply_async( kwargs={"account_ids": [account.pk]}, headers={"trigger_source": PaperlessTask.TriggerSource.MANUAL}, ) return Response({"result": "OK"}) @extend_schema_view( bulk_delete=extend_schema( operation_id="processed_mail_bulk_delete", description="Delete multiple processed mail records by ID.", request=inline_serializer( name="BulkDeleteMailRequest", fields={ "mail_ids": serializers.ListField(child=serializers.IntegerField()), }, ), responses={ (HTTPStatus.OK, "application/json"): inline_serializer( name="BulkDeleteMailResponse", fields={ "result": serializers.CharField(), "deleted_mail_ids": serializers.ListField( child=serializers.IntegerField(), ), }, ), HTTPStatus.BAD_REQUEST: None, HTTPStatus.FORBIDDEN: None, }, ), ) class ProcessedMailViewSet(PassUserMixin, ReadOnlyModelViewSet[ProcessedMail]): permission_classes = (IsAuthenticated, PaperlessObjectPermissions) serializer_class = ProcessedMailSerializer pagination_class = StandardPagination filter_backends = ( DjangoFilterBackend, OrderingFilter, ObjectOwnedOrGrantedPermissionsFilter, ) filterset_class = ProcessedMailFilterSet queryset = ProcessedMail.objects.all().order_by("-processed") @action(methods=["post"], detail=False) def bulk_delete(self, request): mail_ids = request.data.get("mail_ids", []) if not isinstance(mail_ids, list) or not all( isinstance(i, int) for i in mail_ids ): return HttpResponseBadRequest("mail_ids must be a list of integers") mails = ProcessedMail.objects.filter(id__in=mail_ids) for mail in mails: if not has_perms_owner_aware(request.user, "delete_processedmail", mail): return HttpResponseForbidden("Insufficient permissions") mail.delete() return Response({"result": "OK", "deleted_mail_ids": mail_ids}) class MailRuleViewSet(PassUserMixin, ModelViewSet[MailRule]): model = MailRule queryset = MailRule.objects.all().order_by("order") serializer_class = MailRuleSerializer pagination_class = StandardPagination permission_classes = (IsAuthenticated, PaperlessObjectPermissions) filter_backends = (ObjectOwnedOrGrantedPermissionsFilter,) @extend_schema_view( get=extend_schema( description="Callback view for OAuth2 authentication", responses={200: None}, ), ) class OauthCallbackView(GenericAPIView[Any]): permission_classes = (IsAuthenticated,) def get(self, request, format=None): if not ( request.user and request.user.has_perms(["paperless_mail.add_mailaccount"]) ): return HttpResponseBadRequest( "You do not have permission to add mail accounts", ) logger = logging.getLogger("paperless_mail") code = request.query_params.get("code") # Gmail passes scope as a query param, Outlook does not scope = request.query_params.get("scope") if code is None: logger.error( "Invalid oauth callback request: missing code", ) return HttpResponseBadRequest("Invalid request, see logs for more detail") oauth_manager = PaperlessMailOAuth2Manager( state=request.session.get("oauth_state"), ) state = request.query_params.get("state", "") if not oauth_manager.validate_state(state): logger.error( "Invalid oauth callback request: state validation failed", ) return HttpResponseBadRequest("Invalid request, see logs for more detail") try: if scope is not None and "google" in scope: # Google account_type = MailAccount.MailAccountType.GMAIL_OAUTH imap_server = "imap.gmail.com" defaults = { "name": f"Gmail OAuth {timezone.now()}", "username": "", "imap_security": MailAccount.ImapSecurity.SSL, "imap_port": 993, "account_type": account_type, } result = oauth_manager.get_gmail_access_token(code) elif scope is None: # Outlook account_type = MailAccount.MailAccountType.OUTLOOK_OAUTH imap_server = "outlook.office365.com" defaults = { "name": f"Outlook OAuth {timezone.now()}", "username": "", "imap_security": MailAccount.ImapSecurity.SSL, "imap_port": 993, "account_type": account_type, } result = oauth_manager.get_outlook_access_token(code) access_token = result["access_token"] refresh_token = result["refresh_token"] expires_in = result["expires_in"] account, _ = MailAccount.objects.update_or_create( password=access_token, is_token=True, imap_server=imap_server, refresh_token=refresh_token, expiration=timezone.now() + timedelta(seconds=expires_in), owner=request.user, defaults=defaults, ) return HttpResponseRedirect( f"{oauth_manager.oauth_redirect_url}?oauth_success=1&account_id={account.pk}", ) except GetAccessTokenError: logger.error("Error getting access token from OAuth provider") return HttpResponseRedirect( f"{oauth_manager.oauth_redirect_url}?oauth_success=0", )