01 / apply it
yardr pack apply ./protonFrom a clone of yardr-packs: the pack's directory. yardr pack check ./proton shows what it would change first. It goes into the workspace itself, and into no repo · depot.
yardr pack apply https://yardr.dev/packs/proton/From this site, as the files below.
02 / the files
Every file, as it ships.
# Proton Mail as a pack of kind source: the integration, imported into the
# yard and into no depot. What Proton is here: the declared source proton,
# which searches the operator's mail, reads a message and reads the thread a
# message is in, all through Proton Mail Bridge's local IMAP as the one
# account Bridge is signed in to. Proton offers no public API for mail, so each
# call is answered by a program the pack brings beside its declaration,
# proton/proton-imap, which the yard runs (call.run) and which talks IMAP to
# Bridge itself. It reads and changes nothing.
#
# yardr pack apply /Users/mckean/projects/repos/yardr-packs/proton
#
# The source is named proton: a pack is named after the service it talks to.
# A pack of kind source asks no grants at import, so what a crew member may
# call is granted afterwards, by the setup notes. Every agent granted a call
# reads the whole mailbox of that one account: Bridge has no narrower login.
name: proton
kind: source
setup: SETUP.md
sources:
proton:
declaration: proton.yaml
# Suggested grants: named sets of the source's calls, printed numbered by
# pack show. Suggesting grants nothing: the operator grants them, and a
# terminal pack apply ignores them. Every call here is a read, so there is no
# preset for writing.
presets:
read:
about: Search mail, read a message and read a thread; change nothing
grants:
proton: [read]
#!/usr/bin/env python3
# proton-imap: the three calls of the proton source, search, message and
# thread, answered from Proton Mail Bridge's local IMAP. The yard runs it as
# `proton-imap <call>` (the `run` of each call in proton.yaml): the call's
# parameters are on stdin as {"params": {...}}, the Bridge login is in
# PROTON_BRIDGE as <username>:<password>, and where Bridge listens is in
# YARDR_SOURCE_BASE. One JSON object goes to stdout. A failure is one line on
# stderr and a non-zero exit, which is what the yard shows whoever made the
# call, so no line holds the login, and what Bridge or a message wrote is
# flattened to one line before it is quoted.
#
# It reads and changes nothing, and is built so that it cannot do otherwise:
# the only commands it sends are LIST, EXAMINE (a mailbox opened read-only,
# never SELECT), UID SEARCH, UID FETCH of BODY.PEEK and LOGOUT, each from one
# method of Bridge below, and nothing else in this file touches the socket.
#
# A message is named by an id, <mailbox, percent-encoded>:<UIDVALIDITY>:<UID>,
# which is all IMAP needs to find it again, and which Bridge's renumbering of a
# mailbox (a new UIDVALIDITY) turns into a refusal instead of another message.
#
# Bridge's IMAP engine, Gluon, is what this was written against (see
# proton.yaml): STARTTLS, LOGIN, LIST with the \All attribute, EXAMINE, SEARCH
# with CHARSET and literals, FETCH, and none of Gmail's X-GM-THRID or THREAD,
# so a thread is built from References and In-Reply-To.
#
# Needs Python 3.9 or newer, and nothing beyond its standard library.
import datetime
import email
import email.header
import email.policy
import email.utils
import html.parser
import imaplib
import ipaddress
import json
import os
import re
import socket
import ssl
import sys
import time
import traceback
import urllib.parse
DEFAULT_PORT = 1143 # Bridge's, when the base names none
TIMEOUT = 20 # seconds one command may take
BUDGET = 25 # seconds a whole call may take: the yard stops a command at 30
MAX_RESULTS = 50 # a page of search, and the messages of a thread
OUTPUT = 60000 # bytes of answer: the yard takes 64 KB of a command's
TEXT = 48000 # bytes of message text in an answer, shared by a thread's messages
FULL_ONE = 25 << 20 # the largest message read in full
FULL_EACH = 5 << 20 # the largest message of a thread whose text is read
FULL_ALL = 25 << 20 # and the most of them a thread reads in all
# imaplib refuses a response line over 1 MB, and the answer to a SEARCH is one
# line: a mailbox of 150,000 messages is 1 MB of it. A literal has no such bound.
imaplib._MAXLINE = 8 << 20
class Refusal(Exception):
"""What was wrong, in the one line the caller is shown."""
def one_line(text, limit=300):
"""text as one line of printable characters, at most limit of them. What
Bridge says and what a message holds reach stderr and the answer, and may
hold line breaks and terminal controls."""
text = re.sub(r"[\x00-\x1f\x7f-\x9f
]+", " ", str(text))
text = re.sub(r"\s+", " ", text).strip()
return text if len(text) <= limit else text[: limit - 1] + "…"
IMAP_CONTROLS = re.compile(r"[\x00-\x1f\x7f-\x9f]")
def mailbox_name(text):
"""A mailbox name safe to put in an IMAP quoted string."""
# Quoting escapes quotes and backslashes, but a line break would split the command.
if IMAP_CONTROLS.search(text):
raise Refusal("a mailbox name has a control character")
return text
def words(error):
"""What an error says as one line. imaplib's holds bytes where a server
spoke; an OSError's first argument is its number, so it is read whole."""
arg = error.args[0] if isinstance(error, imaplib.IMAP4.error) and error.args else error
if isinstance(arg, bytes):
arg = arg.decode("utf-8", "replace")
return one_line(arg)
def clip_bytes(text, limit):
"""(text cut on a character boundary to limit UTF-8 bytes, whether it was)."""
data = text.encode("utf-8", "replace")
if len(data) <= limit:
return text, False
return data[:limit].decode("utf-8", "ignore"), True
# ---- the query ------------------------------------------------------------
# q is Gmail's syntax in part, since that is what an agent knows, turned into
# IMAP SEARCH keys before anything is sent: the terms are ANDed.
class Atom(str):
"""A search argument that goes as it is, a date. Any other is a string:
quoted, or sent as a literal when it holds more than printable ASCII."""
OPERATORS = "from: to: cc: bcc: subject: after: before: newer_than: older_than: is: in: label:"
HEADERS = {"from": "FROM", "to": "TO", "cc": "CC", "bcc": "BCC", "subject": "SUBJECT"}
FLAGS = {"unread": "UNSEEN", "read": "SEEN", "starred": "FLAGGED", "flagged": "FLAGGED",
"answered": "ANSWERED", "draft": "DRAFT"}
MONTHS = "Jan Feb Mar Apr May Jun Jul Aug Sep Oct Nov Dec".split() # not the locale's
MAX_TERMS = 30
KEY = re.compile(r"[A-Za-z_]+:")
def terms(q):
"""q as (negated, operator or None, value, quoted) for each term. A quote
that is never closed is refused: it would make the rest of q a phrase."""
out, i, n = [], 0, len(q)
while True:
while i < n and q[i].isspace():
i += 1
if i >= n:
return out
negated = q[i] == "-" and i + 1 < n and not q[i + 1].isspace()
if negated:
i += 1
key = None
m = KEY.match(q, i)
if m:
key, i = m.group()[:-1].lower(), m.end()
if i < n and q[i] == '"':
j, buf = i + 1, []
while j < n and q[j] != '"':
if q[j] == "\\" and j + 1 < n:
j += 1
buf.append(q[j])
j += 1
if j >= n:
raise Refusal("q has a quote that is never closed")
value, quoted, i = "".join(buf), True, j + 1
else:
j = i
while j < n and not q[j].isspace():
j += 1
value, quoted, i = q[i:j], False, j
out.append((negated, key, value, quoted))
if len(out) > MAX_TERMS:
raise Refusal("q has more than %d terms" % MAX_TERMS)
def imap_date(day):
return "%d-%s-%d" % (day.day, MONTHS[day.month - 1], day.year)
def plan_search(q, today=None):
"""q as ([(negated, IMAP key, argument or None)], the mailbox in: or label:
names or None). today is the date newer_than and older_than count back
from, UTC's when it is None."""
if today is None:
today = datetime.datetime.now(datetime.timezone.utc).date()
q = q or ""
if "\x00" in q:
raise Refusal("q has a NUL in it")
crits, box = [], None
for negated, key, value, quoted in terms(q):
if key is None:
if value:
crits.append((negated, "TEXT", value))
continue
if key in HEADERS:
if not value:
raise Refusal("%s: needs something to look for" % key)
crits.append((negated, HEADERS[key], value))
elif key in ("after", "before"):
m = re.fullmatch(r"(\d{4})[-/](\d{1,2})[-/](\d{1,2})", value)
try:
day = datetime.date(*map(int, m.groups())) if m else None
except ValueError:
day = None
if day is None:
raise Refusal("%s: wants a date, YYYY-MM-DD or YYYY/MM/DD" % key)
crits.append((negated, "SINCE" if key == "after" else "BEFORE", Atom(imap_date(day))))
elif key in ("newer_than", "older_than"):
m = re.fullmatch(r"(\d{1,4})([dwmy])", value.lower())
if not m:
raise Refusal("%s: wants a count and d, w, m or y, such as 7d or 1y" % key)
ago = today - datetime.timedelta(days=int(m.group(1)) * {"d": 1, "w": 7, "m": 30, "y": 365}[m.group(2)])
crits.append((negated, "SINCE" if key == "newer_than" else "BEFORE", Atom(imap_date(ago))))
elif key == "is":
if value.lower() not in FLAGS:
raise Refusal("is: knows %s, not %s" % (", ".join(sorted(FLAGS)), one_line(value, 40)))
crits.append((negated, FLAGS[value.lower()], None))
elif key in ("in", "label"):
if negated or box is not None or not value:
raise Refusal("%s: names one mailbox, once, and cannot be left out with a -" % key)
box = value
else:
raise Refusal("%s: is no operator of this search (%s): put it in quotes to look for it as text" % (key, OPERATORS))
return crits, box
def quoted_string(text):
return '"' + text.replace("\\", "\\\\").replace('"', '\\"') + '"'
def render(crit):
"""(the SEARCH words of one criterion, the string that goes after them as
a literal or None). Bridge reads a literal only where a client sends one."""
negated, key, arg = crit
out = (["NOT"] if negated else []) + [key]
if arg is None:
return out, None
if isinstance(arg, Atom):
return out + [str(arg)], None
if all(" " <= c <= "~" for c in arg):
return out + [quoted_string(arg)], None
return out, arg
def search(bridge, crits):
"""The UIDs, ascending, of the open mailbox's messages that match every
criterion. imaplib sends one literal to a command, and it must be the
command's last word, so a criterion with one is a command of its own (the
last of them rides on the command of the plain criteria) and the answers
are intersected."""
plain, literals = [], []
for crit in crits:
key, literal = render(crit)
(literals if literal is not None else plain).append((key, literal))
head = [w for key, _ in plain for w in key] or ([] if literals else ["ALL"])
jobs = []
if literals:
key, literal = literals.pop()
jobs.append((head + key, literal))
else:
jobs.append((head, None))
jobs += literals
found = None
for key, literal in jobs:
uids = set(bridge.search(key, literal))
found = uids if found is None else found & uids
return sorted(found)
# ---- the connection -------------------------------------------------------
def loopback(host):
"""Whether host is this machine: localhost, or an address in 127.0.0.0/8 or
::1. No name is looked up, so a name that is not localhost is not it."""
if host == "localhost" or host.endswith(".localhost"):
return True
try:
return ipaddress.ip_address(host).is_loopback
except ValueError:
return False
def tls_context(host):
"""TLS for the connection to Bridge. Bridge makes itself a self-signed
certificate that no client can verify, so it is not verified, and only
where the connection cannot leave the machine. Any other host is verified
as the system verifies every server."""
if not loopback(host):
return ssl.create_default_context()
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
context.minimum_version = ssl.TLSVersion.TLSv1_2
return context
class Box:
"""One mailbox as LIST gives it."""
def __init__(self, name, attrs, delim):
self.name, self.attrs, self.delim = name, attrs, delim
@property
def selectable(self):
return "\\noselect" not in self.attrs and "\\nonexistent" not in self.attrs
LIST_LINE = re.compile(rb'\((?P<attrs>[^)]*)\) (?P<delim>"(?:[^"\\]|\\.)*"|NIL) (?P<name>.*)', re.S)
FETCH_START = re.compile(rb"\s*\d+ \(")
def unquote_imap(text):
if len(text) >= 2 and text[:1] == b'"' and text[-1:] == b'"':
text = re.sub(rb"\\(.)", rb"\1", text[1:-1])
return text.decode("utf-8", "replace")
def parse_list(data):
boxes = []
for item in data or []:
literal = None
if isinstance(item, tuple):
item, literal = item
m = LIST_LINE.match(item) if item else None
if not m:
continue
name = literal.decode("utf-8", "replace") if literal is not None else unquote_imap(m["name"].strip())
delim = None if m["delim"] == b"NIL" else unquote_imap(m["delim"])
attrs = {a.lower() for a in m["attrs"].decode("ascii", "replace").split()}
boxes.append(Box(name, attrs, delim))
return boxes
def parse_fetch(data):
"""imaplib's answer to a FETCH as one dict a message: uid, flags, internal
(INTERNALDATE as Bridge wrote it), size and literals, the bodies Bridge
sent in order. imaplib hands a response as the text up to a literal with
the literal, then what follows it as the next item, and a server may put
UID and FLAGS before the literal or after it, so the text of each response
is joined and read as one."""
messages, current = [], None
for item in data or []:
literal = None
if isinstance(item, tuple):
item, literal = item
if not item:
continue
if FETCH_START.match(item):
current = {"attrs": b"", "literals": []}
messages.append(current)
if current is None:
continue
current["attrs"] += item + b" "
if literal is not None:
current["literals"].append(literal)
out = []
for m in messages:
attrs = m["attrs"]
uid = re.search(rb"\bUID (\d+)", attrs)
flags = re.search(rb"\bFLAGS \(([^)]*)\)", attrs)
internal = re.search(rb'\bINTERNALDATE "([^"]+)"', attrs)
size = re.search(rb"\bRFC822\.SIZE (\d+)", attrs)
if not uid:
continue
out.append({
"uid": int(uid.group(1)),
"flags": flags.group(1).decode("utf-8", "replace").split() if flags else [],
"internal": internal.group(1).decode("ascii", "replace") if internal else None,
"size": int(size.group(1)) if size else None,
"literals": m["literals"],
})
return out
class Bridge:
"""The connection to Bridge: opened and logged in when it is made, and the
only code that sends a command. A command that goes wrong ends the call
with a line that says what, and the deadline of the call is kept between
them."""
def __init__(self, base, credential):
self.deadline = time.monotonic() + BUDGET
self.open_box = None
self.validity = {}
try:
url = urllib.parse.urlsplit(base)
host, port = url.hostname, url.port or DEFAULT_PORT
except ValueError:
host = None
if not host:
raise Refusal("YARDR_SOURCE_BASE is no host and port: %s" % one_line(base, 100))
user, colon, password = credential.partition(":")
if not colon or not user or not password:
raise Refusal("PROTON_BRIDGE is not <username>:<password>: give the username and the password Bridge shows for the account (SETUP.md)")
# LOGIN sends the user as an atom, which has no %, * or quote in it, so
# a user that is not a plain address could not be said; and no line
# break could be in the password.
if not re.fullmatch(r"[A-Za-z0-9._+@=-]+", user) or not all(" " <= c <= "~" for c in password):
raise Refusal("PROTON_BRIDGE holds a character LOGIN cannot carry: a Bridge username is an address, and its password is printable ASCII")
self.where = "%s:%d" % (host, port)
try:
self.imap = imaplib.IMAP4(host, port, timeout=TIMEOUT)
except (socket.timeout, TimeoutError):
raise Refusal("no answer from Bridge at %s in %d seconds: is Proton Mail Bridge running and signed in?" % (self.where, TIMEOUT))
except OSError as e:
raise Refusal("no Bridge at %s (%s): is Proton Mail Bridge running and signed in? If it listens on another port, SETUP.md says how to give the yard that one" % (self.where, one_line(e, 100)))
except imaplib.IMAP4.error as e:
raise Refusal("what answers at %s is no IMAP server, or it is not Bridge speaking STARTTLS: %s" % (self.where, words(e)))
self.imap.sock.settimeout(self._left())
try:
self.imap.starttls(ssl_context=tls_context(host))
except (OSError, imaplib.IMAP4.error) as e:
raise Refusal("Bridge at %s would not start TLS (%s): this speaks IMAP with STARTTLS, so Bridge's IMAP connection mode must be STARTTLS, its default (SETUP.md)" % (self.where, words(e)))
self.imap.sock.settimeout(self._left())
try:
self.imap.login(user, password)
except imaplib.IMAP4.error as e:
# A server's words are quoted, and one that echoed the password would put it in the line.
raise Refusal("Bridge refused the login (%s): PROTON_BRIDGE wants the username and the password Bridge shows for the account under Mailbox details, not the Proton password (SETUP.md)" % words(e).replace(password, "[password]"))
except (socket.timeout, TimeoutError, OSError) as e:
raise Refusal("the login to Bridge failed: %s" % one_line(e, 100))
def _left(self):
"""Seconds the next command may take: its own limit, or what the call has left of its."""
left = self.deadline - time.monotonic()
if left <= 0:
raise Refusal("Bridge took too long: the call is over its %d seconds" % BUDGET)
return min(TIMEOUT, left)
def _run(self, what, fn, *args):
timeout = self._left()
try:
self.imap.sock.settimeout(timeout)
return fn(*args)
except (socket.timeout, TimeoutError):
raise Refusal("Bridge did not answer %s in time: a search reads each message of a mailbox, so in: and a date keep it small" % what)
except imaplib.IMAP4.abort as e:
raise Refusal("Bridge dropped the connection %s: %s" % (what, words(e)))
except imaplib.IMAP4.error as e:
raise Refusal("Bridge refused %s: %s" % (what, words(e)))
except OSError as e:
raise Refusal("the connection to Bridge failed %s: %s" % (what, one_line(e, 100)))
def mailboxes(self):
typ, data = self._run("the list of mailboxes", self.imap.list)
if typ != "OK":
raise Refusal("Bridge would not list its mailboxes")
return parse_list(data)
def examine(self, name):
"""Opens the mailbox read-only and says its UIDVALIDITY."""
mailbox_name(name)
if self.open_box == name:
return self.validity[name]
quoted = quoted_string(name)
typ, data = self._run("to open " + one_line(name, 60), self.imap.select, quoted, True)
if typ != "OK":
raise Refusal("Bridge would not open %s: %s" % (one_line(name, 60), one_line(data[-1] if data else "", 100)))
_, found = self.imap.response("UIDVALIDITY")
self.open_box = name
self.validity[name] = int(found[-1]) if found and found[-1] else 0
return self.validity[name]
def search(self, key, literal=None):
args = (["CHARSET", "UTF-8"] if literal is not None else []) + key
def send():
# Set here, so that a command that is never sent leaves none behind
# for the next one to send.
if literal is not None:
self.imap.literal = literal.encode("utf-8")
return self.imap.uid("SEARCH", *args)
typ, data = self._run("the search", send)
if typ != "OK":
raise Refusal("Bridge refused the search: %s" % one_line(data[-1] if data else "", 100))
return [int(u) for u in (data[0] or b"").split()] if data else []
def fetch(self, uids, items):
typ, data = self._run("a message", self.imap.uid, "FETCH", ",".join(str(u) for u in uids), items)
if typ != "OK":
raise Refusal("Bridge refused to give a message: %s" % one_line(data[-1] if data else "", 100))
return parse_fetch(data)
def close(self):
try:
self.imap.logout()
except Exception:
pass
# ---- mailboxes ------------------------------------------------------------
ALIASES = {"sent": "\\sent", "drafts": "\\drafts", "trash": "\\trash", "spam": "\\junk", "junk": "\\junk",
"archive": "\\archive", "starred": "\\flagged", "all": "\\all", "anywhere": "\\all"}
def all_mail(boxes):
"""The mailbox of all mail, which Bridge lists with \\All unless the
operator has hidden it in Bridge, or None."""
return next((b for b in boxes if b.selectable and "\\all" in b.attrs), None)
def default_box(boxes):
return all_mail(boxes) or next((b for b in boxes if b.name.upper() == "INBOX"), None) \
or next((b for b in boxes if b.selectable), None)
def find_box(boxes, want):
"""The mailbox in: names: its name as LIST gives it in any case, a word
for one Bridge marks (sent, trash, spam, archive, starred, all), or the
last part of a nested name (Receipts for Labels/Receipts) when only one
ends so."""
low = want.lower()
usable = [b for b in boxes if b.selectable]
for b in usable:
if b.name.lower() == low:
return b
special = ALIASES.get(low)
for b in usable:
if special and special in b.attrs:
return b
tails = [b for b in usable if b.name.lower().rsplit(b.delim or "/", 1)[-1] == low]
if len(tails) == 1:
return tails[0]
if tails:
raise Refusal("%s ends the names of several mailboxes, %s: give one in full" % (
one_line(want, 60), ", ".join(one_line(b.name, 60) for b in tails)))
raise Refusal("no mailbox %s: Bridge has %s" % (one_line(want, 60), ", ".join(one_line(b.name, 60) for b in usable)))
def open_box(bridge, name, validity):
"""Opens the mailbox an id names, and refuses the id if Bridge has
renumbered it since."""
now = bridge.examine(name)
if now != validity:
raise Refusal("that id is stale: Bridge has renumbered %s (its UIDVALIDITY was %d and is %d), so a number in it may be another message: search again" % (one_line(name, 60), validity, now))
# ---- messages -------------------------------------------------------------
ID = re.compile(r"([A-Za-z0-9%._~-]+):([0-9]+):([0-9]+)")
MSGID = re.compile(r"<[^<>\s]+>")
def make_id(mailbox, validity, uid):
return "%s:%d:%d" % (urllib.parse.quote(mailbox, safe=""), validity, uid)
def parse_id(text):
m = ID.fullmatch(text) if isinstance(text, str) else None
if not m or int(m.group(3)) < 1:
raise Refusal("that is no id of a message: it is the id search answers, mailbox:validity:uid")
encoded = m.group(1)
if re.search(r"%(?![0-9A-Fa-f]{2})", encoded):
raise Refusal("that is no id of a message: its mailbox name is not percent-encoded")
try:
name = urllib.parse.unquote(encoded, errors="strict")
except UnicodeDecodeError:
raise Refusal("that is no id of a message: its mailbox name is not UTF-8")
try:
mailbox_name(name)
except Refusal:
raise Refusal("that is no id of a message: its mailbox name has a control character")
return name, int(m.group(2)), int(m.group(3))
def parse_message(raw):
return email.message_from_bytes(raw or b"", policy=email.policy.default)
def decoded(raw):
"""A header's raw text with its RFC 2047 words decoded, on one line."""
try:
text = str(email.header.make_header(email.header.decode_header(str(raw))))
except Exception:
text = str(raw)
return one_line(text, 100000)
def hval(msg, name):
"""A header of msg as one line of text, or None. The default policy
decodes it; one it cannot parse is read raw and decoded by hand."""
try:
value = msg[name]
return None if value is None else one_line(value, 100000)
except Exception:
for key, raw in msg.raw_items():
if key.lower() == name.lower():
return decoded(raw)
return None
def clip(text, limit):
return text if len(text) <= limit else text[: limit - 1] + "…"
def parse_internal(text):
"""INTERNALDATE, 06-Nov-2026 09:00:00 +0000, as a datetime, or None."""
m = re.fullmatch(r"\s*(\d{1,2})-([A-Za-z]{3})-(\d{4}) (\d\d):(\d\d):(\d\d) ([+-])(\d\d)(\d\d)", text or "")
if not m or m.group(2).title() not in MONTHS:
return None
offset = datetime.timedelta(hours=int(m.group(8)), minutes=int(m.group(9)))
zone = datetime.timezone(-offset if m.group(7) == "-" else offset)
try:
return datetime.datetime(int(m.group(3)), MONTHS.index(m.group(2).title()) + 1, int(m.group(1)),
int(m.group(4)), int(m.group(5)), int(m.group(6)), tzinfo=zone)
except ValueError:
return None
def header_date(text):
"""A Date header as a datetime, or None."""
try:
return email.utils.parsedate_to_datetime(text)
except Exception:
return None
def stamp(when):
"""Seconds since the epoch; a date with no zone (-0000) is taken as UTC."""
if when is None:
return 0.0
if when.tzinfo is None:
when = when.replace(tzinfo=datetime.timezone.utc)
return when.timestamp()
def ids_of(text):
return MSGID.findall(text or "")
class Row:
"""One message as Bridge gave it: where it is and what it carries, and its
headers or all of it as an email.message.Message."""
def __init__(self, mailbox, validity, entry, raw):
self.mailbox, self.validity, self.uid = mailbox, validity, entry["uid"]
self.id = make_id(mailbox, validity, entry["uid"])
self.flags = [f for f in entry["flags"] if f.lower() != "\\recent"]
self.size = entry["size"]
self.internal = parse_internal(entry["internal"])
self.skipped = False
self.read(raw, False)
def read(self, raw, whole):
"""Takes raw as the message's headers, or as all of it when whole."""
self.msg, self.whole = parse_message(raw), whole
self.text = self.files = None
ids = ids_of(hval(self.msg, "Message-ID"))
self.mid = ids[0] if ids else None
self.sent = header_date(hval(self.msg, "Date")) or self.internal
def body(self):
"""Its text and its attachments, worked out once: fit() asks again for a
smaller cut of the same text."""
if self.text is None:
self.text, self.files = body_text(self.msg), attachments(self.msg)
return self.text, self.files
def read_whole(self, bridge):
"""Reads all of the message: the mailbox it is in must be the one open."""
body = bridge.fetch([self.uid], "(UID BODY.PEEK[])")
entry = next((e for e in body if e["uid"] == self.uid and e["literals"]), None)
if entry is None:
raise Refusal("Bridge gave no text for that message")
self.read(entry["literals"][0], True)
def read_rows(bridge, mailbox, validity, uids, header=True):
"""Rows for the UIDs of the open mailbox that Bridge still has, in the
order of uids. The headers are all of them when header, else the three
that search shows."""
if not uids:
return []
fields = "BODY.PEEK[HEADER]" if header else "BODY.PEEK[HEADER.FIELDS (FROM SUBJECT DATE)]"
got = {e["uid"]: e for e in bridge.fetch(uids, "(UID FLAGS INTERNALDATE RFC822.SIZE %s)" % fields)}
return [Row(mailbox, validity, got[u], got[u]["literals"][0] if got[u]["literals"] else b"")
for u in uids if u in got]
class Plain(html.parser.HTMLParser):
"""HTML as the text a reader sees: no tags, scripts or styles, and a line
break where a block ends."""
BLOCKS = {"p", "div", "br", "li", "tr", "table", "ul", "ol", "blockquote", "pre", "hr", "section",
"article", "header", "footer", "h1", "h2", "h3", "h4", "h5", "h6"}
SKIPPED = {"script", "style", "head", "title"}
def __init__(self):
super().__init__(convert_charrefs=True)
self.parts, self.skipping = [], 0
def handle_starttag(self, tag, attrs):
if tag in self.SKIPPED:
self.skipping += 1
elif tag in self.BLOCKS:
self.parts.append("\n")
def handle_startendtag(self, tag, attrs):
self.handle_starttag(tag, attrs)
def handle_endtag(self, tag):
if tag in self.SKIPPED:
self.skipping = max(0, self.skipping - 1)
elif tag in self.BLOCKS:
self.parts.append("\n")
def handle_data(self, data):
if not self.skipping:
self.parts.append(re.sub(r"\s+", " ", data))
def html_to_text(markup):
parser = Plain()
try:
parser.feed(markup)
parser.close()
return "".join(parser.parts)
except Exception:
return re.sub(r"<[^>]*>", " ", markup)
def tidy(text):
text = text.replace("\r\n", "\n").replace("\r", "\n").replace("\xa0", " ")
text = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", text)
text = "\n".join(line.rstrip() for line in text.split("\n"))
return re.sub(r"\n{3,}", "\n\n", text).strip()
def body_text(msg):
"""The text of a message: its plain part, else its HTML part as text."""
try:
part = msg.get_body(preferencelist=("plain", "html"))
except Exception:
part = None
if part is None:
return ""
try:
content = part.get_content()
except Exception:
content = (part.get_payload(decode=True) or b"").decode("utf-8", "replace")
if part.get_content_type() == "text/html":
content = html_to_text(content)
return tidy(content)
def attachments(msg):
"""The name, type and size of each attachment, and never its contents."""
try:
parts = list(msg.iter_attachments())
except Exception:
parts = []
if not msg.is_multipart() and msg.get_content_maintype() != "text":
parts = [msg]
out = []
for part in parts[:100]:
try:
name = part.get_filename()
except Exception:
name = None
try:
data = part.get_payload(decode=True)
size = len(data) if data is not None else len(part.as_bytes())
except Exception:
size = None
entry = {"name": clip(decoded(name), 200) if name else None, "type": part.get_content_type()}
if size is not None:
entry["size"] = size
out.append(entry)
return out
def describe(row, limit=None, brief=False):
"""One message as the answers show it. With a limit its text is read, cut to
that many bytes. A brief one, a message of a thread, leaves out its
mailbox and references, and shortens its recipients."""
msg = row.msg
cc_to = 300 if brief else 2000
out = {"id": row.id}
if not brief:
out["mailbox"] = row.mailbox
out["flags"] = row.flags
out["from"] = clip(hval(msg, "From") or "", 500)
for key, name in (("to", "To"), ("cc", "Cc")):
value = hval(msg, name)
if value:
out[key] = clip(value, cc_to)
out["subject"] = clip(hval(msg, "Subject") or "", 1000)
when = row.sent
out["date"] = when.isoformat() if when else clip(hval(msg, "Date") or "", 100)
if row.mid:
out["messageId"] = row.mid
reply = ids_of(hval(msg, "In-Reply-To"))
if reply:
out["inReplyTo"] = reply[0]
refs = ids_of(hval(msg, "References"))
if refs and not brief:
out["references"] = refs if len(refs) <= 40 else refs[:1] + refs[-39:]
if row.size is not None:
out["size"] = row.size
if limit is not None:
if row.whole:
text, files = row.body()
text, cut = clip_bytes(text, limit)
out["text"] = text
if cut:
out["truncated"] = True
out["attachments"] = files
elif row.skipped:
out["truncated"] = True
return out
def dump(answer):
return json.dumps(answer, ensure_ascii=False, separators=(",", ":")).encode("utf-8", "replace")
def fit(build, droppable=0):
"""The answer build(limit, drop) gives for the largest limit that makes it
fit in OUTPUT bytes: limit is the bytes of text all the messages of an
answer share, and drop the oldest messages left out once no text is left
to give up. Cutting a byte of text takes at least a byte from the JSON, so
each round gives up the overshoot and a little more; a round is cheap, since
each message's text is worked out once (Row.body)."""
limit, drop = TEXT, 0
while True:
answer = build(limit, drop)
size = len(dump(answer))
if size <= OUTPUT:
return answer
if limit:
limit = max(0, limit - (size - OUTPUT) - 256)
elif drop < droppable:
drop += 1
else:
raise Refusal("the answer is over 60 KB even without its text")
# ---- the calls ------------------------------------------------------------
def want_format(params):
fmt = params.get("format") or "metadata"
if fmt not in ("metadata", "full"):
raise Refusal("format is metadata or full")
return fmt
def call_search(params, connect):
try:
limit = int(params.get("maxResults") or 20)
except (TypeError, ValueError):
limit = 0
if not 1 <= limit <= MAX_RESULTS:
raise Refusal("maxResults is a number from 1 to %d" % MAX_RESULTS)
crits, want = plan_search(params.get("q"))
bridge = connect()
boxes = bridge.mailboxes()
box = find_box(boxes, want) if want else default_box(boxes)
if box is None:
raise Refusal("Bridge has no mailbox to search")
validity = bridge.examine(box.name)
uids = search(bridge, crits)
rows = read_rows(bridge, box.name, validity, uids[::-1][:limit], header=False)
hits = []
for row in rows:
when = row.sent
hits.append({
"id": row.id,
"from": clip(hval(row.msg, "From") or "", 200),
"subject": clip(hval(row.msg, "Subject") or "", 300),
"date": when.isoformat() if when else clip(hval(row.msg, "Date") or "", 100),
"flags": row.flags,
})
# A page is bounded by its fields, and a mailbox with a long name or a
# message with many flags can still make 50 of them too much: the oldest go.
return fit(lambda limit, drop: {"mailbox": box.name, "messages": hits[:len(hits) - drop],
"resultSizeEstimate": len(uids)}, max(0, len(hits) - 1))
def call_message(params, connect):
name, validity, uid = parse_id(params.get("id"))
full = want_format(params) == "full"
bridge = connect()
open_box(bridge, name, validity)
got = bridge.fetch([uid], "(UID FLAGS INTERNALDATE RFC822.SIZE%s)" % ("" if full else " BODY.PEEK[HEADER]"))
entry = next((e for e in got if e["uid"] == uid), None)
if entry is None:
raise Refusal("Bridge has no message of that id: it was moved or deleted, or the id is wrong: search again")
if full and (entry["size"] or 0) > FULL_ONE:
raise Refusal("the message is %d MB, over the %d MB read in full: format metadata reads its headers" % (entry["size"] >> 20, FULL_ONE >> 20))
row = Row(name, validity, entry, b"" if full else (entry["literals"] or [b""])[0])
if full:
row.read_whole(bridge)
return fit(lambda limit, drop: describe(row, limit if full else None))
def thread_criteria(ids):
"""The SEARCH words for every message that is, answers or follows one of
ids by its Message-ID, In-Reply-To or References: one OR of them, since
each search of Bridge's reads each message of the mailbox."""
key = []
for one in ids:
for field in ("Message-ID", "In-Reply-To", "References"):
key += ["HEADER", field, quoted_string(one)]
return ["OR"] * (len(ids) * 3 - 1) + key
def call_thread(params, connect):
name, validity, uid = parse_id(params.get("id"))
full = want_format(params) == "full"
bridge = connect()
open_box(bridge, name, validity)
mine = read_rows(bridge, name, validity, [uid])
if not mine:
raise Refusal("Bridge has no message of that id: it was moved or deleted, or the id is wrong: search again")
me = mine[0]
# The thread's root is the first message of References, or else the one
# In-Reply-To names: clients write the chain from the root down. A search
# for the root, and for the message itself, finds the root, the replies and
# the message's own replies, however far down the chain each cites.
refs = ids_of(hval(me.msg, "References"))
reply = ids_of(hval(me.msg, "In-Reply-To"))
root = refs[0] if refs else (reply[0] if reply else me.mid)
# An id with anything but printable ASCII in it would need a literal of its
# own, and a Message-ID has none in practice.
wanted = [i for i in dict.fromkeys([me.mid, root]) if i and all(" " <= c <= "~" for c in i)]
rows, found, matched = [me], [], 1
if wanted:
boxes = bridge.mailboxes()
scope = all_mail(boxes)
own = next((b for b in boxes if b.name == name), None)
# Mail in the trash or spam is not in all mail: its thread is read where it is.
if scope is None or (own is not None and own.attrs & {"\\trash", "\\junk"}):
scope = own
if scope is not None:
sv = bridge.examine(scope.name)
uids = bridge.search(thread_criteria(wanted))
found = read_rows(bridge, scope.name, sv, uids[-MAX_RESULTS:])
# The message's own copy is among the matches, unless the mailbox
# has none; when more matched than are read it may be one not read.
has_copy = len(uids) > MAX_RESULTS or me.mid in {r.mid for r in found}
matched = len(uids) + (0 if has_copy else 1)
rows = found + [me]
def is_me(row):
return row is me or (me.mid is not None and row.mid == me.mid)
# One row to a Message-ID (a message sent to oneself is in the mailbox
# twice), the mailbox's own copy over the message asked about, so that the
# ids of a thread are those of one mailbox where they can be.
unique, seen = [], set()
for row in sorted(rows, key=lambda r: (r is me, r.uid)):
if row.mid is not None and row.mid in seen:
continue
seen.add(row.mid)
unique.append(row)
rows = sorted(unique, key=lambda r: (stamp(r.sent), r.uid))
while len(rows) > MAX_RESULTS:
rows.remove(next(r for r in rows if not is_me(r)))
# What matched and is not shown: not the copies of one message, which are one.
twins = len(found) - len({r.mid or r.id for r in found})
omitted = max(0, matched - twins - len(rows))
if full:
total = 0
for row in rows:
size = row.size or 0
if size > FULL_EACH or total + size > FULL_ALL:
row.skipped = True
continue
total += size
bridge.examine(row.mailbox)
row.read_whole(bridge)
def build(limit, drop):
kept = rows[drop:]
each = limit // max(1, len(kept))
out = {"id": params["id"]}
if root:
out["threadId"] = root
out["messages"] = [describe(r, each if full else None, brief=True) for r in kept]
if omitted + drop:
out["omitted"] = omitted + drop
return out
return fit(build, len(rows) - 1)
CALLS = {"search": call_search, "message": call_message, "thread": call_thread}
def main(argv):
if len(argv) != 2 or argv[1] not in CALLS:
sys.stderr.write("proton-imap: run as proton-imap %s, with the call's parameters on stdin\n" % "|".join(CALLS))
return 2
try:
params = json.loads(sys.stdin.buffer.read().decode("utf-8"))["params"]
if not isinstance(params, dict):
raise TypeError
except (ValueError, KeyError, TypeError):
sys.stderr.write('proton-imap: stdin is no {"params": {...}} object\n')
return 2
bridges = []
def connect():
base, credential = os.environ.get("YARDR_SOURCE_BASE"), os.environ.get("PROTON_BRIDGE")
if not base:
raise Refusal("YARDR_SOURCE_BASE is not set: it is Bridge's address, from the source's base")
if not credential:
raise Refusal("PROTON_BRIDGE is not set: the source's env file holds <username>:<password> for it (SETUP.md)")
bridges.append(Bridge(base, credential))
return bridges[-1]
try:
answer = CALLS[argv[1]](params, connect)
out = sys.stdout.buffer
out.write(dump(answer))
out.flush()
return 0
except Refusal as e:
sys.stderr.write("proton-imap: %s\n" % one_line(e, 600))
except Exception as e:
# A bug of this program, not of the call: say where, in one line.
frame = traceback.extract_tb(e.__traceback__)[-1]
sys.stderr.write("proton-imap: unexpected %s at line %d: %s\n" % (type(e).__name__, frame.lineno, one_line(e, 200)))
finally:
for bridge in bridges:
bridge.close()
return 1
if __name__ == "__main__":
sys.exit(main(sys.argv))
# Proton Mail as a declared source: what an agent may ask of one mailbox
# through Proton Mail Bridge, which runs on the yard's machine and speaks IMAP
# on a loopback port. Three calls, the same three google-mail has: search,
# message and thread. All are reads, and each is answered by the program
# proton/proton-imap, not by a request: yardr's `run` (call.run) runs it with
# the parameters on stdin and the credential in its environment, and its one
# JSON object is the answer. The yard sends nothing, so the base below is no
# address a request goes to; it reaches the program as YARDR_SOURCE_BASE and
# says where Bridge listens.
#
# Checked against Gluon, Bridge's IMAP engine (ProtonMail/gluon d871584, read
# on 7 November 2026), against the real yardr in a scratch yard with a stand-in
# for Bridge, and not against a live Bridge (README.md, "What was checked",
# says what was and was not). What Gluon does that this file relies on:
#
# It advertises IMAP4rev1 UNSELECT UIDPLUS MOVE ID IDLE AUTH=PLAIN and, once
# it has a certificate, STARTTLS. Nothing else: no X-GM-EXT-1 (so no
# X-GM-THRID), no THREAD, no SORT, no ESEARCH, no LITERAL+. So a thread is
# built from References and In-Reply-To, and a search string that needs
# it goes as a synchronizing literal with CHARSET UTF-8, which Gluon reads
# as any IANA charset.
# Its SEARCH matches FROM, TO, CC, BCC, SUBJECT and HEADER as a case-
# insensitive substring of the raw header (an RFC 2047 word is not
# decoded), TEXT as a substring of the raw message, and SINCE and BEFORE by
# the date Bridge holds. Every key but the dates, the flags and the sizes
# makes it read each message of the mailbox, so a text search grows with
# the mailbox.
# It lists the mailbox of all mail with the \All attribute (the operator can
# hide it in Bridge), and Bridge's own folders and labels as Folders/<name>
# and Labels/<name>, with placeholders for Folders and Labels that
# cannot be opened.
#
# No call opens a mailbox with SELECT, changes a flag or moves, copies or
# deletes a message: the program sends LIST, EXAMINE, UID SEARCH, UID FETCH of
# BODY.PEEK and LOGOUT, and nothing else.
#
# The name is proton: a pack is named after the service it talks to.
#
# A command is a list, run without a shell. A path that does not start
# with / is a file beside this one, which `yardr source add` copies into
# the yard.
name: proton
about: Proton Mail through Proton Mail Bridge's local IMAP. Searches mail, reads a message, reads a thread; reads only, sends and changes nothing.
# The name of the variable, never the value. Bridge shows the account a
# username (the address) and a password of its own for a mail client, and the
# variable holds them as <username>:<password>, split at the first colon. The
# program logs in to Bridge with them. It is a password for Bridge on this
# machine and not the Proton password, which Bridge never gives out. It lives
# in the source's env file.
credential:
env: PROTON_BRIDGE
# ---- what an agent may ask of it, through `yardr call` --------------------
# Calls answered by a command, granted one by one. A command is never
# something an agent calls itself.
#
# base is where Bridge listens for IMAP, in the one form yardr takes: http://
# to this machine (an imap:// base is refused as no address yardr sends to).
# The program takes the host and the port out of it and ignores the scheme: it
# always speaks IMAP with STARTTLS, Bridge's default connection mode. Bridge
# takes the first free port from 1143, so the port may be another: SETUP.md
# says how to give the yard's copy of this file the port Bridge shows. A pack
# applied again replaces the yard's copy and puts this one back.
#
# search is the check call: it is the one call with no required parameter.
# `yardr source check proton` and `yardr doctor --try` run it as it stands,
# with no q and the default 20 results, so a Bridge that is not running, a
# login it refuses or a connection mode that is not STARTTLS shows there and
# not at an agent's first call. `source check` prints what search answers: the
# mailbox and its newest messages.
#
# Every call goes to Bridge's one account. What an agent may read is that
# account's whole mailbox, whatever it passes: the grant is the permission.
base: http://127.0.0.1:1143
check: search
calls:
search:
kind: read
about: >-
Search mail, newest first, at most 50 a page. Each message comes with its
id, from, subject, date and flags, and resultSizeEstimate says how many
matched. q is Gmail-like and every term must match: from: to: cc: bcc:
subject: (text in that header), after: and before: (YYYY-MM-DD or
YYYY/MM/DD), newer_than: and older_than: (a count and d, w, m or y, such
as 7d), is:unread, is:read, is:starred, is:answered, is:draft, in: or
label: (inbox, sent, drafts, trash, spam, archive, starred, all, or a
mailbox name as the answer's mailbox gives it), "a phrase" in quotes, a
leading - to leave a term out, and any other word, which is looked for
anywhere in the message. Without in: it searches All Mail, or the inbox
where Bridge hides All Mail, and the answer says which. It refuses an
operator it does not know. Quote the word to search for it as text.
Bridge matches the raw text of headers and messages, so an accented word
can miss mail that encoded it, and every text search reads each message
of the mailbox. in: and the dates keep a search quick.
params:
# Free text in the syntax above, held to a length. Never kept in the
# record of a call. Left out: every message of the mailbox.
q: { type: string, max: 1000, optional: true }
# At most 50, as google-mail's: a string, since an int has no bound.
maxResults: { type: string, pattern: '^([1-9]|[1-4][0-9]|50)$', default: "20" }
run: [proton-imap, search]
answer: { select: [mailbox, messages, resultSizeEstimate] }
message:
kind: read
about: >-
Read one message by an id from search or thread. format metadata, the
default, answers its headers: from, to, cc, subject, date, messageId,
inReplyTo, references, flags, size. format full adds text (the plain
part, or the HTML part as text, cut at about 48 KB with truncated: true)
and attachments (the name, type and size of each, never the contents).
full refuses a message over 25 MB, and metadata still reads it. An id
from before Bridge renumbered its mailbox is refused as stale, so search
again.
params:
# <mailbox, percent-encoded>:<UIDVALIDITY>:<UID>, as search answers it.
id: { type: string, pattern: '^[A-Za-z0-9%._~-]+:[0-9]+:[0-9]+$', max: 200, log: true }
format: { type: enum, values: [metadata, full], default: metadata }
run: [proton-imap, message]
answer: { select: [id, mailbox, flags, from, to, cc, subject, date, messageId, inReplyTo, references, size, text, truncated, attachments] }
thread:
kind: read
about: >-
Read the conversation a message is in, oldest first, by the message's id.
Each message has the shape message gives. Bridge has no thread of its own
(no X-GM-THRID, no THREAD), so a thread is every message whose
Message-ID, References or In-Reply-To names the message or the thread's
root. It is read from All Mail, or from the message's own mailbox where
Bridge hides All Mail or the message is in the trash or spam. At most 50
messages come, the newest and the one asked about, and omitted says how
many were left out. format full shares one text budget of about 48 KB
between them.
params:
id: { type: string, pattern: '^[A-Za-z0-9%._~-]+:[0-9]+:[0-9]+$', max: 200, log: true }
format: { type: enum, values: [metadata, full], default: metadata }
run: [proton-imap, thread]
answer: { select: [id, threadId, messages, omitted] }
# proton Proton Mail, read only. A yard can search the operator's mail, read a message and read the thread a message is in, as `yardr call proton search|message|thread`. These are the same three reads the `google-mail` source gives. The pack sends no mail, writes no draft and sets no flag. Proton offers no public API for mail, so the pack reads it through [Proton Mail Bridge](https://proton.me/mail/bridge), which runs on the yard's machine and serves the account over IMAP on a loopback port. The pack declares the source `proton` and carries a program, `proton-imap`, that answers each call by talking IMAP to Bridge. yardr runs the program (a call's `run`; see "A call may be answered by a command" in `yardr docs integrations`). The program needs Python 3.9 or newer and nothing else. [SETUP.md](SETUP.md) lists what the operator sets up. `.yardr/test` runs `yardr pack check` and this pack's `test`. | file | what it is | | --- | --- | | `pack.yaml` | the pack: kind `source`, the source it declares, one suggested preset (`read`) | | `proton.yaml` | the source: Bridge's default address, one credential, three read calls each run by `proton-imap` | | `proton-imap` | the program: `proton-imap search\|message\|thread`, parameters on stdin, one JSON object out | | `SETUP.md` | what the operator does: Bridge, the pack, the login, the grant, a first call | | `test`, `testdata/` | the pack's tests, a stand-in for Bridge, its mail and the goldens | ## The calls All three are reads, and `read` grants them together. The yard shows each answer as data from outside the yard. Mail is whatever anyone sent the account, so it carries no instruction. **`search`** takes `q` (optional) and `maxResults` (1 to 50, and 20 if left out). It answers: - `mailbox`, the mailbox it read. - `messages`, newest first. Each has `id`, `from`, `subject`, `date` and `flags`. `date` is ISO 8601 with the message's own offset. `flags` are IMAP's, such as `\Seen`, and a message without `\Seen` is unread. - `resultSizeEstimate`, how many messages matched. The count is exact, because IMAP returns every match. "Newest" means the order messages reached the mailbox, which is Bridge's UID order. That matches date order, except for mail Bridge synchronized out of order. **`message`** takes `id` and `format`, which is `metadata` (the default) or `full`. `metadata` answers the headers: `id`, `mailbox`, `flags`, `from`, `to`, `cc`, `subject`, `date`, `messageId`, `inReplyTo`, `references` and `size`. A header the message lacks is left out. `full` adds two things. `text` is the plain part, or the HTML part with its tags, styles and scripts removed, cut at about 48 KB with `truncated: true`. `attachments` lists the first 100, each with `name`, `type` and `size`, and never the contents. `full` refuses a message over 25 MB, and `metadata` still reads it. **`thread`** takes the `id` of any message in the thread, and `format` as `message` does. It answers `id` as asked, `threadId` (the Message-ID of the thread's root) and `messages`, oldest first. Each message has the shape `message` gives, without `mailbox` and `references`. With `full`, the messages share one 48 KB budget of text, and a message whose text was cut says `truncated: true`. A thread has at most 50 messages: the newest, plus the one asked about. `omitted` says how many were left out. An answer is at most 60 KB, under the 64 KB a command may print. Each call stops after 25 seconds, before the yard's 30-second kill, so a failure arrives as a message from the program. ### Ids A message id is `<mailbox>:<UIDVALIDITY>:<UID>`, with the mailbox name percent-encoded: `INBOX:1111:42`, `All%20Mail:2222:107`. IMAP needs exactly that to find a message again, so `message` and `thread` take nothing else. Bridge gives a mailbox a new UIDVALIDITY when it rebuilds its copy, and after that an old UID may point at another message. The program refuses such an id as stale, and the caller searches again. ### `q` `q` is a Gmail-like subset that the program turns into IMAP SEARCH keys. It is not raw IMAP. Agents know Gmail's operators, and `google-mail`'s `q` uses them. Every term must match. A term is `operator:value` (`subject:report`, `subject:"quarterly report"`), a word, or a `"phrase"` in quotes. A `-` in front of a term leaves it out. | operator | means | | --- | --- | | `from:` `to:` `cc:` `bcc:` | the header contains the text | | `subject:` | the subject contains the text | | `after:` `before:` | received on or after, or before, a date such as `2026-10-01` or `2026/10/01` | | `newer_than:` `older_than:` | received within, or more than, a span ago: a count and `d`, `w`, `m` (30 days) or `y` (365 days), such as `7d` or `1y` | | `is:` | `unread`, `read`, `starred` (or `flagged`), `answered`, `draft` | | `in:` `label:` | the mailbox to search: `inbox`, `sent`, `drafts`, `trash`, `spam`, `archive`, `starred`, `all`, a name as the answer's `mailbox` gives it (`Folders/Work`), or the last path part of a name when only one mailbox ends that way (`Receipts` finds `Labels/Receipts`) | | a word, or a `"phrase"` | the text appears anywhere in the message | Without `in:`, the search reads the mailbox Bridge lists with the `\All` attribute, which is All Mail. If the operator hid All Mail in Bridge, it reads the inbox. The answer's `mailbox` says which. Bridge lists mailbox names in IMAP's modified UTF-7, so `Café` appears as `Caf&AOk-`. Ids and the answer's `mailbox` use that form too. The program refuses an operator it does not know, such as `has:attachment`, and lists the ones it knows. It does this before it connects. To search for such a word as text, put it in quotes. `OR`, brackets and braces are not supported and count as words. **What Bridge's search does.** This comes from Gluon's source (see "What was checked"), and it shapes what an agent should expect. Bridge matches against raw text. `from:`, `to:`, `cc:`, `bcc:`, `subject:` and bare words are case-insensitive substring matches on the raw header or message. Bridge does not decode an RFC 2047 word (`=?UTF-8?B?...?=`) first, and most accented subjects and names are sent that way. So `subject:café` can miss a message whose subject shows as "Café", while `from:alice@example.com` matches. A word inside a base64 part is never found. Each of these searches also makes Bridge read every message in the mailbox from its own store, so the time grows with the mailbox. `in:` and the date operators are answered without opening a message, so they keep a search quick. A string with an accent goes to Bridge as a literal under `CHARSET UTF-8`, which Gluon accepts. A search that takes more than 20 seconds fails with "Bridge did not answer in time". ### Thread Bridge has no thread of its own. Gluon advertises only `IMAP4rev1 UNSELECT UIDPLUS MOVE ID IDLE AUTH=PLAIN STARTTLS`. It has no Gmail extension, so no `X-GM-THRID`, and no `THREAD`, `SORT` or `ESEARCH`. The program builds a thread from `References` and `In-Reply-To`, as mail clients do. 1. The thread's root is the first id in `References`, else the id in `In-Reply-To`, else the message itself. 2. One search finds every message whose `Message-ID`, `In-Reply-To` or `References` names the message or the root. That gives the root, its replies and the message's own replies. 3. The search runs in All Mail, which holds mail from every folder and from Sent. Where All Mail is hidden, it runs in the message's own mailbox. A message in the trash or spam is not in All Mail, so its thread comes from that mailbox, as far as that mailbox has it. A message without a `Message-ID` is a thread of itself. The program never uses subjects, so two unrelated messages with the same subject are two threads. A reply whose `References` a client cut is not found. ## What it does not do It sends no mail and writes no drafts. An update call could follow once reads are trusted, but yardr refuses an update that a command answers today. It changes no flags and moves or deletes nothing. It does not hear new mail into the inbox, because Bridge sends nothing out. A check on a timer is a later bead. It reads no attachment contents, only names, types and sizes. It does not use Proton's own web API, only what Bridge shows a mail client. ## How it keeps to that `proton-imap` sends five commands and no others: `LIST`, `EXAMINE` (which opens a mailbox read-only; the program never sends `SELECT`), `UID SEARCH`, `UID FETCH` of `BODY.PEEK` sections, and `LOGOUT`. One class, `Bridge`, is the only code that touches the connection. The test's stand-in for Bridge answers every other command with `NO` and fails the test, so an edit that sends `STORE`, or fetches a body without `PEEK`, shows at once. The connection uses STARTTLS before the login, and the program never sends the login in the clear. The test's stand-in refuses a clear login and counts it. Bridge makes itself a self-signed certificate that no client can verify. The program skips certificate checks only where the connection cannot leave the machine: an address in `127.0.0.0/8`, `::1`, or the name `localhost`. It decides without looking up any name. Any other host is verified the way the system verifies every server. What remains is a program of another user on the same machine that listens on Bridge's port while Bridge is not running and collects the login. That password opens Bridge on this machine and nothing else, and Proton's own password is never in it. The program does not speak Bridge's SSL connection mode, so leave IMAP on STARTTLS, which is the default. `PROTON_BRIDGE` holds `<username>:<password>` as Bridge shows them for the account, split at the first colon. The password is Bridge's own, not the Proton password. The yard gives it to the program in its environment, and the program sends it only to Bridge's login. A failure prints one line, with Bridge's words flattened to one line, and the test searches every output for the password. The base is `http://127.0.0.1:1143`, Bridge's default address. It says `http://` because yardr accepts only `https://`, or `http://` to this machine, as a source's address, and refuses `imap://`. The program reads the host and port from it and ignores the scheme. [SETUP.md](SETUP.md) shows how to give the yard's copy another port. The source's check call is `search`, the one call with no required parameter. `yardr source check proton` and `yardr doctor --try` make it with no query, so a Bridge that is not running or a login it refuses shows there and not at the first call an agent makes. ## What was checked - Gluon, Bridge's IMAP engine (ProtonMail/gluon `d871584`, read on 7 November 2026): what it advertises, how `SEARCH` matches, how it parses quoted strings and literals, how it writes LIST and FETCH responses, and that it has no thread extension. - proton-bridge (`36d7b6a`): the `\All` mailbox, the `Folders/` and `Labels/` names, the default port, and that its certificate is RSA 2048. - The program and the declaration, against a stand-in for Bridge (`testdata/bridge_stub.py`). The stand-in does STARTTLS with a certificate that `openssl` makes for each run, matches search the way Gluon does, speaks literals, and refuses anything that could change a mailbox. - The real yardr in a scratch yard, against the same stand-in, through `yardr call`. - Python 3.9 (the macOS system Python) and 3.14. **Not checked: a live Bridge.** The stand-in follows Gluon's source, not a running Bridge's answers. The first call against a real Bridge is therefore the real check. Look at how it greets, what `LIST` returns (the program relies on the `\All`, `\Sent`, `\Trash`, `\Junk` and `\Archive` attributes), the `UIDVALIDITY` values, how long a search of a large mailbox takes, and how an old message's headers and parts look. A difference there is a bug in this pack, not in Bridge. ## The tests Run `./test` in this directory, or `sh .yardr/test` from the repository, which also runs `yardr pack check`. The test needs `openssl` (for the stand-in's throwaway certificate), `yardr` and Python 3.9 or newer, and it fails if any is missing instead of skipping. It needs no network and no secret, never touches a live Bridge, and takes under half a minute. It has four parts, cheapest first: 1. The program's own functions, called in the test's process. They cover the query translator (tables of what each `q` becomes and is refused for), how a search with accents splits into commands, which hosts skip certificate checks, the parsers for imaplib's answers, HTML to text, the 60 KB bound, the deadline and the timeout. 2. The program as a process, run the way the yard runs it, against the stand-in. This covers each call's answer (in goldens where the shape matters), big, odd and missing mail, each way a call fails, and an audit of every command the stand-in saw. 3. The test itself, run again as a process that dies after its temp dir is made. The copy gets an `openssl` that fails, so it exits at the certificate, and the test checks that it removed its temp dir. That holds the test to cleaning up on every exit, not only after its last line. 4. The real yardr in a scratch yard. It applies the pack from this directory, reads the declaration back, runs the grants SETUP.md teaches exactly as written, and makes each call through `yardr call`, comparing the answer with what the program gives on its own. `UPDATE_GOLDEN=1 ./test` rewrites the goldens in `testdata/golden/` from what the program prints. Read the diff before you keep it.
# Setting up Proton Mail The pack adds one Proton Mail account to the yard as the declared source `proton`. An agent can search the mailbox, read a message and read the thread a message belongs to. It cannot send mail, write a draft or set a flag. Proton offers no public API for mail, so the yard reads it through Proton Mail Bridge. Bridge is Proton's app for mail programs. It runs on the yard's machine and serves the account over IMAP on a loopback port. A program the pack carries, `proton-imap`, talks to Bridge for the yard. Proton sends nothing to the yard, so there is no hook or webhook to set up. **The yard acts as one account.** Every agent granted a call of this source reads that account's whole mailbox, whatever it asks for. Bridge has no narrower login. Nobody is asked before a call is made, so the grant is the permission. Mail is whatever anyone wrote to the account, so the yard shows each answer as data from outside the yard, never as an instruction. You need: - a Proton account on a plan that includes Bridge; - Bridge, started on the yard's machine, since it serves only while it runs; - `python3` 3.9 or newer on the `PATH` of the yard's server, with nothing installed beside it; - a yardr that answers a call with a command (`yardr docs integrations` has "A call may be answered by a command"). Commands below call the pack's directory (the `proton` directory of the yardr-packs checkout) `<pack dir>`. 1. **Bridge, signed in.** Install Bridge from Proton's site and sign in to the account the yard should read. Bridge then synchronizes the account's mail to the machine. A large mailbox takes a while, and a search sees only what has synchronized so far. Open the account's mailbox details in Bridge. They show the settings a mail program takes: - the IMAP host, `127.0.0.1`; - the IMAP port, `1143` unless that was taken (Bridge takes the first free port from 1143); - the security, STARTTLS. The program speaks IMAP with STARTTLS only, so if you switched Bridge's connection mode to SSL, switch IMAP back; - the username, which is the account's address; - the password. It is Bridge's own, a random string made for mail programs. It is **not the Proton password**, which Bridge never gives out and the yard never needs. Keep Bridge running and signed in while the yard reads. With Bridge stopped, every call says there is no Bridge. If you hid the All Mail folder in Bridge, the pack reads the inbox instead of all mail (README.md says what changes). 2. **The pack.** Check what applying it does, then apply it as the operator: ```sh yardr pack check <pack dir> ``` ```sh yardr pack apply <pack dir> ``` The check and the apply card say a command answers the source's calls. The yard runs it as you, with the login in its environment, and it can make requests of its own with that login. The command is `proton-imap`, one Python file that uses only its standard library, so read it before you apply. It connects to the address in the source's `base` and nowhere else. It sends five kinds of IMAP command and all of them only read (README.md lists them), and the pack's test holds it to that. The source's address is Bridge's default, `127.0.0.1:1143`. 3. **Bridge's port, if it is not 1143.** The yard sends a login only to the address the source's file names. If Bridge shows another port, write it into a copy of the declaration and add the copy to the yard. Keep the copy beside the original, so the program it names is found. Git ignores the copy. The address says `http://` on purpose. yardr accepts only `https://`, or `http://` to this machine, as a source's address, and refuses `imap://`. The program ignores the scheme and reads the host and the port: ```sh sed 's|^base: .*|base: http://127.0.0.1:<port>|' <pack dir>/proton.yaml > <pack dir>/proton.local.yaml ``` ```sh yardr source add <pack dir>/proton.local.yaml ``` ```sh yardr source show proton ``` The first line of `source show` is the address in use. Do this before step 4, because the yard pairs a login with the address it was given for. Applying the pack again puts the shipped address back and holds the login until you add the copy again. 4. **The login.** The credential is one value, `PROTON_BRIDGE`. It is the username and the password from step 1 joined by a colon, such as `you@proton.me:AbCdEfGhIjKlMnOp`. `yardr source key` asks for it and shows nothing as you type. The yard's server then writes it into the source's env file, `~/.yardr/sources/proton.env`, with mode 0600. The value is never an argument or a command's variable, where every process of your user could read it: ```sh yardr source key proton ``` Then check the source. This confirms the login is set and the program can be run, then makes the source's check call, `search` with no query. That logs in to Bridge, so a Bridge that is not running or a login it refuses shows here (step 7 lists what each says), and the check prints the mailbox it read and the newest messages: sender, subject and date. Step 6 tries each call: ```sh yardr source check proton ``` 5. **Who may call what.** Every call is a read, and `read` covers all three: `search`, `message` and `thread`. `yardr pack show proton` prints the pack's one suggested preset. A pack applied from a directory takes it by name, as this line does: ```sh yardr source grant proton read --crew yardmaster ``` A grantee who comes later gets one card: ```sh yardr bead ask <bead> "Let the reviewers read Proton mail" --run "source grant proton read --group reviewers" --context "The reviewers read mail the yard was sent; the card names each call it covers." ``` One caller may make 60 calls of a source in an hour. An agent that reads a mailbox message by message can run out. Raise the limit with `yardr source set proton calls_per_hour=<n>`. To take a grant back, run `yardr source revoke proton read --crew yardmaster`. 6. **Try it.** As the operator, or in a session that was granted `read`: ```sh yardr call proton search maxResults=3 ``` The answer names the mailbox it read (`All Mail`, or the inbox) and lists the newest messages first. Each has an `id`, sender, subject, date and flags, and `resultSizeEstimate` says how many matched. Then use an `id` from that answer: ```sh yardr call proton message id=<an id> format=full ``` ```sh yardr call proton thread id=<an id> ``` `search` takes a Gmail-like `q`, such as `q="from:alice newer_than:7d"` or `q="is:unread in:inbox"`. README.md lists every operator. A call stops after 25 seconds. A search makes Bridge read every message in the mailbox, so `in:` and the date operators keep a search of a large mailbox quick. 7. **When it does not work.** The program says what is wrong in one line, and the yard shows that line under the call. - *no Bridge at 127.0.0.1:1143*: Bridge is not running or not signed in, or it listens on another port (step 3). - *Bridge refused the login*: the password in `PROTON_BRIDGE` is not the one Bridge shows now. Copy the current one and run `yardr source key proton` again. - *would not start TLS ... connection mode must be STARTTLS*, or *is no IMAP server, or it is not Bridge speaking STARTTLS*: Bridge's IMAP connection mode is probably SSL, or something other than Bridge listens on that port. Switch Bridge's IMAP mode to STARTTLS (step 1) and check the port (step 3). - *Bridge did not answer ... in time*: a command took more than 20 seconds. That is usually a search of a large mailbox, or Bridge is still synchronizing. Narrow the search with `in:` or a date, or wait. - *that id is stale*: Bridge renumbered the mailbox, which it does when it rebuilds its copy. Search again for a fresh id. - An empty search soon after you signed in usually means Bridge is still synchronizing. 8. **Ending.** Take the grants back, then remove the source and its program. The file with the login stays at `~/.yardr/sources/proton.env`, so delete it yourself: ```sh yardr source revoke proton read --crew yardmaster ``` ```sh yardr source rm proton ``` Removing the account from Bridge also ends the yard's access, because the password stops working.
#!/usr/bin/env python3
# The proton pack's tests: what the gate runs beside yardr pack check. Needs no
# network and no secret, done in seconds, and never touches a live Bridge: the
# Bridge here is testdata/bridge_stub.py, a socket loop in this process that
# answers the commands proton-imap sends, over STARTTLS with a certificate made
# here and thrown away. Needs openssl and yardr, and fails without them, as
# snowflake's test does: a part that was skipped would leave the pack unpinned.
#
# functions proton-imap's own functions called in this process, the
# cheapest way to hold what needs no server: the query translator
# as tables of what each q becomes and what it is refused for,
# the way a search with accents is split into commands, which
# hosts get no certificate check, the parsers of what imaplib hands
# back, HTML as text, the bound on an answer, and the program's
# deadline and timeout.
# program proton-imap as a process, run as the yard runs it (parameters on
# stdin, the login and the base in its environment and nothing
# else of the yard's) against the stub: each call's answer, in
# goldens where its shape is the point, then what it does with
# mail that is big, odd or missing; each way a call fails, with the
# line it says; and, of every command the stub saw, that all of
# them were reads, the password never went in the clear and no
# output held it.
# exit this file run again, as a process that dies after its temp dir is
# made (an openssl that fails ends it at the certificate), to hold
# that the dir is gone: cleanup runs at every exit, not only after
# the last line.
# yardr the real yardr in a scratch yard (its own home, an environment
# of nothing but HOME and PATH, a short path for its socket): the
# pack applied from its directory, the declaration as the yard reads
# it, the grants SETUP.md teaches run as written, and the calls made
# through yardr call against the stub, each held to what the program
# gives when run on its own.
#
# UPDATE_GOLDEN=1 ./test writes the goldens again from what the program
# prints; read the diff before keeping it.
import atexit
import datetime
import difflib
import importlib.machinery
import importlib.util
import json
import os
import re
import shlex
import shutil
import socket
import ssl
import subprocess
import sys
import tempfile
sys.dont_write_bytecode = True
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.join(HERE, "testdata"))
import bridge_stub as bs # noqa: E402
PROGRAM = os.path.join(HERE, "proton-imap")
USER, PASSWORD = "bob@example.com", "b3f9-bridge-pass"
CREDENTIAL = USER + ":" + PASSWORD
UPDATE = os.environ.get("UPDATE_GOLDEN") == "1"
failed = []
leaks = []
def ok(name, condition, detail=""):
print("test: %s %s" % (name, "ok" if condition else "FAILED" + (": " + str(detail) if detail else "")))
if not condition:
failed.append(name)
def golden(name, value):
"""Holds value, as JSON, to testdata/golden/<name>.json."""
path = os.path.join(HERE, "testdata", "golden", name + ".json")
text = json.dumps(value, indent=2, ensure_ascii=False) + "\n"
if UPDATE:
os.makedirs(os.path.dirname(path), exist_ok=True)
if not os.path.exists(path) or open(path, encoding="utf-8").read() != text:
with open(path, "w", encoding="utf-8") as f:
f.write(text)
print("test: %s golden written" % name)
return
if not os.path.exists(path):
ok("golden " + name, False, "no golden %s (UPDATE_GOLDEN=1 ./test writes it)" % path)
return
want = open(path, encoding="utf-8").read()
diff = "\n".join(list(difflib.unified_diff(want.splitlines(), text.splitlines(), "golden", "answer", lineterm=""))[:30])
ok("golden " + name, want == text, "differs from %s:\n%s" % (path, diff))
for tool in ("openssl",):
if not shutil.which(tool):
print("test: %s is not on PATH: the pack needs it" % tool)
sys.exit(1)
YARDR = os.environ.get("YARDR") or shutil.which("yardr") or "/Users/mckean/go/bin/yardr"
if not os.access(YARDR, os.X_OK):
print("test: no yardr (%s): the declaration and the calls are read through it" % YARDR)
sys.exit(1)
tmp = tempfile.mkdtemp(prefix="protont.", dir="/tmp")
stubs = []
core = None
def cleanup():
for stub in stubs:
try:
stub.stop()
except Exception:
pass
if core:
subprocess.run([YARDR, "--home", core + "/y", "--actor", "tester", "server", "stop"], capture_output=True,
env={"HOME": core + "/h", "PATH": os.environ["PATH"]})
shutil.rmtree(core, ignore_errors=True)
shutil.rmtree(tmp, ignore_errors=True)
# At exit, not as the last statement: an exception or a sys.exit anywhere below ends the run too, and
# would leave the dirs, and from the yardr part on the scratch yard's server, behind.
atexit.register(cleanup)
# ---- the program, loaded to call its functions ---------------------------------
loader = importlib.machinery.SourceFileLoader("proton_imap", PROGRAM)
spec = importlib.util.spec_from_loader("proton_imap", loader)
P = importlib.util.module_from_spec(spec)
loader.exec_module(P)
# ---- a certificate made here and thrown away -----------------------------------
made = subprocess.run(["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-keyout", tmp + "/key.pem",
"-out", tmp + "/cert.pem", "-days", "2", "-subj", "/CN=localhost"], capture_output=True)
if made.returncode != 0:
print("test: the throwaway certificate could not be made: %s" % made.stderr.decode()[-300:])
sys.exit(1)
# ---- the mail the stand-in holds ---------------------------------------------------
NAMES = "welcome report reply news other muller sentreply trashed work".split()
SENT = {"welcome": "02-Nov-2026 08:00:00 +0000", "report": "03-Nov-2026 08:30:00 +0000", "reply": "03-Nov-2026 11:00:00 +0000",
"news": "04-Nov-2026 07:15:00 +0000", "other": "05-Nov-2026 12:00:00 +0000", "muller": "06-Nov-2026 08:30:00 +0000",
"sentreply": "04-Nov-2026 10:00:00 +0000", "trashed": "05-Nov-2026 09:00:00 +0000", "work": "06-Nov-2026 14:00:00 +0000"}
FLAGS = {"welcome": ["\\Seen"], "report": ["\\Seen", "\\Answered"], "reply": [], "news": ["\\Flagged"], "other": [],
"muller": ["\\Seen"], "sentreply": ["\\Seen"], "trashed": ["\\Seen"], "work": []}
RAW = {}
for _name in NAMES:
with open(os.path.join(HERE, "testdata", "mail", _name + ".eml"), "rb") as _f:
RAW[_name] = _f.read()
def copies(first_uid, names):
return [bs.Msg(first_uid + i, RAW[n], FLAGS[n], SENT[n]) for i, n in enumerate(names.split())]
def world(all_mail=True):
"""The account: an inbox, Bridge's special folders, all mail with a copy of what is not in the trash,
and a folder and two labels, each a placeholder (\\Noselect) above them."""
boxes = [
bs.Mailbox("INBOX", ["\\HasNoChildren"], 1111, copies(1, "welcome report reply news other muller")),
bs.Mailbox("Drafts", ["\\HasNoChildren", "\\Drafts"], 1212, []),
bs.Mailbox("Sent", ["\\HasNoChildren", "\\Sent"], 3333, copies(1, "sentreply")),
bs.Mailbox("Starred", ["\\HasNoChildren", "\\Flagged"], 1313, []),
bs.Mailbox("Archive", ["\\HasNoChildren", "\\Archive"], 1414, []),
bs.Mailbox("Spam", ["\\HasNoChildren", "\\Junk"], 1515, []),
bs.Mailbox("Trash", ["\\HasNoChildren", "\\Trash"], 4444, copies(1, "trashed")),
bs.Mailbox("Folders", ["\\Noselect", "\\HasChildren"], 1, []),
bs.Mailbox("Folders/Work", ["\\HasNoChildren"], 5555, copies(1, "work")),
bs.Mailbox("Labels", ["\\Noselect", "\\HasChildren"], 1, []),
bs.Mailbox("Labels/Receipts", ["\\HasNoChildren"], 6666, copies(1, "welcome")),
bs.Mailbox("Labels/Work", ["\\HasNoChildren"], 7777, []),
]
if all_mail:
boxes.append(bs.Mailbox("All Mail", ["\\HasNoChildren", "\\All"], 2222,
copies(101, "welcome report reply news other muller sentreply work")))
return boxes
def gen(mid, subject, body, refs=(), reply=None, date="Mon, 02 Nov 2026 08:00:00 +0000", extra=(), ctype="text/plain; charset=utf-8",
to="Bob <bob@example.com>"):
"""A message made here, as bytes."""
lines = ["From: Gen <gen@example.com>", "To: " + to, "Subject: " + subject, "Date: " + date]
if mid:
lines.append("Message-ID: " + mid)
if reply:
lines.append("In-Reply-To: " + reply)
if refs:
lines.append("References: " + " ".join(refs))
lines += list(extra) + ["MIME-Version: 1.0", "Content-Type: " + ctype, "", body]
return "\n".join(lines).encode("utf-8")
def start(boxes, **kw):
stub = bs.Stub(boxes, USER, PASSWORD, tmp + "/cert.pem", tmp + "/key.pem", **kw)
stub.start()
stubs.append(stub)
return stub
def env_for(stub, **over):
env = {"PATH": os.environ["PATH"], "HOME": tmp, "PROTON_BRIDGE": CREDENTIAL,
"YARDR_SOURCE_BASE": "http://127.0.0.1:%d" % stub.port}
for key, value in over.items():
if value is None:
env.pop(key, None)
else:
env[key] = value
return env
def run(stub, call, params=None, stdin=None, **over):
"""(exit status, stdout, stderr) of proton-imap as the yard runs it."""
given = stdin if stdin is not None else json.dumps({"params": params or {}})
env = env_for(stub, **over)
p = subprocess.run([PROGRAM, call], input=given, capture_output=True, text=True, env=env, timeout=120)
# What no output may hold: the password, and whatever login this run was given
# (a wrong one is no less a login), unless that is too short to tell from words.
secret = env.get("PROTON_BRIDGE", "")
needles = {PASSWORD, CREDENTIAL} | ({secret, secret.partition(":")[2]} if len(secret.partition(":")[2]) >= 8 else set())
for text in (p.stdout, p.stderr):
if any(n in text for n in needles):
leaks.append("%s %s" % (call, params))
return p.returncode, p.stdout, p.stderr
def answer(stub, call, params=None, **over):
rc, out, err = run(stub, call, params, **over)
if rc != 0:
raise AssertionError("%s %s exited %d: %s" % (call, params, rc, err.strip()))
return json.loads(out)
def refused(stub, call, params, want, name, status=1, **over):
rc, out, err = run(stub, call, params, **over)
ok(name, rc == status and out == "" and want in err and err.count("\n") == 1,
"exit %d, stdout %r, stderr %r, wanted %r in one line" % (rc, out[:80], err, want))
def ids(result):
return [m["id"] for m in result["messages"]]
# =====================================================================================
# functions
# =====================================================================================
TODAY = datetime.date(2026, 11, 7)
PLANS = [
("", [], None),
("invoice", [(False, "TEXT", "invoice")], None),
('"two words"', [(False, "TEXT", "two words")], None),
('"a \\"quoted\\" word"', [(False, "TEXT", 'a "quoted" word')], None),
("from:alice", [(False, "FROM", "alice")], None),
("FROM:Alice", [(False, "FROM", "Alice")], None),
('subject:"big news"', [(False, "SUBJECT", "big news")], None),
("to:a cc:b bcc:c", [(False, "TO", "a"), (False, "CC", "b"), (False, "BCC", "c")], None),
("-from:bob", [(True, "FROM", "bob")], None),
("-spam", [(True, "TEXT", "spam")], None),
("after:2026-10-01", [(False, "SINCE", "1-Oct-2026")], None),
("before:2026/10/09", [(False, "BEFORE", "9-Oct-2026")], None),
("newer_than:7d", [(False, "SINCE", "31-Oct-2026")], None),
("older_than:2w", [(False, "BEFORE", "24-Oct-2026")], None),
("newer_than:1m", [(False, "SINCE", "8-Oct-2026")], None),
("older_than:1y", [(False, "BEFORE", "7-Nov-2025")], None),
("is:unread is:read is:starred is:flagged is:answered is:draft",
[(False, "UNSEEN", None), (False, "SEEN", None), (False, "FLAGGED", None), (False, "FLAGGED", None),
(False, "ANSWERED", None), (False, "DRAFT", None)], None),
("-is:read", [(True, "SEEN", None)], None),
("in:sent", [], "sent"),
("label:Receipts", [], "Receipts"),
('in:"Folders/My Stuff"', [], "Folders/My Stuff"),
("from:alice invoice in:inbox", [(False, "FROM", "alice"), (False, "TEXT", "invoice")], "inbox"),
('"has:attachment"', [(False, "TEXT", "has:attachment")], None),
("12:30", [(False, "TEXT", "12:30")], None),
("-", [(False, "TEXT", "-")], None),
]
for q, crits, box in PLANS:
got = P.plan_search(q, today=TODAY)
ok("q %r is %s" % (q, "%s %s" % (crits, box) if crits or box else "everything"), got == (crits, box), got)
REFUSALS = [
("has:attachment", "is no operator"), ("http://example.com", "put it in quotes"), ('"unclosed', "never closed"),
("from:", "needs something"), ("after:yesterday", "wants a date"), ("after:2026-13-45", "wants a date"),
("before:26-10-01", "wants a date"), ("newer_than:7x", "wants a count"), ("older_than:d", "wants a count"),
("is:weird", "is: knows"), ("in:a in:b", "once"), ("-in:sent", "cannot be left out"), ("in:", "names one mailbox"),
("x " * 31, "more than 30 terms"), ("a\x00b", "NUL"),
]
for q, want in REFUSALS:
try:
P.plan_search(q, today=TODAY)
ok("q %r is refused" % q[:20], False, "it was taken")
except P.Refusal as e:
ok("q %r is refused, saying %r" % (q[:20], want), want in str(e), str(e))
class Fake:
"""The part of Bridge that search() sends its commands to: it keeps them and answers in turn."""
def __init__(self, *answers):
self.calls, self.answers = [], list(answers)
def search(self, key, literal=None):
self.calls.append((key, literal))
return self.answers[len(self.calls) - 1]
SPLITS = [
("nothing is ALL", [], [([1, 2], ["ALL"], None)], [1, 2]),
("plain criteria are one command", [(False, "FROM", "alice"), (False, "UNSEEN", None)],
[([3, 1], ["FROM", '"alice"', "UNSEEN"], None)], [1, 3]),
("one with accents is a literal under CHARSET", [(False, "SUBJECT", "Müller")], [([4], ["SUBJECT"], "Müller")], [4]),
("a plain one rides before the literal", [(False, "FROM", "alice"), (False, "SUBJECT", "Müller")],
[([5], ["FROM", '"alice"', "SUBJECT"], "Müller")], [5]),
("two with accents are two commands, intersected", [(False, "SUBJECT", "Müller"), (False, "TEXT", "Café")],
[([1, 2, 3], ["TEXT"], "Café"), ([2, 3, 4], ["SUBJECT"], "Müller")], [2, 3]),
("a negated one keeps its NOT", [(True, "FROM", "José")], [([9], ["NOT", "FROM"], "José")], [9]),
("a quote and a backslash are escaped", [(False, "SUBJECT", 'say "hi" \\ there')],
[([1], ["SUBJECT", '"say \\"hi\\" \\\\ there"'], None)], [1]),
("a tab is a literal", [(False, "TEXT", "a\tb")], [([1], ["TEXT"], "a\tb")], [1]),
]
for name, crits, commands, want in SPLITS:
fake = Fake(*[c[0] for c in commands])
got = P.search(fake, crits)
ok("search: " + name, fake.calls == [(c[1], c[2]) for c in commands] and got == want, "%s -> %s" % (fake.calls, got))
LOOPBACK = {"127.0.0.1": True, "127.1.2.3": True, "::1": True, "localhost": True, "bridge.localhost": True,
"192.0.2.1": False, "10.0.0.5": False, "0.0.0.0": False, "example.com": False,
"localhost.example.com": False, "127.0.0.1.example.com": False, "::ffff:8.8.8.8": False}
for host, want in LOOPBACK.items():
context = P.tls_context(host)
verifies = context.verify_mode == ssl.CERT_REQUIRED and context.check_hostname
skips = context.verify_mode == ssl.CERT_NONE and not context.check_hostname
ok("TLS to %s is %s" % (host, "not verified" if want else "verified"), P.loopback(host) == want and (skips if want else verifies))
LISTS = [
(b'(\\HasNoChildren \\All) "/" "All Mail"', ("All Mail", {"\\hasnochildren", "\\all"}, "/")),
(b'(\\Noselect \\HasChildren) "/" Folders', ("Folders", {"\\noselect", "\\haschildren"}, "/")),
(b'(\\HasNoChildren) NIL INBOX', ("INBOX", {"\\hasnochildren"}, None)),
(b'(\\HasNoChildren) "/" "Say \\"hi\\""', ('Say "hi"', {"\\hasnochildren"}, "/")),
((b'(\\HasNoChildren) "/" {10}', b'Odd "Name"'), ('Odd "Name"', {"\\hasnochildren"}, "/")),
]
for raw, want in LISTS:
got = P.parse_list([raw])
ok("LIST %r" % (raw,), len(got) == 1 and (got[0].name, got[0].attrs, got[0].delim) == want, [(b.name, b.attrs, b.delim) for b in got])
HEADER = b"From: a@example.com\r\n\r\n"
FETCHES = [
(b"FLAGS", [(b'1 (UID 42 FLAGS (\\Seen) INTERNALDATE "06-Nov-2026 09:00:00 +0000" BODY[HEADER.FIELDS (FROM)] {22}', HEADER), b")"],
[{"uid": 42, "flags": ["\\Seen"], "internal": "06-Nov-2026 09:00:00 +0000", "size": None, "literals": [HEADER]}]),
(b"attributes after the literal", [(b"2 (BODY[HEADER.FIELDS (FROM)] {22}", HEADER), b" UID 43 FLAGS () RFC822.SIZE 99)"],
[{"uid": 43, "flags": [], "internal": None, "size": 99, "literals": [HEADER]}]),
(b"no literal at all", [b"3 (UID 44 FLAGS (\\Seen \\Answered))", None],
[{"uid": 44, "flags": ["\\Seen", "\\Answered"], "internal": None, "size": None, "literals": []}]),
(b"a response with no UID is no answer to a UID FETCH", [b"4 (FLAGS (\\Seen))"], []),
(b"nothing", [None], []),
]
for name, data, want in FETCHES:
ok("FETCH parsed: %s" % name.decode(), P.parse_fetch(data) == want, P.parse_fetch(data))
many = P.parse_fetch([(b"1 (UID 1 BODY[HEADER] {22}", HEADER), b")", (b"2 (UID 2 BODY[HEADER] {22}", HEADER), b")"])
ok("FETCH parsed: two messages each with a literal", [e["uid"] for e in many] == [1, 2] and all(e["literals"] == [HEADER] for e in many), many)
INTERNALS = {"06-Nov-2026 09:00:00 +0000": "2026-11-06T09:00:00+00:00", " 6-Nov-2026 09:00:00 -0500": "2026-11-06T09:00:00-05:00",
"06-Foo-2026 09:00:00 +0000": None, "yesterday": None, None: None}
for text, want in INTERNALS.items():
got = P.parse_internal(text)
ok("INTERNALDATE %r" % (text,), (got.isoformat() if got else None) == want, got)
for mailbox in ("INBOX", "All Mail", "Folders/Work", "A&B-é"):
made_id = P.make_id(mailbox, 77, 5)
ok("id of %r reads back" % mailbox, re.fullmatch(r"[A-Za-z0-9%._~-]+:77:5", made_id) and P.parse_id(made_id) == (mailbox, 77, 5), made_id)
for bad in ("x", "INBOX:1", "INBOX:1:0", ":1:1", "INBOX:a:1", "INBOX%0D%0Aevil:1:1",
"INBOX%GG:1:1", "INBOX%FF:1:1", None, 5):
try:
P.parse_id(bad)
ok("id %r is refused" % (bad,), False)
except P.Refusal as e:
ok("id %r is refused" % (bad,), "no id of a message" in str(e), e)
bad_name_bridge = object.__new__(P.Bridge)
bad_name_bridge.open_box, bad_name_bridge.validity = None, {}
try:
P.Bridge.examine(bad_name_bridge, "INBOX\r\nA999 UID STORE 1 +FLAGS (\\Seen)")
ok("EXAMINE refuses a mailbox name that could add an IMAP command", False)
except P.Refusal as e:
ok("EXAMINE refuses a mailbox name that could add an IMAP command", "control character" in str(e), e)
TEXTS = [
("<html><head><title>t</title><style>p{}</style></head><body><p>Hello & welcome</p><div>a<br>b é</div><script>x()</script></body></html>",
"Hello & welcome\n\na\nb é"),
("<p>one<p>two</p>", "one\ntwo"),
# a line break in markup is white space, as in a browser
("plain <b>bold</b> text\n\n\n\n\nmore", "plain bold text more"),
("<div><<<broken & < markup", None),
]
for markup, want in TEXTS:
got = P.tidy(P.html_to_text(markup))
ok("HTML as text: %r" % markup[:30], want is None or got == want, got)
ok("tidy takes the controls out, keeps the lines and no more than one blank between", P.tidy("a\x00b \r\n\xa0c\x1b\r\n\r\n\r\n\r\nd") == "ab\n c\n\nd",
repr(P.tidy("a\x00b \r\n\xa0c\x1b\r\n\r\n\r\n\r\nd")))
ok("one_line flattens a line break and a terminal control", P.one_line("a\r\nb\x1b[31mc") == "a b [31mc", P.one_line("a\r\nb\x1b[31mc"))
ok("an error of imaplib is read as the text it holds", P.words(P.imaplib.IMAP4.error(b"[AUTH] bad\r\nline")) == "[AUTH] bad line")
fitted = P.fit(lambda limit, drop: {"t": "é" * (min(limit, 90000) // 2), "drop": drop})
ok("fit gives up text until the answer is within 60 KB", len(P.dump(fitted)) <= P.OUTPUT and len(fitted["t"]) > 20000, len(P.dump(fitted)))
dropped = P.fit(lambda limit, drop: {"m": ["x" * 20000] * (5 - drop)}, 4)
ok("fit leaves out the oldest messages when no text is left", len(P.dump(dropped)) <= P.OUTPUT and len(dropped["m"]) == 2, len(dropped["m"]))
try:
P.fit(lambda limit, drop: {"m": "x" * 70000}, 0)
ok("fit refuses an answer it cannot bring within 60 KB", False)
except P.Refusal as e:
ok("fit refuses an answer it cannot bring within 60 KB", "over 60 KB" in str(e))
ok("a thread's search is one OR of three keys for each id", P.thread_criteria(["<a>", "<b>"]).count("OR") == 5
and P.thread_criteria(["<a>"]) == ["OR", "OR", "HEADER", "Message-ID", '"<a>"', "HEADER", "In-Reply-To", '"<a>"', "HEADER", "References", '"<a>"'])
# the mailboxes a query may name, as Bridge lists them
listing = start(world())
boxes = P.parse_list([b'(%s) "/" "%s"' % (" ".join(b.attrs).encode(), b.name.encode()) for b in listing.mailboxes])
FOUND = {"inbox": "INBOX", "INBOX": "INBOX", "sent": "Sent", "Sent": "Sent", "trash": "Trash", "spam": "Spam", "junk": "Spam",
"archive": "Archive", "starred": "Starred", "all": "All Mail", "anywhere": "All Mail", "drafts": "Drafts",
"all mail": "All Mail", "FOLDERS/work": "Folders/Work", "receipts": "Labels/Receipts"}
for want, name in FOUND.items():
ok("in:%s is %s" % (want, name), P.find_box(boxes, want).name == name)
for want, said in (("work", "several mailboxes, Folders/Work, Labels/Work"), ("Folders", "no mailbox Folders"), ("nope", "Bridge has INBOX")):
try:
P.find_box(boxes, want)
ok("in:%s is refused" % want, False)
except P.Refusal as e:
ok("in:%s is refused, saying %r" % (want, said), said in str(e), e)
try:
P.find_box(boxes, "nope")
except P.Refusal as e:
named = str(e).split("Bridge has ")[1].split(", ")
ok("a refusal lists the mailboxes that can be opened and not the placeholders above Bridge's folders and labels",
"Folders/Work" in named and "All Mail" in named and "Folders" not in named and "Labels" not in named, named)
ok("the default is all mail, or the inbox where all mail is hidden",
P.default_box(boxes).name == "All Mail" and P.default_box([b for b in boxes if b.name != "All Mail"]).name == "INBOX")
# the deadline and the timeout of a command, in this process
slow = start(world(), delay=1.5)
base = "http://127.0.0.1:%d" % slow.port
saved = P.TIMEOUT, P.BUDGET
try:
P.TIMEOUT = 0.4
bridge = P.Bridge(base, CREDENTIAL)
try:
P.call_search({"q": "from:alice"}, lambda: bridge)
ok("a search Bridge does not answer in time is refused", False)
except P.Refusal as e:
ok("a search Bridge does not answer in time is refused, saying so", "did not answer the search in time" in str(e), e)
P.TIMEOUT = saved[0]
P.BUDGET = -1
try:
P.call_search({}, lambda: P.Bridge(base, CREDENTIAL))
ok("a call over its deadline is refused", False)
except P.Refusal as e:
ok("a call over its deadline is refused, saying so", "took too long" in str(e), e)
finally:
P.TIMEOUT, P.BUDGET = saved
# the sizes past which a message is not read in full, with the limits made small enough for the mail here
real = P.FULL_ONE, P.FULL_EACH, P.FULL_ALL
connect = lambda: P.Bridge("http://127.0.0.1:%d" % listing.port, CREDENTIAL) # noqa: E731
try:
P.FULL_ONE = 100
try:
P.call_message({"id": "INBOX:1111:2", "format": "full"}, connect)
ok("a message over the size read in full is refused as full", False)
except P.Refusal as e:
ok("a message over the size read in full is refused as full, and metadata is named", "format metadata reads its headers" in str(e), e)
ok("and metadata still reads it", P.call_message({"id": "INBOX:1111:2"}, connect)["size"] == 651)
P.FULL_ONE, P.FULL_EACH = real[0], 500
each = {m["messageId"]: m for m in P.call_thread({"id": "INBOX:1111:3", "format": "full"}, connect)["messages"]}
ok("a message of a thread over the size read in full is shown without its text, marked cut",
each["<q1@example.com>"].get("truncated") is True and "text" not in each["<q1@example.com>"]
and each["<r1@example.com>"]["text"].startswith("Thanks Alice") and "truncated" not in each["<r1@example.com>"], each)
P.FULL_EACH, P.FULL_ALL = real[1], 800
total = {m["messageId"]: m for m in P.call_thread({"id": "INBOX:1111:3", "format": "full"}, connect)["messages"]}
ok("a thread stops reading messages in full when the total is reached: the first is read, the rest are cut",
total["<q1@example.com>"]["text"].startswith("Hi Bob") and all(total[i].get("truncated") is True and "text" not in total[i]
for i in ("<r1@example.com>", "<s1@example.com>")), total)
finally:
P.FULL_ONE, P.FULL_EACH, P.FULL_ALL = real
# =====================================================================================
# program
# =====================================================================================
main = start(world())
main_all = {"mailbox": "All Mail"}
default = answer(main, "search")
golden("search-default", default)
ok("search with nothing asks for all mail, newest first", default["mailbox"] == "All Mail" and default["resultSizeEstimate"] == 8
and ids(default) == ["All%20Mail:2222:" + str(u) for u in (108, 107, 106, 105, 104, 103, 102, 101)], ids(default))
ok("a search shows each message's id, sender, subject, date and flags and nothing more",
all(set(m) == {"id", "from", "subject", "date", "flags"} for m in default["messages"]))
ok("a sender and a subject are decoded", {"News Café <news@example.com>", "José Müller <jose@example.net>"} <= {m["from"] for m in default["messages"]}
and "Café news" in [m["subject"] for m in default["messages"]])
paged = answer(main, "search", {"maxResults": "3"})
ok("maxResults gives the newest few and the number that matched", len(paged["messages"]) == 3 and paged["resultSizeEstimate"] == 8
and ids(paged) == ids(default)[:3])
SEARCHES = [
("from:alice", ["All%20Mail:2222:107", "All%20Mail:2222:102"]),
("cc:carol", ["All%20Mail:2222:102"]),
("to:bob -from:alice -from:dave after:2026-11-04", ["All%20Mail:2222:108", "All%20Mail:2222:106", "All%20Mail:2222:104"]),
('subject:"quarterly report"', ["All%20Mail:2222:" + str(u) for u in (107, 105, 103, 102)]),
("after:2026-11-05", ["All%20Mail:2222:" + str(u) for u in (108, 106, 105)]),
("before:2026-11-03", ["All%20Mail:2222:101"]),
("after:2026/11/03 before:2026/11/05", ["All%20Mail:2222:" + str(u) for u in (107, 104, 103, 102)]),
("is:starred", ["All%20Mail:2222:104"]),
("is:answered", ["All%20Mail:2222:102"]),
("-is:read", ["All%20Mail:2222:" + str(u) for u in (108, 105, 104, 103)]),
("welcome", ["All%20Mail:2222:104", "All%20Mail:2222:101"]),
('"numbers are attached"', ["All%20Mail:2222:103", "All%20Mail:2222:102"]),
('"has:attachment"', []),
("subject:Müller", ["All%20Mail:2222:106"]),
("Café", ["All%20Mail:2222:106"]),
("subject:Müller Café", ["All%20Mail:2222:106"]),
("from:alice subject:Müller", []),
("-subject:Müller from:jose", []),
]
for q, want in SEARCHES:
got = ids(answer(main, "search", {"q": q}))
ok("search q=%r finds %d" % (q, len(want)), got == want, got)
INBOX = lambda *uids: ["INBOX:1111:%d" % u for u in uids] # noqa: E731
IN = [
("is:unread in:inbox", "INBOX", INBOX(5, 4, 3)),
("in:INBOX", "INBOX", INBOX(6, 5, 4, 3, 2, 1)),
("in:sent", "Sent", ["Sent:3333:1"]),
("in:trash", "Trash", ["Trash:4444:1"]),
("in:spam", "Spam", []),
("in:archive", "Archive", []),
("in:starred", "Starred", []),
("in:drafts", "Drafts", []),
("in:all is:starred", "All Mail", ["All%20Mail:2222:104"]),
("in:Folders/Work", "Folders/Work", ["Folders%2FWork:5555:1"]),
("label:receipts", "Labels/Receipts", ["Labels%2FReceipts:6666:1"]),
]
for q, box, want in IN:
got = answer(main, "search", {"q": q})
ok("search q=%r reads %s" % (q, box), got["mailbox"] == box and ids(got) == want, (got["mailbox"], ids(got)))
golden("search-unread-inbox", answer(main, "search", {"q": "is:unread in:inbox"}))
# what the stub saw of those searches: plain ones with no literal and no charset, accents with both
sent = main.searches
ok("a search with no accents sends no literal and no charset", all(s["charset"] is None and s["literals"] == 0 for s in sent
if "Müller" not in s["text"] and "Café" not in s["text"]))
ok("a search with accents sends CHARSET UTF-8 and the string as a literal", all(s["charset"] == "UTF-8" and s["literals"] >= 1 for s in sent
if "Müller" in s["text"] or "Café" in s["text"]))
ok("two strings with accents were two commands, and a plain one rode before a literal",
[s["text"] for s in sent if "Müller" in s["text"] or "Café" in s["text"]]
== ["SUBJECT Müller", "TEXT Café", "TEXT Café", "SUBJECT Müller", "FROM alice SUBJECT Müller", "FROM jose NOT SUBJECT Müller"],
[s["text"] for s in sent if "Müller" in s["text"] or "Café" in s["text"]])
for q, want, name in (("has:attachment", "is no operator", "an operator the search does not know"),
('"unclosed', "never closed", "a quote that is never closed"),
("after:soon", "wants a date", "a date that is none"),
("in:nowhere", "no mailbox nowhere: Bridge has INBOX", "a mailbox Bridge does not have"),
("in:work", "several mailboxes", "a mailbox name two end with")):
before = main.connections
rc, out, err = run(main, "search", {"q": q})
ok("search: %s is refused in one line" % name, rc == 1 and out == "" and want in err and err.count("\n") == 1, (rc, err))
# a q the translator refuses is refused before Bridge is asked anything; a mailbox is known only by asking
if name != "a mailbox Bridge does not have" and name != "a mailbox name two end with":
ok("search: %s is refused before it connects" % name, main.connections == before, main.connections - before)
for bad in ("0", "51", "abc", "-1", "5.5"):
refused(main, "search", {"maxResults": bad}, "maxResults is a number from 1 to 50", "search: maxResults %s is refused" % bad)
# message
report = "INBOX:1111:2"
golden("message-metadata", answer(main, "message", {"id": report}))
golden("message-full-report", answer(main, "message", {"id": report, "format": "full"}))
golden("message-full-news", answer(main, "message", {"id": "INBOX:1111:4", "format": "full"}))
meta = answer(main, "message", {"id": report})
ok("message metadata is the default and has no text", "text" not in meta and answer(main, "message", {"id": report, "format": "metadata"}) == meta)
full = answer(main, "message", {"id": report, "format": "full"})
ok("message full is the plain part decoded and the attachment's name, type and size", full["text"] == "Hi Bob,\n\nThe numbers are attached. Revenue is up 12%.\n\nAlice"
and full["attachments"] == [{"name": "report.pdf", "type": "application/pdf", "size": 55}], full)
ok("a message's date is ISO 8601 with its own offset", meta["date"] == "2026-11-03T09:30:00+01:00")
news = answer(main, "message", {"id": "INBOX:1111:4", "format": "full"})
ok("an HTML message is its text, with no tag, style or script", news["text"] == "Hello & welcome\n\nSecond\nline é" and news["attachments"] == [], news["text"])
thread_member = answer(main, "message", {"id": "All%20Mail:2222:107", "format": "full"})
ok("message reads from all mail as from the inbox, with the headers of a reply",
thread_member["inReplyTo"] == "<r1@example.com>" and thread_member["references"] == ["<q1@example.com>", "<r1@example.com>"]
and thread_member["mailbox"] == "All Mail", thread_member)
refused(main, "message", {"id": "INBOX:9:2"}, "that id is stale", "message: an id from before Bridge renumbered the mailbox is refused as stale")
refused(main, "message", {"id": "INBOX:1111:99"}, "Bridge has no message of that id", "message: an id of a message that is gone is refused")
refused(main, "message", {"id": "Nowhere:1:1"}, "Bridge would not open Nowhere", "message: an id of a mailbox that is not there is refused")
refused(main, "message", {"id": "INBOX"}, "no id of a message", "message: a word that is no id is refused")
refused(main, "message", {"id": report, "format": "html"}, "format is metadata or full", "message: a format it does not have is refused")
# thread
thread = answer(main, "thread", {"id": "INBOX:1111:3"})
golden("thread-report-metadata", thread)
golden("thread-report-full", answer(main, "thread", {"id": "INBOX:1111:3", "format": "full"}))
members = lambda t: [m["messageId"] for m in t["messages"]] # noqa: E731
ok("a thread is the root, its replies and the reply in Sent, oldest first, and no unrelated message with the same subject",
members(thread) == ["<q1@example.com>", "<r1@example.com>", "<s1@example.com>"] and thread["threadId"] == "<q1@example.com>", members(thread))
ok("a thread has no unrelated message, and none that is in the trash", "<o1@example.org>" not in members(thread) and "<t1@example.net>" not in members(thread))
ok("a thread's messages are all of one mailbox, all mail", all(m["id"].startswith("All%20Mail:2222:") for m in thread["messages"]), ids(thread))
ok("a thread leaves out the mailbox and the references, which message gives", all("mailbox" not in m and "references" not in m for m in thread["messages"]))
for start_id in ("INBOX:1111:2", "Sent:3333:1", "All%20Mail:2222:107"):
again = answer(main, "thread", {"id": start_id})
ok("the thread of %s is the same thread" % start_id, members(again) == members(thread) and again["threadId"] == thread["threadId"] and again["id"] == start_id, members(again))
alone = answer(main, "thread", {"id": "INBOX:1111:5"})
ok("a message that is in no thread is its own", members(alone) == ["<o1@example.org>"] and alone["threadId"] == "<o1@example.org>", alone)
trashed = answer(main, "thread", {"id": "Trash:4444:1"})
ok("a thread of mail in the trash is read from the trash, where all mail has no copy", members(trashed) == ["<t1@example.net>"], members(trashed))
refused(main, "thread", {"id": "INBOX:9:3"}, "that id is stale", "thread: an id from before Bridge renumbered the mailbox is refused as stale")
refused(main, "thread", {"id": "INBOX:1111:99"}, "Bridge has no message of that id", "thread: an id of a message that is gone is refused")
# the same answers when the stand-in sends attributes after the literal, and read from a mailbox with no all mail
after = start(world(), order="after")
same = [("search", {"q": "from:alice"}), ("search", {"q": "is:unread in:inbox"}), ("message", {"id": report}),
("message", {"id": report, "format": "full"}), ("thread", {"id": "INBOX:1111:3"}), ("thread", {"id": "INBOX:1111:3", "format": "full"})]
ok("a server that sends UID and FLAGS after the body gives the same answers", all(answer(after, c, p) == answer(main, c, p) for c, p in same))
bare = start(world(all_mail=False))
no_all = answer(bare, "search")
ok("with all mail hidden a search reads the inbox, and says so", no_all["mailbox"] == "INBOX" and no_all["resultSizeEstimate"] == 6, no_all["mailbox"])
no_all_thread = answer(bare, "thread", {"id": "INBOX:1111:3"})
ok("with all mail hidden a thread is read from the message's own mailbox: what is in the inbox",
members(no_all_thread) == ["<q1@example.com>", "<r1@example.com>"] and no_all_thread["id"] == "INBOX:1111:3", members(no_all_thread))
# mail that is big, many, odd or without a Message-ID
long_refs = ["<ref%03d@gen>" % i for i in range(100)]
edge_inbox = [
bs.Msg(1, gen("<big@gen>", "Big one", 'say "hi"\n' * 20000)),
bs.Msg(2, gen("<many@gen>", "Many attachments", "--B\nContent-Type: text/plain; charset=utf-8\n\nsee attached\n" + "".join(
'--B\nContent-Type: text/plain; name="f%03d.txt"\nContent-Disposition: attachment; filename="f%03d.txt"\n\ndata\n' % (i, i)
for i in range(150)) + "--B--\n", ctype='multipart/mixed; boundary="B"')),
bs.Msg(3, gen(None, "No Message-ID", "nothing names this one")),
bs.Msg(4, gen("<long@gen>", "x" * 2000, "short", refs=long_refs, to=", ".join("user%d@example.com" % i for i in range(300)))),
bs.Msg(5, gen("<pdf@gen>", "A bare PDF", "JVBERi0xLjQK", ctype='application/pdf; name="x.pdf"',
extra=["Content-Disposition: attachment; filename=\"x.pdf\"", "Content-Transfer-Encoding: base64"])),
bs.Msg(6, b"From: =?UTF-8?B?!!!not-base64?= <bad@example.com>\nSubject: caf\xe9 in a byte of Latin-1\nDate: not a date at all\n"
b"Message-ID: <bad@gen>\nMIME-Version: 1.0\nContent-Type: text/plain; charset=utf-8\n\nbody\n", internal="06-Nov-2026 08:00:00 +0000"),
bs.Msg(7, gen("<charset@gen>", "A charset nobody knows", "its text, in a charset Python has no codec for", ctype="text/plain; charset=x-no-such-charset")),
]
chain = [bs.Msg(1, gen("<root@gen>", "A long thread", "root " + "x" * 3000))]
for i in range(2, 62):
chain.append(bs.Msg(i, gen("<r%d@gen>" % i, "Re: A long thread", "reply %d " % i + "y" * 3000, refs=["<root@gen>"], reply="<root@gen>",
date="Mon, 02 Nov 2026 %02d:%02d:00 +0000" % (8 + i // 60, i % 60))))
twin = bs.Msg(63, gen("<r5@gen>", "Re: A long thread", "a second copy of reply 5, as mail sent to oneself is", refs=["<root@gen>"], reply="<root@gen>"))
edge = start([bs.Mailbox("INBOX", [], 7, edge_inbox), bs.Mailbox("All Mail", ["\\All"], 8, chain[:61])])
big = answer(edge, "message", {"id": "INBOX:7:1", "format": "full"})
rc, out, err = run(edge, "message", {"id": "INBOX:7:1", "format": "full"})
ok("an answer is within the 64 KB a command may print", len(out.encode()) <= 65536 and len(out.encode()) <= P.OUTPUT, len(out.encode()))
ok("a message too long is cut and says so, at the text's budget", big.get("truncated") is True and 40000 < len(big["text"].encode()) <= P.TEXT
and big["text"].startswith('say "hi"\nsay "hi"'), (big.get("truncated"), len(big["text"].encode())))
many_attachments = answer(edge, "message", {"id": "INBOX:7:2", "format": "full"})
ok("attachments are listed up to 100, by name and size", len(many_attachments["attachments"]) == 100
and many_attachments["attachments"][0] == {"name": "f000.txt", "type": "text/plain", "size": 4}, many_attachments["attachments"][:1])
nomid = answer(edge, "message", {"id": "INBOX:7:3"})
ok("a message with no Message-ID has none in its answer", "messageId" not in nomid and nomid["subject"] == "No Message-ID")
before = len(edge.searches)
alone = answer(edge, "thread", {"id": "INBOX:7:3"})
ok("the thread of a message with no Message-ID is itself, with no search", [m["subject"] for m in alone["messages"]] == ["No Message-ID"]
and "threadId" not in alone and len(edge.searches) == before, alone)
long = answer(edge, "message", {"id": "INBOX:7:4"})
ok("long headers are cut: recipients, subject and references",
len(long["to"]) <= 2000 and long["to"].endswith("…") and len(long["subject"]) == 1000 and len(long["references"]) == 40
and long["references"][0] == long_refs[0] and long["references"][-1] == long_refs[-1], (len(long["to"]), len(long["subject"]), len(long["references"])))
pdf = answer(edge, "message", {"id": "INBOX:7:5", "format": "full"})
ok("a message that is one PDF is its own attachment and has no text", pdf["text"] == "" and pdf["attachments"] == [{"name": "x.pdf", "type": "application/pdf", "size": 9}], pdf)
bad = answer(edge, "message", {"id": "INBOX:7:6", "format": "full"})
ok("a message with a bad date, a bad encoded word and a byte of Latin-1 is read as well as it can be",
bad["date"] == "2026-11-06T08:00:00+00:00" and "bad@example.com" in bad["from"] and bad["subject"].startswith("caf") and bad["text"] == "body", bad)
charset = answer(edge, "message", {"id": "INBOX:7:7", "format": "full"})
ok("a message in a charset Python has no codec for is read as UTF-8, as well as it can be", charset["text"] == "its text, in a charset Python has no codec for", charset)
hits = answer(edge, "search", {"q": "in:inbox"})
ok("a search of a mailbox with odd mail still answers", len(hits["messages"]) == 7 and hits["messages"][1]["subject"].startswith("caf"), hits["messages"][:2])
# a thread of 61 messages, each with 3 KB of text: the newest 50 and the message asked about, whose
# copy is one of the older, the rest counted
first = answer(edge, "thread", {"id": "All%20Mail:8:3"})
ok("a thread over 50 messages is the newest and the message asked about, and says how many are left out",
len(first["messages"]) == 50 and first["omitted"] == 11 and "<r3@gen>" in members(first) and "<r13@gen>" in members(first)
and "<r12@gen>" not in members(first) and "<root@gen>" not in members(first), (len(first["messages"]), first.get("omitted")))
rc, out, err = run(edge, "thread", {"id": "All%20Mail:8:30", "format": "full"})
full_thread = json.loads(out)
ok("a thread read in full shares the text budget, within 64 KB", rc == 0 and len(out.encode()) <= P.OUTPUT and len(full_thread["messages"]) == 50
and all(m.get("truncated") is True for m in full_thread["messages"]) and sum(len(m["text"].encode()) for m in full_thread["messages"]) <= P.TEXT,
(rc, len(out.encode()), err))
flooded = start([bs.Mailbox("INBOX", [], 1, [])], flood=200000)
many_found = answer(flooded, "search")
ok("an answer to a search of 200,000 UIDs, over a megabyte on one line, is read", many_found["resultSizeEstimate"] == 200000 and many_found["messages"] == [],
many_found["resultSizeEstimate"])
twins = start([bs.Mailbox("INBOX", [], 1, [chain[0], chain[4]]),
bs.Mailbox("All Mail", ["\\All"], 2, [chain[0], chain[4], twin])])
tw = answer(twins, "thread", {"id": "INBOX:1:5"})
ok("two copies of one message are one in a thread, and are not counted as left out",
members(tw) == ["<root@gen>", "<r5@gen>"] and "omitted" not in tw and ids(tw) == ["All%20Mail:2:1", "All%20Mail:2:5"], (members(tw), tw.get("omitted"), ids(tw)))
# how a call fails
refused(main, "search", {}, "Bridge refused the login", "a wrong password is refused, and never shown",
PROTON_BRIDGE=USER + ":not-the-password")
careless = start(world(), echo=True)
rc, out, err = run(careless, "search", {}, PROTON_BRIDGE=USER + ":not-the-password")
ok("a server that says the password in refusing the login does not get it into the line", rc == 1 and "Bridge refused the login" in err
and "not-the-password" not in err and "for %s:[password]" % USER in err, (rc, err))
refused(main, "search", {}, "PROTON_BRIDGE is not set", "a missing login is refused", PROTON_BRIDGE=None)
refused(main, "search", {}, "PROTON_BRIDGE is not <username>:<password>", "a login with no colon is refused", PROTON_BRIDGE="nocolon")
refused(main, "search", {}, "PROTON_BRIDGE holds a character LOGIN cannot carry", "a username LOGIN cannot carry is refused", PROTON_BRIDGE='a b"c:pw')
refused(main, "search", {}, "PROTON_BRIDGE holds a character LOGIN cannot carry", "a username with a % is refused: it is no atom", PROTON_BRIDGE="a%b@example.com:password1234")
refused(main, "search", {}, "PROTON_BRIDGE holds a character LOGIN cannot carry", "a password with a line break is refused", PROTON_BRIDGE="u@example.com:pw\nrest")
refused(main, "search", {}, "YARDR_SOURCE_BASE is not set", "a missing base is refused", YARDR_SOURCE_BASE=None)
refused(main, "search", {}, "YARDR_SOURCE_BASE is no host and port", "a base with no host is refused", YARDR_SOURCE_BASE="http://:1143x")
gone = socket.socket()
gone.bind(("127.0.0.1", 0))
closed = gone.getsockname()[1]
gone.close()
refused(main, "search", {}, "no Bridge at 127.0.0.1:%d" % closed, "no Bridge listening is refused, saying where and what to do",
YARDR_SOURCE_BASE="http://127.0.0.1:%d" % closed)
no_tls = start(world(), starttls=False)
refused(no_tls, "search", {}, "connection mode must be STARTTLS", "a Bridge that does not offer STARTTLS is refused, naming its connection mode")
ok("the password was never sent to a server that did not offer TLS", no_tls.logins == [] and no_tls.clear_logins == 0, no_tls.logins)
hung = start(world(), hang_up=True)
refused(hung, "search", {}, "is no IMAP server", "a server that hangs up is refused")
rc, out, err = run(main, "nothing")
ok("an unknown call is a usage error", rc == 2 and "run as proton-imap search|message|thread" in err)
rc, out, err = run(main, "search", stdin="not json")
ok("input that is no JSON is a usage error", rc == 2 and "stdin is no" in err)
rc, out, err = run(main, "search", stdin='{"params": []}')
ok("input whose params are no object is a usage error", rc == 2 and "stdin is no" in err)
# what the stand-in saw
everything = [s for s in stubs if s.connections]
verbs = set(v for s in everything for v in s.commands)
ok("every command sent was a read: %s" % ", ".join(sorted(verbs)), verbs <= {"CAPABILITY", "STARTTLS", "LOGIN", "LIST", "EXAMINE", "UID SEARCH", "UID FETCH", "LOGOUT"}, verbs)
ok("no mailbox was opened with SELECT, and nothing was changed, moved or copied: no violations", all(s.violations == [] for s in stubs), [s.violations for s in stubs if s.violations])
ok("no password was sent before TLS", all(s.clear_logins == 0 for s in stubs))
ok("every login was the one in PROTON_BRIDGE, split at its colon", all(login in {(USER, PASSWORD), (USER, "not-the-password")} for s in stubs for login in s.logins), [s.logins for s in stubs if s.logins][:1])
ok("no output of any run held the password or the login", leaks == [], leaks)
# =====================================================================================
# exit
# =====================================================================================
# A run that dies after its temp dir is made must not leave it. Only a process shows that, so this
# file is run again with an openssl that fails: the certificate is the first step after the dir is
# made, and it exits there. The stand-in writes its arguments down, and they name the dir that run made.
# The copy is told it is one, so that if it ever got as far as this it would not start another.
if "PROTON_TEST_COPY" not in os.environ:
standin = tmp + "/standin"
os.mkdir(standin)
with open(standin + "/openssl", "w") as f:
f.write("#!/bin/sh\nprintf '%s\\n' \"$@\" > " + shlex.quote(standin + "/args") + "\nexit 1\n")
os.chmod(standin + "/openssl", 0o755)
# The copy takes a fraction of a second. The timeout is for one that does not stop at the certificate.
exited = subprocess.run([sys.executable, os.path.join(HERE, "test")], capture_output=True, text=True, timeout=60,
env=dict(os.environ, PATH=standin + os.pathsep + os.environ["PATH"], PROTON_TEST_COPY="1"))
asked = open(standin + "/args").read() if os.path.exists(standin + "/args") else ""
keyout = re.search(r"^-keyout\n(/tmp/protont\.[^/\n]+)/key\.pem$", asked, re.M)
ok("a copy of this file with an openssl that fails exits 1 at the certificate, after it made its temp dir",
exited.returncode == 1 and "certificate could not be made" in exited.stdout and keyout is not None,
(exited.returncode, exited.stdout[-200:], exited.stderr[-200:], asked))
ok("that temp dir is gone when the copy has exited", keyout is not None and not os.path.exists(keyout.group(1)), keyout and keyout.group(1))
if keyout:
# what a run that did not clean up would have left: this run does not add to it
shutil.rmtree(keyout.group(1), ignore_errors=True)
# =====================================================================================
# yardr
# =====================================================================================
core = tempfile.mkdtemp(prefix="protonc.", dir="/tmp")
os.mkdir(core + "/h")
def yrun(*args, stdin=None):
p = subprocess.run([YARDR, "--home", core + "/y", "--actor", "tester"] + list(args), input=stdin, capture_output=True, text=True,
env={"HOME": core + "/h", "PATH": os.environ["PATH"]}, timeout=180)
if PASSWORD in p.stdout + p.stderr or CREDENTIAL in p.stdout + p.stderr:
leaks.append("yardr " + " ".join(args))
return p
def holding(root, needle):
"""The regular files under root that hold needle: a socket is no file to read."""
found = []
for here_, _, names in os.walk(root):
for name in names:
path = os.path.join(here_, name)
if os.path.isfile(path) and not os.path.islink(path):
with open(path, "rb") as f:
if needle in f.read():
found.append(path)
return found
init = yrun("init", "--harness", "claude", "--operator", "tester")
if init.returncode != 0:
ok("the scratch yard is made", False, init.stderr[-300:])
else:
applied = yrun("pack", "apply", HERE, "--yes")
ok("the pack applies from its directory", applied.returncode == 0, applied.stderr[-300:])
copy = os.path.join(core, "y", "sources", "proton", "proton-imap")
ok("the program is carried into the yard, byte for byte, at mode 0700",
os.path.exists(copy) and open(copy, "rb").read() == open(PROGRAM, "rb").read() and (os.stat(copy).st_mode & 0o777) == 0o700)
shown = yrun("source", "show", "proton", "--json")
declaration = json.loads(shown.stdout)["declaration"] if shown.returncode == 0 else {}
calls = {c["name"]: c for c in declaration.get("calls", [])}
ok("the declaration as the yard reads it: Bridge's default address, one credential, three read calls each run by proton-imap, and search as the check",
declaration.get("base") == "http://127.0.0.1:1143" and declaration.get("credential") == {"env": "PROTON_BRIDGE"}
and declaration.get("check") == "search" and list(calls) == ["search", "message", "thread"]
and all(c["kind"] == "read" and c["run"] == ["proton-imap", n] for n, c in calls.items()), json.dumps(declaration)[:300])
params = {n: {p["name"]: p for p in c.get("params", [])} for n, c in calls.items()}
ok("search takes q, free text kept out of the record, and maxResults of at most 50, 20 unless given",
params["search"]["q"] == {"name": "q", "type": "string", "max": 1000, "optional": True}
and params["search"]["maxResults"]["pattern"] == "^([1-9]|[1-4][0-9]|50)$" and params["search"]["maxResults"]["default"] == "20", params["search"])
ok("message and thread take an id, kept in the record, and a format of metadata or full, metadata unless given",
all(params[n]["id"]["log"] is True and params[n]["id"]["pattern"] == r"^[A-Za-z0-9%._~-]+:[0-9]+:[0-9]+$"
and params[n]["format"]["values"] == ["metadata", "full"] and params[n]["format"]["default"] == "metadata" for n in ("message", "thread")),
params["message"])
ok("the pack suggests one preset, read, granting every call and nothing else",
"1. read:" in yrun("pack", "show", "proton").stdout and re.search(r"proton: read\n", yrun("pack", "show", "proton").stdout) is not None)
# the grants SETUP.md teaches, run as written (read out of it, not copied here)
with open(os.path.join(HERE, "SETUP.md"), encoding="utf-8") as f:
setup = f.read()
grants = re.findall(r"^\s*yardr (source grant [^\n`\"]+)$", setup, re.M)
ok("SETUP.md teaches a grant", len(grants) >= 1)
for grant in grants:
done = yrun(*shlex.split(grant))
ok("SETUP.md's `yardr %s` is accepted" % grant, done.returncode == 0, done.stderr[-300:])
held = yrun("source", "grants", "--json")
ok("what it granted is read, as read, to the yardmaster and nobody else",
held.returncode == 0 and [(g["holder"] if "holder" in g else None) for g in json.loads(held.stdout)] != [] and "proton" in held.stdout, held.stdout[-300:])
# the operator's copy points at the stand-in, as SETUP.md has the operator point it at Bridge
yard_stub = start(world())
mine = os.path.join(core, "mine")
os.mkdir(mine)
shutil.copy(PROGRAM, mine)
with open(os.path.join(HERE, "proton.yaml"), encoding="utf-8") as f:
text = f.read()
with open(os.path.join(mine, "proton.yaml"), "w", encoding="utf-8") as f:
f.write(re.sub(r"(?m)^base: .*$", "base: http://127.0.0.1:%d" % yard_stub.port, text))
added = yrun("source", "add", os.path.join(mine, "proton.yaml"))
keyed = yrun("source", "key", "proton", stdin="PROTON_BRIDGE=%s\n" % CREDENTIAL)
ok("the operator's copy is added and the login given by source key", added.returncode == 0 and keyed.returncode == 0, (added.stderr + keyed.stderr)[-300:])
# search is the check call, so the check makes it with no parameters: one search, and the first call Bridge sees
seen = yard_stub.connections
checked = yrun("source", "check", "proton", "--json")
made = yard_stub.connections - seen
verdict = json.loads(checked.stdout) if checked.returncode == 0 else {}
ok("source check, as SETUP.md says, makes the check call: one search against Bridge, answered as the program answers it on its own",
checked.returncode == 0 and made == 1 and verdict.get("call") == "search" and verdict.get("status") == 200
and verdict.get("answer") == answer(yard_stub, "search"),
(checked.returncode, made, checked.stdout[-300:], checked.stderr[-300:]))
envfile = os.path.join(core, "y", "sources", "proton.env")
ok("the login is in the source's env file, mode 0600", os.path.exists(envfile) and (os.stat(envfile).st_mode & 0o777) == 0o600
and CREDENTIAL in open(envfile).read(), envfile)
def through(*args):
"""(exit status, the answer, what a refusal said): yardr call --json prints the answer alone, and a refusal as {"error": ...}."""
p = yrun("call", "proton", *args, "--json")
if p.returncode == 0:
return 0, json.loads(p.stdout), ""
try:
said = json.loads(p.stdout).get("error", "")
except ValueError:
said = ""
return p.returncode, None, said + p.stderr
for call, given, direct in (("search", ["q=from:alice"], {"q": "from:alice"}),
("search", ["maxResults=2"], {"maxResults": "2"}),
("search", ["q=is:unread in:inbox", "maxResults=5"], {"q": "is:unread in:inbox", "maxResults": "5"}),
("message", ["id=" + report], {"id": report}),
("message", ["id=" + report, "format=full"], {"id": report, "format": "full"}),
("thread", ["id=INBOX:1111:3"], {"id": "INBOX:1111:3"}),
("thread", ["id=INBOX:1111:3", "format=full"], {"id": "INBOX:1111:3", "format": "full"})):
rc, got, err = through(call, *given)
ok("yardr call proton %s %s answers what the program gives on its own" % (call, " ".join(given)), rc == 0 and got == answer(yard_stub, call, direct), (rc, err.strip()[-200:]))
seen = yard_stub.connections
for given, why in ((["maxResults=51"], "maxResults over 50"), (["id=nonsense"], "an id that is none"),
(["id=INBOX%0D%0Aevil:1111:3"], "an id with an encoded control character"),
(["id=" + report, "format=html"], "a format it does not have")):
call = "search" if given[0].startswith("max") else "message"
rc, got, err = through(call, *given)
ok("a call with %s does not reach Bridge" % why, rc != 0 and yard_stub.connections == seen, (rc, err.strip()[-200:]))
rc, got, err = through("message", "id=INBOX:9:2")
ok("a refusal of the program reaches the caller as the program's line", rc != 0 and "that id is stale" in err, (rc, got, err.strip()[-300:]))
events = yrun("events")
ok("the event log of those calls holds no password, and the free text of a search is not in it", PASSWORD not in events.stdout and "from:alice" not in events.stdout
and "is:unread" not in events.stdout, events.stdout[-300:])
where = holding(core + "/y", PASSWORD.encode()) + holding(core + "/h", PASSWORD.encode())
ok("the password is in no file of the yard or its home but the source's env file", where == [envfile], where)
# the wrong password, through yardr
yrun("source", "key", "proton", stdin="PROTON_BRIDGE=%s:wrong-password\n" % USER)
rc, got, err = through("search")
ok("a wrong password reaches the caller as Bridge's refusal, with no password in it", rc != 0 and "Bridge refused the login" in err and "wrong-password" not in err, (rc, got, err.strip()[-300:]))
checked = yrun("source", "check", "proton")
said = checked.stdout + checked.stderr
ok("source check with the wrong password fails in Bridge's refusal, before any real call, with no password in it",
checked.returncode != 0 and "Bridge refused the login" in said and "wrong-password" not in said, (checked.returncode, said[-300:]))
ok("the login the stand-in saw was never in the clear", yard_stub.clear_logins == 0 and yard_stub.violations == [])
ok("no output of any yardr command held the password or the login", leaks == [], leaks)
print("test: proton: %s" % ("all ok" if not failed else "FAILED (%d)" % len(failed)))
sys.exit(1 if failed else 0)
# A stand-in for Proton Mail Bridge's IMAP, for the proton pack's tests: a
# socket loop that answers the commands proton-imap sends and no others. The
# test starts it in a thread, on 127.0.0.1 and a port the system chooses, so
# nothing here reaches a real Bridge, a network or a secret.
#
# What it takes from Gluon, Bridge's IMAP engine (ProtonMail/gluon d871584),
# because that is what the program is written against: STARTTLS and then
# LOGIN, LIST with attributes (\All, \Sent, \Trash, \Noselect), EXAMINE with
# UIDVALIDITY, UID SEARCH with CHARSET and literals where each of FROM, TO, CC,
# BCC, SUBJECT and HEADER is a case-insensitive substring of the RAW header
# (an RFC 2047 word is not decoded) and TEXT is a substring of the raw
# message, and UID FETCH of FLAGS, INTERNALDATE, RFC822.SIZE and BODY.PEEK
# sections. Like Gluon, no LITERAL+ (a client waits for the "+"), no THREAD
# and no X-GM-THRID.
#
# What it refuses is the point of the test: anything that could change a
# mailbox or leave a trace on it (SELECT, STORE, COPY, MOVE, EXPUNGE, APPEND,
# CREATE, DELETE, RENAME, CLOSE), a body fetched with BODY[] where BODY.PEEK[]
# keeps \Seen alone, and any search key or fetch item it does not know. Each
# is answered NO or BAD and written to Stub.violations, which a test asserts is
# empty, so a read-only program is one a test can hold to it.
#
# Stub.order says where a FETCH response puts its attributes: "normal" has
# UID and FLAGS before the body literal, "after" has them after it, which a
# server may do and a parser of imaplib's answer has to take.
import datetime
import re
import socketserver
import ssl
import threading
import time
CRLF = b"\r\n"
MONTHS = "Jan Feb Mar Apr May Jun Jul Aug Sep Oct Nov Dec".split()
ALLOWED = {"CAPABILITY", "STARTTLS", "LOGIN", "LIST", "EXAMINE", "UID", "LOGOUT", "NOOP"}
class Msg:
def __init__(self, uid, raw, flags=(), internal="03-Nov-2026 10:00:00 +0000"):
self.uid, self.flags, self.internal = uid, list(flags), internal
self.raw = raw.replace(b"\r\n", b"\n").replace(b"\n", CRLF)
head, _, self.body = self.raw.partition(CRLF + CRLF)
self.head = head + CRLF + CRLF
self.lines = re.sub(rb"\r\n[ \t]+", b" ", head).split(CRLF)
def header(self, name):
"""The RAW value of the first header of that name, folding undone, as Gluon's Get."""
prefix = name.lower().encode() + b":"
for line in self.lines:
if line.lower().startswith(prefix):
return line[len(prefix):].strip()
return b""
def fields(self, names):
"""The header block of just these fields, as BODY[HEADER.FIELDS (...)] answers."""
keep = {n.lower().encode() + b":" for n in names}
out, take = [], False
for line in self.head.split(CRLF)[:-2]:
if line[:1] in (b" ", b"\t"):
if take:
out.append(line)
continue
take = any(line.lower().startswith(k) for k in keep)
if take:
out.append(line)
return CRLF.join(out) + CRLF + CRLF
def date(self):
day, mon, year = self.internal.split(" ")[0].split("-")
return datetime.date(int(year), MONTHS.index(mon) + 1, int(day))
class Mailbox:
def __init__(self, name, attrs=(), validity=1, messages=()):
self.name, self.attrs, self.validity, self.messages = name, list(attrs), validity, list(messages)
class Stub:
"""The server. mailboxes are Mailbox objects; user and password are what
LOGIN must give; starttls says whether it offers STARTTLS; order is where
a FETCH response puts its attributes; delay is seconds a UID SEARCH waits
before it answers, for a test of the program's timeout; hang_up closes a
connection as soon as it is made; flood answers every UID SEARCH with that
many UIDs, of messages that are not there, to make a line too long for
imaplib's own bound; echo makes a refused LOGIN say the password it was
given, as a careless server might."""
def __init__(self, mailboxes, user, password, certfile, keyfile, starttls=True, order="normal", delay=0.0, hang_up=False, flood=0, echo=False):
self.mailboxes, self.user, self.password = mailboxes, user, password
self.starttls, self.order, self.delay, self.hang_up, self.flood, self.echo = starttls, order, delay, hang_up, flood, echo
self.context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
self.context.load_cert_chain(certfile, keyfile)
self.lock = threading.Lock()
self.connections = 0
self.commands = [] # each command's verb, "UID SEARCH" for a UID one, as received
self.logins = [] # (user, password) of each LOGIN
self.clear_logins = 0 # LOGINs before TLS: the password sent in the clear
self.searches = [] # {"charset": ..., "literals": n, "text": ...} per UID SEARCH
self.violations = []
self.port = None
def start(self):
class Server(socketserver.ThreadingTCPServer):
allow_reuse_address = True
daemon_threads = True
self.server = Server(("127.0.0.1", 0), Handler)
self.server.stub = self
self.port = self.server.server_address[1]
threading.Thread(target=self.server.serve_forever, daemon=True).start()
return self.port
def stop(self):
self.server.shutdown()
self.server.server_close()
def note(self, verb):
with self.lock:
self.commands.append(verb)
def violate(self, what):
with self.lock:
self.violations.append(what)
def mailbox(self, name):
for box in self.mailboxes:
if box.name == name or (name.upper() == "INBOX" and box.name.upper() == "INBOX"):
return box
return None
class Handler(socketserver.BaseRequestHandler):
def setup(self):
self.stub = self.server.stub
self.sock = self.request
self.rfile = self.sock.makefile("rb")
self.tls = False
self.authed = False
self.box = None
def send(self, data):
self.sock.sendall(data if isinstance(data, bytes) else data.encode())
def line(self, text):
self.send(text.encode() + CRLF if isinstance(text, str) else text + CRLF)
def caps(self):
return "IMAP4rev1 ID UIDPLUS UNSELECT MOVE IDLE AUTH=PLAIN" + (" STARTTLS" if self.stub.starttls and not self.tls else "")
def handle(self):
stub = self.stub
with stub.lock:
stub.connections += 1
if stub.hang_up:
return
self.line("* OK [CAPABILITY %s] Bridge stand-in ready" % self.caps())
while True:
try:
got = self.read_command()
except (OSError, ssl.SSLError):
return
if got is None:
return
tag, verb, text, literals = got
try:
if self.dispatch(tag, verb, text, literals) == "bye":
return
except (OSError, ssl.SSLError):
return
def read_command(self):
"""(tag, verb, the rest of the line with each literal as \\0<n>\\0, the literals)."""
text, literals = b"", []
while True:
line = self.rfile.readline()
if not line:
return None
m = re.search(rb"\{(\d+)(\+?)\}\r\n$", line)
if not m:
text += line.rstrip(b"\r\n")
break
text += line[: m.start()] + b"\x00%d\x00" % len(literals)
if not m.group(2):
self.line("+ Ready for literal data")
literals.append(self.rfile.read(int(m.group(1))))
parts = text.split(b" ", 2)
tag = parts[0].decode("ascii", "replace")
verb = parts[1].decode("ascii", "replace").upper() if len(parts) > 1 else ""
return tag, verb, parts[2] if len(parts) > 2 else b"", literals
@staticmethod
def tokens(text, literals):
"""Atoms as str, quoted strings and literals as bytes, ( and ) as str."""
out, i = [], 0
while i < len(text):
c = text[i : i + 1]
if c == b" ":
i += 1
elif c in (b"(", b")"):
out.append(c.decode())
i += 1
elif c == b'"':
j, buf = i + 1, b""
while text[j : j + 1] != b'"':
if text[j : j + 1] == b"\\":
j += 1
buf += text[j : j + 1]
j += 1
out.append(buf)
i = j + 1
elif c == b"\x00":
j = text.index(b"\x00", i + 1)
out.append(literals[int(text[i + 1 : j])])
i = j + 1
else:
j = i
while j < len(text) and text[j : j + 1] not in (b" ", b"(", b")"):
j += 1
out.append(text[i:j].decode("ascii", "replace"))
i = j
return out
def dispatch(self, tag, verb, text, literals):
stub = self.stub
toks = self.tokens(text, literals)
if verb == "UID" and toks:
stub.note("UID " + str(toks[0]).upper())
else:
stub.note(verb)
if verb not in ALLOWED:
stub.violate("command %s %s" % (verb, text[:60].decode("ascii", "replace")))
self.line("%s NO [CANNOT] this is a read-only stand-in" % tag)
return None
if verb == "CAPABILITY":
self.line("* CAPABILITY %s" % self.caps())
self.line("%s OK CAPABILITY" % tag)
elif verb == "NOOP":
self.line("%s OK NOOP" % tag)
elif verb == "LOGOUT":
self.line("* BYE logging out")
self.line("%s OK LOGOUT" % tag)
return "bye"
elif verb == "STARTTLS":
if not stub.starttls:
self.line("%s NO TLS unavailable" % tag)
return None
self.line("%s OK Begin TLS negotiation now" % tag)
self.sock = stub.context.wrap_socket(self.request, server_side=True)
self.rfile = self.sock.makefile("rb")
self.tls = True
elif verb == "LOGIN":
user, password = (t.decode() if isinstance(t, bytes) else t for t in toks[:2])
with stub.lock:
stub.logins.append((user, password))
if not self.tls:
stub.clear_logins += 1
if not self.tls:
self.line("%s NO [PRIVACYREQUIRED] TLS first" % tag)
elif (user, password) == (stub.user, stub.password):
self.authed = True
self.line("%s OK [CAPABILITY %s] Logged in" % (tag, self.caps()))
else:
self.line("%s NO [AUTHENTICATIONFAILED] Invalid credentials%s" % (tag, " for %s:%s" % (user, password) if stub.echo else ""))
elif not self.authed:
self.line("%s BAD not authenticated" % tag)
elif verb == "LIST":
for box in stub.mailboxes:
self.line('* LIST (%s) "/" "%s"' % (" ".join(box.attrs), box.name))
self.line("%s OK LIST" % tag)
elif verb == "EXAMINE":
box = stub.mailbox(toks[0].decode() if toks and isinstance(toks[0], bytes) else "")
if box is None or "\\Noselect" in box.attrs:
self.line("%s NO [NONEXISTENT] no such mailbox" % tag)
return None
self.box = box
self.line("* FLAGS (\\Seen \\Answered \\Flagged \\Deleted \\Draft)")
self.line("* %d EXISTS" % len(box.messages))
self.line("* 0 RECENT")
self.line("* OK [UIDVALIDITY %d] UIDs valid" % box.validity)
self.line("* OK [UIDNEXT %d] Predicted next UID" % (max([m.uid for m in box.messages] or [0]) + 1))
self.line("%s OK [READ-ONLY] EXAMINE completed" % tag)
elif verb == "UID":
sub = str(toks[0]).upper() if toks else ""
if sub == "SEARCH":
self.search(tag, toks[1:], literals)
elif sub == "FETCH":
self.fetch(tag, text, literals)
else:
stub.violate("UID %s" % sub)
self.line("%s NO [CANNOT] this is a read-only stand-in" % tag)
return None
# ---- UID SEARCH ---------------------------------------------------------
def search(self, tag, toks, literals):
stub = self.stub
charset = None
if toks and isinstance(toks[0], str) and toks[0].upper() == "CHARSET":
charset = toks[1] if isinstance(toks[1], str) else toks[1].decode()
toks = toks[2:]
if charset.upper() not in ("UTF-8", "US-ASCII"):
self.line("%s NO [BADCHARSET (UTF-8 US-ASCII)] unknown charset" % tag)
return
with stub.lock:
stub.searches.append({"charset": charset, "literals": len(literals),
"text": " ".join(t.decode("utf-8", "replace") if isinstance(t, bytes) else t for t in toks)})
if stub.delay:
time.sleep(stub.delay)
it = iter(toks)
try:
tests = []
while True:
try:
tests.append(self.key(it))
except StopIteration:
break
except ValueError as e:
stub.violate("search %s" % e)
self.line("%s BAD %s" % (tag, e))
return
found = [m.uid for m in self.box.messages if all(t(m) for t in tests)]
if stub.flood:
found = list(range(1, stub.flood + 1))
self.line("* SEARCH" + "".join(" %d" % u for u in found))
self.line("%s OK UID SEARCH completed" % tag)
@staticmethod
def string(it):
value = next(it)
if not isinstance(value, bytes):
raise ValueError("search wants a string, got %r" % (value,))
return value
def key(self, it):
word = next(it)
if not isinstance(word, str):
raise ValueError("search wants a key, got a string")
up = word.upper()
if up == "ALL":
return lambda m: True
if up in ("FROM", "TO", "CC", "BCC", "SUBJECT"):
want = self.string(it).lower()
return lambda m: want in m.header(up).lower()
if up == "HEADER":
field = next(it)
field = field if isinstance(field, str) else field.decode()
want = self.string(it).lower()
return lambda m: want in m.header(field).lower()
if up == "TEXT":
want = self.string(it).lower()
return lambda m: want in m.raw.lower()
if up in ("SINCE", "BEFORE"):
day, mon, year = next(it).split("-")
when = datetime.date(int(year), MONTHS.index(mon) + 1, int(day))
return (lambda m: m.date() >= when) if up == "SINCE" else (lambda m: m.date() < when)
flags = {"SEEN": ("\\Seen", True), "UNSEEN": ("\\Seen", False), "FLAGGED": ("\\Flagged", True),
"UNFLAGGED": ("\\Flagged", False), "ANSWERED": ("\\Answered", True),
"UNANSWERED": ("\\Answered", False), "DRAFT": ("\\Draft", True), "UNDRAFT": ("\\Draft", False),
"DELETED": ("\\Deleted", True), "UNDELETED": ("\\Deleted", False)}
if up in flags:
flag, has = flags[up]
return lambda m: (flag in m.flags) == has
if up == "NOT":
inner = self.key(it)
return lambda m: not inner(m)
if up == "OR":
a, b = self.key(it), self.key(it)
return lambda m: a(m) or b(m)
raise ValueError("unknown search key %s" % word)
# ---- UID FETCH ----------------------------------------------------------
def fetch(self, tag, text, literals):
stub = self.stub
m = re.match(rb"FETCH ([0-9,:*]+) \((.*)\)\s*$", text, re.S)
if not m:
stub.violate("fetch %r" % text[:80])
self.line("%s BAD fetch syntax" % tag)
return
wanted = self.uid_set(m.group(1).decode())
items = m.group(2).decode("ascii", "replace")
sections = re.findall(r"BODY(\.PEEK)?\[([^\]]*)\]", items)
for peek, section in sections:
if not peek:
stub.violate("BODY[%s] without PEEK" % section)
plain = re.sub(r"BODY(\.PEEK)?\[[^\]]*\]", "", items).split()
for item in plain:
if item not in ("UID", "FLAGS", "INTERNALDATE", "RFC822.SIZE"):
stub.violate("fetch item %s" % item)
for seq, msg in enumerate(self.box.messages, 1):
if msg.uid not in wanted:
continue
attrs = ["UID %d" % msg.uid]
if "FLAGS" in plain:
attrs.append("FLAGS (%s)" % " ".join(msg.flags))
if "INTERNALDATE" in plain:
attrs.append('INTERNALDATE "%s"' % msg.internal)
if "RFC822.SIZE" in plain:
attrs.append("RFC822.SIZE %d" % len(msg.raw))
bodies = []
for _, section in sections:
name = section.upper()
if name == "":
data = msg.raw
elif name == "HEADER":
data = msg.head
elif name.startswith("HEADER.FIELDS ("):
data = msg.fields(name[len("HEADER.FIELDS ("):-1].split())
else:
stub.violate("body section %s" % section)
data = b""
bodies.append(("BODY[%s]" % section, data))
self.respond(seq, attrs, bodies)
self.line("%s OK UID FETCH completed" % tag)
def respond(self, seq, attrs, bodies):
"""One FETCH response: the attributes before the literal, or after it."""
stub = self.stub
if not bodies:
self.line("* %d FETCH (%s)" % (seq, " ".join(attrs)))
return
out = b"* %d FETCH (" % seq
if stub.order == "normal":
out += " ".join(attrs).encode() + b" "
for i, (name, data) in enumerate(bodies):
out += (b" " if i else b"") + name.encode() + b" {%d}\r\n" % len(data) + data
if stub.order == "after":
out += b" " + " ".join(attrs).encode()
self.send(out + b")" + CRLF)
@staticmethod
def uid_set(text):
out = set()
for part in text.split(","):
lo, _, hi = part.partition(":")
if hi:
out.update(range(int(lo), (10**9 if hi == "*" else int(hi)) + 1))
else:
out.add(int(lo))
return out
{
"id": "INBOX:1111:4",
"mailbox": "INBOX",
"flags": [
"\\Flagged"
],
"from": "News Café <news@example.com>",
"to": "bob@example.com",
"subject": "Café news",
"date": "2026-11-04T07:15:00+00:00",
"messageId": "<news1@example.com>",
"size": 489,
"text": "Hello & welcome\n\nSecond\nline é",
"attachments": []
}
{
"id": "INBOX:1111:2",
"mailbox": "INBOX",
"flags": [
"\\Seen",
"\\Answered"
],
"from": "Alice Example <alice@example.com>",
"to": "Bob <bob@example.com>",
"cc": "carol@example.com",
"subject": "Quarterly report",
"date": "2026-11-03T09:30:00+01:00",
"messageId": "<q1@example.com>",
"size": 651,
"text": "Hi Bob,\n\nThe numbers are attached. Revenue is up 12%.\n\nAlice",
"attachments": [
{
"name": "report.pdf",
"type": "application/pdf",
"size": 55
}
]
}
{
"id": "INBOX:1111:2",
"mailbox": "INBOX",
"flags": [
"\\Seen",
"\\Answered"
],
"from": "Alice Example <alice@example.com>",
"to": "Bob <bob@example.com>",
"cc": "carol@example.com",
"subject": "Quarterly report",
"date": "2026-11-03T09:30:00+01:00",
"messageId": "<q1@example.com>",
"size": 651
}
{
"mailbox": "All Mail",
"messages": [
{
"id": "All%20Mail:2222:108",
"from": "Boss <boss@example.com>",
"subject": "Planning",
"date": "2026-11-06T14:00:00+00:00",
"flags": []
},
{
"id": "All%20Mail:2222:107",
"from": "Alice Example <alice@example.com>",
"subject": "Re: Re: Quarterly report",
"date": "2026-11-04T10:00:00+00:00",
"flags": [
"\\Seen"
]
},
{
"id": "All%20Mail:2222:106",
"from": "José Müller <jose@example.net>",
"subject": "Gespräch mit Müller",
"date": "2026-11-06T08:30:00+00:00",
"flags": [
"\\Seen"
]
},
{
"id": "All%20Mail:2222:105",
"from": "Dave <dave@example.org>",
"subject": "Quarterly report",
"date": "2026-11-05T12:00:00+00:00",
"flags": []
},
{
"id": "All%20Mail:2222:104",
"from": "News Café <news@example.com>",
"subject": "Café news",
"date": "2026-11-04T07:15:00+00:00",
"flags": [
"\\Flagged"
]
},
{
"id": "All%20Mail:2222:103",
"from": "Bob <bob@example.com>",
"subject": "Re: Quarterly report",
"date": "2026-11-03T11:00:00+00:00",
"flags": []
},
{
"id": "All%20Mail:2222:102",
"from": "Alice Example <alice@example.com>",
"subject": "Quarterly report",
"date": "2026-11-03T09:30:00+01:00",
"flags": [
"\\Seen",
"\\Answered"
]
},
{
"id": "All%20Mail:2222:101",
"from": "Proton Team <team@proton.example>",
"subject": "Welcome to Proton",
"date": "2026-11-02T08:00:00+00:00",
"flags": [
"\\Seen"
]
}
],
"resultSizeEstimate": 8
}
{
"mailbox": "INBOX",
"messages": [
{
"id": "INBOX:1111:5",
"from": "Dave <dave@example.org>",
"subject": "Quarterly report",
"date": "2026-11-05T12:00:00+00:00",
"flags": []
},
{
"id": "INBOX:1111:4",
"from": "News Café <news@example.com>",
"subject": "Café news",
"date": "2026-11-04T07:15:00+00:00",
"flags": [
"\\Flagged"
]
},
{
"id": "INBOX:1111:3",
"from": "Bob <bob@example.com>",
"subject": "Re: Quarterly report",
"date": "2026-11-03T11:00:00+00:00",
"flags": []
}
],
"resultSizeEstimate": 3
}
{
"id": "INBOX:1111:3",
"threadId": "<q1@example.com>",
"messages": [
{
"id": "All%20Mail:2222:102",
"flags": [
"\\Seen",
"\\Answered"
],
"from": "Alice Example <alice@example.com>",
"to": "Bob <bob@example.com>",
"cc": "carol@example.com",
"subject": "Quarterly report",
"date": "2026-11-03T09:30:00+01:00",
"messageId": "<q1@example.com>",
"size": 651,
"text": "Hi Bob,\n\nThe numbers are attached. Revenue is up 12%.\n\nAlice",
"attachments": [
{
"name": "report.pdf",
"type": "application/pdf",
"size": 55
}
]
},
{
"id": "All%20Mail:2222:103",
"flags": [],
"from": "Bob <bob@example.com>",
"to": "Alice Example <alice@example.com>",
"subject": "Re: Quarterly report",
"date": "2026-11-03T11:00:00+00:00",
"messageId": "<r1@example.com>",
"inReplyTo": "<q1@example.com>",
"size": 434,
"text": "Thanks Alice, I will read it today.\n\nOn Tue, 3 Nov 2026, Alice Example wrote:\n> Hi Bob,\n>\n> The numbers are attached. Revenue is up 12%.",
"attachments": []
},
{
"id": "All%20Mail:2222:107",
"flags": [
"\\Seen"
],
"from": "Alice Example <alice@example.com>",
"to": "Bob <bob@example.com>",
"subject": "Re: Re: Quarterly report",
"date": "2026-11-04T10:00:00+00:00",
"messageId": "<s1@example.com>",
"inReplyTo": "<r1@example.com>",
"size": 348,
"text": "Great, let me know what you think.",
"attachments": []
}
]
}
{
"id": "INBOX:1111:3",
"threadId": "<q1@example.com>",
"messages": [
{
"id": "All%20Mail:2222:102",
"flags": [
"\\Seen",
"\\Answered"
],
"from": "Alice Example <alice@example.com>",
"to": "Bob <bob@example.com>",
"cc": "carol@example.com",
"subject": "Quarterly report",
"date": "2026-11-03T09:30:00+01:00",
"messageId": "<q1@example.com>",
"size": 651
},
{
"id": "All%20Mail:2222:103",
"flags": [],
"from": "Bob <bob@example.com>",
"to": "Alice Example <alice@example.com>",
"subject": "Re: Quarterly report",
"date": "2026-11-03T11:00:00+00:00",
"messageId": "<r1@example.com>",
"inReplyTo": "<q1@example.com>",
"size": 434
},
{
"id": "All%20Mail:2222:107",
"flags": [
"\\Seen"
],
"from": "Alice Example <alice@example.com>",
"to": "Bob <bob@example.com>",
"subject": "Re: Re: Quarterly report",
"date": "2026-11-04T10:00:00+00:00",
"messageId": "<s1@example.com>",
"inReplyTo": "<r1@example.com>",
"size": 348
}
]
}
From: José Müller <jose@example.net> To: bob@example.com Subject: Gespräch mit Müller Date: Fri, 06 Nov 2026 08:30:00 +0000 Message-ID: <mueller1@example.net> MIME-Version: 1.0 Content-Type: text/plain; charset=utf-8 Content-Transfer-Encoding: 8bit Guten Tag, wir sprechen über Café und Müller.
From: =?UTF-8?B?TmV3cyBDYWbDqQ==?= <news@example.com>
To: bob@example.com
Subject: =?UTF-8?B?Q2Fmw6kgbmV3cw==?=
Date: Wed, 04 Nov 2026 07:15:00 +0000
Message-ID: <news1@example.com>
MIME-Version: 1.0
Content-Type: text/html; charset=utf-8
Content-Transfer-Encoding: 7bit
<html><head><title>a title nobody reads</title><style>p { color: red }</style></head>
<body><p>Hello & welcome</p><div>Second<br>line é</div>
<script>alert("never shown")</script></body></html>
From: Dave <dave@example.org> To: bob@example.com Subject: Quarterly report Date: Thu, 05 Nov 2026 12:00:00 +0000 Message-ID: <o1@example.org> MIME-Version: 1.0 Content-Type: text/plain; charset=utf-8 A different report, from Dave: it shares a subject with the thread and nothing else.
From: Bob <bob@example.com> To: Alice Example <alice@example.com> Subject: Re: Quarterly report Date: Tue, 03 Nov 2026 11:00:00 +0000 Message-ID: <r1@example.com> In-Reply-To: <q1@example.com> References: <q1@example.com> MIME-Version: 1.0 Content-Type: text/plain; charset=utf-8 Thanks Alice, I will read it today. On Tue, 3 Nov 2026, Alice Example wrote: > Hi Bob, > > The numbers are attached. Revenue is up 12%.
From: Alice Example <alice@example.com> To: Bob <bob@example.com> Cc: carol@example.com Subject: Quarterly report Date: Tue, 03 Nov 2026 09:30:00 +0100 Message-ID: <q1@example.com> MIME-Version: 1.0 Content-Type: multipart/mixed; boundary="b1" --b1 Content-Type: text/plain; charset=utf-8 Content-Transfer-Encoding: quoted-printable Hi Bob, The numbers are attached. Revenue is up 12=25. Alice --b1 Content-Type: application/pdf; name="report.pdf" Content-Disposition: attachment; filename="report.pdf" Content-Transfer-Encoding: base64 JVBERi0xLjQKJSBhIHN0dWIgcGRmIGZvciB0aGUgcHJvdG9uIHBhY2sncyB0ZXN0CiUlRU9GCg== --b1--
From: Alice Example <alice@example.com> To: Bob <bob@example.com> Subject: Re: Re: Quarterly report Date: Wed, 04 Nov 2026 10:00:00 +0000 Message-ID: <s1@example.com> In-Reply-To: <r1@example.com> References: <q1@example.com> <r1@example.com> MIME-Version: 1.0 Content-Type: text/plain; charset=utf-8 Great, let me know what you think.
From: Eve <eve@example.net> To: Bob <bob@example.com> Subject: Re: Quarterly report Date: Thu, 05 Nov 2026 09:00:00 +0000 Message-ID: <t1@example.net> In-Reply-To: <r1@example.com> References: <q1@example.com> <r1@example.com> MIME-Version: 1.0 Content-Type: text/plain; charset=utf-8 I would have deleted this: it is in the thread, and in the trash.
From: Proton Team <team@proton.example> To: Bob <bob@example.com> Subject: Welcome to Proton Date: Mon, 02 Nov 2026 08:00:00 +0000 Message-ID: <welcome@proton.example> MIME-Version: 1.0 Content-Type: text/plain; charset=utf-8 Content-Transfer-Encoding: 7bit Welcome to Proton Mail. Your mailbox is ready.
From: Boss <boss@example.com> To: Bob <bob@example.com> Subject: Planning Date: Fri, 06 Nov 2026 14:00:00 +0000 Message-ID: <w1@example.com> MIME-Version: 1.0 Content-Type: text/plain; charset=utf-8 Planning for next quarter: this one is filed in Folders/Work.