From 5334d0d2a41801981040eae37344a1dd1b0d872c Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Fri, 19 Dec 2025 19:38:59 +0000 Subject: [PATCH] =?UTF-8?q?=F0=9F=9B=A1=EF=B8=8F=20Sentinel:=20Fix=20DoS?= =?UTF-8?q?=20risk=20and=20redact=20PII=20in=20IMAP=20handler?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Limit processed emails to 10 per cycle (MAX_EMAIL_LIMIT) to prevent DoS. - Redact PII (username, sender email) from debug logs. - Add test case for email limit. - Update Sentinel journal. --- .jules/sentinel.md | 6 ++++ .../satcom_forecast/imap_handler.py | 30 +++++++++++++++---- tests/test_imap_handler_pytest.py | 28 +++++++++++++++++ 3 files changed, 58 insertions(+), 6 deletions(-) create mode 100644 .jules/sentinel.md diff --git a/.jules/sentinel.md b/.jules/sentinel.md new file mode 100644 index 0000000..7310bb0 --- /dev/null +++ b/.jules/sentinel.md @@ -0,0 +1,6 @@ +## 2024-02-14 - [IMAP Handler Security Hardening] +**Vulnerability:** Unbounded IMAP email processing allowed potential Denial of Service (DoS) via mailbox flooding. Debug logs exposed PII (usernames, email addresses). +**Learning:** IMAP search results can be arbitrarily large. Processing them all in a loop without limits can hang the application. Standard logging of email headers can inadvertently leak PII. +**Prevention:** +1. Enforce strict limits on processed items (e.g., `MAX_EMAIL_LIMIT`). +2. Redact sensitive fields in logs (mask emails, remove passwords/usernames). diff --git a/custom_components/satcom_forecast/imap_handler.py b/custom_components/satcom_forecast/imap_handler.py index 1d2378e..5ce7c6c 100644 --- a/custom_components/satcom_forecast/imap_handler.py +++ b/custom_components/satcom_forecast/imap_handler.py @@ -11,6 +11,9 @@ _LOGGER = logging.getLogger(__name__) +# Limit the number of emails processed per run to prevent DoS/timeouts +MAX_EMAIL_LIMIT = 10 + async def check_imap_for_gps( host: str, @@ -21,11 +24,10 @@ async def check_imap_for_gps( security: str = "SSL", ) -> List[Dict[str, Any]]: _LOGGER.debug( - "Checking IMAP for GPS coordinates - Host: %s, Port: %s, User: %s, " + "Checking IMAP for GPS coordinates - Host: %s, Port: %s, User: ***, " "Folder: %s, Security: %s", host, port, - username, folder, security, ) @@ -80,7 +82,7 @@ def _check_imap_sync( mail = imaplib.IMAP4(host, port) _LOGGER.debug("IMAP unencrypted connection established successfully") - _LOGGER.debug("Attempting login for user: %s", username) + _LOGGER.debug("Attempting login for user: ***") mail.login(username, password) _LOGGER.debug("IMAP login successful") @@ -129,11 +131,21 @@ def _check_imap_sync( _LOGGER.error("Search error details: %s", messages) return [] - message_count = len(messages[0].split()) if messages and messages[0] else 0 + all_messages = messages[0].split() if messages and messages[0] else [] + message_count = len(all_messages) _LOGGER.debug("Found %d unread messages", message_count) + # Apply limit to prevent DoS + if message_count > MAX_EMAIL_LIMIT: + _LOGGER.warning( + "Too many messages (%d), processing only first %d to prevent overload", + message_count, + MAX_EMAIL_LIMIT, + ) + all_messages = all_messages[:MAX_EMAIL_LIMIT] + result: List[Dict[str, Any]] = [] - for num in messages[0].split(): + for num in all_messages: _LOGGER.debug("Processing message number: %s", num) typ, data = mail.fetch(num, "(RFC822)") # type: ignore[assignment] @@ -163,7 +175,13 @@ def _check_imap_sync( msg: Message = email.message_from_bytes(payload) sender_header = msg.get("From", "") from_email = parseaddr(sender_header)[1] if sender_header else "unknown" - _LOGGER.debug("Processing message from: %s", from_email) + # Redact email for privacy in logs + redacted_email = ( + f"{from_email[:3]}***@***{from_email.split('@')[-1]}" + if "@" in from_email + else "***" + ) + _LOGGER.debug("Processing message from: %s", redacted_email) body = "" if msg.is_multipart(): diff --git a/tests/test_imap_handler_pytest.py b/tests/test_imap_handler_pytest.py index c8657f2..ad4bc40 100644 --- a/tests/test_imap_handler_pytest.py +++ b/tests/test_imap_handler_pytest.py @@ -101,3 +101,31 @@ def test_successful_imap_operation(self): # Verify proper cleanup mock_connection.logout.assert_called_once() + + @pytest.mark.skipif(not HAS_HA, reason="Home Assistant not available") + def test_message_limit(self): + """Test that only MAX_EMAIL_LIMIT messages are processed.""" + from imap_handler import MAX_EMAIL_LIMIT + + with patch("imap_handler.imaplib.IMAP4_SSL") as mock_imap: + mock_connection = Mock() + mock_connection.login.return_value = None + mock_connection.select.return_value = ("OK", [b"1"]) + + # Create a list of messages exceeding the limit + messages_indices = [str(i).encode() for i in range(1, MAX_EMAIL_LIMIT + 5)] + messages_response = b" ".join(messages_indices) + mock_connection.search.return_value = ("OK", [messages_response]) + + # Mock fetch to avoid errors when processing + mock_connection.fetch.return_value = ( + "OK", + [(b"1 (RFC822 {100}", b"From: sender@example.com\r\n\r\nBody text")], + ) + + mock_imap.return_value = mock_connection + + asyncio.run(check_imap_for_gps("test.com", 993, "user", "pass")) + + # Verify that fetch was called exactly MAX_EMAIL_LIMIT times + assert mock_connection.fetch.call_count == MAX_EMAIL_LIMIT