diff --git a/mcp_servers/email_server.py b/mcp_servers/email_server.py
index b43eeca87..227c89939 100644
--- a/mcp_servers/email_server.py
+++ b/mcp_servers/email_server.py
@@ -33,6 +33,7 @@ from mcp.types import Tool, TextContent
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+from src.email_mime import decode_text_part as _decode_text_part
server = Server("email")
EMAIL_SOCKET_TIMEOUT = float(os.environ.get("EMAIL_SOCKET_TIMEOUT", "20"))
from src.constants import DATA_DIR as _DATA_DIR, APP_DB, EMAIL_CACHE_DB, SETTINGS_FILE as _SETTINGS_FILE, MAIL_ATTACHMENTS_DIR
@@ -923,29 +924,25 @@ def _extract_text(msg):
"""Extract plain text body from email message."""
if msg.is_multipart():
text_parts = []
+ html_parts = []
for part in msg.walk():
ct = part.get_content_type()
cd = str(part.get("Content-Disposition", ""))
if ct == "text/plain" and "attachment" not in cd:
- payload = part.get_payload(decode=True)
- if payload:
- charset = part.get_content_charset() or "utf-8"
- text_parts.append(payload.decode(charset, errors="replace"))
- elif ct == "text/html" and not text_parts and "attachment" not in cd:
- payload = part.get_payload(decode=True)
- if payload:
- charset = part.get_content_charset() or "utf-8"
- raw_html = payload.decode(charset, errors="replace")
+ text = _decode_text_part(part)
+ if text:
+ text_parts.append(text)
+ elif ct == "text/html" and not html_parts and "attachment" not in cd:
+ raw_html = _decode_text_part(part)
+ if raw_html:
text = re.sub(r"
", "\n", raw_html, flags=re.I)
text = re.sub(r"<[^>]+>", "", text)
text = html.unescape(text)
- text_parts.append(text.strip())
- return "\n".join(text_parts)
+ html_parts.append(text.strip())
+ return "\n".join(text_parts or html_parts)
else:
- payload = msg.get_payload(decode=True)
- if payload:
- charset = msg.get_content_charset() or "utf-8"
- text = payload.decode(charset, errors="replace")
+ text = _decode_text_part(msg)
+ if text:
if msg.get_content_type() == "text/html":
text = re.sub(r"
", "\n", text, flags=re.I)
text = re.sub(r"(?:p|div|li|tr|h[1-6])\s*>", "\n", text, flags=re.I)
diff --git a/routes/email/email_helpers.py b/routes/email/email_helpers.py
index d697b2fe7..d6987f7b2 100644
--- a/routes/email/email_helpers.py
+++ b/routes/email/email_helpers.py
@@ -36,6 +36,7 @@ from fastapi import Query, HTTPException, Request
from pydantic import BaseModel
from typing import Optional, List
+from src.email_mime import decode_text_part as _decode_text_part
from src.auth_helpers import _auth_disabled, get_current_user
from src.secret_storage import decrypt as _decrypt
from src.url_safety import OutboundAddressBlocked, connect_outbound_tcp
@@ -1776,44 +1777,38 @@ def _extract_html(msg):
ct = part.get_content_type()
cd = str(part.get("Content-Disposition", ""))
if ct == "text/html" and "attachment" not in cd:
- payload = part.get_payload(decode=True)
- if payload:
- charset = part.get_content_charset() or "utf-8"
- return payload.decode(charset, errors="replace")
+ text = _decode_text_part(part)
+ if text:
+ return text
elif msg.get_content_type() == "text/html":
- payload = msg.get_payload(decode=True)
- if payload:
- charset = msg.get_content_charset() or "utf-8"
- return payload.decode(charset, errors="replace")
+ text = _decode_text_part(msg)
+ if text:
+ return text
return None
def _extract_text(msg):
if msg.is_multipart():
text_parts = []
+ html_parts = []
for part in msg.walk():
ct = part.get_content_type()
cd = str(part.get("Content-Disposition", ""))
if ct == "text/plain" and "attachment" not in cd:
- payload = part.get_payload(decode=True)
- if payload:
- charset = part.get_content_charset() or "utf-8"
- text_parts.append(payload.decode(charset, errors="replace"))
- elif ct == "text/html" and not text_parts and "attachment" not in cd:
- payload = part.get_payload(decode=True)
- if payload:
- charset = part.get_content_charset() or "utf-8"
- raw_html = payload.decode(charset, errors="replace")
+ text = _decode_text_part(part)
+ if text:
+ text_parts.append(text)
+ elif ct == "text/html" and not html_parts and "attachment" not in cd:
+ raw_html = _decode_text_part(part)
+ if raw_html:
text = re.sub(r"
", "\n", raw_html, flags=re.I)
text = re.sub(r"<[^>]+>", "", text)
text = html.unescape(text)
- text_parts.append(text.strip())
- return "\n".join(text_parts)
+ html_parts.append(text.strip())
+ return "\n".join(text_parts or html_parts)
else:
- payload = msg.get_payload(decode=True)
- if payload:
- charset = msg.get_content_charset() or "utf-8"
- text = payload.decode(charset, errors="replace")
+ text = _decode_text_part(msg)
+ if text:
if msg.get_content_type() == "text/html":
text = re.sub(r"
", "\n", text, flags=re.I)
text = re.sub(r"(?:p|div|li|tr|h[1-6])\s*>", "\n", text, flags=re.I)
diff --git a/src/email_mime.py b/src/email_mime.py
new file mode 100644
index 000000000..66945437d
--- /dev/null
+++ b/src/email_mime.py
@@ -0,0 +1,13 @@
+"""Shared MIME text decoding for the web email helpers and MCP server."""
+
+
+def decode_text_part(part) -> str:
+ payload = part.get_payload(decode=True)
+ if not payload:
+ return ""
+ charset = part.get_content_charset() or "utf-8"
+ try:
+ return payload.decode(charset, errors="replace")
+ except (LookupError, ValueError):
+ # Broken senders can label valid UTF-8 with an unsupported codec.
+ return payload.decode("utf-8", errors="replace")
diff --git a/tests/test_email_mime_body_decoding.py b/tests/test_email_mime_body_decoding.py
new file mode 100644
index 000000000..ea90663ad
--- /dev/null
+++ b/tests/test_email_mime_body_decoding.py
@@ -0,0 +1,80 @@
+"""MIME alternatives are selected independently of part order and codec labels."""
+from email import message_from_bytes
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+
+import pytest
+
+from mcp_servers.email_server import _extract_text as mcp_text
+from routes.email_helpers import _extract_html, _extract_text as route_text
+
+
+def alternative(order, plain='Body text', html='
Body text
'): + msg = MIMEMultipart('alternative') + parts = {'plain': MIMEText(plain, 'plain', 'utf-8'), 'html': MIMEText(html, 'html', 'utf-8')} + for kind in order: + msg.attach(parts[kind]) + return message_from_bytes(msg.as_bytes()) + + +@pytest.mark.parametrize('extract', [route_text, mcp_text], ids=['route', 'mcp']) +@pytest.mark.parametrize('order', [('html', 'plain'), ('plain', 'html')]) +def test_plain_alternative_wins_in_either_order(extract, order): + assert extract(alternative(order)) == 'Body text' + + +@pytest.mark.parametrize('extract', [route_text, mcp_text], ids=['route', 'mcp']) +def test_empty_plain_allows_html_fallback(extract): + assert extract(alternative(('plain', 'html'), plain='', html='HelloBody text
', 'html')) + assert extract(msg) == 'Body text' + + +@pytest.mark.parametrize('extract', [route_text, mcp_text], ids=['route', 'mcp']) +def test_multiple_plain_body_parts_remain_joined(extract): + msg = MIMEMultipart('mixed') + msg.attach(MIMEText('Part one', 'plain')) + msg.attach(MIMEText('Part two', 'plain')) + assert extract(msg) == 'Part one\nPart two' + + +@pytest.mark.parametrize('extract', [route_text, mcp_text], ids=['route', 'mcp']) +@pytest.mark.parametrize('kind', ['plain', 'html']) +@pytest.mark.parametrize('multipart', [False, True]) +def test_unknown_charset_falls_back_to_utf8(extract, kind, multipart): + body = 'Hello café' if kind == 'plain' else 'Hello café
' + raw = f'Content-Type: text/{kind}; charset=x-unknown\r\nContent-Transfer-Encoding: 8bit\r\n\r\n'.encode()+body.encode() + part = message_from_bytes(raw) + if multipart: + msg = MIMEMultipart('mixed');msg.attach(part) + else: + msg = part + assert extract(msg) == 'Hello café' + + +@pytest.mark.parametrize('extract', [route_text, mcp_text], ids=['route', 'mcp']) +def test_declared_latin1_is_preserved(extract): + assert extract(MIMEText('café', 'plain', 'iso-8859-1')) == 'café' + + +@pytest.mark.parametrize('multipart', [False, True]) +def test_raw_html_unknown_charset_is_readable(multipart): + part = message_from_bytes(b'Content-Type: text/html; charset=x-unknown\r\n\r\nHello caf\xc3\xa9
') + if multipart: + msg=MIMEMultipart('mixed');msg.attach(part) + else:msg=part + assert _extract_html(msg) == 'Hello café
' + + +@pytest.mark.parametrize('extract', [route_text, mcp_text], ids=['route', 'mcp']) +def test_invalid_bytes_are_replaced_in_unknown_charset(extract): + msg=message_from_bytes(b'Content-Type: text/plain; charset=x-unknown\r\n\r\nHello\xff') + assert extract(msg) == 'Hello\ufffd'