fix(mail): robust MIME filename decoding with fallback for attachments
- Add _decode_mime_filename() helper with Q-encoding fallback decoder - Use helper for both new and existing email attachment saving - Handles edge cases where email.header.decode_header fails on partially processed filenames
This commit is contained in:
@@ -47,6 +47,36 @@ logger = logging.getLogger(__name__)
|
|||||||
MAX_ATTACHMENT_SIZE = 25 * 1024 * 1024 # 25 MB
|
MAX_ATTACHMENT_SIZE = 25 * 1024 * 1024 # 25 MB
|
||||||
|
|
||||||
|
|
||||||
|
def _decode_mime_filename(filename: str) -> str:
|
||||||
|
"""Decode MIME-encoded filename, handling =?charset?Q?...?= and =?charset?B?...?= patterns."""
|
||||||
|
if not filename:
|
||||||
|
return "attachment"
|
||||||
|
# If no MIME encoding pattern, return as-is
|
||||||
|
if "=?" not in filename:
|
||||||
|
return filename
|
||||||
|
try:
|
||||||
|
from email.header import decode_header, make_header
|
||||||
|
return str(make_header(decode_header(filename)))
|
||||||
|
except Exception:
|
||||||
|
# Fallback: manually decode Q-encoding if decode_header fails
|
||||||
|
# This handles cases where the email parser partially processed the filename
|
||||||
|
try:
|
||||||
|
import re
|
||||||
|
def decode_q(match):
|
||||||
|
charset, encoding, encoded = match.group(1), match.group(2).upper(), match.group(3)
|
||||||
|
if encoding == 'B':
|
||||||
|
import base64
|
||||||
|
decoded = base64.b64decode(encoded).decode(charset or 'utf-8', errors='replace')
|
||||||
|
else: # Q encoding
|
||||||
|
decoded = encoded.replace('_', ' ')
|
||||||
|
decoded = re.sub(r'=([0-9A-Fa-f]{2})', lambda m: chr(int(m.group(1), 16)), decoded)
|
||||||
|
decoded = decoded.encode('latin-1').decode(charset or 'utf-8', errors='replace')
|
||||||
|
return decoded
|
||||||
|
return re.sub(r'=\?([^?]+)\?([BbQq])\?([^?]*)\?=', decode_q, filename)
|
||||||
|
except Exception:
|
||||||
|
return filename
|
||||||
|
|
||||||
|
|
||||||
def _sanitize_filename(filename: str) -> str:
|
def _sanitize_filename(filename: str) -> str:
|
||||||
"""Sanitize a filename to prevent path traversal attacks."""
|
"""Sanitize a filename to prevent path traversal attacks."""
|
||||||
# Remove any path components — keep only the basename
|
# Remove any path components — keep only the basename
|
||||||
@@ -773,11 +803,7 @@ async def imap_sync_account(
|
|||||||
for att_data in attachments:
|
for att_data in attachments:
|
||||||
try:
|
try:
|
||||||
raw_filename = att_data["filename"] or "attachment"
|
raw_filename = att_data["filename"] or "attachment"
|
||||||
try:
|
decoded_filename = _decode_mime_filename(raw_filename)
|
||||||
from email.header import decode_header, make_header
|
|
||||||
decoded_filename = str(make_header(decode_header(raw_filename)))
|
|
||||||
except Exception:
|
|
||||||
decoded_filename = raw_filename
|
|
||||||
storage_path = await _save_attachment_to_storage(
|
storage_path = await _save_attachment_to_storage(
|
||||||
existing_mail.id, decoded_filename, att_data["content"]
|
existing_mail.id, decoded_filename, att_data["content"]
|
||||||
)
|
)
|
||||||
@@ -831,11 +857,7 @@ async def imap_sync_account(
|
|||||||
try:
|
try:
|
||||||
# Decode MIME-encoded filenames (e.g. =?utf-8?q?...?=)
|
# Decode MIME-encoded filenames (e.g. =?utf-8?q?...?=)
|
||||||
raw_filename = att_data["filename"] or "attachment"
|
raw_filename = att_data["filename"] or "attachment"
|
||||||
try:
|
decoded_filename = _decode_mime_filename(raw_filename)
|
||||||
from email.header import decode_header, make_header
|
|
||||||
decoded_filename = str(make_header(decode_header(raw_filename)))
|
|
||||||
except Exception:
|
|
||||||
decoded_filename = raw_filename
|
|
||||||
storage_path = await _save_attachment_to_storage(
|
storage_path = await _save_attachment_to_storage(
|
||||||
mail.id, decoded_filename, att_data["content"]
|
mail.id, decoded_filename, att_data["content"]
|
||||||
)
|
)
|
||||||
|
|||||||
Reference in New Issue
Block a user