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