"""Service layer for the Mail plugin: encryption, IMAP sync, SMTP send, rules, vacation, PGP.""" from __future__ import annotations import base64 import json import logging import mimetypes import os import re import uuid from datetime import UTC, datetime, timedelta from email import message_from_bytes from email.message import EmailMessage from email.utils import formataddr, formatdate, make_msgid import aiofiles import aioimaplib import aiosmtplib import nh3 import pgpy from cryptography.fernet import Fernet from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC from sqlalchemy import and_, or_, select, text from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings from app.core.notifications import create_notification from app.plugins.builtins.mail.models import ( Mail, MailAccount, MailAttachment, MailFolder, MailLabel, MailLabelAssignment, MailRule, MailSignature, MailTemplate, VacationSentLog, ) logger = logging.getLogger(__name__) # ─── 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 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: """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, } # ─── AES-256 Encryption (Fernet) ─── MAIL_ENCRYPTION_KEY = os.environ.get("MAIL_ENCRYPTION_KEY", "leocrm-mail-encryption-key-2024") def _derive_key(password: str, salt: bytes = b"leocrm-mail-salt") -> bytes: """Derive a 32-byte Fernet key from a password using PBKDF2.""" kdf = PBKDF2HMAC( algorithm=hashes.SHA256(), length=32, salt=salt, iterations=480000, ) return base64.urlsafe_b64encode(kdf.derive(password.encode())) _fernet = Fernet(_derive_key(MAIL_ENCRYPTION_KEY)) def encrypt_password(plaintext: str) -> str: """Encrypt a password using AES-256 (Fernet). Returns base64 ciphertext.""" return _fernet.encrypt(plaintext.encode()).decode() def decrypt_password(ciphertext: str) -> str: """Decrypt a password encrypted with encrypt_password.""" return _fernet.decrypt(ciphertext.encode()).decode() # ─── HTML Sanitization (F-MAIL: no script tags) ─── def sanitize_html(raw_html: str) -> str: """Sanitize HTML using nh3 — removes script tags and dangerous attributes.""" if not raw_html: return "" return nh3.clean( raw_html, tags={ "a", "b", "br", "div", "em", "h1", "h2", "h3", "h4", "h5", "h6", "hr", "i", "img", "li", "ol", "p", "span", "strong", "table", "tbody", "td", "th", "thead", "tr", "u", "ul", "blockquote", "code", "pre", "font", "center", }, attributes={ "a": {"href", "title", "target"}, "img": {"src", "alt", "width", "height"}, "span": {"style"}, "div": {"style"}, "font": {"color", "size", "face"}, "p": {"style"}, "td": {"style"}, "th": {"style"}, }, ) # ─── IMAP Quota Parser ─── def _parse_imap_quota_response(response) -> int | None: """Parse an IMAP GETQUOTAROOT response and return usage percentage. Looks for lines like: * QUOTA "INBOX" (STORAGE 12345 67890) where 12345 is used and 67890 is limit. Returns the usage percentage as an int, or None if parsing fails. """ try: lines = response.lines if hasattr(response, "lines") else response for line in lines: if isinstance(line, (bytes, bytearray)): line = line.decode("utf-8", errors="replace") if not isinstance(line, str): continue if "QUOTA" not in line.upper(): continue # Extract the parenthesized storage values # Pattern: (STORAGE ) match = re.search(r"\(STORAGE\s+(\d+)\s+(\d+)\)", line, re.IGNORECASE) if match: used = int(match.group(1)) limit = int(match.group(2)) if limit > 0: return int((used / limit) * 100) except Exception: pass return None # ─── Mail Account Service ─── async def create_mail_account( db: AsyncSession, *, tenant_id: uuid.UUID, user_id: uuid.UUID, data: dict ) -> MailAccount: """Create a new mail account with encrypted password.""" account = MailAccount( tenant_id=tenant_id, user_id=user_id, email_address=data["email_address"], display_name=data.get("display_name", ""), imap_host=data["imap_host"], imap_port=data.get("imap_port", 993), imap_ssl=data.get("imap_ssl", True), smtp_host=data["smtp_host"], smtp_port=data.get("smtp_port", 587), smtp_tls=data.get("smtp_tls", True), username=data.get("username") or data["email_address"], encrypted_password=encrypt_password(data["password"]), is_shared=data.get("is_shared", False), is_active=True, sent_folder_imap_name=data.get("sent_folder_imap_name"), drafts_folder_imap_name=data.get("drafts_folder_imap_name"), spam_folder_imap_name=data.get("spam_folder_imap_name"), trash_folder_imap_name=data.get("trash_folder_imap_name"), ) db.add(account) await db.flush() # Create INBOX first so subfolders can reference it as parent inbox_folder = MailFolder( tenant_id=tenant_id, account_id=account.id, name="Posteingang", imap_name="INBOX", is_standard=True, ) db.add(inbox_folder) await db.flush() # Create standard subfolders under INBOX (IMAP server uses '.' delimiter) for fname, imap_name in [ ("Gesendet", "INBOX.Sent"), ("Entwürfe", "INBOX.Drafts"), ("Spam", "INBOX.spam"), ]: folder = MailFolder( tenant_id=tenant_id, account_id=account.id, name=fname, imap_name=imap_name, parent_id=inbox_folder.id, is_standard=True, ) db.add(folder) await db.flush() return account async def update_mail_account(db: AsyncSession, account: MailAccount, data: dict) -> MailAccount: """Update a mail account, encrypting password if changed.""" field_map = { "email": "email_address", "email_address": "email_address", "display_name": "display_name", "imap_host": "imap_host", "imap_port": "imap_port", "imap_ssl": "imap_ssl", "smtp_host": "smtp_host", "smtp_port": "smtp_port", "smtp_tls": "smtp_tls", "username": "username", "is_shared": "is_shared", "is_active": "is_active", "sent_folder_imap_name": "sent_folder_imap_name", "drafts_folder_imap_name": "drafts_folder_imap_name", "spam_folder_imap_name": "spam_folder_imap_name", "trash_folder_imap_name": "trash_folder_imap_name", } for api_field, model_field in field_map.items(): if api_field in data and data[api_field] is not None: setattr(account, model_field, data[api_field]) if "password" in data and data["password"] is not None: account.encrypted_password = encrypt_password(data["password"]) await db.flush() await db.refresh(account) return account async def get_account_password(account: MailAccount) -> str: """Decrypt and return the account password (internal use only).""" return decrypt_password(account.encrypted_password) def account_to_response(account: MailAccount) -> dict: """Convert MailAccount to response dict, NEVER including password.""" return { "id": str(account.id), "email": account.email_address, "email_address": account.email_address, "display_name": account.display_name, "imap_host": account.imap_host, "imap_port": account.imap_port, "imap_ssl": account.imap_ssl, "smtp_host": account.smtp_host, "smtp_port": account.smtp_port, "smtp_tls": account.smtp_tls, "username": account.username, "is_shared": account.is_shared, "is_active": account.is_active, "sent_folder_imap_name": account.sent_folder_imap_name, "drafts_folder_imap_name": account.drafts_folder_imap_name, "spam_folder_imap_name": account.spam_folder_imap_name, "trash_folder_imap_name": account.trash_folder_imap_name, "created_at": account.created_at, "updated_at": account.updated_at, } # ─── IMAP Sync Service (F-MAIL-01) ─── # German display names for standard IMAP folders. # Keys are full IMAP paths (with dot delimiter) and also bare leaf names. IMAP_FOLDER_NAME_MAP = { "INBOX": "Posteingang", "INBOX.Sent": "Gesendet", "INBOX.Drafts": "Entwürfe", "INBOX.Trash": "Papierkorb", "INBOX.Archive": "Archiv", "INBOX.spam": "Spam", "INBOX.Spam": "Spam", # Bare names (fallback for servers that don't nest under INBOX) "Sent": "Gesendet", "Sent Items": "Gesendet", "Sent Mail": "Gesendet", "Drafts": "Entwürfe", "Draft": "Entwürfe", "Spam": "Spam", "Junk": "Spam", "Junk Email": "Spam", "Junk E-mail": "Spam", "Trash": "Papierkorb", "Deleted": "Papierkorb", "Deleted Items": "Papierkorb", "Archive": "Archiv", } # Standard IMAP folders that are always considered "standard" STANDARD_IMAP_FOLDERS = { "INBOX", "INBOX.Sent", "INBOX.Drafts", "INBOX.Trash", "INBOX.Archive", "INBOX.spam", "INBOX.Spam", "Sent", "Sent Items", "Sent Mail", "Drafts", "Draft", "Spam", "Junk", "Junk Email", "Junk E-mail", "Trash", "Deleted", "Deleted Items", "Archive", } MAX_EMAILS_PER_FOLDER = 2000 def _get_german_folder_name(imap_name: str) -> str: """Return the German display name for a standard IMAP folder, or the original name.""" if imap_name in IMAP_FOLDER_NAME_MAP: return IMAP_FOLDER_NAME_MAP[imap_name] # Try the leaf component (after last dot) for unknown nested folders leaf = imap_name.rsplit(".", 1)[-1] if "." in imap_name else imap_name if leaf in IMAP_FOLDER_NAME_MAP: return IMAP_FOLDER_NAME_MAP[leaf] return imap_name def _parse_imap_list_response(response) -> tuple[list[tuple[str, str]], str]: """Parse IMAP LIST response into list of (flags, folder_name) tuples. Handles both "/" and "." delimiters. The IMAP LIST response format is: * LIST (\\HasChildren) "." "INBOX" * LIST (\\HasNoChildren) "." "INBOX.Sent" Returns (folders, delimiter) where delimiter is the hierarchy separator detected from the LIST response (defaults to '.' if not found). """ folders: list[tuple[str, str]] = [] delimiter = '.' lines = response.lines if hasattr(response, 'lines') else response for line in lines: if isinstance(line, (bytes, bytearray)): text = line.decode('utf-8', errors='replace') elif isinstance(line, str): text = line else: continue if 'LIST' not in text: continue # Extract quoted segments — the delimiter is the first quoted string, # the folder name is the second. parts = text.split('"') if len(parts) >= 4: delimiter = parts[1] folder_name = parts[3] flags = parts[0] if parts[0] else '' folders.append((flags, folder_name)) elif len(parts) >= 2: folder_name = parts[-2] if len(parts) >= 2 else '' if folder_name: folders.append(('', folder_name)) return folders, delimiter def _build_folder_hierarchy( imap_folders: list[tuple[str, str]], account_id: uuid.UUID, tenant_id: uuid.UUID, existing_folders: dict[str, MailFolder], delimiter: str = '.', folder_mapping: dict[str, str | None] | None = None, ) -> list[MailFolder]: """Create or update MailFolder records from IMAP LIST response. Handles delimiter-separated hierarchies (e.g. INBOX.Sent → parent=INBOX). Updates existing folders in-place (name, is_standard, parent_id) so that stale DB records with wrong imap_name values get corrected. """ result: list[MailFolder] = [] # Build reverse mapping: imap_name -> standard type (sent/drafts/spam/trash) mapping_by_imap: dict[str, str] = {} if folder_mapping: for std_type, imap_name_val in folder_mapping.items(): if imap_name_val: mapping_by_imap[imap_name_val] = std_type # First pass: create or update folder records for flags, imap_name in imap_folders: if not imap_name: continue display_name = _get_german_folder_name(imap_name) is_standard = imap_name in STANDARD_IMAP_FOLDERS # If account has explicit folder mapping, mark mapped folders as standard if imap_name in mapping_by_imap: is_standard = True std_type = mapping_by_imap[imap_name] if std_type == "sent": display_name = "Gesendet" elif std_type == "drafts": display_name = "Entwürfe" elif std_type == "spam": display_name = "Spam" elif std_type == "trash": display_name = "Papierkorb" if imap_name in existing_folders: folder = existing_folders[imap_name] folder.name = display_name folder.is_standard = is_standard result.append(folder) else: folder = MailFolder( tenant_id=tenant_id, account_id=account_id, name=display_name, imap_name=imap_name, is_standard=is_standard, ) result.append(folder) # Second pass: set parent_id based on IMAP hierarchy folder_by_imap_name = {f.imap_name: f for f in result} for folder in result: if delimiter in folder.imap_name: parts = folder.imap_name.split(delimiter) parent_imap = delimiter.join(parts[:-1]) if parent_imap in folder_by_imap_name: parent = folder_by_imap_name[parent_imap] folder.parent_id = parent.id # may be None for new folders; fixed after flush else: folder.parent_id = None else: folder.parent_id = None return result async def imap_sync_account( db: AsyncSession, account_id: uuid.UUID, tenant_id: uuid.UUID, ) -> dict: """Sync mail folders and messages from IMAP server. Syncs ALL folders from the IMAP server (not just INBOX). For each folder, fetches the last MAX_EMAILS_PER_FOLDER emails (sorted by date descending) to avoid timeouts on large mailboxes. Creates/updates mail_folders records with German display names. """ account = ( await db.execute( select(MailAccount).where( and_(MailAccount.id == account_id, MailAccount.tenant_id == tenant_id) ) ) ).scalar_one_or_none() if not account: return {"synced": 0, "error": "Account not found"} if not account.is_active: try: await create_notification( db, account.tenant_id, account.user_id, "mail_account", "Mail-Account deaktiviert", f"Account {account.email_address} ist deaktiviert und wird nicht synchronisiert.", ) await db.flush() except Exception: pass return {"synced": 0, "error": "Account is not active"} password = await get_account_password(account) # ── IMAP connection ── try: client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port) await client.wait_hello_from_server() except Exception as e: try: await create_notification( db, account.tenant_id, account.user_id, "mail_error", "IMAP-Verbindung fehlgeschlagen", f"Account {account.email_address}: {e}", ) await db.flush() except Exception: pass return {"synced": 0, "error": f"IMAP connection failed: {e}"} # ── IMAP login ── try: await client.login(account.username, password) except Exception as e: try: await create_notification( db, account.tenant_id, account.user_id, "mail_auth", "IMAP-Login fehlgeschlagen", f"Account {account.email_address}: Passwort oder Anmeldedaten prüfen", ) await db.flush() except Exception: pass try: await client.logout() except Exception: pass return {"synced": 0, "error": f"IMAP login failed: {e}"} # ── Quota check (non-critical, not all servers support QUOTA) ── try: quota_resp = await client.getquotaroot('INBOX') usage_pct = _parse_imap_quota_response(quota_resp) if usage_pct is not None and usage_pct > 80: if usage_pct > 95: quota_title = "Postfach voll – keine neuen Mails empfangbar" else: quota_title = "Postfach fast voll" try: await create_notification( db, account.tenant_id, account.user_id, "mail_quota", quota_title, f"Account {account.email_address}: {usage_pct}% belegt", ) await db.flush() except Exception: pass except Exception: pass try: # 1) LIST all folders from IMAP server list_response = await client.list('""', '"*"') imap_folders, imap_delimiter = _parse_imap_list_response(list_response) # If LIST returned nothing, fall back to standard folders (dot-delimited) if not imap_folders: imap_folders = [ ('', 'INBOX'), ('', 'INBOX.Sent'), ('', 'INBOX.Drafts'), ('', 'INBOX.spam'), ('', 'INBOX.Trash'), ] # 2) Load existing folders from DB for this account existing_db_folders = ( await db.execute( select(MailFolder).where( and_(MailFolder.account_id == account.id, MailFolder.tenant_id == tenant_id) ) ) ).scalars().all() existing_by_imap: dict[str, MailFolder] = { f.imap_name: f for f in existing_db_folders } # 2a) Migrate stale folder names: if DB has 'Sent' but IMAP returns # 'INBOX.Sent', update the DB record's imap_name so it matches. imap_names_from_server = {name for _, name in imap_folders if name} for db_folder in existing_db_folders: if db_folder.imap_name not in imap_names_from_server: # Try matching by leaf component using detected delimiter leaf = db_folder.imap_name.rsplit(imap_delimiter, 1)[-1] for srv_name in imap_names_from_server: srv_leaf = srv_name.rsplit(imap_delimiter, 1)[-1] if srv_leaf.lower() == leaf.lower(): db_folder.imap_name = srv_name existing_by_imap[srv_name] = db_folder break # 3) Create/update folders in DB folder_mapping = { "sent": account.sent_folder_imap_name, "drafts": account.drafts_folder_imap_name, "spam": account.spam_folder_imap_name, "trash": account.trash_folder_imap_name, } db_folders = _build_folder_hierarchy( imap_folders, account.id, tenant_id, existing_by_imap, imap_delimiter, folder_mapping=folder_mapping, ) for folder in db_folders: if folder.id is None: db.add(folder) await db.flush() # 3a) Re-set parent_id now that new folders have IDs after flush folder_by_imap_post_flush = {f.imap_name: f for f in db_folders} for folder in db_folders: if imap_delimiter in folder.imap_name: parts = folder.imap_name.split(imap_delimiter) parent_imap = imap_delimiter.join(parts[:-1]) if parent_imap in folder_by_imap_post_flush: parent = folder_by_imap_post_flush[parent_imap] if parent.id: folder.parent_id = parent.id await db.flush() # Build a map of imap_name -> folder_id for email sync folder_by_imap = {f.imap_name: f for f in db_folders} synced_count = 0 new_mails: list[dict] = [] # 4) Sync emails for each folder (limit to last 50 per folder) for imap_name, folder in folder_by_imap.items(): try: # Select the folder on the IMAP server — use the raw IMAP # name without extra quoting (aioimaplib handles it) select_resp = await client.select(imap_name) if select_resp.result != 'OK': continue # Fetch UIDs and sort by date descending — get last 50 search_resp = await client.uid_search('ALL') uids_raw = search_resp[1][0] if search_resp[1] and search_resp[1][0] else b'' if isinstance(uids_raw, (bytes, bytearray)): uids = uids_raw.split() else: uids = [] # Limit to last MAX_EMAILS_PER_FOLDER UIDs # UIDs are monotonically increasing, so the last ones are the newest if len(uids) > MAX_EMAILS_PER_FOLDER: uids = uids[-MAX_EMAILS_PER_FOLDER:] for uid in uids: uid_str = uid.decode() if isinstance(uid, bytes) else str(uid) fetch_resp = await client.uid('fetch', uid_str, '(RFC822)') raw_email = None for line in (fetch_resp.lines if hasattr(fetch_resp, 'lines') else fetch_resp): if isinstance(line, bytearray): raw_email = bytes(line) break if raw_email is None: continue if isinstance(raw_email, str): raw_email = raw_email.encode() msg = message_from_bytes(raw_email) body_text = "" body_html = "" attachments = [] if msg.is_multipart(): for part in msg.walk(): ct = part.get_content_type() if ct == "text/plain": payload = part.get_payload(decode=True) if payload: body_text = payload.decode("utf-8", errors="replace") elif ct == "text/html": payload = part.get_payload(decode=True) if payload: body_html = payload.decode("utf-8", errors="replace") elif part.get_filename(): payload_bytes = part.get_payload(decode=True) or b"" content_id = part.get("Content-ID", None) if content_id: content_id = content_id.strip("<>") attachments.append( { "filename": part.get_filename(), "mime_type": ct, "size": len(payload_bytes), "content": payload_bytes, "content_id": content_id, } ) else: ct = msg.get_content_type() payload = msg.get_payload(decode=True) if payload: decoded = payload.decode("utf-8", errors="replace") if ct == "text/html": body_html = decoded else: body_text = decoded from email.header import decode_header, make_header def _decode_mime_header(value: str) -> str: """Decode MIME encoded-words (=?UTF-8?Q?...?=) to readable text.""" if not value: return "" try: return str(make_header(decode_header(value))) except Exception: return value raw_msg_id = msg.get("Message-ID") if raw_msg_id: message_id = raw_msg_id else: # Deterministic ID so re-syncs don't create duplicates message_id = f"generated-{account.id}-{folder.id}-{uid_str}" subject = _decode_mime_header(msg.get("Subject", "")) from_addr = _decode_mime_header(msg.get("From", "")) to_addrs = _decode_mime_header(msg.get("To", "")) cc_addrs = _decode_mime_header(msg.get("Cc", "")) refs = msg.get("References", "") in_reply_to = msg.get("In-Reply-To") date_str = msg.get("Date", "") # Parse date for received_at received_at = datetime.now(UTC) if date_str: try: from email.utils import parsedate_to_datetime parsed = parsedate_to_datetime(date_str) if parsed: received_at = parsed.astimezone(UTC) if parsed.tzinfo else parsed.replace(tzinfo=UTC) except Exception: pass # Compute thread_id from References/In-Reply-To thread_id = _compute_thread_id(message_id, refs, in_reply_to) # Dedup: first check by (account_id, folder_id, imap_uid), # then fall back to (account_id, message_id) for mails synced # before the imap_uid column existed. existing_mail = ( await db.execute( select(Mail).where( and_( Mail.account_id == account.id, Mail.folder_id == folder.id, Mail.imap_uid == uid_str, Mail.tenant_id == tenant_id, ) ) ) ).scalar_one_or_none() if not existing_mail: existing_mail = ( await db.execute( select(Mail).where( and_( Mail.account_id == account.id, Mail.message_id == message_id, Mail.tenant_id == tenant_id, ) ) ) ).scalar_one_or_none() if existing_mail: # Track folder moves: update folder_id and imap_uid if the # mail now appears in a different IMAP folder. if existing_mail.folder_id != folder.id: existing_mail.folder_id = folder.id existing_mail.imap_uid = uid_str elif not existing_mail.imap_uid: existing_mail.imap_uid = uid_str # Save attachments for existing emails that have has_attachments but no records if attachments and existing_mail.has_attachments: existing_att_count = ( await db.execute( select(text("COUNT(*)")).where( and_( MailAttachment.mail_id == existing_mail.id, MailAttachment.tenant_id == tenant_id, ) ) ) ).scalar() or 0 if existing_att_count == 0: for att_data in attachments: try: raw_filename = att_data["filename"] or "attachment" decoded_filename = _decode_mime_filename(raw_filename) storage_path = await _save_attachment_to_storage( existing_mail.id, decoded_filename, att_data["content"] ) attachment = MailAttachment( tenant_id=tenant_id, mail_id=existing_mail.id, filename=_sanitize_filename(decoded_filename), mime_type=att_data["mime_type"], size_bytes=att_data["size"], storage_path=storage_path, content_id=att_data.get("content_id"), ) db.add(attachment) except Exception as att_err: logger.warning(f"Failed to save attachment for existing mail {existing_mail.id}: {att_err}") continue await db.flush() continue mail = Mail( tenant_id=tenant_id, account_id=account.id, folder_id=folder.id, imap_uid=uid_str, message_id=message_id, thread_id=thread_id, in_reply_to=in_reply_to, references_header=refs, subject=subject, from_address=from_addr, to_addresses=to_addrs, cc_addresses=cc_addrs, body_text=body_text, body_html=body_html, body_html_sanitized=sanitize_html(body_html), has_attachments=len(attachments) > 0, size_bytes=len(raw_email), received_at=received_at, ) db.add(mail) await db.flush() synced_count += 1 # Collect for new-mail notifications new_mails.append({ "from": from_addr, "subject": subject, }) # Save attachments to storage and DB for att_data in attachments: try: # Decode MIME-encoded filenames (e.g. =?utf-8?q?...?=) raw_filename = att_data["filename"] or "attachment" decoded_filename = _decode_mime_filename(raw_filename) storage_path = await _save_attachment_to_storage( mail.id, decoded_filename, att_data["content"] ) attachment = MailAttachment( tenant_id=tenant_id, mail_id=mail.id, filename=_sanitize_filename(decoded_filename), mime_type=att_data["mime_type"], size_bytes=att_data["size"], storage_path=storage_path, content_id=att_data.get("content_id"), ) db.add(attachment) except Exception as att_err: logger.warning(f"Failed to save attachment for new mail {mail.id}: {att_err}") continue await db.flush() # Update folder counts total = ( await db.execute( select(text("COUNT(*)")).where( and_(Mail.folder_id == folder.id, Mail.tenant_id == tenant_id) ) ) ).scalar() unread = ( await db.execute( select(text("COUNT(*)")).where( and_( Mail.folder_id == folder.id, Mail.tenant_id == tenant_id, not Mail.is_seen, ) ) ) ).scalar() folder.total_count = total or 0 folder.unread_count = unread or 0 await db.flush() except Exception: # Skip folders that can't be selected (e.g. no select permission) continue # ── New-mail notifications (max 10, then summary) ── if new_mails: try: if len(new_mails) <= 10: for nm in new_mails: try: await create_notification( db, account.tenant_id, account.user_id, "mail_new", f"Neue E-Mail von {nm['from']}", nm["subject"], ) except Exception: pass else: try: await create_notification( db, account.tenant_id, account.user_id, "mail_new", f"Neue E-Mails: {len(new_mails)} neue Nachrichten", f"Account {account.email_address} hat {len(new_mails)} neue E-Mails empfangen.", ) except Exception: pass await db.flush() except Exception: pass await db.flush() await client.logout() return {"synced": synced_count} except Exception as e: return {"synced": 0, "error": str(e)} def _compute_thread_id(message_id: str, references: str, in_reply_to: str | None) -> str: """Compute thread ID from References/In-Reply-To headers (F-MAIL-05).""" ref_parts: list[str] = [] if references: ref_parts = [r.strip() for r in references.split() if r.strip()] if in_reply_to and in_reply_to.strip() not in ref_parts: ref_parts.append(in_reply_to.strip()) if ref_parts: return ref_parts[0] return message_id or str(uuid.uuid4()) # ─── SMTP Send Service (F-MAIL-02) ─── async def send_mail_via_smtp( db: AsyncSession, *, tenant_id: uuid.UUID, user_id: uuid.UUID, account: MailAccount, to_addrs: list[str], cc_addrs: list[str] = None, bcc_addrs: list[str] = None, subject: str = "", body_html: str = "", body_text: str = "", in_reply_to: str | None = None, references_header: str | None = None, signature: MailSignature | None = None, attachment_paths: list[dict] | None = None, ) -> dict: """Send an email via SMTP using aiosmtplib. Args: attachment_paths: list of dicts with keys 'path', 'filename', 'mime_type' pointing to files on disk to attach. """ cc_addrs = cc_addrs or [] bcc_addrs = bcc_addrs or [] attachment_paths = attachment_paths or [] # Apply signature if provided if signature and signature.body_html: body_html = body_html + f"

{signature.body_html}" if body_text: body_text = body_text + "\n\n-- \n" + _strip_html(signature.body_html) # Build email message msg = EmailMessage() msg["From"] = formataddr((account.display_name or "", account.email_address)) msg["To"] = ", ".join(to_addrs) if cc_addrs: msg["Cc"] = ", ".join(cc_addrs) msg["Subject"] = subject msg["Date"] = formatdate(localtime=True) msg_id = make_msgid( domain=account.email_address.split("@")[-1] if "@" in account.email_address else "localhost" ) msg["Message-ID"] = msg_id if in_reply_to: msg["In-Reply-To"] = in_reply_to if references_header: msg["References"] = references_header if body_html: msg.set_content(body_text or _strip_html(body_html), subtype="plain") msg.add_alternative(body_html, subtype="html") else: msg.set_content(body_text, subtype="plain") # Add attachments to the message for att_info in attachment_paths: file_path = att_info.get("path", "") filename = att_info.get("filename", os.path.basename(file_path)) mime_type = att_info.get("mime_type", "application/octet-stream") if not file_path or not os.path.exists(file_path): continue with open(file_path, "rb") as f: # noqa: ASYNC230 content = f.read() # Determine maintype/subtype from mime_type if "/" in mime_type: maintype, subtype = mime_type.split("/", 1) else: maintype, subtype = "application", "octet-stream" msg.add_attachment( content, maintype=maintype, subtype=subtype, filename=filename, ) # Send via SMTP password = await get_account_password(account) try: smtp = aiosmtplib.SMTP( hostname=account.smtp_host, port=account.smtp_port, use_tls=account.smtp_tls, ) await smtp.connect() await smtp.login(account.username, password) recipients = to_addrs + cc_addrs + bcc_addrs await smtp.send_message(msg, recipients=recipients) await smtp.quit() # Store sent mail in Sent folder — use configured mapping if set, # otherwise flexible lookup to handle different IMAP naming conventions if account.sent_folder_imap_name: sent_folder = ( await db.execute( select(MailFolder).where( and_( MailFolder.account_id == account.id, MailFolder.imap_name == account.sent_folder_imap_name, ) ) ) ).scalar_one_or_none() else: sent_folder = ( await db.execute( select(MailFolder).where( and_( MailFolder.account_id == account.id, MailFolder.imap_name.in_( ["Sent", "INBOX.Sent", "Sent Items", "Sent Mail"] ), ) ) ) ).scalar_one_or_none() if not sent_folder: sent_folder = ( await db.execute( select(MailFolder).where( and_( MailFolder.account_id == account.id, MailFolder.is_standard == True, MailFolder.imap_name.ilike("%sent%"), ) ) ) ).scalar_one_or_none() if sent_folder: thread_id = _compute_thread_id(msg_id, references_header or "", in_reply_to) sent_mail = Mail( tenant_id=tenant_id, account_id=account.id, folder_id=sent_folder.id, message_id=msg_id, thread_id=thread_id, in_reply_to=in_reply_to, references_header=references_header, subject=subject, from_address=account.email_address, to_addresses=", ".join(to_addrs), cc_addresses=", ".join(cc_addrs), bcc_addresses=", ".join(bcc_addrs), body_text=body_text or _strip_html(body_html), body_html=body_html, body_html_sanitized=sanitize_html(body_html), is_seen=True, is_answered=bool(in_reply_to), sent_at=datetime.now(UTC), ) db.add(sent_mail) await db.flush() # ── Notification: mail sent ── try: await create_notification( db, tenant_id, user_id, "mail_sent", "E-Mail gesendet", subject, ) await db.flush() except Exception: pass return {"status": "sent", "message_id": msg_id} except Exception as e: # ── Notification: send error ── try: await create_notification( db, tenant_id, user_id, "mail_send_error", "E-Mail konnte nicht gesendet werden", str(e), ) await db.flush() except Exception: pass return {"status": "error", "error": str(e)} async def reply_to_mail( db: AsyncSession, *, tenant_id: uuid.UUID, user_id: uuid.UUID, original_mail: Mail, account: MailAccount, body_html: str, body_text: str = "", reply_to_all: bool = False, signature: MailSignature | None = None, ) -> dict: """Reply to a mail, setting In-Reply-To and References headers (F-MAIL-02).""" to_addrs = [original_mail.from_address] if reply_to_all and original_mail.cc_addresses: to_addrs.extend([a.strip() for a in original_mail.cc_addresses.split(",") if a.strip()]) refs = original_mail.references_header or "" new_refs = f"{refs} {original_mail.message_id}".strip() result = await send_mail_via_smtp( db, tenant_id=tenant_id, user_id=user_id, account=account, to_addrs=to_addrs, subject=f"Re: {original_mail.subject}".replace("Re: Re: ", "Re: "), body_html=body_html, body_text=body_text, in_reply_to=original_mail.message_id, references_header=new_refs, signature=signature, ) # Mark original as answered original_mail.is_answered = True await db.flush() return result async def forward_mail( db: AsyncSession, *, tenant_id: uuid.UUID, user_id: uuid.UUID, original_mail: Mail, account: MailAccount, to_addrs: list[str], cc_addrs: list[str] = None, body_html: str = "", body_text: str = "", signature: MailSignature | None = None, ) -> dict: """Forward a mail with original as forwarded content (F-MAIL-02).""" fwd_subject = f"Fwd: {original_mail.subject}".replace("Fwd: Fwd: ", "Fwd: ") fwd_body = ( f"

----- Original Message -----
" f"From: {original_mail.from_address}
" f"Subject: {original_mail.subject}

" f"{original_mail.body_html or original_mail.body_text}" ) full_html = body_html + fwd_body full_text = (body_text or _strip_html(body_html)) + "\n\n----- Original Message -----\n" result = await send_mail_via_smtp( db, tenant_id=tenant_id, user_id=user_id, account=account, to_addrs=to_addrs, cc_addrs=cc_addrs or [], subject=fwd_subject, body_html=full_html, body_text=full_text, signature=signature, ) original_mail.is_forwarded = True await db.flush() return result # ─── Template Service (F-MAIL-06) ─── def substitute_template_vars(template_body: str, variables: dict[str, str]) -> str: """Replace {{placeholder}} variables in template body.""" result = template_body for key, value in variables.items(): result = result.replace(f"{{{{{key}}}}}", value) result = result.replace(f"{{{{{key.lower()}}}}}", value) result = result.replace(f"{{{{{key.upper()}}}}}", value) return result # ─── Mail Rule Engine (F-MAIL-07) ─── def matches_condition(mail: Mail, conditions: dict) -> bool: """Check if a mail matches all rule conditions.""" for field, expected in conditions.items(): if field == "from_contains": if expected.lower() not in mail.from_address.lower(): return False elif field == "subject_contains": if expected.lower() not in mail.subject.lower(): return False elif field == "to_contains": if expected.lower() not in mail.to_addresses.lower(): return False elif field == "body_contains": body = (mail.body_text + mail.body_html).lower() if expected.lower() not in body: return False elif field == "has_attachments": if mail.has_attachments != bool(expected): return False elif field == "is_flagged": if mail.is_flagged != bool(expected): return False return True async def execute_rule_actions( db: AsyncSession, mail: Mail, actions: dict, tenant_id: uuid.UUID ) -> dict: """Execute rule actions on a matching mail.""" results = {} for action, value in actions.items(): if action == "move_to_folder": folder_id = uuid.UUID(value) if isinstance(value, str) else value folder = ( await db.execute( select(MailFolder).where( and_(MailFolder.id == folder_id, MailFolder.tenant_id == tenant_id) ) ) ).scalar_one_or_none() if folder: mail.folder_id = folder.id results["moved"] = str(folder.id) elif action == "label": label_id = uuid.UUID(value) if isinstance(value, str) else value existing = ( await db.execute( select(MailLabelAssignment).where( and_( MailLabelAssignment.mail_id == mail.id, MailLabelAssignment.label_id == label_id, ) ) ) ).scalar_one_or_none() if not existing: assignment = MailLabelAssignment( tenant_id=tenant_id, mail_id=mail.id, label_id=label_id, ) db.add(assignment) results["labeled"] = str(label_id) elif action == "mark_seen": mail.is_seen = bool(value) results["seen"] = bool(value) elif action == "mark_flagged": mail.is_flagged = bool(value) results["flagged"] = bool(value) elif action == "forward_to": results["forward_to"] = value await db.flush() return results async def apply_rules_to_mail(db: AsyncSession, mail: Mail, tenant_id: uuid.UUID) -> list[dict]: """Find and apply all matching rules to a mail, sorted by priority.""" rules = ( ( await db.execute( select(MailRule) .where( and_( MailRule.tenant_id == tenant_id, MailRule.is_active, or_( MailRule.account_id == mail.account_id, MailRule.account_id.is_(None), ), ) ) .order_by(MailRule.priority) ) ) .scalars() .all() ) applied = [] for rule in rules: conditions = json.loads(rule.conditions) if rule.conditions else {} actions = json.loads(rule.actions) if rule.actions else {} if matches_condition(mail, conditions): result = await execute_rule_actions(db, mail, actions, tenant_id) applied.append({"rule_id": str(rule.id), "rule_name": rule.name, "actions": result}) return applied # ─── Vacation Auto-Reply (F-MAIL-08) ─── VACATION_DEDUP_HOURS = 24 async def should_send_vacation_reply( db: AsyncSession, account_id: uuid.UUID, sender_address: str, tenant_id: uuid.UUID, ) -> bool: """Check if vacation auto-reply should be sent (dedup within 24h).""" cutoff = datetime.now(UTC) - timedelta(hours=VACATION_DEDUP_HOURS) existing = ( await db.execute( select(VacationSentLog).where( and_( VacationSentLog.account_id == account_id, VacationSentLog.sender_address == sender_address, VacationSentLog.sent_at >= cutoff, VacationSentLog.tenant_id == tenant_id, ) ) ) ).scalar_one_or_none() return existing is None async def log_vacation_sent( db: AsyncSession, account_id: uuid.UUID, sender_address: str, tenant_id: uuid.UUID, ) -> None: """Log that a vacation auto-reply was sent to a sender.""" log = VacationSentLog( tenant_id=tenant_id, account_id=account_id, sender_address=sender_address, sent_at=datetime.now(UTC), ) db.add(log) await db.flush() # ─── PGP Service (F-MAIL-12) ─── def import_pgp_private_key(private_key_armored: str, passphrase: str = "") -> tuple[str, str]: """Import a PGP private key. Returns (key_id, public_key_armored).""" key, _ = pgpy.PGPKey.from_blob(private_key_armored) if key.is_protected: with key.unlock(passphrase): pub_key = key.pubkey key_id = str(key.fingerprint).upper()[-16:] return key_id, str(pub_key) pub_key = key.pubkey key_id = str(key.fingerprint).upper()[-16:] return key_id, str(pub_key) def import_pgp_public_key(public_key_armored: str) -> str: """Import a PGP public key. Returns key_id.""" key, _ = pgpy.PGPKey.from_blob(public_key_armored) return str(key.fingerprint).upper()[-16:] def pgp_encrypt_message(plaintext: str, recipient_public_key_armored: str) -> str: """Encrypt a message with recipient's public PGP key.""" pub_key, _ = pgpy.PGPKey.from_blob(recipient_public_key_armored) msg = pgpy.PGPMessage.new(plaintext) encrypted = pub_key.encrypt(msg) return str(encrypted) def pgp_decrypt_message(ciphertext: str, private_key_armored: str, passphrase: str = "") -> str: """Decrypt a PGP-encrypted message.""" key, _ = pgpy.PGPKey.from_blob(private_key_armored) enc_msg = pgpy.PGPMessage.from_blob(ciphertext) if key.is_protected: with key.unlock(passphrase): decrypted = key.decrypt(enc_msg) return decrypted.message.decode("utf-8") decrypted = key.decrypt(enc_msg) return decrypted.message.decode("utf-8") # ─── Contact Linking (F-MAIL-10) ─── def extract_email_addresses(text: str) -> list[str]: """Extract email addresses from a text string.""" if not text: return [] return re.findall(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", text) # ─── Utility ─── def _strip_html(html: str) -> str: """Simple HTML to text conversion for plain text fallback.""" if not html: return "" # Remove tags text = re.sub(r"<[^>]+>", "", html) # Replace HTML entities text = ( text.replace(" ", " ") .replace("&", "&") .replace("<", "<") .replace(">", ">") .replace(""", '"') ) return text.strip() def mail_to_response( mail: Mail, attachments: list[MailAttachment] | None = None, labels: list[MailLabel] | None = None, ) -> dict: """Convert a Mail ORM object to a response dict.""" resp = { "id": str(mail.id), "account_id": str(mail.account_id), "folder_id": str(mail.folder_id), "message_id": mail.message_id, "thread_id": mail.thread_id, "in_reply_to": mail.in_reply_to, "subject": mail.subject, "from_address": mail.from_address, "from_name": mail.from_address.split("<")[0].strip().strip('"') if "<" in mail.from_address else mail.from_address, "to_addresses": extract_email_addresses(mail.to_addresses) if mail.to_addresses else [], "cc_addresses": extract_email_addresses(mail.cc_addresses) if mail.cc_addresses else [], "bcc_addresses": extract_email_addresses(mail.bcc_addresses) if mail.bcc_addresses else [], "body_text": mail.body_text, "body_html": mail.body_html if mail.body_html else None, "body_html_sanitized": mail.body_html_sanitized, "sanitized_html": mail.body_html_sanitized, "date": mail.received_at.isoformat() if mail.received_at else (mail.sent_at.isoformat() if mail.sent_at else None), "is_seen": mail.is_seen, "is_flagged": mail.is_flagged, "flag_type": mail.flag_type, "is_draft": mail.is_draft, "is_answered": mail.is_answered, "is_forwarded": mail.is_forwarded, "has_attachments": mail.has_attachments, "size_bytes": mail.size_bytes, "received_at": mail.received_at, "sent_at": mail.sent_at, "contact_id": str(mail.contact_id) if mail.contact_id else None, "company_id": str(mail.company_id) if mail.company_id else None, "attachments": [], "labels": [], } if attachments: resp["attachments"] = [attachment_to_response(a) for a in attachments] if labels: resp["labels"] = [ {"id": str(lbl.id), "name": lbl.name, "color": lbl.color} for lbl in labels ] return resp def folder_to_response(folder: MailFolder) -> dict: """Convert MailFolder to response dict.""" return { "id": str(folder.id), "account_id": str(folder.account_id), "name": folder.name, "imap_name": folder.imap_name, "parent_id": str(folder.parent_id) if folder.parent_id else None, "is_standard": folder.is_standard, "unread_count": folder.unread_count, "total_count": folder.total_count, } def rule_to_response(rule: MailRule) -> dict: """Convert MailRule to response dict.""" return { "id": str(rule.id), "name": rule.name, "account_id": str(rule.account_id) if rule.account_id else None, "priority": rule.priority, "is_active": rule.is_active, "conditions": json.loads(rule.conditions) if rule.conditions else {}, "actions": json.loads(rule.actions) if rule.actions else {}, } def template_to_response(template: MailTemplate) -> dict: """Convert MailTemplate to response dict.""" return { "id": str(template.id), "name": template.name, "subject": template.subject, "body_html": template.body_html, } def signature_to_response(sig: MailSignature) -> dict: """Convert MailSignature to response dict.""" return { "id": str(sig.id), "name": sig.name, "body_html": sig.body_html, "account_id": str(sig.account_id) if sig.account_id else None, "is_default": sig.is_default, } def label_to_response(label: MailLabel) -> dict: """Convert MailLabel to response dict.""" return { "id": str(label.id), "name": label.name, "color": label.color, } # ─── IMAP Flag Sync ─── async def imap_sync_mail_flags( db: AsyncSession, mail_id: uuid.UUID, tenant_id: uuid.UUID, ) -> None: """Sync is_seen/is_flagged flags from DB to IMAP server. Connects to the IMAP server, selects the mail's folder, and uses UID STORE to set/remove \\Seen and \\Flagged flags. Non-critical: logs warnings on failure but does not raise. """ import logging logger = logging.getLogger(__name__) # Load the mail with its folder and account mail = ( await db.execute( select(Mail).where( and_(Mail.id == mail_id, Mail.tenant_id == tenant_id) ) ) ).scalar_one_or_none() if not mail: logger.warning("imap_sync_mail_flags: mail %s not found", mail_id) return folder = ( await db.execute( select(MailFolder).where( and_(MailFolder.id == mail.folder_id, MailFolder.tenant_id == tenant_id) ) ) ).scalar_one_or_none() if not folder: logger.warning("imap_sync_mail_flags: folder %s not found for mail %s", mail.folder_id, mail_id) return account = ( await db.execute( select(MailAccount).where( and_(MailAccount.id == folder.account_id, MailAccount.tenant_id == tenant_id) ) ) ).scalar_one_or_none() if not account: logger.warning("imap_sync_mail_flags: account not found for mail %s", mail_id) return if not mail.message_id: logger.warning("imap_sync_mail_flags: mail %s has no message_id, cannot sync", mail_id) return password = await get_account_password(account) client = None try: client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port) await client.wait_hello_from_server() await client.login(account.username, password) # Select the folder select_resp = await client.select(folder.imap_name) if select_resp.result != 'OK': logger.warning("imap_sync_mail_flags: cannot select folder %s", folder.imap_name) return # Find the UID by searching for the Message-ID header search_resp = await client.uid_search(f'HEADER Message-ID "{mail.message_id}"') uids_raw = search_resp[1][0] if search_resp[1] and search_resp[1][0] else b'' if isinstance(uids_raw, (bytes, bytearray)): uids = uids_raw.split() else: uids = [] if not uids: logger.warning("imap_sync_mail_flags: no UID found for Message-ID %s", mail.message_id) return uid_str = uids[0].decode() if isinstance(uids[0], bytes) else str(uids[0]) # Sync \Seen flag if mail.is_seen: await client.uid('store', uid_str, '+FLAGS (\\Seen)') else: await client.uid('store', uid_str, '-FLAGS (\\Seen)') # Sync \Flagged flag if mail.is_flagged: await client.uid('store', uid_str, '+FLAGS (\\Flagged)') else: await client.uid('store', uid_str, '-FLAGS (\\Flagged)') logger.info("imap_sync_mail_flags: synced flags for mail %s (UID %s)", mail_id, uid_str) except Exception as exc: logger.warning("imap_sync_mail_flags: failed for mail %s: %s", mail_id, exc) finally: if client is not None: try: await client.logout() except Exception: pass # ─── IMAP Delete ─── async def imap_delete_mail( db: AsyncSession, mail_id: uuid.UUID, tenant_id: uuid.UUID ) -> None: """Delete mail from IMAP server. Connects to IMAP, selects the mail's folder, finds UID by Message-ID, marks it as \\Deleted and expunges. Non-critical: errors are logged, DB soft-delete still succeeds. """ import logging logger = logging.getLogger(__name__) mail = ( await db.execute( select(Mail).where( and_(Mail.id == mail_id, Mail.tenant_id == tenant_id) ) ) ).scalar_one_or_none() if not mail: logger.warning("imap_delete_mail: mail %s not found", mail_id) return folder = ( await db.execute( select(MailFolder).where( and_(MailFolder.id == mail.folder_id, MailFolder.tenant_id == tenant_id) ) ) ).scalar_one_or_none() if not folder: logger.warning("imap_delete_mail: folder %s not found for mail %s", mail.folder_id, mail_id) return account = ( await db.execute( select(MailAccount).where( and_(MailAccount.id == folder.account_id, MailAccount.tenant_id == tenant_id) ) ) ).scalar_one_or_none() if not account: logger.warning("imap_delete_mail: account not found for mail %s", mail_id) return if not mail.message_id: logger.warning("imap_delete_mail: mail %s has no message_id, cannot sync", mail_id) return password = await get_account_password(account) client = None try: client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port) await client.wait_hello_from_server() await client.login(account.username, password) select_resp = await client.select(folder.imap_name) if select_resp.result != 'OK': logger.warning("imap_delete_mail: cannot select folder %s", folder.imap_name) return search_resp = await client.uid_search(f'HEADER Message-ID "{mail.message_id}"') uids_raw = search_resp[1][0] if search_resp[1] and search_resp[1][0] else b'' if isinstance(uids_raw, (bytes, bytearray)): uids = uids_raw.split() else: uids = [] if not uids: logger.warning("imap_delete_mail: no UID found for Message-ID %s", mail.message_id) return uid_str = uids[0].decode() if isinstance(uids[0], bytes) else str(uids[0]) await client.uid('store', uid_str, r'+FLAGS (\\Deleted)') await client.expunge() logger.info("imap_delete_mail: deleted mail %s (UID %s) from IMAP", mail_id, uid_str) except Exception as exc: logger.warning("imap_delete_mail: failed for mail %s: %s", mail_id, exc) finally: if client is not None: try: await client.logout() except Exception: pass # ─── IMAP Move ─── async def imap_move_mail( db: AsyncSession, mail_id: uuid.UUID, target_folder_id: uuid.UUID, tenant_id: uuid.UUID, ) -> None: """Move mail to another folder on IMAP server. Uses UID MOVE if supported, otherwise COPY + STORE \\Deleted + EXPUNGE. Non-critical: errors are logged, DB update still succeeds. """ import logging logger = logging.getLogger(__name__) mail = ( await db.execute( select(Mail).where( and_(Mail.id == mail_id, Mail.tenant_id == tenant_id) ) ) ).scalar_one_or_none() if not mail: logger.warning("imap_move_mail: mail %s not found", mail_id) return source_folder = ( await db.execute( select(MailFolder).where( and_(MailFolder.id == mail.folder_id, MailFolder.tenant_id == tenant_id) ) ) ).scalar_one_or_none() if not source_folder: logger.warning("imap_move_mail: source folder not found for mail %s", mail_id) return target_folder = ( await db.execute( select(MailFolder).where( and_(MailFolder.id == target_folder_id, MailFolder.tenant_id == tenant_id) ) ) ).scalar_one_or_none() if not target_folder: logger.warning("imap_move_mail: target folder %s not found", target_folder_id) return account = ( await db.execute( select(MailAccount).where( and_(MailAccount.id == source_folder.account_id, MailAccount.tenant_id == tenant_id) ) ) ).scalar_one_or_none() if not account: logger.warning("imap_move_mail: account not found for mail %s", mail_id) return if not mail.message_id: logger.warning("imap_move_mail: mail %s has no message_id, cannot sync", mail_id) return password = await get_account_password(account) client = None try: client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port) await client.wait_hello_from_server() await client.login(account.username, password) select_resp = await client.select(source_folder.imap_name) if select_resp.result != 'OK': logger.warning("imap_move_mail: cannot select folder %s", source_folder.imap_name) return search_resp = await client.uid_search(f'HEADER Message-ID "{mail.message_id}"') uids_raw = search_resp[1][0] if search_resp[1] and search_resp[1][0] else b'' if isinstance(uids_raw, (bytes, bytearray)): uids = uids_raw.split() else: uids = [] if not uids: logger.warning("imap_move_mail: no UID found for Message-ID %s", mail.message_id) return uid_str = uids[0].decode() if isinstance(uids[0], bytes) else str(uids[0]) # Try UID MOVE first; fall back to COPY + STORE \\Deleted + EXPUNGE try: move_resp = await client.uid('move', uid_str, target_folder.imap_name) if move_resp.result == 'OK': logger.info("imap_move_mail: moved mail %s (UID %s) via UID MOVE", mail_id, uid_str) return except Exception as move_exc: logger.info("imap_move_mail: UID MOVE not supported, falling back: %s", move_exc) # Fallback: COPY + STORE \\Deleted + EXPUNGE copy_resp = await client.uid('copy', uid_str, target_folder.imap_name) if copy_resp.result != 'OK': logger.warning("imap_move_mail: COPY failed for mail %s", mail_id) return await client.uid('store', uid_str, r'+FLAGS (\\Deleted)') await client.expunge() logger.info("imap_move_mail: moved mail %s (UID %s) via COPY+DELETE", mail_id, uid_str) except Exception as exc: logger.warning("imap_move_mail: failed for mail %s: %s", mail_id, exc) finally: if client is not None: try: await client.logout() except Exception: pass # ─── Draft Save / Update ─── async def save_draft( db: AsyncSession, account_id: uuid.UUID, tenant_id: uuid.UUID, user_id: uuid.UUID, data: dict, ) -> Mail: """Save a new draft mail to DB and IMAP Drafts folder.""" import logging logger = logging.getLogger(__name__) # 1. Find the Drafts folder (imap_name contains 'Drafts') account = ( await db.execute( select(MailAccount).where( and_(MailAccount.id == account_id, MailAccount.tenant_id == tenant_id) ) ) ).scalar_one_or_none() if not account: raise ValueError("Account not found") folders = ( await db.execute( select(MailFolder).where( and_(MailFolder.account_id == account_id, MailFolder.tenant_id == tenant_id) ) ) ).scalars().all() drafts_folder = None for f in folders: if 'draft' in f.imap_name.lower(): drafts_folder = f break if not drafts_folder: raise ValueError("No Drafts folder found for this account") # 2. Create Mail record to_str = ', '.join(data.get('to', [])) cc_str = ', '.join(data.get('cc', [])) bcc_str = ', '.join(data.get('bcc', [])) subject = data.get('subject', '') body_text = data.get('body_text', '') body_html = data.get('body_html', '') msg_id = make_msgid() now = datetime.now(UTC) mail = Mail( tenant_id=tenant_id, account_id=account_id, folder_id=drafts_folder.id, message_id=msg_id, thread_id=msg_id, subject=subject, from_address=account.email_address, to_addresses=to_str, cc_addresses=cc_str, bcc_addresses=bcc_str, body_text=body_text, body_html=body_html, body_html_sanitized=body_html, is_seen=True, is_flagged=False, is_draft=True, is_answered=False, is_forwarded=False, has_attachments=False, size_bytes=len(body_text.encode('utf-8')), received_at=now, sent_at=None, ) db.add(mail) await db.flush() # ── Notification: draft saved ── try: await create_notification( db, tenant_id, user_id, "mail_draft", "Entwurf gespeichert", subject or "Ohne Betreff", ) await db.flush() except Exception: pass # 3. Build RFC822 message and APPEND to IMAP Drafts folder password = await get_account_password(account) client = None try: client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port) await client.wait_hello_from_server() await client.login(account.username, password) # Build RFC822 message email_msg = EmailMessage() email_msg['From'] = formataddr((account.display_name, account.email_address)) if to_str: email_msg['To'] = to_str if cc_str: email_msg['Cc'] = cc_str email_msg['Subject'] = subject email_msg['Date'] = formatdate(localtime=True) email_msg['Message-ID'] = msg_id email_msg.set_content(body_text if body_text else '') if body_html: email_msg.add_alternative(body_html, subtype='html') rfc822_bytes = email_msg.as_bytes() # APPEND to Drafts folder append_resp = await client.append( drafts_folder.imap_name, r'(\\Draft)', str(int(now.timestamp())), rfc822_bytes, ) if append_resp.result != 'OK': logger.warning("save_draft: IMAP APPEND failed for drafts folder %s", drafts_folder.imap_name) else: logger.info("save_draft: appended draft %s to IMAP Drafts", mail.id) except Exception as exc: logger.warning("save_draft: IMAP append failed (non-critical): %s", exc) finally: if client is not None: try: await client.logout() except Exception: pass return mail async def update_draft( db: AsyncSession, mail_id: uuid.UUID, tenant_id: uuid.UUID, data: dict, ) -> Mail: """Update an existing draft.""" import logging logger = logging.getLogger(__name__) mail = ( await db.execute( select(Mail).where( and_(Mail.id == mail_id, Mail.tenant_id == tenant_id) ) ) ).scalar_one_or_none() if not mail: raise ValueError("Mail not found") if not mail.is_draft: raise ValueError("Mail is not a draft") # 2. Update fields to_str = ', '.join(data.get('to', [])) cc_str = ', '.join(data.get('cc', [])) bcc_str = ', '.join(data.get('bcc', [])) mail.to_addresses = to_str mail.cc_addresses = cc_str mail.bcc_addresses = bcc_str mail.subject = data.get('subject', '') mail.body_text = data.get('body_text', '') mail.body_html = data.get('body_html', '') mail.body_html_sanitized = data.get('body_html', '') mail.size_bytes = len(mail.body_text.encode('utf-8')) await db.flush() # 3. Delete old IMAP copy and APPEND new one folder = ( await db.execute( select(MailFolder).where( and_(MailFolder.id == mail.folder_id, MailFolder.tenant_id == tenant_id) ) ) ).scalar_one_or_none() if not folder: return mail account = ( await db.execute( select(MailAccount).where( and_(MailAccount.id == folder.account_id, MailAccount.tenant_id == tenant_id) ) ) ).scalar_one_or_none() if not account: return mail password = await get_account_password(account) client = None try: client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port) await client.wait_hello_from_server() await client.login(account.username, password) # Delete old copy from IMAP select_resp = await client.select(folder.imap_name) if select_resp.result == 'OK' and mail.message_id: search_resp = await client.uid_search(f'HEADER Message-ID "{mail.message_id}"') uids_raw = search_resp[1][0] if search_resp[1] and search_resp[1][0] else b'' if isinstance(uids_raw, (bytes, bytearray)): uids = uids_raw.split() else: uids = [] if uids: uid_str = uids[0].decode() if isinstance(uids[0], bytes) else str(uids[0]) await client.uid('store', uid_str, r'+FLAGS (\\Deleted)') await client.expunge() # APPEND new copy email_msg = EmailMessage() email_msg['From'] = formataddr((account.display_name, account.email_address)) if to_str: email_msg['To'] = to_str if cc_str: email_msg['Cc'] = cc_str email_msg['Subject'] = mail.subject email_msg['Date'] = formatdate(localtime=True) email_msg['Message-ID'] = mail.message_id email_msg.set_content(mail.body_text if mail.body_text else '') if mail.body_html: email_msg.add_alternative(mail.body_html, subtype='html') rfc822_bytes = email_msg.as_bytes() now = datetime.now(UTC) append_resp = await client.append( folder.imap_name, r'(\\Draft)', str(int(now.timestamp())), rfc822_bytes, ) if append_resp.result != 'OK': logger.warning("update_draft: IMAP APPEND failed for drafts folder %s", folder.imap_name) else: logger.info("update_draft: appended updated draft %s to IMAP Drafts", mail.id) except Exception as exc: logger.warning("update_draft: IMAP sync failed (non-critical): %s", exc) finally: if client is not None: try: await client.logout() except Exception: pass return mail # ─── IMAP Folder Create / Delete ─── async def imap_create_folder( db: AsyncSession, account_id: uuid.UUID, folder_name: str, tenant_id: uuid.UUID ) -> None: """Create folder on IMAP server. Connects to IMAP, creates folder with CREATE command. Non-critical: errors are logged, DB operation still succeeds. """ import logging logger = logging.getLogger(__name__) account = ( await db.execute( select(MailAccount).where( and_(MailAccount.id == account_id, MailAccount.tenant_id == tenant_id) ) ) ).scalar_one_or_none() if not account: logger.warning("imap_create_folder: account %s not found", account_id) return password = await get_account_password(account) client = None try: client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port) await client.wait_hello_from_server() await client.login(account.username, password) resp = await client.create(folder_name) if resp.result != "OK": logger.warning( "imap_create_folder: CREATE failed for %s: %s", folder_name, resp ) else: logger.info("imap_create_folder: created folder %s on IMAP", folder_name) try: await create_notification( db, account.tenant_id, account.user_id, "mail_folder", "Ordner erstellt", folder_name, ) await db.flush() except Exception: pass except Exception as exc: logger.warning("imap_create_folder: failed (non-critical): %s", exc) finally: if client is not None: try: await client.logout() except Exception: pass async def imap_delete_folder( db: AsyncSession, folder_id: uuid.UUID, tenant_id: uuid.UUID ) -> None: """Delete folder from IMAP server. Connects to IMAP, deletes folder with DELETE command. Non-critical: errors are logged, DB operation still succeeds. """ import logging logger = logging.getLogger(__name__) folder = ( await db.execute( select(MailFolder).where( and_(MailFolder.id == folder_id, MailFolder.tenant_id == tenant_id) ) ) ).scalar_one_or_none() if not folder: logger.warning("imap_delete_folder: folder %s not found", folder_id) return account = ( await db.execute( select(MailAccount).where( and_( MailAccount.id == folder.account_id, MailAccount.tenant_id == tenant_id, ) ) ) ).scalar_one_or_none() if not account: logger.warning("imap_delete_folder: account not found for folder %s", folder_id) return password = await get_account_password(account) client = None try: client = aioimaplib.IMAP4_SSL(host=account.imap_host, port=account.imap_port) await client.wait_hello_from_server() await client.login(account.username, password) resp = await client.delete(folder.imap_name) if resp.result != "OK": logger.warning( "imap_delete_folder: DELETE failed for %s: %s", folder.imap_name, resp ) else: logger.info("imap_delete_folder: deleted folder %s on IMAP", folder.imap_name) try: await create_notification( db, account.tenant_id, account.user_id, "mail_folder", "Ordner gelöscht", folder.imap_name, ) await db.flush() except Exception: pass except Exception as exc: logger.warning("imap_delete_folder: failed (non-critical): %s", exc) finally: if client is not None: try: await client.logout() except Exception: pass # ─── Auto-Sync ─── async def auto_sync_all_accounts() -> None: """Auto-sync all active mail accounts. Called periodically by the background scheduler. Iterates all active mail accounts and syncs each one. """ import logging from app.core.db import get_session_factory logger = logging.getLogger(__name__) factory = get_session_factory() async with factory() as db: accounts = ( await db.execute( select(MailAccount).where(MailAccount.is_active.is_(True)) ) ).scalars().all() if not accounts: return logger.info("auto_sync_all_accounts: syncing %d active account(s)", len(accounts)) for account in accounts: try: result = await imap_sync_account(db, account.id, account.tenant_id) logger.info( "auto_sync_all_accounts: synced account %s (%s): %s", account.id, account.username, result, ) except Exception as exc: logger.warning( "auto_sync_all_accounts: failed for account %s: %s", account.id, exc, ) try: await create_notification( db, account.tenant_id, account.user_id, "mail_sync_error", "Synchronisierung fehlgeschlagen", f"Account {account.email_address}: {exc}", ) except Exception: pass # commit per-account so partial progress is saved try: await db.commit() except Exception: await db.rollback()