feat: add Postfix/pneumatube container rig with live recipient rejection

src/pneumatube/__main__.py

@@ -10,7 +10,11 @@ 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,
+only), PNEUMATUBE_LISTEN_HOST/PORT, and PNEUMATUBE_LOOKUP_HOST/PORT. In lmtp
+mode, the lookup service answers Postfix's check_recipient_access (a
+tcp_table(5) query backed by the same RESOLVE check as handle_RCPT) so an
+unknown user is rejected live at RCPT TO instead of bouncing after Postfix
+has already queued the message. 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):
@@ -30,6 +34,7 @@ from aiosmtpd.lmtp import LMTP
 from aiosmtpd.smtp import SMTP as PlainSMTP
 
 from . import config as config_module
+from . import lookup_server
 from . import vendor
 from .email_to_smol import EmailToSmolBridge
 from .smol_to_email import SmolToEmailBridge
@@ -85,9 +90,17 @@ async def _run(cfg: config_module.Config) -> None:
                                      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)
+
+    tasks = [asyncio.create_task(_poll_forever(worker, outbound, cfg.fetch_interval))]
+    if cfg.listen_protocol == "lmtp":
+        # Only meaningful behind Postfix's check_recipient_access; the plain
+        # SMTP dev mode already rejects unknown users live in handle_RCPT.
+        tasks.append(asyncio.create_task(lookup_server.serve(inbound, worker, cfg)))
     try:
-        await _poll_forever(worker, outbound, cfg.fetch_interval)
+        await asyncio.gather(*tasks)
     finally:
+        for task in tasks:
+            task.cancel()
         controller.stop()
         worker.shutdown()
 

src/pneumatube/config.py

@@ -24,6 +24,7 @@ def _int(name: str, default: int) -> int:
 
 _LISTEN_PROTOCOLS = ("lmtp", "smtp")
 _LMTP_DEFAULT_PORT, _SMTP_DEFAULT_PORT = 2424, 1025
+_LOOKUP_DEFAULT_PORT = 2526
 
 
 @dataclass(frozen=True)
@@ -39,6 +40,12 @@ class Config:
     listen_host: str
     listen_port: int
 
+    # Postfix tcp_table(5) lookup service backing check_recipient_access, so
+    # an unknown user is rejected live during RCPT TO instead of bouncing
+    # after Postfix has already accepted the message (lmtp mode only).
+    lookup_host: str
+    lookup_port: int
+
     smtp_relay_host: str
     smtp_relay_port: int
     smtp_relay_user: str | None
@@ -65,6 +72,8 @@ def load() -> Config:
         listen_protocol=listen_protocol,
         listen_host=os.environ.get("PNEUMATUBE_LISTEN_HOST", "127.0.0.1"),
         listen_port=_int("PNEUMATUBE_LISTEN_PORT", default_port),
+        lookup_host=os.environ.get("PNEUMATUBE_LOOKUP_HOST", "127.0.0.1"),
+        lookup_port=_int("PNEUMATUBE_LOOKUP_PORT", _LOOKUP_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"),

src/pneumatube/lookup_server.py

@@ -0,0 +1,76 @@
+"""Postfix tcp_table(5) lookup service, backing check_recipient_access.
+
+Without this, Postfix's smtpd accepts a recipient into its queue before ever
+asking pneumatube whether the username exists — the live RESOLVE check only
+runs later, when the queue manager delivers over LMTP, so an unknown user
+gets a backscatter-prone bounce instead of a live SMTP-time reject. Wiring
+this service into smtpd_recipient_restrictions moves that same RESOLVE check
+to RCPT TO time, before Postfix commits to accepting the message.
+
+Protocol (tcp_table(5)): one request per line, `get <key>`, where key and
+the response text are %XX-quoted. A response code of 200 carries a result
+that Postfix applies as an access(5) action (e.g. "OK" or "550 ... text");
+500 means "no opinion, fall through to the next restriction"; 400 is a
+temporary error (Postfix defers the command rather than rejecting it).
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+from urllib.parse import quote, unquote
+
+from .config import Config
+from .email_to_smol import EmailToSmolBridge
+from .worker import SmolWorker
+
+logger = logging.getLogger(__name__)
+
+
+async def _lookup(bridge: EmailToSmolBridge, worker: SmolWorker, config: Config, key: str) -> str:
+    username, _, domain = key.partition("@")
+    if domain.lower() != config.email_domain.lower():
+        # Not a key shape we own (e.g. check_recipient_access's own fallback
+        # to a bare-domain lookup after a miss) — no opinion.
+        return "500 not found"
+
+    exists = await worker.run(bridge.user_exists, username.lower())
+    if exists is None:
+        return "400 temporary failure resolving recipient"
+    if exists:
+        return "200 " + quote("OK")
+    return "200 " + quote("550 5.1.1 no such user")
+
+
+async def _handle_connection(bridge, worker, config, reader, writer) -> None:
+    peer = writer.get_extra_info("peername")
+    try:
+        while True:
+            line = await reader.readline()
+            if not line:
+                break
+            request = line.decode("ascii", "replace").strip()
+            if not request:
+                continue
+            command, _, rest = request.partition(" ")
+            if command != "get":
+                writer.write(b"400 unsupported request\n")
+                await writer.drain()
+                continue
+            reply = await _lookup(bridge, worker, config, unquote(rest))
+            writer.write(reply.encode("ascii") + b"\n")
+            await writer.drain()
+    except (ConnectionError, asyncio.IncompleteReadError):
+        pass
+    except Exception:
+        logger.exception("lookup connection from %s failed", peer)
+    finally:
+        writer.close()
+
+
+async def serve(bridge: EmailToSmolBridge, worker: SmolWorker, config: Config) -> None:
+    server = await asyncio.start_server(
+        lambda r, w: _handle_connection(bridge, worker, config, r, w),
+        host=config.lookup_host, port=config.lookup_port)
+    logger.info("Postfix recipient-lookup service on %s:%s", config.lookup_host, config.lookup_port)
+    async with server:
+        await server.serve_forever()