🛡️ Sentinel: Fix DoS risk and redact PII in IMAP handler

.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).

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():

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