From 3130fc3a7c13dba2b76bdc895dc99b5bd01be560 Mon Sep 17 00:00:00 2001 From: stumpylog <797416+stumpylog@users.noreply.github.com> Date: Tue, 22 Sep 2026 09:20:21 -0700 Subject: [PATCH] Use shared outbound resolution for IMAP host pinning get_mailbox validates the IMAP host with resolve_public_addresses and keeps its existing error messages; the pinned client dials typed addresses. Co-Authored-By: Claude Opus 5 --- src/paperless_mail/mail.py | 55 ++++++++++++-------- src/paperless_mail/tests/test_mail.py | 72 ++++++++++++++++++++++----- 2 files changed, 94 insertions(+), 33 deletions(-) diff --git a/src/paperless_mail/mail.py b/src/paperless_mail/mail.py index aa141bb8e..c237fc73c 100644 --- a/src/paperless_mail/mail.py +++ b/src/paperless_mail/mail.py @@ -1,6 +1,5 @@ import datetime import imaplib -import ipaddress import itertools import logging import socket @@ -46,8 +45,11 @@ from documents.models import Correspondent from documents.models import PaperlessTask 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.network import HostResolutionError +from paperless.network import IPAddress +from paperless.network import OutboundRequestBlockedError +from paperless.network import blocked_message +from paperless.network import resolve_public_addresses from paperless_mail.models import MailAccount from paperless_mail.models import MailRule from paperless_mail.models import ProcessedMail @@ -448,16 +450,24 @@ class PinnedIMAP4(imaplib.IMAP4): class, this behaves exactly like imaplib.IMAP4 / imaplib.IMAP4_SSL. """ - def __init__(self, host, port, pinned_ips, ssl_context=None, timeout=None) -> None: + def __init__( + self, + host: str, + port: int | None, + pinned_ips: tuple[IPAddress, ...] | None, + ssl_context: ssl.SSLContext | None = None, + timeout: float | None = None, + ) -> None: self._pinned_ips = pinned_ips self.ssl_context = ssl_context super().__init__(host, port, timeout=timeout) - def _connect_pinned(self, timeout): + def _connect_pinned(self, timeout: float | None) -> socket.socket: + assert self._pinned_ips is not None last_error: OSError | None = None - for ip_str in self._pinned_ips: + for ip in self._pinned_ips: try: - address = (ip_str, self.port) + address = (str(ip), self.port) if timeout is not None: return socket.create_connection(address, timeout) return socket.create_connection(address) @@ -465,8 +475,8 @@ class PinnedIMAP4(imaplib.IMAP4): last_error = e raise last_error or OSError(f"Could not connect to {self.host}") - def _create_socket(self, timeout): - if self._pinned_ips: + def _create_socket(self, timeout: float | None) -> socket.socket: + if self._pinned_ips is not None: sock = self._connect_pinned(timeout) else: sock = super()._create_socket(timeout) @@ -478,7 +488,12 @@ class PinnedIMAP4(imaplib.IMAP4): class PinnedClientMixin: """Builds the imaplib client against the pre-resolved addresses, if any.""" - def __init__(self, *args, pinned_ips: list[str] | None, **kwargs) -> None: + def __init__( + self, + *args, + pinned_ips: tuple[IPAddress, ...] | None, + **kwargs, + ) -> None: self._pinned_ips = pinned_ips super().__init__(*args, **kwargs) @@ -516,22 +531,20 @@ class PinnedMailBoxStartTls(PinnedClientMixin, MailBoxStartTls): return client -def get_mailbox(server, port, security) -> MailBox: +def get_mailbox( + server: str, + port: int, + security: MailAccount.ImapSecurity, +) -> MailBox: """ Returns the correct MailBox instance for the given configuration. """ - pinned_ips: list[str] | None = None + pinned_ips: tuple[IPAddress, ...] | None = None if not settings.EMAIL_ALLOW_INTERNAL_HOSTS: try: - pinned_ips = resolve_hostname_ips(server) - except ValueError as e: - raise MailError(str(e)) from e - - for ip_str in pinned_ips: - if not is_public_ip(ipaddress.ip_address(ip_str)): - raise MailError( - f"Connection blocked: {server} resolves to a non-public address", - ) + pinned_ips = resolve_public_addresses(server, port) + except (OutboundRequestBlockedError, HostResolutionError) as e: + raise MailError(blocked_message(e)) from e ssl_context = ssl.create_default_context() if settings.EMAIL_CERTIFICATE_FILE is not None: # pragma: no cover diff --git a/src/paperless_mail/tests/test_mail.py b/src/paperless_mail/tests/test_mail.py index c249a9d62..b55bfb1d5 100644 --- a/src/paperless_mail/tests/test_mail.py +++ b/src/paperless_mail/tests/test_mail.py @@ -1,9 +1,12 @@ import dataclasses +import ipaddress +import socket import time import uuid from collections import namedtuple from datetime import timedelta from unittest import mock +from unittest.mock import MagicMock import pytest from django.contrib.auth.models import Permission @@ -25,6 +28,7 @@ from documents.models import MatchingModel from paperless_mail import tasks from paperless_mail.mail import MailAccountHandler from paperless_mail.mail import MailError +from paperless_mail.mail import PinnedIMAP4 from paperless_mail.mail import TagMailAction from paperless_mail.mail import apply_mail_action from paperless_mail.mail import error_callback @@ -2045,10 +2049,13 @@ class TestMailAccountTestView(APITestCase): 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"]) + @mock.patch( + "paperless.network._getaddrinfo", + return_value=[(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 993))], + ) def test_mail_account_test_view_blocks_internal_host_when_disabled( self, - _mock_resolve_hostname_ips, + _mock_getaddrinfo: MagicMock, ) -> None: data = { "imap_server": "internal.example", @@ -2205,10 +2212,10 @@ class TestGetMailboxHostPinning(TestCase): @override_settings(EMAIL_ALLOW_INTERNAL_HOSTS=False) @mock.patch( - "paperless_mail.mail.resolve_hostname_ips", - return_value=["93.184.216.34"], + "paperless_mail.mail.resolve_public_addresses", + return_value=(ipaddress.ip_address("93.184.216.34"),), ) - def test_connects_to_validated_ip(self, _mock_resolve) -> None: + def test_connects_to_validated_ip(self, _mock_resolve: MagicMock) -> None: with mock.patch( "paperless_mail.mail.socket.create_connection", side_effect=OSError("no connection in tests"), @@ -2225,10 +2232,13 @@ class TestGetMailboxHostPinning(TestCase): @override_settings(EMAIL_ALLOW_INTERNAL_HOSTS=False) @mock.patch( - "paperless_mail.mail.resolve_hostname_ips", - return_value=["93.184.216.34"], + "paperless_mail.mail.resolve_public_addresses", + return_value=(ipaddress.ip_address("93.184.216.34"),), ) - def test_ssl_pins_ip_but_keeps_hostname_for_sni(self, _mock_resolve) -> None: + def test_ssl_pins_ip_but_keeps_hostname_for_sni( + self, + _mock_resolve: MagicMock, + ) -> None: ssl_context = mock.MagicMock() ssl_context.wrap_socket.return_value.makefile.side_effect = OSError( "no connection in tests", @@ -2259,13 +2269,51 @@ class TestGetMailboxHostPinning(TestCase): @override_settings(EMAIL_ALLOW_INTERNAL_HOSTS=False) @mock.patch( - "paperless_mail.mail.resolve_hostname_ips", - return_value=["93.184.216.34", "127.0.0.1"], + "paperless.network._getaddrinfo", + return_value=[ + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 993)), + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 993)), + ], ) - def test_blocks_when_any_resolved_address_is_internal(self, _mock_resolve) -> None: - with self.assertRaises(MailError): + def test_blocks_when_any_resolved_address_is_internal( + self, + _mock_resolve: MagicMock, + ) -> None: + """ + GIVEN: + - A mail host resolving to one public and one loopback address + - EMAIL_ALLOW_INTERNAL_HOSTS is False + WHEN: + - A mailbox is requested + THEN: + - The whole host is blocked with the existing message + """ + with self.assertRaisesMessage( + MailError, + "Connection blocked: mail.example.com resolves to a non-public address", + ): get_mailbox("mail.example.com", 993, MailAccount.ImapSecurity.SSL) + def test_empty_pin_list_never_falls_back_to_hostname_lookup(self) -> None: + """ + GIVEN: + - A pinned IMAP client given an empty tuple of addresses + WHEN: + - It connects + THEN: + - It fails without opening any socket, rather than resolving the + hostname itself + """ + with ( + mock.patch("paperless_mail.mail.socket.create_connection") as pinned, + mock.patch("imaplib.IMAP4._create_socket") as unpinned, + self.assertRaises(OSError), + ): + PinnedIMAP4("mail.example.com", 143, ()) + + pinned.assert_not_called() + unpinned.assert_not_called() + class TestMailAccountProcess(APITestCase): def setUp(self) -> None: