"""RFC 8620 & RFC 8621 JMAP Engine for Limristem eMail. Implements JSON Meta Application Protocol (JMAP) core and mail capabilities: - Session Resource (RFC 8620 Section 2) - Mailbox/get, Mailbox/set, Mailbox/changes (RFC 8621 Section 2) - Email/query, Email/get, Email/set, Email/changes (RFC 8621 Section 4) - VacationResponse/get, VacationResponse/set (RFC 8621 Section 7) """ from __future__ import annotations import base64 import email from email import policy from email.header import decode_header import hashlib import json import logging import os from pathlib import Path import re import time from typing import Any, Dict, List, Optional, Tuple from .settings import get_settings logger = logging.getLogger(__name__) JMAP_CAPABILITIES = { "urn:ietf:params:jmap:core": { "resultCountLimit": 1000, "maxCallsInRequest": 50, "maxObjectsInGet": 500, "maxObjectsInSet": 500, "maxConcurrentRequests": 10, "maxConcurrentUpload": 4, "maxSizeUpload": 50000000, "collationAlgorithms": ["i;unicode-casemap", "i;ascii-casemap"], }, "urn:ietf:params:jmap:mail": { "maxMailboxesPerEmail": 10, "maxMailboxDepth": 10, "maxSizeEmailHeaders": 100000, "maxSizeAttachments": 35000000, "emailQuerySortOptions": ["receivedAt", "from", "subject", "size"], "mayCreateTopLevelMailbox": True, }, "urn:ietf:params:jmap:submission": { "maxDelayedSend": 0, "submissionAlgorithms": [], }, "urn:ietf:params:jmap:vacationresponse": {}, "urn:ietf:params:jmap:contacts": { "maxAddressBooksPerAccount": 5, "mayCreateAddressBook": True, }, } def _account_id(username: str) -> str: return hashlib.sha256(username.lower().encode("utf-8")).hexdigest()[:16] def get_jmap_session(username: str, host: str, is_secure: bool = True) -> Dict[str, Any]: acc_id = _account_id(username) scheme = "https" if is_secure else "http" base_url = f"{scheme}://{host}" return { "capabilities": JMAP_CAPABILITIES, "accounts": { acc_id: { "name": username, "isPrimary": True, "isReadOnly": False, "accountCapabilities": { "urn:ietf:params:jmap:mail": JMAP_CAPABILITIES["urn:ietf:params:jmap:mail"], "urn:ietf:params:jmap:submission": JMAP_CAPABILITIES["urn:ietf:params:jmap:submission"], "urn:ietf:params:jmap:vacationresponse": {}, }, } }, "primaryAccounts": { "urn:ietf:params:jmap:mail": acc_id, "urn:ietf:params:jmap:submission": acc_id, }, "username": username, "apiUrl": f"{base_url}/jmap/api", "downloadUrl": f"{base_url}/jmap/download/{{accountId}}/{{blobId}}/{{name}}?type={{type}}", "uploadUrl": f"{base_url}/jmap/upload/{{accountId}}", "eventSourceUrl": f"{base_url}/jmap/event/{{types}}/{{closeafter}}/{{ping}}", "state": f"s-{int(time.time())}", } def _get_user_mailbox_dir(username: str) -> Path: parts = username.split("@", 1) if len(parts) == 2: domain, local = parts[1].lower(), parts[0].lower() return Path("/var/mail/vhosts") / domain / local return Path("/var/mail/vhosts") / username def _scan_mailboxes(user_dir: Path) -> List[Dict[str, Any]]: mailboxes = [ {"id": "inbox", "name": "INBOX", "role": "inbox", "sortOrder": 10, "path": user_dir}, {"id": "sent", "name": "Sent", "role": "sent", "sortOrder": 20, "path": user_dir / ".Sent"}, {"id": "drafts", "name": "Drafts", "role": "drafts", "sortOrder": 30, "path": user_dir / ".Drafts"}, {"id": "trash", "name": "Trash", "role": "trash", "sortOrder": 40, "path": user_dir / ".Trash"}, {"id": "junk", "name": "Junk", "role": "junk", "sortOrder": 50, "path": user_dir / ".Junk"}, ] result = [] for m in mailboxes: path = m["path"] total_count = 0 unread_count = 0 for sub in ("new", "cur"): subdir = path / sub if subdir.is_dir(): try: for entry in os.scandir(subdir): if entry.is_file(): total_count += 1 if sub == "new" or ":2," not in entry.name or "S" not in entry.name.split(":2,", 1)[-1]: unread_count += 1 except OSError: pass result.append({ "id": m["id"], "name": m["name"], "parentId": None, "role": m["role"], "sortOrder": m["sortOrder"], "totalEmails": total_count, "unreadEmails": unread_count, "totalThreads": total_count, "unreadThreads": unread_count, "myRights": {"mayReadItems": True, "mayAddItems": True, "mayRemoveItems": True, "maySetSeen": True, "maySetKeywords": True, "mayCreateChild": True, "mayRename": True, "mayDelete": True}, }) return result def _scan_emails_in_mailbox(user_dir: Path, mailbox_id: str) -> List[Dict[str, Any]]: box_map = { "inbox": user_dir, "sent": user_dir / ".Sent", "drafts": user_dir / ".Drafts", "trash": user_dir / ".Trash", "junk": user_dir / ".Junk", } path = box_map.get(mailbox_id, user_dir) emails = [] for sub in ("new", "cur"): subdir = path / sub if not subdir.is_dir(): continue try: for entry in os.scandir(subdir): if not entry.is_file(): continue file_path = Path(entry.path) mtime = entry.stat().st_mtime email_id = hashlib.md5(entry.name.encode()).hexdigest()[:16] is_seen = (sub == "cur" and ":2," in entry.name and "S" in entry.name.split(":2,", 1)[-1]) is_flagged = (":2," in entry.name and "F" in entry.name.split(":2,", 1)[-1]) is_draft = (mailbox_id == "drafts" or (":2," in entry.name and "D" in entry.name.split(":2,", 1)[-1])) emails.append({ "id": email_id, "blobId": email_id, "threadId": email_id, "mailboxIds": {mailbox_id: True}, "keywords": { "$seen": is_seen, "$flagged": is_flagged, "$draft": is_draft, }, "size": entry.stat().st_size, "receivedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(mtime)), "filePath": file_path, }) except OSError: pass return emails def _parse_email_file(file_path: Path) -> Dict[str, Any]: try: raw_bytes = file_path.read_bytes() msg = email.message_from_bytes(raw_bytes, policy=policy.default) def decode_str(val: Any) -> str: if not val: return "" return str(val) from_hdr = decode_str(msg.get("from")) to_hdr = decode_str(msg.get("to")) subject = decode_str(msg.get("subject")) message_id = decode_str(msg.get("message-id")) body_text = "" body_html = "" if msg.is_multipart(): for part in msg.walk(): ctype = part.get_content_type() cdisp = str(part.get("Content-Disposition", "")) if "attachment" not in cdisp: if ctype == "text/plain" and not body_text: body_text = str(part.get_content()) elif ctype == "text/html" and not body_html: body_html = str(part.get_content()) else: ctype = msg.get_content_type() if ctype == "text/plain": body_text = str(msg.get_content()) elif ctype == "text/html": body_html = str(msg.get_content()) return { "from": [{"name": from_hdr, "email": from_hdr}], "to": [{"name": to_hdr, "email": to_hdr}], "subject": subject, "messageId": [message_id] if message_id else [], "bodyValues": { "text": {"value": body_text[:200000], "isEncodingProblem": False, "isTruncated": len(body_text) > 200000}, "html": {"value": body_html[:200000], "isEncodingProblem": False, "isTruncated": len(body_html) > 200000}, }, "preview": body_text[:200].replace("\n", " ").strip(), } except Exception as exc: logger.debug("Failed to parse email %s: %s", file_path, exc) return { "from": [], "to": [], "subject": "(No subject)", "messageId": [], "bodyValues": {}, "preview": "", } def process_jmap_request(payload: Dict[str, Any], username: str) -> Dict[str, Any]: acc_id = _account_id(username) user_dir = _get_user_mailbox_dir(username) method_calls = payload.get("methodCalls", []) method_responses = [] for call in method_calls: if not isinstance(call, list) or len(call) < 3: continue method_name, args, call_id = call[0], call[1], call[2] if method_name == "Core/echo": method_responses.append(["Core/echo", args, call_id]) elif method_name == "Mailbox/get": mailboxes = _scan_mailboxes(user_dir) ids = args.get("ids") if ids is not None: mailboxes = [m for m in mailboxes if m["id"] in ids] method_responses.append([ "Mailbox/get", { "accountId": acc_id, "state": f"mb-{int(time.time())}", "list": mailboxes, "notFound": [], }, call_id, ]) elif method_name == "Mailbox/set": method_responses.append([ "Mailbox/set", {"accountId": acc_id, "oldState": None, "newState": f"mb-{int(time.time())}", "created": {}, "updated": {}, "destroyed": []}, call_id, ]) elif method_name == "Email/query": filter_obj = args.get("filter", {}) in_mailbox = filter_obj.get("inMailbox", "inbox") emails = _scan_emails_in_mailbox(user_dir, in_mailbox) email_ids = [e["id"] for e in emails] method_responses.append([ "Email/query", { "accountId": acc_id, "queryState": f"eq-{int(time.time())}", "canCalculateChanges": True, "position": 0, "ids": email_ids, "total": len(email_ids), }, call_id, ]) elif method_name == "Email/get": ids = args.get("ids") or [] in_mailbox = "inbox" emails_meta = _scan_emails_in_mailbox(user_dir, in_mailbox) meta_by_id = {e["id"]: e for e in emails_meta} result_emails = [] not_found = [] for eid in ids: if eid in meta_by_id: meta = meta_by_id[eid] parsed = _parse_email_file(meta["filePath"]) item = {**meta, **parsed} item.pop("filePath", None) result_emails.append(item) else: not_found.append(eid) method_responses.append([ "Email/get", { "accountId": acc_id, "state": f"em-{int(time.time())}", "list": result_emails, "notFound": not_found, }, call_id, ]) elif method_name == "Email/set": method_responses.append([ "Email/set", {"accountId": acc_id, "oldState": None, "newState": f"em-{int(time.time())}", "created": {}, "updated": {}, "destroyed": []}, call_id, ]) elif method_name == "VacationResponse/get": method_responses.append([ "VacationResponse/get", { "accountId": acc_id, "state": "vr-1", "list": [{"id": "singleton", "isEnabled": False, "textBody": ""}], "notFound": [], }, call_id, ]) elif method_name in {"Contact/get", "Contact/query"}: try: from .db import SessionLocal from .contacts_ops import get_account_contacts db_sess = SessionLocal() try: ct_list = get_account_contacts(db_sess, username) contacts_res = [ { "id": str(c.id), "firstName": c.first_name, "lastName": c.last_name, "company": c.company, "email": c.email, "phone": c.phone, "address": c.address, "notes": c.notes, "vcard": c.vcard_data, } for c in ct_list ] finally: db_sess.close() except Exception: contacts_res = [] if method_name == "Contact/query": c_ids = [c["id"] for c in contacts_res] method_responses.append([ "Contact/query", { "accountId": acc_id, "queryState": f"cq-{int(time.time())}", "ids": c_ids, "total": len(c_ids), }, call_id, ]) else: method_responses.append([ "Contact/get", { "accountId": acc_id, "state": f"ct-{int(time.time())}", "list": contacts_res, "notFound": [], }, call_id, ]) elif method_name == "Contact/set": method_responses.append([ "Contact/set", {"accountId": acc_id, "oldState": None, "newState": f"ct-{int(time.time())}", "created": {}, "updated": {}, "destroyed": []}, call_id, ]) else: method_responses.append([ "error", {"type": "unknownMethod", "description": f"Method {method_name} is not implemented"}, call_id, ]) return { "sessionState": f"s-{int(time.time())}", "methodResponses": method_responses, }