"""Attachment storage helpers for the Mail plugin. Extracted from services.py as part of the God-object split (BUG-018 pilot). Re-exported by ``app.plugins.builtins.mail.services``. """ from __future__ import annotations import os import re import uuid import aiofiles from app.config import settings from app.plugins.builtins.mail.models import MailAttachment # ─── Attachment Storage Helpers ─── 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 base64 def decode_q(match): charset, encoding, encoded = ( match.group(1), match.group(2).upper(), match.group(3), ) if encoding == "B": 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: """Sanitize a filename to prevent path traversal attacks.""" # Remove any path components — keep only the basename filename = os.path.basename(filename or "attachment") # Replace potentially dangerous characters filename = re.sub(r"[^a-zA-Z0-9._-]", "_", filename) # Ensure non-empty if not filename: filename = "attachment" # Limit length if len(filename) > 200: name, ext = os.path.splitext(filename) filename = name[:200 - len(ext)] + ext return filename def _attachment_storage_path(mail_id: uuid.UUID, filename: str) -> str: """Build the on-disk storage path for a mail attachment.""" safe_name = _sanitize_filename(filename) return os.path.join( settings.storage_path, "mail_attachments", str(mail_id), safe_name, ) async def _save_attachment_to_storage( mail_id: uuid.UUID, filename: str, content: bytes ) -> str: """Save attachment content to disk and return the storage path.""" storage_path = _attachment_storage_path(mail_id, filename) os.makedirs(os.path.dirname(storage_path), exist_ok=True) async with aiofiles.open(storage_path, "wb") as f: await f.write(content) return storage_path def attachment_to_response(att: MailAttachment) -> dict: """Convert a MailAttachment ORM object to a response dict.""" return { "id": str(att.id), "mail_id": str(att.mail_id), "filename": att.filename, "mime_type": att.mime_type, "size_bytes": att.size_bytes, "size": att.size_bytes, # alias for frontend compatibility "content_id": att.content_id, "is_inline": bool(att.content_id), "dms_file_id": str(att.dms_file_id) if att.dms_file_id else None, }