872 lines
31 KiB
Python
872 lines
31 KiB
Python
from __future__ import annotations
|
|
|
|
import imaplib
|
|
import re
|
|
import socket
|
|
import ssl
|
|
import time
|
|
from dataclasses import dataclass
|
|
from email import policy
|
|
from email.message import EmailMessage
|
|
from email.parser import BytesParser
|
|
from typing import Any
|
|
|
|
from govoplan_mail.backend.config import ImapConfig, TransportSecurity
|
|
from govoplan_mail.backend.dev.mock_mailbox import (
|
|
MOCK_IMAP_FOLDERS,
|
|
consume_fail_next_imap,
|
|
get_record,
|
|
is_mock_imap_host,
|
|
list_records,
|
|
record_imap_append,
|
|
)
|
|
|
|
|
|
class ImapConfigurationError(ValueError):
|
|
"""Raised when IMAP settings are incomplete or inconsistent."""
|
|
|
|
|
|
class ImapAppendError(RuntimeError):
|
|
"""Raised when APPENDing to Sent fails.
|
|
|
|
temporary=True means retrying later may help. temporary=False means the
|
|
configuration or mailbox choice probably needs user/admin attention.
|
|
"""
|
|
|
|
def __init__(self, message: str, *, temporary: bool | None = None):
|
|
super().__init__(message)
|
|
self.temporary = temporary
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ImapLoginTestResult:
|
|
host: str
|
|
port: int
|
|
security: str
|
|
authenticated: bool
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ImapMailboxInfo:
|
|
name: str
|
|
flags: list[str]
|
|
message_count: int | None = None
|
|
unseen_count: int | None = None
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ImapFolderListResult:
|
|
host: str
|
|
port: int
|
|
security: str
|
|
folders: list[ImapMailboxInfo]
|
|
detected_sent_folder: str | None = None
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ImapMailboxAttachmentInfo:
|
|
filename: str | None
|
|
content_type: str
|
|
size_bytes: int
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ImapMailboxMessageSummary:
|
|
uid: str
|
|
folder: str
|
|
subject: str | None
|
|
from_header: str | None
|
|
to_header: str | None
|
|
cc_header: str | None
|
|
date: str | None
|
|
message_id: str | None
|
|
flags: list[str]
|
|
size_bytes: int | None
|
|
body_preview: str | None
|
|
attachment_count: int
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ImapMailboxMessageDetail:
|
|
uid: str
|
|
folder: str
|
|
subject: str | None
|
|
from_header: str | None
|
|
to_header: str | None
|
|
cc_header: str | None
|
|
date: str | None
|
|
message_id: str | None
|
|
flags: list[str]
|
|
size_bytes: int | None
|
|
body_preview: str | None
|
|
body_text: str | None
|
|
body_html: str | None
|
|
headers: dict[str, str]
|
|
attachments: list[ImapMailboxAttachmentInfo]
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ImapMailboxMessageListResult:
|
|
host: str
|
|
port: int
|
|
security: str
|
|
folder: str
|
|
messages: list[ImapMailboxMessageSummary]
|
|
total_count: int
|
|
offset: int
|
|
limit: int
|
|
uidvalidity: str | None = None
|
|
cursor_reset: bool = False
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ImapMailboxMessageResult:
|
|
host: str
|
|
port: int
|
|
security: str
|
|
folder: str
|
|
message: ImapMailboxMessageDetail
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ImapAppendResult:
|
|
host: str
|
|
port: int
|
|
security: str
|
|
folder: str
|
|
bytes_appended: int
|
|
response: str | None = None
|
|
|
|
|
|
def _require_imap_config(config: ImapConfig) -> tuple[str, int]:
|
|
if not config.host:
|
|
raise ImapConfigurationError("IMAP host is required")
|
|
if not config.port:
|
|
raise ImapConfigurationError("IMAP port is required")
|
|
if bool(config.username) != bool(config.password):
|
|
raise ImapConfigurationError("IMAP username and password must be provided together, or both omitted")
|
|
return config.host, config.port
|
|
|
|
|
|
def _open_imap(config: ImapConfig) -> imaplib.IMAP4:
|
|
host, port = _require_imap_config(config)
|
|
context = ssl.create_default_context()
|
|
|
|
try:
|
|
if config.security == TransportSecurity.TLS:
|
|
client: imaplib.IMAP4 = imaplib.IMAP4_SSL(host=host, port=port, timeout=config.timeout_seconds, ssl_context=context)
|
|
else:
|
|
client = imaplib.IMAP4(host=host, port=port, timeout=config.timeout_seconds)
|
|
if config.security == TransportSecurity.STARTTLS:
|
|
typ, data = client.starttls(ssl_context=context)
|
|
if typ != "OK":
|
|
raise ImapAppendError(f"IMAP STARTTLS failed: {data!r}", temporary=True)
|
|
|
|
if config.username and config.password:
|
|
typ, data = client.login(config.username, config.password)
|
|
if typ != "OK":
|
|
raise ImapAppendError(f"IMAP login failed: {data!r}", temporary=False)
|
|
return client
|
|
except Exception:
|
|
try:
|
|
client.logout() # type: ignore[possibly-undefined]
|
|
except Exception:
|
|
pass
|
|
raise
|
|
|
|
|
|
def _decode_item(item: bytes | str | None) -> str:
|
|
if item is None:
|
|
return ""
|
|
if isinstance(item, bytes):
|
|
return item.decode("utf-8", errors="replace")
|
|
return item
|
|
|
|
|
|
def _unquote_imap_token(value: str) -> str:
|
|
value = value.strip()
|
|
if len(value) >= 2 and value[0] == '"' and value[-1] == '"':
|
|
value = value[1:-1]
|
|
value = value.replace('\\"', '"').replace('\\\\', '\\')
|
|
return value
|
|
|
|
|
|
def _extract_mailbox_name(list_response_line: bytes | str) -> tuple[str, set[str]] | None:
|
|
r"""Best-effort parser for IMAP LIST response lines.
|
|
|
|
RFC 3501 LIST responses contain attributes, hierarchy delimiter, then mailbox
|
|
name. Some servers quote both delimiter and mailbox::
|
|
|
|
(\HasNoChildren \Sent) "/" "Sent"
|
|
|
|
Others quote only the delimiter and leave the mailbox as an atom::
|
|
|
|
(\HasNoChildren \Sent) "/" Sent
|
|
|
|
The parser must therefore parse the delimiter token separately instead of
|
|
blindly taking the last quoted value.
|
|
"""
|
|
|
|
line = _decode_item(list_response_line).strip()
|
|
match = re.match(
|
|
r'^\((?P<flags>[^)]*)\)\s+'
|
|
r'(?P<delimiter>"(?:[^"\\]|\\.)*"|NIL|[^\s]+)\s+'
|
|
r'(?P<mailbox>.+?)\s*$',
|
|
line,
|
|
re.IGNORECASE,
|
|
)
|
|
if match:
|
|
flags = {part.lower() for part in match.group("flags").split()}
|
|
mailbox = _unquote_imap_token(match.group("mailbox"))
|
|
if mailbox:
|
|
return mailbox, flags
|
|
return None
|
|
|
|
# Fallback for non-standard server lines: prefer the final token.
|
|
parts = line.split(maxsplit=2)
|
|
if len(parts) >= 3:
|
|
flags_text = parts[0].strip("()")
|
|
flags = {part.lower() for part in flags_text.split()}
|
|
mailbox = _unquote_imap_token(parts[2])
|
|
if mailbox:
|
|
return mailbox, flags
|
|
return None
|
|
|
|
|
|
def _detect_sent_folder(parsed: list[tuple[str, set[str]]]) -> str | None:
|
|
for name, flags in parsed:
|
|
if "\\sent" in flags or "\\sentmail" in flags:
|
|
return name
|
|
|
|
common_names = [
|
|
"Sent",
|
|
"Sent Items",
|
|
"Sent Messages",
|
|
"Gesendet",
|
|
"Gesendete Elemente",
|
|
"INBOX.Sent",
|
|
"INBOX/Sent",
|
|
]
|
|
names = {name.lower(): name for name, _ in parsed}
|
|
for candidate in common_names:
|
|
if candidate.lower() in names:
|
|
return names[candidate.lower()]
|
|
return None
|
|
|
|
|
|
def discover_sent_folder(client: imaplib.IMAP4) -> str | None:
|
|
typ, data = client.list()
|
|
if typ != "OK" or not data:
|
|
return None
|
|
|
|
parsed: list[tuple[str, set[str]]] = []
|
|
for item in data:
|
|
extracted = _extract_mailbox_name(item)
|
|
if extracted:
|
|
parsed.append(extracted)
|
|
|
|
return _detect_sent_folder(parsed)
|
|
|
|
|
|
def _effective_sent_folder(*, config: ImapConfig, requested_folder: str | None, client: imaplib.IMAP4) -> str:
|
|
if requested_folder and requested_folder != "auto":
|
|
return requested_folder
|
|
if config.sent_folder and config.sent_folder != "auto":
|
|
return config.sent_folder
|
|
discovered = discover_sent_folder(client)
|
|
if discovered:
|
|
return discovered
|
|
raise ImapConfigurationError("Could not discover Sent folder; configure delivery.imap_append_sent.folder or server.imap.sent_folder")
|
|
|
|
|
|
def test_imap_login(*, imap_config: ImapConfig) -> ImapLoginTestResult:
|
|
"""Open an IMAP connection and authenticate if credentials are configured.
|
|
|
|
This is a side-effect-free connection test for the WebUI. It does not select
|
|
a mailbox and does not append any message.
|
|
"""
|
|
|
|
host, port = _require_imap_config(imap_config)
|
|
if is_mock_imap_host(imap_config.host):
|
|
return ImapLoginTestResult(
|
|
host=host,
|
|
port=port,
|
|
security=imap_config.security.value,
|
|
authenticated=bool(imap_config.username and imap_config.password),
|
|
)
|
|
|
|
client = _open_imap(imap_config)
|
|
try:
|
|
return ImapLoginTestResult(
|
|
host=host,
|
|
port=port,
|
|
security=imap_config.security.value,
|
|
authenticated=bool(imap_config.username and imap_config.password),
|
|
)
|
|
finally:
|
|
try:
|
|
client.logout()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def list_imap_folders(*, imap_config: ImapConfig) -> ImapFolderListResult:
|
|
"""Return folders visible through IMAP LIST and the best sent-folder guess."""
|
|
|
|
host, port = _require_imap_config(imap_config)
|
|
if is_mock_imap_host(imap_config.host):
|
|
records = list_records(limit=500)
|
|
folders = []
|
|
for item in MOCK_IMAP_FOLDERS:
|
|
name = str(item["name"])
|
|
count = sum(1 for record in records if _mock_folder_matches(record, name))
|
|
folders.append(ImapMailboxInfo(name=name, flags=list(item.get("flags") or []), message_count=count, unseen_count=None))
|
|
return ImapFolderListResult(
|
|
host=host,
|
|
port=port,
|
|
security=imap_config.security.value,
|
|
folders=folders,
|
|
detected_sent_folder="Sent",
|
|
)
|
|
|
|
client = _open_imap(imap_config)
|
|
try:
|
|
typ, data = client.list()
|
|
if typ != "OK":
|
|
raise ImapAppendError(f"IMAP folder listing failed: {data!r}", temporary=True)
|
|
|
|
parsed: list[tuple[str, set[str]]] = []
|
|
folders: list[ImapMailboxInfo] = []
|
|
for item in data or []:
|
|
extracted = _extract_mailbox_name(item)
|
|
if not extracted:
|
|
continue
|
|
name, flags = extracted
|
|
parsed.append((name, flags))
|
|
message_count, unseen_count = (None, None) if _has_folder_flag(flags, "noselect") else _imap_folder_status(client, name)
|
|
folders.append(ImapMailboxInfo(name=name, flags=sorted(flags), message_count=message_count, unseen_count=unseen_count))
|
|
|
|
return ImapFolderListResult(
|
|
host=host,
|
|
port=port,
|
|
security=imap_config.security.value,
|
|
folders=folders,
|
|
detected_sent_folder=_detect_sent_folder(parsed),
|
|
)
|
|
finally:
|
|
try:
|
|
client.logout()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _has_folder_flag(flags: set[str], flag: str) -> bool:
|
|
wanted = flag.casefold().lstrip("\\")
|
|
return any(item.casefold().lstrip("\\") == wanted for item in flags)
|
|
|
|
|
|
def _quote_mailbox_name(name: str) -> str:
|
|
return "\"" + name.replace("\\", "\\\\").replace("\"", "\\\"") + "\""
|
|
|
|
|
|
def _imap_folder_status(client: imaplib.IMAP4, folder: str) -> tuple[int | None, int | None]:
|
|
try:
|
|
typ, data = client.status(_quote_mailbox_name(folder), "(MESSAGES UNSEEN)")
|
|
except Exception:
|
|
return None, None
|
|
if typ != "OK":
|
|
return None, None
|
|
text = " ".join(_decode_item(item) for item in data or [])
|
|
messages_match = re.search(r"\bMESSAGES\s+(\d+)", text, re.IGNORECASE)
|
|
unseen_match = re.search(r"\bUNSEEN\s+(\d+)", text, re.IGNORECASE)
|
|
return (
|
|
int(messages_match.group(1)) if messages_match else None,
|
|
int(unseen_match.group(1)) if unseen_match else None,
|
|
)
|
|
|
|
|
|
def _parse_message(raw: bytes) -> EmailMessage:
|
|
return BytesParser(policy=policy.default).parsebytes(raw)
|
|
|
|
|
|
def _header_text(message: EmailMessage, name: str) -> str | None:
|
|
value = message.get(name)
|
|
return str(value) if value is not None else None
|
|
|
|
|
|
def _body_content(message: EmailMessage, prefer: tuple[str, ...]) -> str | None:
|
|
try:
|
|
body = message.get_body(preferencelist=prefer) if message.is_multipart() else message
|
|
if body is None:
|
|
return None
|
|
content = body.get_content()
|
|
except Exception:
|
|
return None
|
|
text = str(content).strip()
|
|
return text or None
|
|
|
|
|
|
def _body_preview(message: EmailMessage) -> str | None:
|
|
text = _body_content(message, ("plain", "html"))
|
|
if not text:
|
|
return None
|
|
preview = re.sub(r"\s+", " ", text).strip()
|
|
return preview[:600] or None
|
|
|
|
|
|
def _attachment_infos(message: EmailMessage) -> list[ImapMailboxAttachmentInfo]:
|
|
attachments: list[ImapMailboxAttachmentInfo] = []
|
|
for part in message.iter_attachments():
|
|
payload = part.get_payload(decode=True) or b""
|
|
attachments.append(
|
|
ImapMailboxAttachmentInfo(
|
|
filename=part.get_filename(),
|
|
content_type=part.get_content_type(),
|
|
size_bytes=len(payload),
|
|
)
|
|
)
|
|
return attachments
|
|
|
|
|
|
def _message_summary_from_headers(*, uid: str, folder: str, headers: bytes, flags: list[str], size_bytes: int | None) -> ImapMailboxMessageSummary:
|
|
message = _parse_message(headers.rstrip() + b"\r\n\r\n")
|
|
return ImapMailboxMessageSummary(
|
|
uid=uid,
|
|
folder=folder,
|
|
subject=_header_text(message, "Subject"),
|
|
from_header=_header_text(message, "From"),
|
|
to_header=_header_text(message, "To"),
|
|
cc_header=_header_text(message, "Cc"),
|
|
date=_header_text(message, "Date"),
|
|
message_id=_header_text(message, "Message-ID"),
|
|
flags=flags,
|
|
size_bytes=size_bytes,
|
|
body_preview=None,
|
|
attachment_count=0,
|
|
)
|
|
|
|
|
|
def _message_summary_from_raw(*, uid: str, folder: str, raw: bytes, flags: list[str], size_bytes: int | None) -> ImapMailboxMessageSummary:
|
|
message = _parse_message(raw)
|
|
attachments = _attachment_infos(message)
|
|
return ImapMailboxMessageSummary(
|
|
uid=uid,
|
|
folder=folder,
|
|
subject=_header_text(message, "Subject"),
|
|
from_header=_header_text(message, "From"),
|
|
to_header=_header_text(message, "To"),
|
|
cc_header=_header_text(message, "Cc"),
|
|
date=_header_text(message, "Date"),
|
|
message_id=_header_text(message, "Message-ID"),
|
|
flags=flags,
|
|
size_bytes=size_bytes if size_bytes is not None else len(raw),
|
|
body_preview=_body_preview(message),
|
|
attachment_count=len(attachments),
|
|
)
|
|
|
|
|
|
def _message_detail_from_raw(*, uid: str, folder: str, raw: bytes, flags: list[str], size_bytes: int | None) -> ImapMailboxMessageDetail:
|
|
message = _parse_message(raw)
|
|
attachments = _attachment_infos(message)
|
|
return ImapMailboxMessageDetail(
|
|
uid=uid,
|
|
folder=folder,
|
|
subject=_header_text(message, "Subject"),
|
|
from_header=_header_text(message, "From"),
|
|
to_header=_header_text(message, "To"),
|
|
cc_header=_header_text(message, "Cc"),
|
|
date=_header_text(message, "Date"),
|
|
message_id=_header_text(message, "Message-ID"),
|
|
flags=flags,
|
|
size_bytes=size_bytes if size_bytes is not None else len(raw),
|
|
body_preview=_body_preview(message),
|
|
body_text=_body_content(message, ("plain",)),
|
|
body_html=_body_content(message, ("html",)),
|
|
headers={str(key): str(value) for key, value in message.items()},
|
|
attachments=attachments,
|
|
)
|
|
|
|
|
|
def _parse_fetch_metadata(metadata: str) -> tuple[str | None, list[str], int | None]:
|
|
uid_match = re.search(r"\bUID\s+(\d+)", metadata, re.IGNORECASE)
|
|
size_match = re.search(r"\bRFC822\.SIZE\s+(\d+)", metadata, re.IGNORECASE)
|
|
flags_match = re.search(r"\bFLAGS\s+\(([^)]*)\)", metadata, re.IGNORECASE)
|
|
flags = flags_match.group(1).split() if flags_match else []
|
|
return uid_match.group(1) if uid_match else None, flags, int(size_match.group(1)) if size_match else None
|
|
|
|
|
|
def _parse_fetch_response(data: list[Any] | tuple[Any, ...] | None) -> tuple[str | None, list[str], int | None, bytes]:
|
|
metadata_parts: list[str] = []
|
|
raw: bytes | None = None
|
|
for item in data or []:
|
|
if isinstance(item, tuple):
|
|
metadata_parts.append(_decode_item(item[0]))
|
|
if isinstance(item[1], bytes):
|
|
raw = item[1]
|
|
else:
|
|
metadata_parts.append(_decode_item(item))
|
|
metadata = " ".join(part for part in metadata_parts if part)
|
|
uid, flags, size_bytes = _parse_fetch_metadata(metadata)
|
|
if raw is None:
|
|
raise ImapAppendError(f"IMAP message fetch returned no message body: {metadata!r}", temporary=True)
|
|
return uid, flags, size_bytes, raw
|
|
|
|
|
|
def _parse_fetch_sequence(metadata: str) -> str | None:
|
|
match = re.match(r"\s*(\d+)\s+\(", metadata)
|
|
return match.group(1) if match else None
|
|
|
|
|
|
def _parse_fetch_parts_with_sequence(data: list[Any] | tuple[Any, ...] | None) -> list[tuple[str | None, str | None, list[str], int | None, bytes]]:
|
|
parts: list[tuple[str | None, str | None, list[str], int | None, bytes]] = []
|
|
for item in data or []:
|
|
if not isinstance(item, tuple) or not isinstance(item[1], bytes):
|
|
continue
|
|
metadata = _decode_item(item[0])
|
|
sequence = _parse_fetch_sequence(metadata)
|
|
uid, flags, size_bytes = _parse_fetch_metadata(metadata)
|
|
parts.append((sequence, uid, flags, size_bytes, item[1]))
|
|
return parts
|
|
|
|
|
|
def _select_readonly(client: imaplib.IMAP4, folder: str) -> tuple[int, str | None]:
|
|
typ, data = client.select(folder, readonly=True)
|
|
if typ != "OK":
|
|
raise ImapAppendError(f"IMAP folder {folder!r} could not be opened read-only: {data!r}", temporary=False)
|
|
selected_count = _decode_item(data[0] if data else None).strip()
|
|
try:
|
|
total_count = max(0, int(selected_count))
|
|
except ValueError:
|
|
total_count = 0
|
|
uidvalidity = None
|
|
try:
|
|
_, uidvalidity_data = client.response("UIDVALIDITY")
|
|
if uidvalidity_data:
|
|
uidvalidity = _decode_item(uidvalidity_data[0]).strip() or None
|
|
except Exception:
|
|
uidvalidity = None
|
|
return total_count, uidvalidity
|
|
|
|
|
|
def _fetch_message_by_uid(client: imaplib.IMAP4, uid: str) -> tuple[str, list[str], int | None, bytes]:
|
|
typ, data = client.uid("fetch", str(uid), "(UID FLAGS RFC822.SIZE BODY.PEEK[])")
|
|
if typ != "OK":
|
|
raise ImapAppendError(f"IMAP message {uid!r} fetch failed: {data!r}", temporary=True)
|
|
fetched_uid, flags, size_bytes, raw = _parse_fetch_response(data)
|
|
return fetched_uid or str(uid), flags, size_bytes, raw
|
|
|
|
|
|
def _fetch_message_by_sequence(client: imaplib.IMAP4, sequence: str) -> tuple[str, list[str], int | None, bytes]:
|
|
typ, data = client.fetch(str(sequence), "(UID FLAGS RFC822.SIZE BODY.PEEK[])")
|
|
if typ != "OK":
|
|
raise ImapAppendError(f"IMAP message sequence {sequence!r} fetch failed: {data!r}", temporary=True)
|
|
fetched_uid, flags, size_bytes, raw = _parse_fetch_response(data)
|
|
if not fetched_uid:
|
|
raise ImapAppendError(f"IMAP message sequence {sequence!r} fetch returned no UID", temporary=True)
|
|
return fetched_uid, flags, size_bytes, raw
|
|
|
|
|
|
def _sequence_set(sequences: list[str]) -> str:
|
|
sequence_numbers = sorted({int(sequence) for sequence in sequences if str(sequence).isdigit()})
|
|
if not sequence_numbers:
|
|
return ",".join(str(sequence) for sequence in sequences)
|
|
expected_range = list(range(sequence_numbers[0], sequence_numbers[-1] + 1))
|
|
if sequence_numbers == expected_range:
|
|
return f"{sequence_numbers[0]}:{sequence_numbers[-1]}"
|
|
return ",".join(str(sequence) for sequence in sequence_numbers)
|
|
|
|
|
|
def _fetch_message_summaries_by_sequence(client: imaplib.IMAP4, sequences: list[str], folder: str) -> list[ImapMailboxMessageSummary]:
|
|
if not sequences:
|
|
return []
|
|
typ, data = client.fetch(_sequence_set(sequences), "(UID FLAGS RFC822.SIZE BODY.PEEK[HEADER.FIELDS (SUBJECT FROM TO CC DATE MESSAGE-ID)])")
|
|
if typ != "OK":
|
|
raise ImapAppendError(f"IMAP message summary fetch failed: {data!r}", temporary=True)
|
|
|
|
by_sequence: dict[str, ImapMailboxMessageSummary] = {}
|
|
for sequence, fetched_uid, flags, size_bytes, headers in _parse_fetch_parts_with_sequence(data):
|
|
if not sequence or not fetched_uid:
|
|
continue
|
|
by_sequence[sequence] = _message_summary_from_headers(uid=fetched_uid, folder=folder, headers=headers, flags=flags, size_bytes=size_bytes)
|
|
|
|
summaries: list[ImapMailboxMessageSummary] = []
|
|
for sequence in sequences:
|
|
summary = by_sequence.get(str(sequence))
|
|
if summary is not None:
|
|
summaries.append(summary)
|
|
continue
|
|
fetched_uid, flags, size_bytes, raw = _fetch_message_by_sequence(client, sequence)
|
|
summaries.append(_message_summary_from_raw(uid=fetched_uid, folder=folder, raw=raw, flags=flags, size_bytes=size_bytes))
|
|
return summaries
|
|
|
|
|
|
def _search_message_uids(client: imaplib.IMAP4) -> list[str]:
|
|
typ, data = client.uid("search", None, "ALL")
|
|
if typ != "OK":
|
|
raise ImapAppendError(f"IMAP UID search failed: {data!r}", temporary=True)
|
|
uids: set[int] = set()
|
|
for item in data or []:
|
|
for token in _decode_item(item).split():
|
|
if token.isdigit():
|
|
uids.add(int(token))
|
|
return [str(uid) for uid in sorted(uids, reverse=True)]
|
|
|
|
|
|
def _paged_descending_uids(
|
|
uids: list[str],
|
|
*,
|
|
offset: int,
|
|
limit: int,
|
|
after_uid: str | None = None,
|
|
) -> tuple[list[str], int, bool]:
|
|
if not uids:
|
|
return [], 0, False
|
|
cursor_reset = False
|
|
if after_uid:
|
|
try:
|
|
start = uids.index(str(after_uid)) + 1
|
|
except ValueError:
|
|
start = 0
|
|
cursor_reset = True
|
|
else:
|
|
start = min(max(0, offset), len(uids))
|
|
return uids[start:start + limit], start, cursor_reset
|
|
|
|
|
|
def _fetch_message_summaries_by_uid(client: imaplib.IMAP4, uids: list[str], folder: str) -> list[ImapMailboxMessageSummary]:
|
|
if not uids:
|
|
return []
|
|
typ, data = client.uid("fetch", _sequence_set(uids), "(UID FLAGS RFC822.SIZE BODY.PEEK[HEADER.FIELDS (SUBJECT FROM TO CC DATE MESSAGE-ID)])")
|
|
if typ != "OK":
|
|
raise ImapAppendError(f"IMAP message summary fetch failed: {data!r}", temporary=True)
|
|
|
|
by_uid: dict[str, ImapMailboxMessageSummary] = {}
|
|
for _sequence, fetched_uid, flags, size_bytes, headers in _parse_fetch_parts_with_sequence(data):
|
|
if not fetched_uid:
|
|
continue
|
|
by_uid[fetched_uid] = _message_summary_from_headers(uid=fetched_uid, folder=folder, headers=headers, flags=flags, size_bytes=size_bytes)
|
|
|
|
summaries: list[ImapMailboxMessageSummary] = []
|
|
for uid in uids:
|
|
summary = by_uid.get(str(uid))
|
|
if summary is not None:
|
|
summaries.append(summary)
|
|
continue
|
|
fetched_uid, flags, size_bytes, raw = _fetch_message_by_uid(client, uid)
|
|
summaries.append(_message_summary_from_raw(uid=fetched_uid, folder=folder, raw=raw, flags=flags, size_bytes=size_bytes))
|
|
return summaries
|
|
|
|
|
|
def _mock_record_folder(record: dict[str, Any]) -> str:
|
|
folder = str(record.get("folder") or "").strip()
|
|
if folder:
|
|
return folder
|
|
return "Sent" if record.get("kind") == "imap_append" else "INBOX"
|
|
|
|
|
|
def _mock_folder_matches(record: dict[str, Any], folder: str) -> bool:
|
|
return _mock_record_folder(record).casefold() == folder.casefold()
|
|
|
|
|
|
def _mock_raw_bytes(record: dict[str, Any]) -> bytes:
|
|
raw = record.get("raw_eml")
|
|
if isinstance(raw, str) and raw:
|
|
return raw.encode("utf-8", errors="replace")
|
|
headers = record.get("headers") if isinstance(record.get("headers"), dict) else {}
|
|
lines = [f"{key}: {value}" for key, value in headers.items()]
|
|
lines.append("")
|
|
lines.append(str(record.get("body_preview") or ""))
|
|
return "\n".join(lines).encode("utf-8", errors="replace")
|
|
|
|
|
|
def list_imap_messages(
|
|
*,
|
|
imap_config: ImapConfig,
|
|
folder: str = "INBOX",
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
after_uid: str | None = None,
|
|
expected_uidvalidity: str | None = None,
|
|
) -> ImapMailboxMessageListResult:
|
|
"""List mailbox messages without mutating read/unread state."""
|
|
|
|
host, port = _require_imap_config(imap_config)
|
|
folder = (folder or "INBOX").strip() or "INBOX"
|
|
limit = max(1, min(limit, 100))
|
|
offset = max(0, offset)
|
|
if is_mock_imap_host(imap_config.host):
|
|
records = [record for record in list_records(limit=500) if _mock_folder_matches(record, folder)]
|
|
cursor_reset = False
|
|
if expected_uidvalidity and expected_uidvalidity != "mock-v1":
|
|
effective_offset = 0
|
|
cursor_reset = True
|
|
elif after_uid:
|
|
record_ids = [str(record.get("id") or "") for record in records]
|
|
try:
|
|
effective_offset = record_ids.index(str(after_uid)) + 1
|
|
except ValueError:
|
|
effective_offset = 0
|
|
cursor_reset = True
|
|
else:
|
|
effective_offset = min(offset, len(records))
|
|
page_records = records[effective_offset:effective_offset + limit]
|
|
messages = [
|
|
_message_summary_from_raw(
|
|
uid=str(record.get("id") or ""),
|
|
folder=_mock_record_folder(record),
|
|
raw=_mock_raw_bytes(record),
|
|
flags=["\\Seen"] if record.get("kind") == "imap_append" else [],
|
|
size_bytes=int(record.get("size_bytes") or 0) or None,
|
|
)
|
|
for record in page_records
|
|
]
|
|
return ImapMailboxMessageListResult(
|
|
host=host,
|
|
port=port,
|
|
security=imap_config.security.value,
|
|
folder=folder,
|
|
messages=messages,
|
|
total_count=len(records),
|
|
offset=effective_offset,
|
|
limit=limit,
|
|
uidvalidity="mock-v1",
|
|
cursor_reset=cursor_reset,
|
|
)
|
|
|
|
client = _open_imap(imap_config)
|
|
try:
|
|
total_count, uidvalidity = _select_readonly(client, folder)
|
|
uids = _search_message_uids(client)
|
|
cursor_reset = False
|
|
effective_after_uid = after_uid
|
|
effective_offset = offset
|
|
if expected_uidvalidity and expected_uidvalidity != uidvalidity:
|
|
effective_after_uid = None
|
|
effective_offset = 0
|
|
cursor_reset = True
|
|
page_uids, effective_offset, anchor_missing = _paged_descending_uids(
|
|
uids,
|
|
offset=effective_offset,
|
|
limit=limit,
|
|
after_uid=effective_after_uid,
|
|
)
|
|
messages = _fetch_message_summaries_by_uid(client, page_uids, folder)
|
|
return ImapMailboxMessageListResult(
|
|
host=host,
|
|
port=port,
|
|
security=imap_config.security.value,
|
|
folder=folder,
|
|
messages=messages,
|
|
total_count=len(uids) if uids else total_count,
|
|
offset=effective_offset,
|
|
limit=limit,
|
|
uidvalidity=uidvalidity,
|
|
cursor_reset=cursor_reset or anchor_missing,
|
|
)
|
|
finally:
|
|
try:
|
|
client.logout()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _paged_descending_sequences(total_count: int, *, offset: int, limit: int) -> list[str]:
|
|
if total_count <= 0 or offset >= total_count:
|
|
return []
|
|
end = total_count - offset
|
|
start = max(1, end - limit + 1)
|
|
return [str(sequence) for sequence in range(end, start - 1, -1)]
|
|
|
|
|
|
def get_imap_message(*, imap_config: ImapConfig, folder: str, uid: str) -> ImapMailboxMessageResult:
|
|
"""Fetch one message body using BODY.PEEK in a read-only selected mailbox."""
|
|
|
|
host, port = _require_imap_config(imap_config)
|
|
folder = (folder or "INBOX").strip() or "INBOX"
|
|
uid = str(uid).strip()
|
|
if not uid:
|
|
raise ImapConfigurationError("Message UID is required")
|
|
|
|
if is_mock_imap_host(imap_config.host):
|
|
record = get_record(uid, include_raw=True)
|
|
if not record or not _mock_folder_matches(record, folder):
|
|
raise ImapAppendError(f"Mock mailbox message {uid!r} was not found in folder {folder!r}", temporary=False)
|
|
message = _message_detail_from_raw(
|
|
uid=uid,
|
|
folder=_mock_record_folder(record),
|
|
raw=_mock_raw_bytes(record),
|
|
flags=["\\Seen"] if record.get("kind") == "imap_append" else [],
|
|
size_bytes=int(record.get("size_bytes") or 0) or None,
|
|
)
|
|
return ImapMailboxMessageResult(host=host, port=port, security=imap_config.security.value, folder=folder, message=message)
|
|
|
|
client = _open_imap(imap_config)
|
|
try:
|
|
_select_readonly(client, folder)
|
|
fetched_uid, flags, size_bytes, raw = _fetch_message_by_uid(client, uid)
|
|
message = _message_detail_from_raw(uid=fetched_uid, folder=folder, raw=raw, flags=flags, size_bytes=size_bytes)
|
|
return ImapMailboxMessageResult(host=host, port=port, security=imap_config.security.value, folder=folder, message=message)
|
|
finally:
|
|
try:
|
|
client.logout()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def append_message_to_sent(
|
|
message_bytes: bytes,
|
|
*,
|
|
imap_config: ImapConfig,
|
|
folder: str | None = None,
|
|
) -> ImapAppendResult:
|
|
"""Append a sent MIME message to the configured IMAP Sent folder.
|
|
|
|
The SMTP send remains authoritative. APPEND is a separate best-effort step
|
|
and should not be used to decide whether an email was sent.
|
|
"""
|
|
|
|
host, port = _require_imap_config(imap_config)
|
|
if is_mock_imap_host(imap_config.host):
|
|
if consume_fail_next_imap():
|
|
raise ImapAppendError("Mock IMAP configured to fail the next append", temporary=False)
|
|
target_folder = folder or (imap_config.sent_folder if imap_config.sent_folder and imap_config.sent_folder != "auto" else "Sent")
|
|
record = record_imap_append(message_bytes, folder=target_folder, imap_host=imap_config.host)
|
|
return ImapAppendResult(
|
|
host=host,
|
|
port=port,
|
|
security=imap_config.security.value,
|
|
folder=target_folder,
|
|
bytes_appended=len(message_bytes),
|
|
response=f"mock append stored as {record.id}",
|
|
)
|
|
|
|
client: imaplib.IMAP4 | None = None
|
|
try:
|
|
client = _open_imap(imap_config)
|
|
target_folder = _effective_sent_folder(config=imap_config, requested_folder=folder, client=client)
|
|
internal_date = imaplib.Time2Internaldate(time.time())
|
|
typ, data = client.append(target_folder, "\\Seen", internal_date, message_bytes)
|
|
if typ != "OK":
|
|
raise ImapAppendError(f"IMAP APPEND failed for folder {target_folder!r}: {data!r}", temporary=False)
|
|
response = "; ".join(_decode_item(item) for item in (data or [])) or None
|
|
return ImapAppendResult(
|
|
host=host,
|
|
port=port,
|
|
security=imap_config.security.value,
|
|
folder=target_folder,
|
|
bytes_appended=len(message_bytes),
|
|
response=response,
|
|
)
|
|
except ImapAppendError:
|
|
raise
|
|
except (OSError, socket.timeout, imaplib.IMAP4.abort) as exc:
|
|
raise ImapAppendError(f"IMAP append failed: {exc}", temporary=True) from exc
|
|
except imaplib.IMAP4.error as exc:
|
|
raise ImapAppendError(f"IMAP append failed: {exc}", temporary=False) from exc
|
|
finally:
|
|
if client is not None:
|
|
try:
|
|
client.logout()
|
|
except Exception:
|
|
pass
|