Files
leocrm/app/plugins/builtins/mail/services.py
T

1298 lines
45 KiB
Python

"""Service layer for the Mail plugin: encryption, IMAP sync, SMTP send, rules, vacation, PGP."""
from __future__ import annotations
import base64
import json
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.plugins.builtins.mail.models import (
Mail,
MailAccount,
MailAttachment,
MailFolder,
MailLabel,
MailLabelAssignment,
MailRule,
MailSignature,
MailTemplate,
VacationSentLog,
)
# ─── Attachment Storage Helpers ───
MAX_ATTACHMENT_SIZE = 25 * 1024 * 1024 # 25 MB
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"},
},
)
# ─── 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,
)
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",
}
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,
"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 = 50
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) -> list[tuple[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"
"""
folders: list[tuple[str, str]] = []
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
def _build_folder_hierarchy(
imap_folders: list[tuple[str, str]],
account_id: uuid.UUID,
tenant_id: uuid.UUID,
existing_folders: dict[str, MailFolder],
) -> list[MailFolder]:
"""Create or update MailFolder records from IMAP LIST response.
Handles dot-delimited 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] = []
delimiter = '.'
# 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 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"}
password = await get_account_password(account)
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)
# 1) LIST all folders from IMAP server
list_response = await client.list('""', '"*"')
imap_folders = _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
leaf = db_folder.imap_name.rsplit(".", 1)[-1]
for srv_name in imap_names_from_server:
srv_leaf = srv_name.rsplit(".", 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
db_folders = _build_folder_hierarchy(
imap_folders, account.id, tenant_id, existing_by_imap
)
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
delimiter = '.'
folder_by_imap_post_flush = {f.imap_name: f for f in db_folders}
for folder in db_folders:
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_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
# 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
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")
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)
# Check if email already exists (by message_id + folder)
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:
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=received_at,
)
db.add(mail)
await db.flush()
synced_count += 1
# Save attachments to storage and DB
for att_data in attachments:
try:
storage_path = await _save_attachment_to_storage(
mail.id, att_data["filename"], att_data["content"]
)
attachment = MailAttachment(
tenant_id=tenant_id,
mail_id=mail.id,
filename=_sanitize_filename(att_data["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:
# Skip attachments that fail to save
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
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"<br><br>{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
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"<br><br>----- Original Message -----<br>"
f"From: {original_mail.from_address}<br>"
f"Subject: {original_mail.subject}<br><br>"
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("&nbsp;", " ")
.replace("&amp;", "&")
.replace("&lt;", "<")
.replace("&gt;", ">")
.replace("&quot;", '"')
)
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"] = [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,
}