From 682662d36fd21645caca4a7e9ec93427db053609 Mon Sep 17 00:00:00 2001 From: red person Date: Tue, 3 Nov 2026 03:39:55 +0300 Subject: [PATCH] fix(email): decode MIME alternatives without duplicating bodies (#6953) --- mcp_servers/email_server.py | 27 ++++----- routes/email/email_helpers.py | 41 ++++++------- src/email_mime.py | 13 +++++ tests/test_email_mime_body_decoding.py | 80 ++++++++++++++++++++++++++ 4 files changed, 123 insertions(+), 38 deletions(-) create mode 100644 src/email_mime.py create mode 100644 tests/test_email_mime_body_decoding.py 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"", "\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"", "\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='Hello
world & all')) == 'Hello\nworld & all' + + +@pytest.mark.parametrize('extract', [route_text, mcp_text], ids=['route', 'mcp']) +def test_text_attachments_are_not_body_alternatives(extract): + msg = MIMEMultipart('mixed') + attachment = MIMEText('Attachment text', 'plain') + attachment.add_header('Content-Disposition', 'attachment', filename='note.txt') + msg.attach(attachment) + msg.attach(MIMEText('

Body 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\n

Hello 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'