Harden mail connector network boundaries

This commit is contained in:
2026-07-21 12:10:32 +02:00
parent b2c013174a
commit e67da95df6
5 changed files with 209 additions and 12 deletions

View File

@@ -3,11 +3,16 @@ from __future__ import annotations
import unittest
from unittest.mock import patch
from govoplan_core.security.outbound_http import OutboundHttpBlocked
from govoplan_mail.backend.config import ImapConfig
from govoplan_mail.backend.sending.imap import (
ImapAppendError,
ImapConfigurationError,
_detect_sent_folder,
_extract_mailbox_name,
_fetch_message_by_uid,
_normalize_mailbox_page,
_open_imap,
_paged_descending_sequences,
_parse_fetch_sequence,
_select_readonly,
@@ -17,6 +22,31 @@ from govoplan_mail.backend.sending.imap import (
class ImapFolderParserTests(unittest.TestCase):
def test_real_imap_connections_honor_deployment_egress_policy(self):
config = ImapConfig(host="imap.internal", port=993, security="tls")
with patch(
"govoplan_mail.backend.sending.imap.validate_outbound_host",
side_effect=OutboundHttpBlocked("private network blocked"),
), self.assertRaisesRegex(ImapConfigurationError, "private network blocked"):
_open_imap(config)
def test_imap_revalidates_and_pins_at_connection_time(self):
config = ImapConfig(host="imap.example.test", port=993, security="tls")
public = [(2, 1, 6, "", ("93.184.216.34", 993))]
private = [(2, 1, 6, "", ("127.0.0.1", 993))]
with patch.dict(
"os.environ",
{"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "false"},
), patch(
"govoplan_core.security.outbound_http.socket.getaddrinfo",
side_effect=(public, private),
), patch("govoplan_core.security.outbound_http.socket.socket") as socket_factory, self.assertRaisesRegex(
OutboundHttpBlocked,
"non-public network",
):
_open_imap(config)
socket_factory.assert_not_called()
def test_extracts_quoted_mailbox_after_quoted_delimiter(self):
self.assertEqual(
_extract_mailbox_name(b'(\\HasNoChildren \\Sent) "/" "Sent Items"'),
@@ -49,6 +79,23 @@ class ImapFolderParserTests(unittest.TestCase):
class ImapMessagePaginationTests(unittest.TestCase):
def test_full_message_fetch_uses_partial_range_and_deployment_limit(self):
class Client:
command = ""
def uid(self, command, uid, fetch_spec):
del command, uid
self.command = fetch_spec
return "OK", [(b"1 (UID 1 RFC822.SIZE 11)", b"12345678901")]
client = Client()
with patch.dict("os.environ", {"GOVOPLAN_CONNECTOR_MAX_FILE_TRANSFER_BYTES": "10"}), self.assertRaisesRegex(
ImapAppendError,
"deployment limit",
):
_fetch_message_by_uid(client, "1")
self.assertIn("BODY.PEEK[]<0.11>", client.command)
def test_paginates_sequence_numbers_newest_first(self):
self.assertEqual(
_paged_descending_sequences(120, offset=0, limit=5),

View File

@@ -1,16 +1,44 @@
from __future__ import annotations
import unittest
from unittest.mock import patch
from govoplan_core.security.outbound_http import OutboundHttpBlocked
from govoplan_mail.backend.config import SmtpConfig
from govoplan_mail.backend.sending.smtp import (
SmtpConfigurationError,
_open_smtp,
_prepare_smtp_send,
_smtp_send_result,
)
class SmtpSendHelperTests(unittest.TestCase):
def test_real_smtp_connections_honor_deployment_egress_policy(self):
config = SmtpConfig(host="smtp.internal", port=587, security="starttls")
with patch(
"govoplan_mail.backend.sending.smtp.validate_outbound_host",
side_effect=OutboundHttpBlocked("private network blocked"),
), self.assertRaisesRegex(SmtpConfigurationError, "private network blocked"):
_open_smtp(config)
def test_smtp_revalidates_and_pins_at_connection_time(self):
config = SmtpConfig(host="smtp.example.test", port=587, security="starttls")
public = [(2, 1, 6, "", ("93.184.216.34", 587))]
private = [(2, 1, 6, "", ("127.0.0.1", 587))]
with patch.dict(
"os.environ",
{"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "false"},
), patch(
"govoplan_core.security.outbound_http.socket.getaddrinfo",
side_effect=(public, private),
), patch("govoplan_core.security.outbound_http.socket.socket") as socket_factory, self.assertRaisesRegex(
OutboundHttpBlocked,
"non-public network",
):
_open_smtp(config)
socket_factory.assert_not_called()
def test_prepare_smtp_send_validates_envelope(self):
config = SmtpConfig(host="smtp.example.org", port=587, security="starttls")
with self.assertRaisesRegex(SmtpConfigurationError, "envelope sender"):
@@ -45,4 +73,3 @@ class SmtpSendHelperTests(unittest.TestCase):
if __name__ == "__main__":
unittest.main()