fix(blocks): validate email recipients in Gmail blocks before API call

Addresses #11954 — GmailSendBlock crashes with an opaque "Invalid To
header" HttpError 400 when the LLM (or user) supplies a malformed
recipient such as a bare username, a JSON string, or an empty value.

Add a lightweight `validate_email_recipients()` check in the shared
`create_mime_message()` path and in `_build_reply_message()` so that
every Gmail block that sends or drafts email gets upfront validation
with a clear, actionable error message listing the invalid entries.
This commit is contained in:
Krishna Chaitanya Balusu
2026-03-24 23:17:38 -04:00
parent 550fa5a319
commit aa749c347d
2 changed files with 78 additions and 1 deletions

View File

@@ -1,5 +1,6 @@
import asyncio
import base64
import re
from abc import ABC
from email import encoders
from email.mime.base import MIMEBase
@@ -46,6 +47,27 @@ def serialize_email_recipients(recipients: list[str]) -> str:
return ", ".join(recipients)
# RFC 5322 simplified pattern: local@domain where domain has at least one dot
_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
def validate_email_recipients(
recipients: list[str], field_name: str = "to"
) -> None:
"""Validate that all recipients are plausible email addresses.
Raises ``ValueError`` with a user-friendly message listing every
invalid entry so the caller (or LLM) can correct them in one pass.
"""
invalid = [addr for addr in recipients if not _EMAIL_RE.match(addr.strip())]
if invalid:
formatted = ", ".join(f"'{a}'" for a in invalid)
raise ValueError(
f"Invalid email address(es) in '{field_name}': {formatted}. "
f"Each entry must be a valid email address (e.g. user@example.com)."
)
def _make_mime_text(
body: str,
content_type: Optional[Literal["auto", "plain", "html"]] = None,
@@ -100,6 +122,13 @@ async def create_mime_message(
) -> str:
"""Create a MIME message with attachments and return base64-encoded raw message."""
# Validate all recipient lists before building the MIME message
validate_email_recipients(input_data.to, "to")
if input_data.cc:
validate_email_recipients(input_data.cc, "cc")
if input_data.bcc:
validate_email_recipients(input_data.bcc, "bcc")
message = MIMEMultipart()
message["to"] = serialize_email_recipients(input_data.to)
message["subject"] = input_data.subject
@@ -1167,6 +1196,14 @@ async def _build_reply_message(
references.append(headers["message-id"])
# Create MIME message
# Validate all recipient lists before building the MIME message
if input_data.to:
validate_email_recipients(input_data.to, "to")
if input_data.cc:
validate_email_recipients(input_data.cc, "cc")
if input_data.bcc:
validate_email_recipients(input_data.bcc, "bcc")
msg = MIMEMultipart()
if input_data.to:
msg["To"] = ", ".join(input_data.to)

View File

@@ -3,7 +3,7 @@ from unittest.mock import Mock, patch
import pytest
from backend.blocks.google.gmail import GmailReadBlock
from backend.blocks.google.gmail import GmailReadBlock, validate_email_recipients
class TestGmailReadBlock:
@@ -250,3 +250,43 @@ class TestGmailReadBlock:
result = await self.gmail_block._get_email_body(msg, self.mock_service)
assert result == "This email does not contain a readable body."
class TestValidateEmailRecipients:
"""Test cases for validate_email_recipients."""
def test_valid_single_email(self):
validate_email_recipients(["user@example.com"])
def test_valid_multiple_emails(self):
validate_email_recipients(["a@b.com", "x@y.org", "test@sub.domain.co"])
def test_invalid_missing_at(self):
with pytest.raises(ValueError, match="Invalid email address"):
validate_email_recipients(["not-an-email"])
def test_invalid_missing_domain_dot(self):
with pytest.raises(ValueError, match="Invalid email address"):
validate_email_recipients(["user@localhost"])
def test_invalid_empty_string(self):
with pytest.raises(ValueError, match="Invalid email address"):
validate_email_recipients([""])
def test_invalid_json_object_string(self):
with pytest.raises(ValueError, match="Invalid email address"):
validate_email_recipients(['{"email": "user@example.com"}'])
def test_mixed_valid_and_invalid(self):
with pytest.raises(ValueError, match="'bad-addr'"):
validate_email_recipients(["good@example.com", "bad-addr"])
def test_field_name_in_error(self):
with pytest.raises(ValueError, match="'cc'"):
validate_email_recipients(["nope"], field_name="cc")
def test_whitespace_trimmed(self):
validate_email_recipients([" user@example.com "])
def test_empty_list_passes(self):
validate_email_recipients([])