mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-11-03 03:32:25 +01:00
fix(email): decode MIME alternatives without duplicating bodies (#6953)
This commit is contained in:
red person
+12
-15
@@ -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"<br\s*/?>", "\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"<br\s*/?>", "\n", text, flags=re.I)
|
||||
text = re.sub(r"</(?:p|div|li|tr|h[1-6])\s*>", "\n", text, flags=re.I)
|
||||
|
||||
@@ -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"<br\s*/?>", "\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"<br\s*/?>", "\n", text, flags=re.I)
|
||||
text = re.sub(r"</(?:p|div|li|tr|h[1-6])\s*>", "\n", text, flags=re.I)
|
||||
|
||||
@@ -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")
|
||||
@@ -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='<p>Body text</p>'):
|
||||
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<br>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('<p>Body text</p>', '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 '<p>Hello café</p>'
|
||||
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<p>Hello caf\xc3\xa9</p>')
|
||||
if multipart:
|
||||
msg=MIMEMultipart('mixed');msg.attach(part)
|
||||
else:msg=part
|
||||
assert _extract_html(msg) == '<p>Hello café</p>'
|
||||
|
||||
|
||||
@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'
|
||||
Reference in New Issue
Block a user