Enhancement: allow opt-in blocking internal mail hosts

docs/configuration.md

@@ -1420,6 +1420,14 @@ ports.
 
 ## Incoming Mail {#incoming_mail}
 
+#### [`PAPERLESS_EMAIL_ALLOW_INTERNAL_HOSTS=<bool>`](#PAPERLESS_EMAIL_ALLOW_INTERNAL_HOSTS) {#PAPERLESS_EMAIL_ALLOW_INTERNAL_HOSTS}
+
+: If set to false, incoming mail account connections are blocked when the
+configured IMAP hostname resolves to a non-public address (for example,
+localhost, link-local, or RFC1918 private ranges).
+
+    Defaults to true, which allows internal hosts.
+
 ### Email OAuth {#email_oauth}
 
 #### [`PAPERLESS_OAUTH_CALLBACK_BASE_URL=<str>`](#PAPERLESS_OAUTH_CALLBACK_BASE_URL) {#PAPERLESS_OAUTH_CALLBACK_BASE_URL}

src/paperless/settings/__init__.py

@@ -497,6 +497,10 @@ SESSION_COOKIE_NAME = f"{COOKIE_PREFIX}sessionid"
 LANGUAGE_COOKIE_NAME = f"{COOKIE_PREFIX}django_language"
 
 EMAIL_CERTIFICATE_FILE = get_path_from_env("PAPERLESS_EMAIL_CERTIFICATE_LOCATION")
+EMAIL_ALLOW_INTERNAL_HOSTS = get_bool_from_env(
+    "PAPERLESS_EMAIL_ALLOW_INTERNAL_HOSTS",
+    "true",
+)
 
 
 ###############################################################################

src/paperless_mail/mail.py

@@ -39,6 +39,8 @@ from documents.loggers import LoggingMixin
 from documents.models import Correspondent
 from documents.parsers import is_mime_type_supported
 from documents.tasks import consume_file
+from paperless.network import is_public_ip
+from paperless.network import resolve_hostname_ips
 from paperless_mail.models import MailAccount
 from paperless_mail.models import MailRule
 from paperless_mail.models import ProcessedMail
@@ -412,6 +414,13 @@ def get_mailbox(server, port, security) -> MailBox:
     """
     Returns the correct MailBox instance for the given configuration.
     """
+    if not settings.EMAIL_ALLOW_INTERNAL_HOSTS:
+        for ip_str in resolve_hostname_ips(server):
+            if not is_public_ip(ip_str):
+                raise MailError(
+                    f"Connection blocked: {server} resolves to a non-public address",
+                )
+
     ssl_context = ssl.create_default_context()
     if settings.EMAIL_CERTIFICATE_FILE is not None:  # pragma: no cover
         ssl_context.load_verify_locations(cafile=settings.EMAIL_CERTIFICATE_FILE)

src/paperless_mail/tests/test_mail.py

@@ -13,6 +13,7 @@ from django.contrib.auth.models import User
 from django.core.management import call_command
 from django.db import DatabaseError
 from django.test import TestCase
+from django.test import override_settings
 from django.utils import timezone
 from imap_tools import NOT
 from imap_tools import EmailAddress
@@ -1846,6 +1847,25 @@ class TestMailAccountTestView(APITestCase):
         self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
         self.assertEqual(response.content.decode(), "Unable to connect to server")
 
+    @override_settings(EMAIL_ALLOW_INTERNAL_HOSTS=False)
+    @mock.patch("paperless_mail.mail.resolve_hostname_ips", return_value=["127.0.0.1"])
+    def test_mail_account_test_view_blocks_internal_host_when_disabled(
+        self,
+        _mock_resolve_hostname_ips,
+    ) -> None:
+        data = {
+            "imap_server": "internal.example",
+            "imap_port": 993,
+            "imap_security": MailAccount.ImapSecurity.SSL,
+            "username": "admin",
+            "password": "secret",
+            "account_type": MailAccount.MailAccountType.IMAP,
+            "is_token": False,
+        }
+        response = self.client.post(self.url, data, format="json")
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(response.content.decode(), "Unable to connect to server")
+
     @mock.patch(
         "paperless_mail.oauth.PaperlessMailOAuth2Manager.refresh_account_oauth_token",
     )

src/paperless_mail/views.py

@@ -120,12 +120,12 @@ class MailAccountViewSet(ModelViewSet, PassUserMixin):
             serializer.validated_data["expiration"] = existing_account.expiration
 
         account = MailAccount(**serializer.validated_data)
-        with get_mailbox(
-            account.imap_server,
-            account.imap_port,
-            account.imap_security,
-        ) as M:
-            try:
+        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
@@ -145,11 +145,11 @@ class MailAccountViewSet(ModelViewSet, PassUserMixin):
 
                 mailbox_login(M, account)
                 return Response({"success": True})
-            except MailError:
-                logger.error(
-                    "Mail account connectivity test failed",
-                )
-                return HttpResponseBadRequest("Unable to connect to server")
+        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):