"""Email: IMAP to read and listen, SMTP to send, with the standard library.
References:
- ``email:``: the whole folder (``INBOX`` unless configured), which is what ``listen`` watches.
- ``email:<address>``: the correspondence with one address: its messages in the folder,
and sends to it.
- ``?folder=<name>`` after either picks the folder for this reference alone, so one process
reads several (Gmail serves labels as folders: ``email:?folder=Newsletters/AI``).
``?from=<sender>,<sender>`` on ``email:`` keeps the messages from any of several senders,
each an address or a whole domain (``@example.org``). Both together:
``email:?folder=News&from=ada@example.org,@example.net``. A value is percent-encoded
where it has a space, ``&``, ``?`` or ``%`` (``email:?folder=%5BGmail%5D/All%20Mail``), and
so is an address with one of those. A folder may be named as written (``Énergie``) or as
the server lists it in modified UTF-7 (``&AMk-nergie``); both are one reference. On a
command line, quote a reference with ``?`` or ``&``: ``correspond read 'email:?folder=News'``.
- ``?gmail=<search>`` adds Gmail's own search syntax (``X-GM-RAW``) on a server that
advertises ``X-GM-EXT-1``, and is refused elsewhere:
``email:?gmail=category:promotions+newer_than:7d``. As in Gmail's own addresses, ``+``
is a space (a plus sign is ``%2B``). It narrows the other keys; it replaces none of them,
so ``?from=`` still checks each From exactly. Listening delivers a message only if it
matches when the listen first passes it: a search whose answer changes later
(``is:starred``, a label added afterwards, ``is:unread``) misses the mail that comes to
match after newer mail has moved the cursor on.
A message's conversation is the correspondence with its sender (``email:<address>``), in
whichever folder it was read; ``native["folder"]`` says which.
Message ids are ``Message-ID`` headers, angle brackets included. A draft's ``reply_to`` is
one, and the send carries ``In-Reply-To`` and ``References``. Reading never marks a message
seen.
Authenticity is ``claimed`` unless the **topmost** ``Authentication-Results`` header was
added by a server you trust (``trusted_authserv_ids``) and records ``dmarc=pass`` for the
From domain: then it is ``domain``. Only the topmost header counts, even when the trusted
server added several: a lower one carrying the same authserv-id may have been written by the
sender (not every server strips those), so a result recorded lower down fails safe as
``claimed``.
>>> Email().parse_ref("Ada@Example.org").encoded
'email:ada@example.org'
>>> Email().parse_ref("?from=@Example.NET,ada@example.org&folder=News/AI").encoded
'email:?folder=News/AI&from=@example.net,ada@example.org'
"""
from __future__ import annotations
import base64
import hashlib
import imaplib
import re
import smtplib
import ssl
import subprocess
from collections.abc import Callable, Iterable, Iterator
from contextlib import contextmanager
from dataclasses import dataclass, replace
from datetime import datetime, timedelta, timezone
from email import policy
from email.message import EmailMessage
from email.parser import BytesParser
from email.utils import formatdate, getaddresses, make_msgid, parsedate_to_datetime
from html.parser import HTMLParser
from typing import Any
from urllib.parse import quote, unquote
from correspond.channels._http import file_name
from correspond.channels._http import media_type as _media_type
from correspond.errors import ChannelError, InvalidRef, NotSupported
from correspond.model import (
Account,
Attachment,
Audience,
Authenticity,
Capabilities,
ChannelIdentity,
ConversationRef,
Draft,
Event,
Grade,
HistoryDepth,
Message,
Scope,
SendResult,
Support,
)
from correspond.notify import with_copies
from correspond.ops import window, with_final_cursor
from correspond.registry import require, value
__all__ = ["LIST_LOCAL_PARTS", "Email", "authenticity"]
NAME = "email"
ADDRESS_RE = re.compile(
r"^[A-Za-z0-9.!#$%&'*+/=?^_`{|}~-]+@[A-Za-z0-9](?:[A-Za-z0-9-]*[A-Za-z0-9])?(?:\.[A-Za-z0-9](?:[A-Za-z0-9-]*[A-Za-z0-9])?)+$"
)
MESSAGE_ID_RE = re.compile(r"^<[^<>\s]+@[^<>\s]+>$")
DOMAIN_RE = re.compile(
r"^@[A-Za-z0-9](?:[A-Za-z0-9-]*[A-Za-z0-9])?(?:\.[A-Za-z0-9](?:[A-Za-z0-9-]*[A-Za-z0-9])?)+$"
)
#: The keys a reference takes after ``?``, in their canonical order.
REF_KEYS = ("folder", "from", "gmail")
#: The IMAP capability of a server that takes Gmail's search syntax (``X-GM-RAW``).
GMAIL_CAPABILITY = "X-GM-EXT-1"
_MUTF7_RUN_RE = re.compile(r"&([A-Za-z0-9+,]*)-")
#: Characters an address keeps unescaped in a canonical reference (``%``, ``?`` and ``&`` are escaped).
_ADDRESS_SAFE = "!#$*+-/=^_`{|}~.@"
#: Characters a folder name keeps unescaped in a canonical reference.
_FOLDER_SAFE = "/[]._-+:@!$()*,;="
#: Characters a Gmail search keeps unescaped in a canonical reference; a space is written ``+``.
_GMAIL_SAFE = ':/[]._-@!$()*,;"<>{}|~^'
#: Words that, as a part of an address's local part (``dev-team@``), mark it as a possible mailing list.
LIST_LOCAL_PARTS = ("list", "all", "team", "dev", "announce", "info")
_LOCAL_PART_SPLIT = re.compile(r"[-._+]")
#: About 25 MB once base64-encoded, the message limit of Gmail and most providers.
MAX_ATTACHMENT_BYTES = 18_000_000
#: How many files one email carries; together they stay within MAX_ATTACHMENT_BYTES.
MAX_ATTACHMENTS = 20
FETCH_BATCH = 50
DEFAULT_READ_LIMIT = 50
LISTEN_LOOKBACK = timedelta(days=1)
TIMEOUT_S = 60
_UID_RE = re.compile(rb"UID (\d+)")
_MONTHS = (
"Jan",
"Feb",
"Mar",
"Apr",
"May",
"Jun",
"Jul",
"Aug",
"Sep",
"Oct",
"Nov",
"Dec",
)
def _quote(text: str) -> str:
return '"' + text.replace("\\", "\\\\").replace('"', '\\"') + '"'
def _mailbox(name: str) -> str:
"""``name`` (a Unicode folder name) as IMAP says it on the wire: modified UTF-7 (RFC 3501 5.1.3).
>>> _mailbox("News/AI"), _mailbox("Énergie"), _mailbox("R&D"), _mailbox("日本語")
('News/AI', '&AMk-nergie', 'R&-D', '&ZeVnLIqe-')
"""
out, pending = [], []
def flush():
if pending:
raw = "".join(pending).encode("utf-16-be")
out.append(
"&"
+ base64.b64encode(raw).decode("ascii").rstrip("=").replace("/", ",")
+ "-"
)
pending.clear()
for char in name:
if "\x20" <= char <= "\x7e":
flush()
out.append("&-" if char == "&" else char)
else:
pending.append(char)
flush()
return "".join(out)
def _folder_name(name: str) -> str:
"""The Unicode name of a folder given either way: as written, or as the server lists it.
A name counts as the server's modified UTF-7 only when decoding it and encoding it back
gives it exactly, so ``Q&A-2024`` stays literal while ``&AMk-nergie`` is ``Énergie``.
>>> _folder_name("&AMk-nergie"), _folder_name("Q&A-2024"), _folder_name("R&-D")
('Énergie', 'Q&A-2024', 'R&D')
"""
if "&" not in name or not name.isascii():
return name
def decode(match: re.Match) -> str:
run = match.group(1)
if not run:
return "&"
padded = run.replace(",", "/") + "=" * (-len(run) % 4)
return base64.b64decode(padded, validate=True).decode("utf-16-be")
try:
decoded = _MUTF7_RUN_RE.sub(decode, name)
except (ValueError, UnicodeDecodeError):
return name
return decoded if _mailbox(decoded) == name else name
@dataclass(frozen=True)
class _Scope:
"""What an email reference selects: an address (or none, the folder), a folder, senders."""
address: str = ""
folder: str | None = None
senders: tuple[str, ...] = ()
gmail: str | None = None
@property
def encoded(self) -> str:
query = []
if self.folder is not None:
query.append(f"folder={quote(self.folder, safe=_FOLDER_SAFE)}")
if self.senders:
query.append(f"from={','.join(self.senders)}")
if self.gmail is not None:
query.append(
f"gmail={quote(self.gmail, safe=_GMAIL_SAFE).replace('%20', '+')}"
)
address = quote(self.address, safe=_ADDRESS_SAFE)
return address + ("?" + "&".join(query) if query else "")
def keeps(self, senders: Iterable[str]) -> bool:
"""Whether a message from any of ``senders`` (its From addresses) is in scope.
IMAP's ``FROM`` matches a substring of the header, display name included, so a
search for ``ada@example.org`` also finds ``bada@example.org``; this check is exact.
"""
wanted = self.senders or ((self.address,) if self.address else ())
return not wanted or any(
sender.lower().endswith(w) if w.startswith("@") else sender.lower() == w
for sender in senders
for w in wanted
)
def _scope(id: str) -> _Scope:
"""Parse an email reference's id; :class:`InvalidRef` says what is wrong.
>>> _scope("ada@example.org?folder=inbox")
_Scope(address='ada@example.org', folder='INBOX', senders=(), gmail=None)
"""
raw_address, _, query = id.partition("?")
try:
address = unquote(raw_address, errors="strict")
except UnicodeDecodeError:
raise InvalidRef(f"email:{id} has a %-escape that is not UTF-8") from None
if address and not ADDRESS_RE.match(address):
raise InvalidRef(
f"email references are email: (the folder) or email:<address>, not email:{id}"
)
fields: dict[str, str] = {}
for item in query.split("&") if query else ():
key, equals, raw = item.partition("=")
if key not in REF_KEYS or not equals:
raise InvalidRef(
f"an email reference takes ?folder=<name>, ?from=<senders> and ?gmail=<search>, not {item!r} (in email:{id})"
)
if key in fields:
raise InvalidRef(f"{key} is given twice in email:{id}")
if key == "gmail":
raw = raw.replace("+", " ") # as in Gmail's own addresses; a plus is %2B
try:
text = unquote(raw, errors="strict")
except UnicodeDecodeError:
raise InvalidRef(
f"{key} in email:{id} has a %-escape that is not UTF-8"
) from None
if not text.strip() or any(ord(c) < 0x20 or c == "\x7f" for c in text):
raise InvalidRef(f"{key} is empty or holds a control character in email:{id}")
fields[key] = text
folder = fields.get("folder")
if folder is not None:
folder = _folder_name(folder)
if folder is not None and folder.upper() == "INBOX":
folder = "INBOX" # the one case-insensitive name (RFC 3501)
senders: tuple[str, ...] = ()
if "from" in fields:
if address:
raise InvalidRef(
f"email:{address} already names the sender; ?from= goes on email: (in email:{id})"
)
names = [part.strip().lower() for part in fields["from"].split(",")]
bad = [n for n in names if not (ADDRESS_RE.match(n) or DOMAIN_RE.match(n))]
if bad:
raise InvalidRef(
f"?from= takes addresses or domains (@example.org), not {', '.join(bad) or 'an empty name'}"
)
senders = tuple(sorted(set(names)))
# Gmail reads a run of spaces as one, so one search has one reference (and one cursor).
gmail = " ".join(fields["gmail"].split()) if "gmail" in fields else None
return _Scope(address=address.lower(), folder=folder, senders=senders, gmail=gmail)
def _imap_date(moment: datetime) -> str:
return f"{moment.day:02d}-{_MONTHS[moment.month - 1]}-{moment.year}"
def _csv(text: str | None) -> set[str]:
return {part.strip().lower() for part in (text or "").split(",") if part.strip()}
def _domain(address: str) -> str:
return address.rpartition("@")[2].lower()
def _is_own(address: str, own_domains: set[str]) -> bool:
domain = _domain(address)
return any(domain == own or domain.endswith(f".{own}") for own in own_domains)
def _list_reason(address: str, lists: set[str]) -> str | None:
"""Why ``address`` may be a mailing list, or ``None``.
>>> _list_reason("dev-team@example.org", set())
"its local part names a group ('team')"
>>> _list_reason("ada@example.org", {"@example.org"}) is None
False
"""
address = address.lower()
if address in lists or f"@{_domain(address)}" in lists:
return "it is under lists in the config"
words = set(_LOCAL_PART_SPLIT.split(address.rpartition("@")[0]))
found = [w for w in LIST_LOCAL_PARTS if w in words]
return f"its local part names a group ({found[0]!r})" if found else None
def _last(response: Any) -> str | None:
values = [v for v in (response[1] if response else []) if v]
if not values:
return None
found = values[-1]
return found.decode() if isinstance(found, bytes) else str(found)
class _TextOnly(HTMLParser):
_BREAKS = {"br", "p", "div", "li", "tr", "h1", "h2", "h3", "h4", "h5", "h6"}
def __init__(self):
super().__init__()
self.parts: list[str] = []
self._hidden = 0
def handle_starttag(self, tag, attrs):
if tag in ("script", "style"):
self._hidden += 1
elif tag in self._BREAKS:
self.parts.append("\n")
def handle_endtag(self, tag):
if tag in ("script", "style") and self._hidden:
self._hidden -= 1
def handle_data(self, data):
if not self._hidden:
self.parts.append(data)
def _html_text(html: str) -> str:
parser = _TextOnly()
parser.feed(html)
parser.close()
return re.sub(r"\n{3,}", "\n\n", "".join(parser.parts)).strip()
def _from_addresses(raw: bytes) -> list[str]:
"""Every address in a raw message's From header, lower-cased (the headers alone are parsed)."""
parsed = BytesParser(policy=policy.default).parsebytes(raw, headersonly=True)
return [address for _, address in _addresses(parsed, "From")]
def _addresses(message: EmailMessage, header: str) -> list[tuple[str, str]]:
return [
(name, addr.lower())
for name, addr in getaddresses([str(v) for v in message.get_all(header, [])])
if addr
]
[docs]
def authenticity(
message: EmailMessage, trusted_authserv_ids: Iterable[str]
) -> Authenticity:
"""``domain`` only when the topmost Authentication-Results comes from a trusted server and DMARC passed for the From domain."""
trusted = {t.lower() for t in trusted_authserv_ids}
senders = _addresses(message, "From")
results = message.get_all("Authentication-Results") or []
if not results:
return Authenticity(
grade=Grade.CLAIMED, evidence={"authentication_results": "none"}
)
top = " ".join(str(results[0]).split())
authserv_id = (top.split(";", 1)[0].split() or [""])[0].lower()
evidence: dict[str, Any] = {
"authserv_id": authserv_id,
"trusted": authserv_id in trusted,
}
if authserv_id not in trusted:
return Authenticity(
grade=Grade.CLAIMED,
evidence={
**evidence,
"reason": "the topmost Authentication-Results header is not from a trusted server",
},
)
dmarc = re.search(r"\bdmarc=([a-z]+)", top, re.I)
header_from = re.search(r"\bheader\.from=\"?([^\s;\"]+)", top, re.I)
domain = senders[0][1].rpartition("@")[2] if len(senders) == 1 else ""
evidence["dmarc"] = dmarc.group(1).lower() if dmarc else "none"
evidence["header_from"] = header_from.group(1).lower() if header_from else ""
if domain and evidence["dmarc"] == "pass" and evidence["header_from"] == domain:
return Authenticity(grade=Grade.DOMAIN, evidence=evidence)
return Authenticity(
grade=Grade.CLAIMED,
evidence={**evidence, "reason": "DMARC did not pass for the one From domain"},
)
def _content(part: EmailMessage) -> str:
try:
return part.get_content()
except (LookupError, UnicodeError):
return (part.get_payload(decode=True) or b"").decode("utf-8", "replace")
_CONTROL_CHARS = re.compile(r"[\x00-\x1f\x7f]")
#: Mailing-list headers kept in ``native`` (RFC 2369 / 2919), as ``native`` key -> header.
LIST_HEADERS = {
"list_id": "List-Id",
"list_unsubscribe": "List-Unsubscribe",
"list_post": "List-Post",
}
def _other_html(message: EmailMessage, body_format: str) -> str | None:
"""The raw HTML part when plain text won as the body; None when there is none.
When HTML is the body (``body_format == "html"``) it is already in ``body``,
so it is not carried a second time in ``native``.
"""
if body_format == "html":
return None
part = message.get_body(preferencelist=("html",))
return None if part is None else _content(part)
def _list_headers(message: EmailMessage) -> dict[str, str]:
"""The List-* headers present, unfolded and control-free; repeated ones are
joined with ", "; absent or empty ones are left out."""
found = {}
for key, header in LIST_HEADERS.items():
values = (
" ".join(_CONTROL_CHARS.sub(" ", str(v)).split())
for v in message.get_all(header, [])
)
joined = ", ".join(v for v in values if v)
if joined:
found[key] = joined
return found
def _body(message: EmailMessage) -> tuple[str, str | None, str]:
part = message.get_body(preferencelist=("plain", "html"))
if part is None:
return "", None, "plain"
content = _content(part)
if part.get_content_type() == "text/html":
return _html_text(content), content, "html"
return content.strip(), content, "plain"
def _part(attachment: Attachment, *, load: bool = True) -> tuple[str, Any, str, str]:
"""A draft's attachment as ``(name, bytes, maintype, subtype)``, its name and type checked.
``load=False`` (a dry run) fetches nothing: the bytes are replaced by the attachment
itself, whose metadata the plan shows.
"""
name = file_name(attachment.name or "file")
maintype, _, subtype = _media_type(attachment.media_type).partition("/")
return name, attachment.content() if load else attachment, maintype, subtype
def _with_notify(to: str, draft: Draft) -> tuple[Draft, tuple[str, ...]]:
"""``draft`` with its ``notify`` addresses in Cc (and out of Bcc), and the addresses added."""
cc = with_copies(to, draft.cc, draft.notify)
copied = {a.lower() for a in cc}
bcc = tuple(a for a in draft.bcc if a.lower() not in copied)
return replace(draft, cc=cc, bcc=bcc), cc[len(draft.cc) :]
[docs]
class Email:
"""A mailbox over IMAP and SMTP."""
name = NAME
def __init__(
self,
*,
imap: Callable[[], Any] | None = None,
smtp: Callable[[], Any] | None = None,
run: Callable[..., Any] = subprocess.run,
):
self.run = run
self._imap_factory = imap or self._connect_imap
self._smtp_factory = smtp or self._connect_smtp
@property
def capabilities(self) -> Capabilities:
"""Read and listen to a folder per reference; send with threading headers."""
return Capabilities(
channel=NAME,
read=Support.FULL,
listen=Support.FULL,
send=Support.FULL,
audience=Support.FULL,
initiate=Support.FULL,
reply=Support.FULL,
priority=Support.PARTIAL,
cc=Support.FULL,
notify=Support.FULL,
upload=Support.FULL,
max_attachments=MAX_ATTACHMENTS,
history_depth=HistoryDepth.FULL,
listen_modes=("poll",),
grades=(Grade.DOMAIN, Grade.CLAIMED),
formats=("plain", "html"),
max_upload_bytes=MAX_ATTACHMENT_BYTES,
native_fields=(
"subject",
"to",
"cc",
"folder",
"uid",
"references",
"html",
"list_id",
"list_unsubscribe",
"list_post",
),
notes=(
"read and listen look at one folder: the reference's ?folder= (a Gmail label is a folder), else the configured one (INBOX by default); your own sent mail is in another",
"?from= on email: keeps the messages from any of several addresses or domains (@example.org, not its subdomains): one IMAP SEARCH per sender, then an exact check of the From addresses (any of them counts), since IMAP matches FROM as a substring of the header; email:<address> gets the same exact check, so a list's rewritten From (ada via group@) is not hers",
"a reference with ? or & must be quoted on a command line",
"?gmail=<search> adds Gmail's search syntax (X-GM-RAW) on a server advertising X-GM-EXT-1 (one CAPABILITY call to check), NotSupported elsewhere; + is a space, as in Gmail's addresses; it narrows ?folder= and ?from=, and the exact From check still applies",
"listen with ?gmail= delivers a message only if it matches when the listen first passes it: a search whose answer changes over time (is:starred, a label added later, is:unread) misses mail that comes to match after newer mail moved the cursor on",
"listen polls by UID; IMAP IDLE is not used",
"upload emails the file to the address as an attachment, its name as the subject (up to 18 MB, about 25 MB once encoded; a server may take less); a send carries the draft's attachments the same way",
"priority sets Importance and X-Priority, which mail clients may ignore",
"native html is the HTML part when plain text won as the body (an HTML body is in body only); list_id, list_unsubscribe and list_post are the List-* headers when present",
"audience: the address plus cc and bcc; a list-shaped address has members nobody can list",
),
)
[docs]
def parse_ref(self, id: str) -> ConversationRef:
"""``""`` for the folder, or an address; either may add ``?folder=`` (and the folder ``?from=``)."""
scope = _scope(id or "")
folder = ConversationRef(
channel=NAME, id=replace(scope, address="", senders=()).encoded, kind="folder"
)
if not scope.address:
return replace(folder, id=scope.encoded)
return ConversationRef(
channel=NAME, id=scope.encoded, kind="address", parent=folder
)
# ---------------------------------------------------------------- connections
def _connect_imap(self) -> imaplib.IMAP4:
host = require(NAME, "imap_host", run=self.run)
port = int(value(NAME, "imap_port") or 993)
user = require(NAME, "user", run=self.run)
password = require(NAME, "password", run=self.run)
try:
conn = imaplib.IMAP4_SSL(
host, port, ssl_context=ssl.create_default_context(), timeout=TIMEOUT_S
)
except (OSError, imaplib.IMAP4.error) as error:
raise ChannelError(
f"could not reach the IMAP server {host}: {error}",
kind="network",
retryable=True,
) from None
try:
conn.login(user, password)
except imaplib.IMAP4.error as error:
try:
conn.logout()
except (OSError, imaplib.IMAP4.error):
pass
raise ChannelError(
f"the IMAP server refused the login for {user}: {error}", kind="auth"
) from None
return conn
@contextmanager
def _imap(self) -> Iterator[Any]:
conn = self._imap_factory()
try:
yield conn
except imaplib.IMAP4.abort as error:
raise ChannelError(
f"the IMAP connection dropped: {error}", kind="network", retryable=True
) from None
except imaplib.IMAP4.error as error:
raise ChannelError(
f"the IMAP server said: {error}", kind="unavailable"
) from None
except (OSError, TimeoutError) as error:
raise ChannelError(
f"IMAP failed: {error}", kind="network", retryable=True
) from None
finally:
try:
conn.logout()
except (OSError, imaplib.IMAP4.error):
pass
def _connect_smtp(self) -> smtplib.SMTP:
host = require(NAME, "smtp_host", run=self.run)
port = int(value(NAME, "smtp_port") or 465)
user = require(NAME, "user", run=self.run)
password = require(NAME, "password", run=self.run)
context = ssl.create_default_context()
try:
if port == 465:
server = smtplib.SMTP_SSL(host, port, timeout=TIMEOUT_S, context=context)
else:
server = smtplib.SMTP(host, port, timeout=TIMEOUT_S)
server.starttls(context=context)
server.login(user, password)
except smtplib.SMTPAuthenticationError as error:
raise ChannelError(
f"the SMTP server refused the login for {user}: {error.smtp_code}",
kind="auth",
) from None
except (OSError, smtplib.SMTPException) as error:
raise ChannelError(
f"could not use the SMTP server {host}: {error}",
kind="network",
retryable=True,
) from None
return server
@contextmanager
def _smtp(self) -> Iterator[Any]:
server = self._smtp_factory()
try:
yield server
except smtplib.SMTPRecipientsRefused as error:
raise ChannelError(
f"the server refused the recipient: {error.recipients}", kind="validation"
) from None
except smtplib.SMTPSenderRefused as error:
raise ChannelError(
f"the server refused the sender: {error.smtp_error!r}", kind="permission"
) from None
except smtplib.SMTPDataError as error:
transient = 400 <= error.smtp_code < 500
raise ChannelError(
f"the server refused the message ({error.smtp_code})",
kind="unavailable" if transient else "validation",
retryable=transient,
) from None
except smtplib.SMTPServerDisconnected as error:
raise ChannelError(
f"the SMTP server disconnected: {error}", kind="network", retryable=True
) from None
except smtplib.SMTPException as error:
raise ChannelError(f"SMTP failed: {error}", kind="unavailable") from None
except OSError as error:
raise ChannelError(
f"SMTP failed: {error}", kind="network", retryable=True
) from None
finally:
try:
server.quit()
except (OSError, smtplib.SMTPException):
pass
# ------------------------------------------------------------------ IMAP work
@staticmethod
def _select(conn: Any, folder: str) -> tuple[str, int | None]:
typ, data = conn.select(_quote(_mailbox(folder)), readonly=True)
if typ != "OK":
raise ChannelError(
f"the folder {folder!r} could not be opened: {data!r}", kind="not_found"
)
validity = _last(conn.response("UIDVALIDITY")) or "0"
uidnext = _last(conn.response("UIDNEXT"))
return validity, int(uidnext) if uidnext and uidnext.isdigit() else None
@staticmethod
def _search(conn: Any, criteria: list[str], *, gmail: str | None = None) -> list[int]:
"""``UID SEARCH`` on ``criteria``, plus Gmail's ``X-GM-RAW`` search when ``gmail`` is given.
A non-ASCII Gmail search goes as a UTF-8 literal, which IMAP sends last, under
``CHARSET UTF-8``, through ``imaplib``'s ``literal`` attribute: an ``imap=`` transport
other than ``imaplib`` must honour it for such a search.
"""
args = list(criteria)
if gmail is not None:
if gmail.isascii():
args += ["X-GM-RAW", _quote(gmail)]
else:
args = ["CHARSET", "UTF-8", *args, "X-GM-RAW"]
conn.literal = gmail.encode("utf-8")
try:
typ, data = conn.uid("SEARCH", *(args or ["ALL"]))
except imaplib.IMAP4.error as error:
if gmail is None:
raise
raise ChannelError(
f"the server refused the Gmail search {gmail!r}: {error}",
kind="validation",
) from None
if typ != "OK":
raise ChannelError(f"the IMAP search failed: {data!r}", kind="unavailable")
return sorted(int(x) for x in ((data or [b""])[0] or b"").split())
@staticmethod
def _fetch(conn: Any, uids: list[int]) -> list[tuple[int, bytes]]:
found: list[tuple[int, bytes]] = []
for start in range(0, len(uids), FETCH_BATCH):
chunk = uids[start : start + FETCH_BATCH]
typ, data = conn.uid(
"FETCH", ",".join(str(u) for u in chunk), "(UID BODY.PEEK[])"
)
if typ != "OK":
raise ChannelError(f"the IMAP fetch failed: {data!r}", kind="unavailable")
data = data or []
for index, item in enumerate(data):
if not isinstance(item, tuple) or len(item) < 2:
continue
match = _UID_RE.search(item[0])
if (
match is None
and index + 1 < len(data)
and isinstance(data[index + 1], bytes)
):
match = _UID_RE.search(data[index + 1])
if match:
found.append((int(match.group(1)), item[1]))
found.sort(key=lambda pair: pair[0])
return found
def _message(
self, uid: int, raw: bytes, *, folder: str, user: str, trusted: set[str]
) -> Message:
parsed: EmailMessage = BytesParser(policy=policy.default).parsebytes(raw)
senders = _addresses(parsed, "From")
name, address = senders[0] if senders else ("", "")
to = [a for _, a in _addresses(parsed, "To")]
cc = [a for _, a in _addresses(parsed, "Cc")]
is_self = bool(user) and address == user.lower()
counterpart = to[0] if is_self and to else address
conversation = (
self.parse_ref(counterpart)
if counterpart and ADDRESS_RE.match(counterpart)
else self.parse_ref("")
)
message_id = (
str(parsed.get("Message-ID") or "").strip()
or f"<uid-{uid}@correspond.invalid>"
)
try:
sent_at = (
parsedate_to_datetime(str(parsed["Date"])) if parsed["Date"] else None
)
except (TypeError, ValueError):
sent_at = None
if sent_at is None:
sent_at = datetime.fromtimestamp(0, tz=timezone.utc)
elif sent_at.tzinfo is None:
sent_at = sent_at.replace(tzinfo=timezone.utc)
text, body, body_format = _body(parsed)
html = _other_html(parsed, body_format)
references = str(parsed.get("References") or "").split()
attachments = []
for index, part in enumerate(parsed.iter_attachments()):
data = part.get_payload(decode=True) or b""
attachments.append(
Attachment(
ref=f"imap:{uid}/{index}",
media_type=part.get_content_type(),
name=part.get_filename(),
size=len(data),
sha256=hashlib.sha256(data).hexdigest(),
loader=lambda data=data: data,
)
)
return Message(
id=message_id,
conversation=conversation,
author=ChannelIdentity(
channel=NAME,
native_id=address,
handle=address or None,
display_name=name or None,
is_self=is_self,
),
authenticity=authenticity(parsed, trusted),
sent_at=sent_at,
text=text,
body=body,
body_format=body_format,
attachments=tuple(attachments),
reply_to=str(parsed.get("In-Reply-To") or "").strip() or None,
thread_root=references[0] if references else None,
native={
"subject": str(parsed.get("Subject") or ""),
"to": to,
"cc": cc,
"folder": folder,
"uid": uid,
"references": references,
**({"html": html} if html is not None else {}),
**_list_headers(parsed),
},
)
def _context(self, scope: _Scope) -> tuple[str, str, set[str]]:
"""``(folder, user, trusted authserv-ids)``: the reference's folder, else the configured one."""
user = value(NAME, "user") or ""
return (
scope.folder or _folder_name(value(NAME, "folder") or "INBOX"),
user,
_csv(value(NAME, "trusted_authserv_ids")),
)
def _search_scope(self, conn: Any, scope: _Scope, criteria: list[str]) -> list[int]:
"""UIDs matching ``criteria`` in the scope: one ``SEARCH`` per sender, merged; the Gmail search in each."""
if scope.gmail is not None:
self._require_gmail(conn)
senders = scope.senders or ((scope.address,) if scope.address else ())
if not senders:
return self._search(conn, criteria, gmail=scope.gmail)
found: set[int] = set()
for sender in senders:
found.update(
self._search(conn, [*criteria, "FROM", _quote(sender)], gmail=scope.gmail)
)
return sorted(found)
@staticmethod
def _require_gmail(conn: Any) -> None:
"""Refuse ``?gmail=`` unless the server says, once logged in, that it takes ``X-GM-RAW``."""
typ, data = conn.capability()
words = b" ".join(d for d in (data or []) if isinstance(d, bytes)).upper().split()
if typ != "OK" or GMAIL_CAPABILITY.encode() not in words:
raise NotSupported(
"gmail search",
NAME,
alternatives=(
f"this server does not advertise {GMAIL_CAPABILITY}: Gmail's search runs on Gmail only",
"?folder= and ?from= work on every server",
),
)
[docs]
def read(
self,
ref: ConversationRef,
*,
since: datetime | None = None,
limit: int | None = None,
) -> list[Message]:
"""Messages in the reference's folder (from its address or senders, if it names any), oldest first; never marked seen."""
scope = _scope(ref.id)
folder, user, trusted = self._context(scope)
criteria = ["SINCE", _imap_date(since)] if since is not None else []
wanted = limit or DEFAULT_READ_LIMIT
kept: list[Message] = []
with self._imap() as conn:
self._select(conn, folder)
uids = self._search_scope(conn, scope, criteria)
# Newest first, a batch at a time, until enough pass the exact sender check:
# mail that matched FROM only as a substring must not crowd out the rest.
end = len(uids)
while end > 0 and len(kept) < wanted:
start = max(0, end - wanted)
batch = [
self._message(uid, raw, folder=folder, user=user, trusted=trusted)
for uid, raw in self._fetch(conn, uids[start:end])
if scope.keeps(_from_addresses(raw))
]
kept = batch + kept
end = start
return window(kept, since=since, limit=limit)
[docs]
def poll(
self, ref: ConversationRef, *, cursor: str | None = None, limit: int | None = None
):
"""New messages since ``cursor`` (``<uidvalidity>:<uid>``); a first poll, or a changed UIDVALIDITY, looks back a day."""
scope = _scope(ref.id)
folder, user, trusted = self._context(scope)
seen_validity, _, seen_uid = (cursor or "").partition(":")
with self._imap() as conn:
validity, uidnext = self._select(conn, folder)
resume = bool(cursor) and seen_validity == validity and seen_uid.isdigit()
if resume:
criteria = ["UID", f"{int(seen_uid) + 1}:*"]
else:
criteria = [
"SINCE",
_imap_date(datetime.now(timezone.utc) - LISTEN_LOOKBACK),
]
# "n:*" always includes the highest UID, even when it is below n.
uids = [
u
for u in self._search_scope(conn, scope, criteria)
if not resume or u > int(seen_uid)
]
if limit:
uids = uids[:limit]
fetched = self._fetch(conn, uids)
events = []
for uid, raw in fetched:
if not scope.keeps(_from_addresses(raw)):
continue # FROM matched a substring of another sender
message = self._message(uid, raw, folder=folder, user=user, trusted=trusted)
events.append(
Event(
kind="message.created",
channel=NAME,
delivery_id=f"email:{folder}:{validity}:{uid}",
cursor=f"{validity}:{uid}",
message=message,
payload={"folder": folder, "uid": uid},
)
)
# Every UID up to the last one fetched has been looked at, delivered or not, so a
# cut list resumes after it, and a sender filtered out cannot stall the cursor.
final = None
if uids:
final = f"{validity}:{uids[-1]}"
elif not resume and uidnext:
final = f"{validity}:{uidnext - 1}"
return with_final_cursor(events, final)
[docs]
def audience(self, ref: ConversationRef, *, draft: Draft | None = None) -> Audience:
"""Who reads an email: its address plus the draft's ``cc`` and ``bcc``, asked of nothing but the config.
Scope ``named``, never ``complete``: any address may be an alias, a shared mailbox or
an auto-forward, which nothing here can see. A recipient whose domain is not in
``own_domains`` makes it ``external``; a list-shaped address (under ``lists`` in the
config, or a local part naming a group, see :data:`LIST_LOCAL_PARTS`) adds a class
and ``list_expansion``, and, when every address is in an own domain, leaves
``external`` unknown, since a list can have outside members. A Bcc address is a
class of its own, so moving it to Cc changes the hash. Every email is pushed to its
recipients, can be forwarded, and cannot be recalled.
"""
to = _scope(ref.id).address
if not to:
return Audience.unknown(
ref.encoded, "email: is the folder, not a correspondence: name an address"
)
if draft is not None and draft.notify:
draft, _ = _with_notify(to, draft) # the Cc a send will really carry
copies = {"cc": draft.cc if draft else (), "bcc": draft.bcc if draft else ()}
malformed = [
a for a in (*copies["cc"], *copies["bcc"]) if not ADDRESS_RE.match(a)
]
if malformed:
return Audience.unknown(
ref.encoded, f"not an email address: {', '.join(malformed)}"
)
# Exact addresses: a local part may be case-sensitive, and SMTP receives each as given.
recipients = list(dict.fromkeys((to, *copies["cc"], *copies["bcc"])))
own, lists = _csv(value(NAME, "own_domains")), _csv(value(NAME, "lists"))
evidence = [
f"to {to}" + "".join(f"; {k} {', '.join(v)}" for k, v in copies.items() if v)
]
evidence.append(
f"own domains: {', '.join(sorted(own))}"
if own
else "no own_domains in the config, so every recipient counts as external"
)
classes = [
"whoever a recipient's mailbox delivers or forwards to (aliases, shared mailboxes, auto-forwards)"
]
classes += [
f"{a} is copied blind: the other recipients do not see it"
for a in copies["bcc"]
]
widening, listed = ["forwarding"], False
for address in recipients:
reason = _list_reason(address, lists)
if reason:
listed = True
classes.append(f"everyone behind {address}, which may be a mailing list")
evidence.append(f"{address} may be a mailing list: {reason}")
if listed:
widening.append("list_expansion")
if any(not _is_own(a, own) for a in recipients):
external = True
else:
external = None if listed else False
return Audience(
ref=ref.encoded,
scope=Scope.NAMED,
readers=tuple(
ChannelIdentity(channel=NAME, native_id=a, handle=a) for a in recipients
),
complete=False,
classes=tuple(classes),
external=external,
retractable=False,
durability=("copies_pushed",),
widening=tuple(widening),
evidence=tuple(evidence),
)
[docs]
def prepare(self, ref: ConversationRef, draft: Draft) -> Draft:
"""The draft as it will go out: ``notify`` addresses checked and put in Cc (out of Bcc). Pure."""
if not draft.notify:
return draft
malformed = [a for a in draft.notify if not ADDRESS_RE.match(a)]
if malformed:
raise ChannelError(
f"notify on email takes addresses, not: {', '.join(malformed)}",
kind="validation",
)
return _with_notify(_scope(ref.id).address, draft)[0]
[docs]
def send(
self, ref: ConversationRef, draft: Draft, *, dry_run: bool = False
) -> SendResult:
"""Send to ``ref``'s address, with ``In-Reply-To`` and ``References`` when replying; ``cc`` in a header, ``bcc`` only on the envelope.
A ``draft.notify`` address that is neither the To address nor already in Cc is
added to Cc (and so leaves Bcc, which would hide it); the plan lists it under
``cc_added``.
"""
return self._send(ref, draft, dry_run=dry_run)
[docs]
def upload(
self,
ref: ConversationRef,
name: str,
data: bytes,
*,
media_type: str = "application/octet-stream",
dry_run: bool = False,
) -> SendResult:
"""Email a file to ``ref``'s address: a message with the file attached and its name as the subject."""
name = file_name(name)
maintype, _, subtype = _media_type(media_type).partition("/")
return self._send(
ref,
Draft(text="", title=name),
attachments=[(name, data, maintype, subtype)],
operation="upload",
dry_run=dry_run,
)
def _send(
self,
ref: ConversationRef,
draft: Draft,
*,
attachments: list[tuple[str, bytes, str, str]] | None = None,
operation: str = "send",
dry_run: bool = False,
) -> SendResult:
if (
attachments is None
): # a send: the draft's own files, fetched only to send them
total = sum(a.size or 0 for a in draft.attachments)
if total > MAX_ATTACHMENT_BYTES:
raise ChannelError(
f"the attachments come to {total} bytes; one email carries at most {MAX_ATTACHMENT_BYTES}",
kind="validation",
)
attachments = (
[_part(a, load=False) for a in draft.attachments]
if dry_run
else [_part(a) for a in draft.attachments]
)
to = _scope(ref.id).address
if not to:
raise ChannelError("send to an address: email:<address>", kind="validation")
cc_before = draft.cc
draft = self.prepare(ref, draft)
cc_added = draft.cc[len(cc_before) :]
if draft.reply_to and not MESSAGE_ID_RE.match(draft.reply_to):
raise ChannelError(
f"reply_to is a Message-ID such as <abc@example.org>, not {draft.reply_to!r}",
kind="validation",
)
malformed = [a for a in (*draft.cc, *draft.bcc) if not ADDRESS_RE.match(a)]
if malformed:
raise ChannelError(
f"not an email address: {', '.join(malformed)}", kind="validation"
)
user = value(NAME, "user") if dry_run else require(NAME, "user", run=self.run)
sender = value(NAME, "from_address") or user or "(not set)"
message = EmailMessage()
try:
message["From"] = sender
message["To"] = to
if draft.cc:
message["Cc"] = ", ".join(draft.cc)
if draft.title:
message["Subject"] = " ".join(draft.title.split())
message["Date"] = formatdate(usegmt=True)
message["Message-ID"] = make_msgid(domain=sender.rpartition("@")[2] or None)
if draft.reply_to:
message["In-Reply-To"] = draft.reply_to
message["References"] = draft.reply_to
if draft.priority in ("high", "urgent"):
message["Importance"], message["X-Priority"] = "high", "1"
elif draft.priority == "low":
message["Importance"], message["X-Priority"] = "low", "5"
message.set_content(draft.text)
for name, data, maintype, subtype in attachments:
if isinstance(data, bytes): # a dry run builds the message without them
message.add_attachment(
data, maintype=maintype, subtype=subtype, filename=name
)
except (ValueError, TypeError) as error:
raise ChannelError(
f"the message could not be built: {error}", kind="validation"
) from None
plan = {
"action": "send an email",
"from": sender,
"to": to,
"cc": ", ".join(draft.cc) or None,
"bcc": ", ".join(draft.bcc) or None,
"subject": draft.title,
"in_reply_to": draft.reply_to,
"priority": draft.priority,
"server": value(NAME, "smtp_host") or "(not set)",
"text": draft.text,
}
if draft.notify:
plan.update(notify=list(draft.notify), cc_added=list(cc_added))
if attachments:
plan["attachments"] = [
{
"name": name,
"media_type": f"{maintype}/{subtype}",
"bytes": len(data) if isinstance(data, bytes) else data.size,
"sha256": hashlib.sha256(data).hexdigest()
if isinstance(data, bytes)
else data.sha256,
}
for name, data, maintype, subtype in attachments
]
if operation == "upload":
plan["attachment"] = plan["attachments"][0]
if dry_run:
return SendResult(
ok=True,
channel=NAME,
conversation=ref.encoded,
operation=operation,
dry_run=True,
plan=plan,
)
with self._smtp() as server:
if draft.bcc: # never in a header: only the envelope names them
server.send_message(message, to_addrs=[to, *draft.cc, *draft.bcc])
else:
server.send_message(message)
return SendResult(
ok=True,
channel=NAME,
conversation=ref.encoded,
operation=operation,
message_id=str(message["Message-ID"]),
account=Account(channel=NAME, id=user, acts_as="user"),
plan=plan,
)