feat: add pneumatube gateway bridging smolmail and SMTP via LMTP

src/pneumatube/__main__.py

@@ -0,0 +1,102 @@
+"""pneumatube: a gateway bridging one smolmail account and an SMTP mailbox.
+
+smolmail -> email: a message the gateway receives with an Email-To
+frontmatter field is relayed to that address over SMTP.
+
+email -> smolmail: mail sent to <username>@PNEUMATUBE_EMAIL_DOMAIN is
+relayed as a smolmail message to <username>@PNEUMATUBE_SMOLMAIL_DOMAIN, if
+that username is registered there, with the original sender recorded in an
+Email-From frontmatter field.
+
+See config.py for the full list of environment variables, including
+PNEUMATUBE_LISTEN_PROTOCOL (lmtp, behind a real MTA, or smtp for local-dev
+only), PNEUMATUBE_LISTEN_HOST and PNEUMATUBE_LISTEN_PORT. Before running,
+register the gateway's own smolmail account the same way any smolmail client
+would, against the same --key/--db paths this reads (PNEUMATUBE_SMOLMAIL_KEY
+/ PNEUMATUBE_SMOLMAIL_DB):
+
+    uv run <smolmail-repo>/smolmail.py --key gateway.key --db gateway.db keygen
+    uv run <smolmail-repo>/smolmail.py --key gateway.key --db gateway.db trust <host> <server-key>
+    uv run <smolmail-repo>/smolmail.py --key gateway.key --db gateway.db register <user>@<host>
+"""
+from __future__ import annotations
+
+import argparse
+import asyncio
+import logging
+
+from aiosmtpd.controller import Controller
+from aiosmtpd.lmtp import LMTP
+from aiosmtpd.smtp import SMTP as PlainSMTP
+
+from . import config as config_module
+from . import vendor
+from .email_to_smol import EmailToSmolBridge
+from .smol_to_email import SmolToEmailBridge
+from .smtp_server import GatewayHandler
+from .worker import SmolWorker
+
+logger = logging.getLogger(__name__)
+
+
+class _GatewayController(Controller):
+    """A Controller that can speak either LMTP or plain SMTP, per config.
+
+    Mirrors aiosmtpd's own pattern for an LMTP controller (aiosmtpd/tests/
+    test_lmtp.py): override factory() to instantiate the chosen protocol
+    class instead of the default SMTP.
+    """
+
+    def __init__(self, handler, protocol_class, **kwargs) -> None:
+        self._protocol_class = protocol_class
+        super().__init__(handler, **kwargs)
+
+    def factory(self):
+        return self._protocol_class(self.handler, **self.SMTP_kwargs)
+
+
+async def _poll_forever(worker: SmolWorker, bridge: SmolToEmailBridge, interval: int) -> None:
+    while True:
+        try:
+            await worker.run(bridge.poll_once)
+        except Exception:
+            logger.exception("smolmail poll failed")
+        await asyncio.sleep(interval)
+
+
+def _open_account(smol, cfg: config_module.Config):
+    # Runs on the worker thread: Store() binds a sqlite3 connection to
+    # whichever thread creates it (worker.py).
+    store = smol.Store(str(cfg.smolmail_db))
+    account = smol.load_account(argparse.Namespace(key=str(cfg.smolmail_key)), store)
+    return store, account
+
+
+async def _run(cfg: config_module.Config) -> None:
+    smol = vendor.load(cfg.smolmail_repo)
+    worker = SmolWorker()
+    store, account = await worker.run(_open_account, smol, cfg)
+
+    outbound = SmolToEmailBridge(smol, store, account, cfg)
+    inbound = EmailToSmolBridge(smol, store, cfg)
+
+    protocol_class = LMTP if cfg.listen_protocol == "lmtp" else PlainSMTP
+    controller = _GatewayController(GatewayHandler(inbound, worker, cfg), protocol_class,
+                                     hostname=cfg.listen_host, port=cfg.listen_port)
+    controller.start()
+    logger.info("%s listener on %s:%s", cfg.listen_protocol.upper(), cfg.listen_host, cfg.listen_port)
+    try:
+        await _poll_forever(worker, outbound, cfg.fetch_interval)
+    finally:
+        controller.stop()
+        worker.shutdown()
+
+
+def main() -> None:
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
+    cfg = config_module.load()
+    asyncio.run(_run(cfg))
+
+
+if __name__ == "__main__":
+    main()

src/pneumatube/config.py

@@ -0,0 +1,74 @@
+"""Runtime configuration, read from environment variables."""
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from pathlib import Path
+
+# Frontmatter is case-insensitive and flat (smolmail SPEC.md S5.5); these two
+# custom keys are the only contract between the two directions of the bridge.
+FRONTMATTER_EMAIL_KEY = "Email-To"      # smolmail -> email: where to deliver
+FRONTMATTER_SENDER_KEY = "Email-From"   # email -> smolmail: who originally sent it
+
+
+def _env(name: str) -> str:
+    try:
+        return os.environ[name]
+    except KeyError:
+        raise RuntimeError(f"missing required environment variable {name}") from None
+
+
+def _int(name: str, default: int) -> int:
+    return int(os.environ.get(name, default))
+
+
+_LISTEN_PROTOCOLS = ("lmtp", "smtp")
+_LMTP_DEFAULT_PORT, _SMTP_DEFAULT_PORT = 2424, 1025
+
+
+@dataclass(frozen=True)
+class Config:
+    smolmail_repo: Path    # checkout of the sibling smolmail project
+    smolmail_key: Path     # gateway's own master secret file
+    smolmail_db: Path      # gateway's own smolmail client state
+    smolmail_domain: str   # smolmail server address used in "user@..." addresses
+    email_domain: str      # domain this gateway accepts inbound mail for
+    fetch_interval: int    # seconds between smolmail inbox polls
+
+    listen_protocol: str    # "lmtp" (default; behind a real MTA) or "smtp" (local dev only)
+    listen_host: str
+    listen_port: int
+
+    smtp_relay_host: str
+    smtp_relay_port: int
+    smtp_relay_user: str | None
+    smtp_relay_password: str | None
+    smtp_relay_starttls: bool
+
+    gateway_from_address: str   # From: header used on mail forwarded to SMTP
+
+
+def load() -> Config:
+    listen_protocol = os.environ.get("PNEUMATUBE_LISTEN_PROTOCOL", "lmtp")
+    if listen_protocol not in _LISTEN_PROTOCOLS:
+        raise RuntimeError(
+            f"PNEUMATUBE_LISTEN_PROTOCOL must be one of {_LISTEN_PROTOCOLS}, got {listen_protocol!r}")
+    default_port = _LMTP_DEFAULT_PORT if listen_protocol == "lmtp" else _SMTP_DEFAULT_PORT
+
+    return Config(
+        smolmail_repo=Path(os.environ.get("PNEUMATUBE_SMOLMAIL_REPO", "../smolmail")),
+        smolmail_key=Path(os.environ.get("PNEUMATUBE_SMOLMAIL_KEY", "gateway.key")),
+        smolmail_db=Path(os.environ.get("PNEUMATUBE_SMOLMAIL_DB", "gateway.db")),
+        smolmail_domain=_env("PNEUMATUBE_SMOLMAIL_DOMAIN"),
+        email_domain=_env("PNEUMATUBE_EMAIL_DOMAIN"),
+        fetch_interval=_int("PNEUMATUBE_FETCH_INTERVAL", 30),
+        listen_protocol=listen_protocol,
+        listen_host=os.environ.get("PNEUMATUBE_LISTEN_HOST", "127.0.0.1"),
+        listen_port=_int("PNEUMATUBE_LISTEN_PORT", default_port),
+        smtp_relay_host=_env("PNEUMATUBE_SMTP_RELAY_HOST"),
+        smtp_relay_port=_int("PNEUMATUBE_SMTP_RELAY_PORT", 587),
+        smtp_relay_user=os.environ.get("PNEUMATUBE_SMTP_RELAY_USER"),
+        smtp_relay_password=os.environ.get("PNEUMATUBE_SMTP_RELAY_PASSWORD"),
+        smtp_relay_starttls=os.environ.get("PNEUMATUBE_SMTP_RELAY_STARTTLS", "1") != "0",
+        gateway_from_address=_env("PNEUMATUBE_GATEWAY_FROM_ADDRESS"),
+    )

src/pneumatube/email_to_smol.py

@@ -0,0 +1,66 @@
+"""email -> smolmail direction.
+
+Confirms a username is registered on the configured smolmail server and, if
+so, relays an inbound email to it as a smolmail message, carrying the
+original sender's email address in the Email-From frontmatter field
+(SPEC.md S5.5).
+"""
+from __future__ import annotations
+
+import argparse
+import logging
+from email.message import Message
+from types import ModuleType
+
+from .config import Config, FRONTMATTER_SENDER_KEY
+
+logger = logging.getLogger(__name__)
+
+
+def _single_line(text: str) -> str:
+    return text.replace("\r", " ").replace("\n", " ").strip()
+
+
+class EmailToSmolBridge:
+    def __init__(self, smol: ModuleType, store: object, config: Config) -> None:
+        self.smol = smol
+        self.store = store
+        self.config = config
+
+    def user_exists(self, username: str) -> bool | None:
+        """True/False for a definite answer; None when resolving itself
+        failed (pinning/network), which is not the same as "no such user"
+        and should not be treated as a permanent bounce."""
+        try:
+            addr = self.smol.Address.parse(f"{username}@{self.config.smolmail_domain}")
+            with self.smol.connect(self.store, addr, require_pin=True) as conn:
+                self.smol.do_resolve(conn, username)
+        except self.smol.SmolError as exc:
+            if "unknown user" in str(exc):
+                return False
+            logger.warning("resolve for %s failed: %s", username, exc)
+            return None
+        return True
+
+    def forward(self, username: str, sender: str, subject: str, body: str) -> None:
+        # Subject and sender end up as frontmatter values (SPEC.md S5.5: one
+        # `Key: value` per line); a bare CR/LF smuggled through email headers
+        # would otherwise let a sender inject extra frontmatter fields.
+        send_args = argparse.Namespace(
+            key=str(self.config.smolmail_key), db=str(self.config.smolmail_db),
+            address=f"{username}@{self.config.smolmail_domain}",
+            body=body, subject=_single_line(subject), reply_to=None,
+            header=[f"{FRONTMATTER_SENDER_KEY}: {_single_line(sender)}"],
+            anonymous=False, no_pad=False)
+        self.smol.cmd_send(send_args, self.store)
+
+    @staticmethod
+    def extract_text(message: Message) -> str:
+        if not message.is_multipart():
+            payload = message.get_payload(decode=True) or b""
+            return payload.decode(message.get_content_charset() or "utf-8", "replace")
+        for part in message.walk():
+            if part.get_content_type() == "text/plain" and not part.is_multipart():
+                payload = part.get_payload(decode=True) or b""
+                return payload.decode(part.get_content_charset() or "utf-8", "replace")
+        return "(no plain-text part)"

src/pneumatube/smol_to_email.py

@@ -0,0 +1,60 @@
+"""smolmail -> email direction.
+
+Polls the gateway's own smolmail inbox and relays any message that carries
+the Email-To frontmatter field (SPEC.md S5.5) to a real mailbox over SMTP.
+A message without that field is left in the local inbox rather than dropped,
+so it stays visible to `smolmail.py list`/`read` against the gateway's own
+db for manual follow-up.
+"""
+from __future__ import annotations
+
+import argparse
+import logging
+from types import ModuleType
+
+from . import smtp_client
+from .config import Config, FRONTMATTER_EMAIL_KEY
+
+logger = logging.getLogger(__name__)
+
+
+class SmolToEmailBridge:
+    def __init__(self, smol: ModuleType, store: object, account: object, config: Config) -> None:
+        self.smol = smol
+        self.store = store
+        self.account = account
+        self.config = config
+        self.fetch_args = argparse.Namespace(
+            key=str(config.smolmail_key), db=str(config.smolmail_db), reset=False, keep=False)
+
+    def poll_once(self) -> None:
+        self.smol.cmd_fetch(self.fetch_args, self.store)
+        # "all" rather than "inbox": a gateway must relay unsolicited senders
+        # too, so requests-tier mail (SPEC.md S5.8) is not silently skipped.
+        for mid, envelope, _received_at, _recipient in self.store.mail("all"):
+            self._relay(mid, envelope)
+
+    def _relay(self, mid: bytes, envelope: bytes) -> None:
+        try:
+            opened = self.smol.describe(self.store, envelope, self.account.keys)
+        except self.smol.SmolError as exc:
+            logger.warning("dropping unreadable %s: %s", mid.hex()[:16], exc)
+            self.store.run("DELETE FROM inbox WHERE id = ?", mid)
+            return
+
+        to_addr = opened["fields"].get(FRONTMATTER_EMAIL_KEY.lower())
+        if not to_addr:
+            logger.info("%s from %s has no %s field; left for manual review",
+                        mid.hex()[:16], opened["from"], FRONTMATTER_EMAIL_KEY)
+            return
+
+        try:
+            smtp_client.send(self.config, to_addr=to_addr,
+                              subject=opened["subject"] or "(no subject)",
+                              body=opened["text"], info_header=opened["from"])
+        except (smtp_client.InvalidAddress, OSError) as exc:
+            logger.warning("could not relay %s to %r: %s", mid.hex()[:16], to_addr, exc)
+            return
+
+        self.store.run("DELETE FROM inbox WHERE id = ?", mid)
+        logger.info("relayed %s to %s", mid.hex()[:16], to_addr)

src/pneumatube/smtp_client.py

@@ -0,0 +1,40 @@
+"""Outbound SMTP: relay a decrypted smolmail message to a real mailbox."""
+from __future__ import annotations
+
+import re
+import smtplib
+from email.message import EmailMessage
+
+from .config import Config
+
+# Deliberately conservative: this only needs to reject header-injection
+# attempts and obvious garbage, not validate every RFC 5321 edge case.
+_EMAIL_RE = re.compile(r"^[^\s@]+@[^\s@]+\.[^\s@]+$")
+
+
+class InvalidAddress(ValueError):
+    pass
+
+
+def validate(address: str) -> str:
+    address = address.strip()
+    if "\n" in address or "\r" in address or not _EMAIL_RE.match(address):
+        raise InvalidAddress(address)
+    return address
+
+
+def send(config: Config, *, to_addr: str, subject: str, body: str, info_header: str) -> None:
+    to_addr = validate(to_addr)
+    message = EmailMessage()
+    message["From"] = config.gateway_from_address
+    message["To"] = to_addr
+    message["Subject"] = subject
+    message["X-Smolmail-From"] = info_header
+    message.set_content(body)
+
+    with smtplib.SMTP(config.smtp_relay_host, config.smtp_relay_port, timeout=30) as client:
+        if config.smtp_relay_starttls:
+            client.starttls()
+        if config.smtp_relay_user:
+            client.login(config.smtp_relay_user, config.smtp_relay_password or "")
+        client.send_message(message)

src/pneumatube/smtp_server.py

@@ -0,0 +1,63 @@
+"""Inbound mail, accepting delivery for this gateway's email domain.
+
+By default (PNEUMATUBE_LISTEN_PROTOCOL=lmtp) this speaks LMTP on a loopback
+port, meant to sit behind a real MTA (Postfix) that owns port 25, TLS and
+abuse limits, and hands accepted mail to this listener as the final delivery
+agent. The plain-SMTP mode is a local-dev fallback only: bound directly to
+the internet it has no TLS, rate limiting or rDNS checks.
+
+handle_RCPT does a live smolmail RESOLVE per recipient, so an unknown
+username is rejected inline during the conversation rather than bounced
+later. handle_DATA replies once per accepted recipient (joined by CRLF)
+instead of a single pass/fail for the whole message, which is what makes
+LMTP's per-recipient status semantics real rather than cosmetic.
+"""
+from __future__ import annotations
+
+import email as email_module
+import logging
+
+from .config import Config
+from .email_to_smol import EmailToSmolBridge
+from .worker import SmolWorker
+
+logger = logging.getLogger(__name__)
+
+
+class GatewayHandler:
+    def __init__(self, bridge: EmailToSmolBridge, worker: SmolWorker, config: Config) -> None:
+        self.bridge = bridge
+        self.worker = worker
+        self.config = config
+
+    async def handle_RCPT(self, server, session, envelope, address, rcpt_options):
+        username, _, domain = address.partition("@")
+        username = username.lower()
+        if domain.lower() != self.config.email_domain.lower():
+            return "550 relaying not permitted"
+
+        exists = await self.worker.run(self.bridge.user_exists, username)
+        if exists is None:
+            return "451 temporary failure resolving recipient"
+        if not exists:
+            return "550 no such user"
+
+        envelope.rcpt_tos.append(f"{username}@{domain}")
+        return "250 OK"
+
+    async def handle_DATA(self, server, session, envelope):
+        message = email_module.message_from_bytes(envelope.content)
+        sender = envelope.mail_from
+        subject = str(message.get("Subject", "(no subject)"))
+        body = self.bridge.extract_text(message)
+
+        replies = []
+        for address in envelope.rcpt_tos:
+            username = address.split("@", 1)[0]
+            try:
+                await self.worker.run(self.bridge.forward, username, sender, subject, body)
+                replies.append("250 2.1.5 delivered")
+            except Exception:
+                logger.exception("failed to forward mail for %s", username)
+                replies.append("451 4.3.0 temporary failure relaying to smolmail")
+        return "\r\n".join(replies)

src/pneumatube/vendor.py

@@ -0,0 +1,27 @@
+"""Loads the sibling smolmail reference client as a library.
+
+smolmail.py is a single-file `uv run --script` program, not an installed
+package, so it is loaded by path from its checkout rather than declared as a
+project dependency. Its own inline dependencies (pynacl, noiseprotocol) are
+declared directly in this project's pyproject.toml so the exec'd module
+resolves its imports normally.
+"""
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+from types import ModuleType
+
+
+def load(repo_path: Path) -> ModuleType:
+    module_path = repo_path / "smolmail.py"
+    if not module_path.is_file():
+        raise FileNotFoundError(
+            f"no smolmail.py at {module_path}; set PNEUMATUBE_SMOLMAIL_REPO "
+            "to a smolmail checkout")
+    spec = importlib.util.spec_from_file_location("smolmail", module_path)
+    module = importlib.util.module_from_spec(spec)
+    sys.modules["smolmail"] = module
+    spec.loader.exec_module(module)
+    return module

src/pneumatube/worker.py

@@ -0,0 +1,26 @@
+"""Serializes all smolmail protocol work onto one dedicated thread.
+
+smolmail.py's Store wraps a sqlite3 connection bound to the thread that
+created it, and its Noise session is one-operation-at-a-time per identity, so
+sharing a single background thread for every smolmail call (fetch polling,
+RESOLVE checks, SEND) is both required for correctness and a natural fit.
+"""
+from __future__ import annotations
+
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
+from typing import Callable, TypeVar
+
+T = TypeVar("T")
+
+
+class SmolWorker:
+    def __init__(self) -> None:
+        self._executor = ThreadPoolExecutor(max_workers=1)
+
+    async def run(self, fn: Callable[..., T], *args: object) -> T:
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(self._executor, fn, *args)
+
+    def shutdown(self) -> None:
+        self._executor.shutdown(wait=False)