"""Service layer for the Mail plugin: encryption, IMAP sync, SMTP send, rules, vacation, PGP.""" from __future__ import annotations import base64 import json 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 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.plugins.builtins.mail.models import ( Mail, MailAccount, MailAttachment, MailFolder, MailLabel, MailLabelAssignment, MailRule, MailSignature, MailTemplate, VacationSentLog, ) # ─── 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"}, }, ) # ─── 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["username"], encrypted_password=encrypt_password(data["password"]), is_shared=data.get("is_shared", False), is_active=True, ) db.add(account) await db.flush() # Create standard folders for fname, imap_name in [ ("Posteingang", "INBOX"), ("Postausgang", "Sent"), ("Entwürfe", "Drafts"), ("Spam", "Spam"), ]: folder = MailFolder( tenant_id=tenant_id, account_id=account.id, name=fname, imap_name=imap_name, 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.""" for field in [ "email_address", "display_name", "imap_host", "imap_port", "imap_ssl", "smtp_host", "smtp_port", "smtp_tls", "username", "is_shared", "is_active", ]: if field in data and data[field] is not None: setattr(account, field, data[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_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, "created_at": account.created_at, "updated_at": account.updated_at, } # ─── IMAP Sync Service (F-MAIL-01) ─── async def imap_sync_account( db: AsyncSession, account_id: uuid.UUID, tenant_id: uuid.UUID, ) -> dict: """Sync mail folders and messages from IMAP server. This function is designed to be called as an ARQ background job. In tests it is mocked — real implementation connects via aioimaplib. """ 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"} password = await get_account_password(account) # Import aioimaplib here so tests can mock it 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) await client.select("INBOX") # Fetch all message UIDs response = await client.uid_search("ALL") uids = response[1][0].split() if response[1] and response[1][0] else [] synced_count = 0 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 # aioimaplib Response lines: [header_bytes, data_bytearray, closing_bytes, completion_bytes] 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": body_text = part.get_payload(decode=True).decode("utf-8", errors="replace") elif ct == "text/html": body_html = part.get_payload(decode=True).decode("utf-8", errors="replace") elif part.get_filename(): attachments.append( { "filename": part.get_filename(), "mime_type": ct, "size": len(part.get_payload(decode=True) or b""), } ) 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 message_id = msg.get("Message-ID", make_msgid()) subject = msg.get("Subject", "") from_addr = msg.get("From", "") to_addrs = msg.get("To", "") cc_addrs = msg.get("Cc", "") refs = msg.get("References", "") in_reply_to = msg.get("In-Reply-To") msg.get("Date", "") # Compute thread_id from References/In-Reply-To thread_id = _compute_thread_id(message_id, refs, in_reply_to) # Get INBOX folder for this account folder = ( await db.execute( select(MailFolder).where( and_( MailFolder.account_id == account.id, MailFolder.imap_name == "INBOX", ) ) ) ).scalar_one_or_none() if folder is None: continue mail = Mail( tenant_id=tenant_id, account_id=account.id, folder_id=folder.id, 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=datetime.now(UTC), ) db.add(mail) synced_count += 1 await db.flush() # Update folder counts folder = ( await db.execute( select(MailFolder).where( and_( MailFolder.account_id == account.id, MailFolder.imap_name == "INBOX", ) ) ) ).scalar_one_or_none() if folder: 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 folder.unread_count = unread 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, ) -> dict: """Send an email via SMTP using aiosmtplib.""" cc_addrs = cc_addrs or [] bcc_addrs = bcc_addrs 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") # 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 sent_folder = ( await db.execute( select(MailFolder).where( and_( MailFolder.account_id == account.id, MailFolder.imap_name == "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() return {"status": "sent", "message_id": msg_id} except Exception as e: 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, "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"] = [ { "id": str(a.id), "filename": a.filename, "mime_type": a.mime_type, "size_bytes": a.size_bytes, "dms_file_id": str(a.dms_file_id) if a.dms_file_id else None, } 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, }