Agent Zero
6702d69f7c
IMAP Sync Block (F-MAIL-01) aus services.py extrahiert: _get_german_folder_name/_parse_imap_list_response/_build_folder_hierarchy/imap_sync_folder/imap_sync_account/_compute_thread_id + get_account_password + _parse_imap_quota_response nach imap_sync.py (1148 Z.). services.py jetzt ~1680 Z. (von 3087). Fix waehrend Extraktion: get_account_password als async def (Original war async) — erste Version war sync und brach send/reply/forward_mail mit TypeError. Beweis: mail+sig_label_routes 51/51 passed in 106.60s; ruff clean.
118 lines
3.9 KiB
Python
118 lines
3.9 KiB
Python
"""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,
|
|
}
|