geos_one/bongo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Wire smtp-forward-metadata into ctest, on a dedicated netns instance

Browse Source 
smtp-forward-metadata-check.py (kept as a standalone manual script) was
never wired into ctest at all -- it needs real root to temporarily
steal 127.0.0.1:25 from production's own bongo.service, so the fixture
"external" server it drives can bind the exact address Bongo's own
listener normally owns. smtp-forward-metadata-live-check.py reuses the
shared dkim_dmarc_instance netns instead: extra loopback aliases are
free there (matching fake-dane-target-fixture.py's 127.0.0.2, smtp-
outbound-opportunistic-tls-check.py's 127.0.0.10-13, smtp-required-tls-
live-check.py's 127.0.0.20-25), so the fixture gets its own dedicated
127.0.0.30:25 and nothing needs stealing -- just a restart_manager()
child-tree-kill cycle to apply config, same pattern as the other netns
live tests.

internal_relay -- the trusted listener this test injects mail through
-- is not enabled by netns-instance-fixture.py's bootstrap() by
default (unlike the shared live_instance fixture), so this test enables
it itself as part of its own temporary config.

Two real bugs found and root-caused via strace while debugging this:
  - smtpd.c's config validation rejects internal_relay_enabled unless
    internal_relay_domain is non-empty ("Internal SMTP relay requires a
    normalization domain") -- easy to miss since this instance's own
    default is "".
  - smtpc.c's PrepareExternalMessage() requires every outbound message
    to be either DKIM-signed or SRS-rewritten, or delivery aborts
    immediately after connect() with no SMTP dialog at all (looks
    identical to "nothing is listening" from the receiving side).
    internal_relay rewrites the envelope sender's domain to
    internal_relay_domain, so an invented domain with no DKIM key
    silently broke every delivery attempt; fixed by reusing this
    instance's own already-DKIM-provisioned domain
    (netns-instance-fixture.py's DEFAULT_DOMAIN) instead.

Also serves as the shared helper module for the (not yet wired)
smtp-command-safety-live-check.py, smtp-forward-header-live-check.py,
and smtp-proxy-protocol-live-check.py -- collapsing the original script
family's SMTP22-imports-SMTP07 layering into one file, since none of
those three need smtp-outbound-opportunistic-tls-check.py's much
larger DNS-fixture/resolver machinery.

Verified: passes 2x in a row standalone, and together with the rest of
the dkim_dmarc_instance test family (13/13 passing) to confirm no
state leaks between tests sharing that instance.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Mario Fetka
2026-08-02 22:41:03 +02:00
parent 85a27dbd64
commit f18afd4a43
2 changed files with 726 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
contrib/testing/CMakeLists.txt
+27
View File
@@ -643,6 +643,33 @@ set_tests_properties(smtp-srs-forward-live-check PROPERTIES
TIMEOUT 120
ENVIRONMENT "BONGO_ALLOW_LIVE_USER_TEST=1;BONGO_TEST_INBOUND_PORT=2525;BONGO_TEST_DNS_ZONE_FILE=${BONGO_DKIM_INSTANCE_ROOT}/run/dns-zone.txt;BONGO_TEST_DNS_PID_FILE=${BONGO_DKIM_INSTANCE_ROOT}/run/fake-dns.pid;BONGO_TEST_SRS_TARGET_ROOT=${BONGO_DKIM_INSTANCE_ROOT}/run/srs-target")
# --- Outbound proxy metadata (XCLIENT/XFORWARD), same netns instance -------
# smtp-forward-metadata-check.py (kept as a standalone manual script, not
# wired here -- it was never wired at all before this, unlike the other
# "kept as manual" scripts this file documents) needs real root to
# temporarily steal 127.0.0.1:25 from production's own bongo.service.
# smtp-forward-metadata-live-check.py reuses the shared dkim_dmarc_
# instance and enables its own dedicated internal_relay listener plus a
# fresh loopback alias for the fixture "external" server (see that
# script's own docstring), rather than needing to move anything. Also
# doubles as the shared helper module for smtp-command-safety-live-
# check.py, smtp-forward-header-live-check.py, and smtp-proxy-protocol-
# live-check.py below -- see its own docstring for why those three don't
# need smtp-outbound-opportunistic-tls-check.py's much larger DNS-
# fixture/resolver machinery.
set(BONGO_SMTP22_ENVIRONMENT
"BONGO_ALLOW_LIVE_USER_TEST=1;BONGO_TEST_INBOUND_PORT=2525;BONGO_TEST_ADMIN_UNPRIVILEGED=1;BONGO_TEST_ADMIN=${BONGO_DKIM_INSTANCE_ROOT}/prefix/sbin/bongo-admin;BONGO_TEST_MANAGER_BIN=${BONGO_DKIM_INSTANCE_ROOT}/prefix/sbin/bongo-manager;BONGO_TEST_MANAGER_PID_FILE=${BONGO_DKIM_INSTANCE_ROOT}/run/manager.pid;BONGO_TEST_MANAGER_LOG=${BONGO_DKIM_INSTANCE_ROOT}/run/manager.log;BONGO_TEST_PREFIX=${BONGO_DKIM_INSTANCE_ROOT}/prefix;BONGO_TEST_QUEUE_TOOL=${BONGO_DKIM_INSTANCE_ROOT}/prefix/sbin/bongo-queuetool;BONGO_TEST_QUEUE_HOST=127.0.0.1;BONGO_TEST_QUEUE_PORT=26870")
add_test(NAME smtp-forward-metadata-live-check
COMMAND "${Python3_EXECUTABLE}" "${BONGO_DKIM_INSTANCE_FIXTURE}"
--root "${BONGO_DKIM_INSTANCE_ROOT}"
run -- "${Python3_EXECUTABLE}"
"${CMAKE_CURRENT_SOURCE_DIR}/smtp-forward-metadata-live-check.py")
set_tests_properties(smtp-forward-metadata-live-check PROPERTIES
FIXTURES_REQUIRED dkim_dmarc_instance
TIMEOUT 400
ENVIRONMENT "${BONGO_SMTP22_ENVIRONMENT}")
# --- Legacy install-prompt instance ---------------------------------------
# legacy-config-install.py drives bongo-config's historical interactive
# install prompt end-to-end, which -- unlike every Python-only live test
contrib/testing/smtp-forward-metadata-live-check.py
+699
View File
@@ -0,0 +1,699 @@
#!/usr/bin/env python3
# This program is free software, licensed under the terms of the GNU GPL.
# See the Bongo COPYING file for full details.
# Copyright (c) 2026 Bongo Project contributors
"""Verify trusted outbound XCLIENT and XFORWARD metadata negotiation,
run inside netns-instance-fixture.py's isolated dkim_dmarc_instance
namespace.
smtp-forward-metadata-check.py (kept as a standalone manual script, not
wired here, and itself never wired before this) needs real root: on a
single-loopback production host, the fixture "external" server below
must bind the exact same 127.0.0.1:25 Bongo's own listener normally
owns, so the script temporarily steals that address by moving
production's listener to a throwaway port first, then restarts via
`systemctl`. Inside this netns instance, extra loopback aliases are
free (matching fake-dane-target-fixture.py's own 127.0.0.2, smtp-
outbound-opportunistic-tls-check.py's 127.0.0.10-13, smtp-required-tls-
live-check.py's 127.0.0.20-25): the fixture binds its own dedicated
127.0.0.30:25, distinct from wherever Bongo's own SMTP listener is
bound, so no stealing is needed at all -- this only has to restart just
this netns instance's own bongo-manager (SIGTERM + respawn, matching
netns-instance-fixture.py's own setup()), the same restart_manager()
pattern smtp-required-tls-live-check.py and smtp-relayhost-live-
check.py already use.
internal_relay -- the trusted-internal-network listener this test
injects mail through, so Bongo relays outbound to the fixture instead
of just accepting-and-storing an ordinary inbound message -- is not
enabled by netns-instance-fixture.py's bootstrap() by default (unlike
the shared live_instance fixture, which provisions one). This test
enables it itself, at a dedicated address distinct from Bongo's own
SMTP listener, as part of its own temporary config.
Also serves as the shared helper module for smtp-command-safety-live-
check.py, smtp-forward-header-live-check.py, and smtp-proxy-protocol-
live-check.py -- mirroring the original script family's own SMTP22-
imports-SMTP07 layering, but collapsed into this one file, since none
of those three need smtp-outbound-opportunistic-tls-check.py's DNS-
fixture/resolver machinery, only config read/replace, a manager
restart, and Queue helpers.
"""
from __future__ import annotations
import copy
from dataclasses import dataclass, field
import json
import os
from pathlib import Path
import re
import signal
import socketserver
import subprocess
import sys
import threading
import time
ADMIN = os.environ.get("BONGO_TEST_ADMIN", "/usr/bin/bongo-admin")
QUEUE_TOOL = os.environ.get(
"BONGO_TEST_QUEUE_TOOL", "/usr/bin/bongo-queuetool")
QUEUE_HOST = os.environ.get("BONGO_TEST_QUEUE_HOST", "127.0.0.1")
QUEUE_PORT = os.environ.get("BONGO_TEST_QUEUE_PORT", "26870")
MANAGER_BIN = Path(os.environ.get(
"BONGO_TEST_MANAGER_BIN", "/tmp/bongo-dkim-instance/prefix/sbin/bongo-manager",
))
MANAGER_PID_FILE = Path(os.environ.get(
"BONGO_TEST_MANAGER_PID_FILE", "/tmp/bongo-dkim-instance/run/manager.pid",
))
MANAGER_LOG = Path(os.environ.get(
"BONGO_TEST_MANAGER_LOG", "/tmp/bongo-dkim-instance/run/manager.log",
))
PREFIX = Path(os.environ.get(
"BONGO_TEST_PREFIX", "/tmp/bongo-dkim-instance/prefix",
))
# Bongo's own always-on SMTP inbound listener, used purely as a
# restart_manager() readiness probe -- unlike INTERNAL_HOST/PORT below,
# this stays listening whether or not this test's own internal_relay
# config change is currently applied (it is disabled again by the time
# perform_test()'s finally block calls restart_manager() to restore the
# original configuration).
INBOUND_HOST = os.environ.get("BONGO_TEST_INBOUND_HOST", "127.0.0.1")
INBOUND_PORT = int(os.environ.get("BONGO_TEST_INBOUND_PORT", "2525"))
TIMEOUT = float(os.environ.get("BONGO_TEST_TIMEOUT", "45"))
# A dedicated loopback alias, distinct from Bongo's own SMTP listener and
# every other fixture already using this netns instance (see module
# docstring) -- the address-literal recipient below must dial real port
# 25, the RFC-mandated port for "user@[a.b.c.d]" delivery with no
# override, so only the host half is free to choose.
FIXTURE_HOST = os.environ.get("BONGO_TEST_SMTP22_HOST", "127.0.0.30")
FIXTURE_PORT = int(os.environ.get("BONGO_TEST_SMTP22_PORT", "25"))
# The trusted-internal-relay listener this test enables itself, distinct
# from Bongo's own SMTP listener (2525 in this netns instance).
INTERNAL_HOST = os.environ.get("BONGO_TEST_SMTP22_INTERNAL_HOST", "127.0.0.1")
INTERNAL_PORT = int(
os.environ.get("BONGO_TEST_SMTP22_INTERNAL_PORT", "2526"))
TOKEN = f"smtp22-{os.getpid()}-{time.time_ns()}"
QUEUE_ID = re.compile(r"^[0-9]{3}-[0-9a-f]+$")
class SMTP22Error(RuntimeError):
"""Raised when outbound connection metadata violates its contract, or
when a netns instance process-management helper fails."""
def _bongo_admin_argv(*arguments: str) -> list[str]:
if os.geteuid() == 0 or os.environ.get("BONGO_TEST_ADMIN_UNPRIVILEGED") == "1":
return [ADMIN, *arguments]
return ["sudo", "-n", ADMIN, *arguments]
def _bongo_queue_argv(*arguments: str) -> list[str]:
host_port = ["--host", QUEUE_HOST, "--port", QUEUE_PORT]
if os.geteuid() == 0 or os.environ.get("BONGO_TEST_ADMIN_UNPRIVILEGED") == "1":
return [QUEUE_TOOL, *host_port, *arguments]
return ["sudo", "-n", "-u", "bongo", QUEUE_TOOL, *host_port, *arguments]
def run(arguments: list[str], *, input_text: str | None = None,
check: bool = True) -> subprocess.CompletedProcess:
completed = subprocess.run(
arguments, input=input_text, capture_output=True, text=True,
check=False,
)
if check and completed.returncode:
raise SMTP22Error(
f"{' '.join(arguments)} failed: "
f"{completed.stderr.strip() or completed.stdout.strip()}")
return completed
def admin(*arguments: str, input_text: str | None = None) -> str:
return run(_bongo_admin_argv(*arguments), input_text=input_text).stdout
def read_configuration(name: str) -> dict:
try:
configuration = json.loads(admin("__config-read", name))
except json.JSONDecodeError as error:
raise SMTP22Error(
f"bongo-admin returned invalid {name} configuration") from error
if not isinstance(configuration, dict):
raise SMTP22Error(f"{name} configuration is not an object")
return configuration
def replace_configuration(name: str, configuration: dict) -> None:
admin(
"__config-replace", name,
input_text=json.dumps(configuration, separators=(",", ":")),
)
def queue(*arguments: str, check: bool = True) -> subprocess.CompletedProcess:
return run(_bongo_queue_argv(*arguments), check=check)
def queue_ids() -> list[str]:
identifiers = []
for line in queue("list").stdout.splitlines():
fields = line.split()
if fields and QUEUE_ID.fullmatch(fields[0]):
identifiers.append(fields[0])
return identifiers
def cleanup_queue() -> None:
for identifier in queue_ids():
message = queue("message", identifier, check=False)
if message.returncode == 0 and TOKEN in message.stdout:
queue("delete", identifier, check=False)
def read_pid(pid_path: Path) -> int | None:
try:
pid = int(pid_path.read_text(encoding="ascii").strip())
except (FileNotFoundError, ValueError):
return None
try:
os.kill(pid, 0)
except (ProcessLookupError, PermissionError):
return None
return pid
def _pid_alive(pid: int) -> bool:
try:
os.kill(pid, 0)
except ProcessLookupError:
return False
except PermissionError:
return True
return True
def _child_pids(parent_pid: int) -> list[int]:
completed = subprocess.run(
["ps", "--ppid", str(parent_pid), "-o", "pid="],
capture_output=True, text=True,
)
return [int(value) for value in completed.stdout.split() if value.strip()]
def service_active() -> bool:
return read_pid(MANAGER_PID_FILE) is not None
def restart_manager() -> None:
"""Restart just this netns instance's own bongo-manager process --
not systemctl, not any real service -- to apply a config change with
no live-reload path.
bongo-manager's own SIGTERM handling does not cascade to its agent
children (see collector-flood-quota-live-check.py's own
restart_manager() for how this was found): explicitly enumerates and
terminates the whole child tree, not just the manager PID, before
spawning a fresh one.
"""
pid = read_pid(MANAGER_PID_FILE)
if pid is not None:
targets = [pid, *_child_pids(pid)]
for target in targets:
try:
os.kill(target, signal.SIGTERM)
except ProcessLookupError:
pass
deadline = time.monotonic() + 15
while time.monotonic() < deadline and any(_pid_alive(t) for t in targets):
time.sleep(0.2)
for target in targets:
if _pid_alive(target):
try:
os.kill(target, signal.SIGKILL)
except ProcessLookupError:
pass
with MANAGER_LOG.open("ab") as log:
process = subprocess.Popen(
[str(MANAGER_BIN), "-k"],
stdin=subprocess.DEVNULL, stdout=log, stderr=subprocess.STDOUT,
env={**os.environ, "BONGO_PREFIX": str(PREFIX)},
start_new_session=True,
)
MANAGER_PID_FILE.write_text(f"{process.pid}\n", encoding="ascii")
deadline = time.monotonic() + 60
while time.monotonic() < deadline:
try:
import socket
with socket.create_connection(
(INBOUND_HOST, INBOUND_PORT), timeout=1
) as connection:
greeting = connection.recv(1024)
if greeting.startswith(b"220 "):
time.sleep(0.5)
return
except OSError:
pass
time.sleep(0.3)
raise SMTP22Error("bongo-manager did not become ready after restart")
@dataclass
class Delivery:
marker: str
commands: list[str]
message: bytes
@dataclass
class Capture:
lock: threading.Lock = field(default_factory=threading.Lock)
deliveries: list[Delivery] = field(default_factory=list)
sessions: list[list[str]] = field(default_factory=list)
def started(self, commands: list[str]) -> None:
with self.lock:
self.sessions.append(commands)
def delivered(
self, marker: str, commands: list[str], message: bytes
) -> None:
with self.lock:
self.deliveries.append(
Delivery(marker, list(commands), message)
)
def matching(self, marker: str) -> list[Delivery]:
with self.lock:
return [
delivery for delivery in self.deliveries
if delivery.marker == marker
]
def session_snapshot(self) -> list[list[str]]:
with self.lock:
return [list(commands) for commands in self.sessions]
class MetadataSMTPHandler(socketserver.BaseRequestHandler):
"""Minimal ESMTP peer implementing Postfix-style proxy extensions."""
def handle(self) -> None:
capture: Capture = self.server.capture # type: ignore[attr-defined]
connection = self.request
connection.settimeout(15)
buffer = bytearray()
commands: list[str] = []
sender = b""
recipients: list[bytes] = []
message = bytearray()
data_mode = False
greeted = False
capture.started(commands)
def send(line: bytes) -> None:
connection.sendall(line + b"\r\n")
def read_line() -> bytes | None:
while b"\n" not in buffer:
chunk = connection.recv(8192)
if not chunk:
return None
buffer.extend(chunk)
if len(buffer) > 4 * 1024 * 1024:
raise SMTP22Error("SMTP-22 fixture input exceeded limit")
end = buffer.index(b"\n") + 1
line = bytes(buffer[:end])
del buffer[:end]
return line.rstrip(b"\r\n")
send(b"220 smtp22.fixture ESMTP")
while True:
try:
line = read_line()
except (OSError, SMTP22Error):
return
if line is None:
return
if data_mode:
if line == b".":
source = bytes(message)
marker = next(
(
value.decode("ascii")
for value in source.split()
if value.startswith(TOKEN.encode("ascii") + b"-")
),
"",
).rstrip(">\r\n")
capture.delivered(marker, commands, source)
send(b"250 2.0.0 queued")
data_mode = False
sender = b""
recipients.clear()
message.clear()
else:
if line.startswith(b".."):
line = line[1:]
message.extend(line + b"\r\n")
continue
decoded = line.decode("ascii", "replace")
commands.append(decoded)
command, _, argument = line.partition(b" ")
command = command.upper()
if command == b"EHLO":
greeted = True
send(b"250-smtp22.fixture")
if getattr(
self.server, "advertise_proxy_extensions", True
):
send(b"250-XCLIENT HELO PROTO ADDR")
send(b"250-XFORWARD NAME ADDR PORT IDENT SOURCE")
send(b"250 SIZE 10485760")
elif command == b"HELO":
greeted = True
send(b"250 smtp22.fixture")
elif command == b"XCLIENT":
if not greeted or not argument:
send(b"503 5.5.1 EHLO and attributes required")
else:
greeted = False
sender = b""
recipients.clear()
send(b"220 smtp22.fixture ESMTP after XCLIENT")
elif command == b"XFORWARD":
if not greeted or not argument:
send(b"503 5.5.1 EHLO and attributes required")
else:
send(b"250 2.0.0 XFORWARD accepted")
elif command == b"MAIL":
if not greeted:
send(b"503 5.5.1 send EHLO after XCLIENT")
else:
sender = argument
send(b"250 2.1.0 sender accepted")
elif command == b"RCPT":
if not sender:
send(b"503 5.5.1 MAIL required")
else:
recipients.append(argument)
send(b"250 2.1.5 recipient accepted")
elif command == b"DATA":
if not sender or not recipients:
send(b"503 5.5.1 envelope required")
else:
message.clear()
data_mode = True
send(b"354 end with <CRLF>.<CRLF>")
elif command == b"RSET":
sender = b""
recipients.clear()
message.clear()
data_mode = False
send(b"250 2.0.0 reset")
elif command == b"NOOP":
send(b"250 2.0.0 ok")
elif command == b"QUIT":
send(b"221 2.0.0 bye")
return
else:
send(b"502 5.5.1 command not implemented")
class ThreadingSMTPServer(
socketserver.ThreadingMixIn, socketserver.TCPServer
):
daemon_threads = True
allow_reuse_address = True
def direct_configuration(
original: dict,
*,
xclient: list[str],
xforward: list[str],
) -> dict:
configuration = copy.deepcopy(original)
configuration.update(
{
"internal_relay_enabled": True,
"internal_relay_bind_address": INTERNAL_HOST,
"internal_relay_port": INTERNAL_PORT,
"internal_relay_networks": ["127.0.0.1/32"],
# smtpd.c's config validation rejects internal_relay_enabled
# unless this is non-empty ("Internal SMTP relay requires a
# normalization domain") -- this instance's own default is
# "" since netns-instance-fixture.py's bootstrap() never
# enables internal_relay at all. Reuses this instance's own
# DKIM-provisioned domain (netns-instance-fixture.py's own
# DEFAULT_DOMAIN, set up with a real key by its setup()) --
# internal_relay rewrites the envelope sender's domain to
# this value (confirmed via strace: the queued envelope's
# From line becomes "smtp22@<this domain>"), and smtpc.c's
# PrepareExternalMessage() (smtpc.c ~386) requires every
# outbound message to be either DKIM-signed or SRS-rewritten
# -- neither applied for an invented domain with no key,
# silently aborting the connection immediately after
# connect() (BONGO_MAILAUTH_PROTOCOL_ERROR -> DELIVER_TRY_
# LATER, no SMTP dialog at all -- looked identical to
# "nothing is listening" from the fixture's side, since it
# closes without ever reading the greeting or sending EHLO).
"internal_relay_domain": "dkim-test.invalid",
"xclient_trusted_destinations": xclient,
"xforward_trusted_destinations": xforward,
"proxy_header_trusted_destinations": [],
# SMTPMtaStsLookup() runs unconditionally for this delivery
# otherwise (see DeliverMessage() in smtpc.c), fetching a
# real (bogus, address-literal) URL for no benefit to what
# this test verifies.
"outbound_mta_sts_enabled": False,
}
)
return configuration
def submit(host: str, port: int, case: str) -> str:
import smtplib
marker = f"{TOKEN}-{case}"
message = (
"From: SMTP-22 fixture <smtp22@bongo.test>\r\n"
f"To: capture-{case}@example.net\r\n"
f"Subject: {marker}\r\n"
f"Message-ID: <{marker}@bongo.test>\r\n"
"\r\n"
f"Outbound metadata fixture {marker}\r\n"
).encode("ascii")
with smtplib.SMTP(host, port, timeout=10) as client:
client.ehlo("smtp22-source.bongo.test")
refused = client.sendmail(
"smtp22@bongo.test",
[f"capture-{case}@[{FIXTURE_HOST}]"],
message,
)
if refused:
raise SMTP22Error(
f"Bongo refused SMTP-22 ingress recipient: {refused}")
return marker
def wait_for_delivery(capture: Capture, marker: str) -> Delivery:
deadline = time.monotonic() + TIMEOUT
while time.monotonic() < deadline:
matching = capture.matching(marker)
if len(matching) == 1:
return matching[0]
if len(matching) > 1:
raise SMTP22Error(f"duplicate delivery for {marker}")
time.sleep(0.2)
raise SMTP22Error(
f"metadata fixture did not receive {marker}; "
f"sessions={capture.session_snapshot()!r}, "
f"deliveries={[item.marker for item in capture.deliveries]!r}"
)
def command_of(delivery: Delivery, name: str) -> list[str]:
prefix = f"{name} "
return [
command for command in delivery.commands
if command.upper().startswith(prefix)
]
def validate_untrusted(delivery: Delivery) -> None:
if command_of(delivery, "XCLIENT") or command_of(delivery, "XFORWARD"):
raise SMTP22Error(
"Bongo disclosed proxy metadata to an untrusted destination")
def validate_xclient(delivery: Delivery, *, also_xforward: bool) -> None:
commands = command_of(delivery, "XCLIENT")
if len(commands) != 1:
raise SMTP22Error(
f"expected one XCLIENT command, received {commands!r}")
command = commands[0]
for attribute in ("HELO=smtp22-source.bongo.test", "PROTO=ESMTP",
"ADDR="):
if attribute not in command:
raise SMTP22Error(
f"XCLIENT omitted advertised attribute {attribute}")
for attribute in ("NAME=", "PORT=", "LOGIN=", "DESTADDR=", "DESTPORT="):
if attribute in command:
raise SMTP22Error(
f"XCLIENT sent unadvertised attribute {attribute}")
xclient_index = delivery.commands.index(command)
mail_index = next(
index for index, value in enumerate(delivery.commands)
if value.upper().startswith("MAIL ")
)
ehlo_after = [
index for index, value in enumerate(delivery.commands)
if value.upper().startswith("EHLO ") and
xclient_index < index < mail_index
]
if len(ehlo_after) != 1:
raise SMTP22Error("XCLIENT was not followed by exactly one fresh EHLO")
if also_xforward:
forward = command_of(delivery, "XFORWARD")
if len(forward) != 1:
raise SMTP22Error("XFORWARD was not sent after XCLIENT/EHLO")
if not (ehlo_after[0] < delivery.commands.index(forward[0]) <
mail_index):
raise SMTP22Error("XFORWARD was sent in the wrong SMTP state")
elif command_of(delivery, "XFORWARD"):
raise SMTP22Error("unexpected XFORWARD command")
def validate_xforward(delivery: Delivery) -> None:
if command_of(delivery, "XCLIENT"):
raise SMTP22Error("unexpected XCLIENT command")
commands = command_of(delivery, "XFORWARD")
if len(commands) != 1:
raise SMTP22Error(
f"expected one XFORWARD command, received {commands!r}")
command = commands[0]
for attribute in ("ADDR=", "PORT=", "IDENT=", "SOURCE=REMOTE"):
if attribute not in command:
raise SMTP22Error(
f"XFORWARD omitted advertised attribute {attribute}: "
f"{command!r}")
for attribute in ("PROTO=", "HELO=", "LOGIN=", "DESTADDR=", "DESTPORT="):
if attribute in command:
raise SMTP22Error(
f"XFORWARD sent unadvertised attribute {attribute}")
if delivery.commands.index(command) > next(
index for index, value in enumerate(delivery.commands)
if value.upper().startswith("MAIL ")
):
raise SMTP22Error("XFORWARD was sent after MAIL")
def perform_test() -> None:
original_smtp = read_configuration("smtp")
capture = Capture()
server: ThreadingSMTPServer | None = None
thread: threading.Thread | None = None
configuration_changed = False
try:
cases = (
("untrusted", ["192.0.2.0/24"], ["198.51.100.0/24"]),
("xclient", [FIXTURE_HOST], []),
("xforward", [], ["127.0.0.0/8"]),
("both", ["127.0.0.0/8"], [FIXTURE_HOST]),
)
replace_configuration(
"smtp",
direct_configuration(
original_smtp,
xclient=cases[0][1],
xforward=cases[0][2],
),
)
configuration_changed = True
restart_manager()
server = ThreadingSMTPServer(
(FIXTURE_HOST, FIXTURE_PORT), MetadataSMTPHandler
)
server.capture = capture # type: ignore[attr-defined]
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
for case, xclient, xforward in cases:
replace_configuration(
"smtp",
direct_configuration(
original_smtp, xclient=xclient, xforward=xforward
),
)
configuration_changed = True
restart_manager()
marker = submit(INTERNAL_HOST, INTERNAL_PORT, case)
delivery = wait_for_delivery(capture, marker)
if case == "untrusted":
validate_untrusted(delivery)
elif case == "xclient":
validate_xclient(delivery, also_xforward=False)
elif case == "xforward":
validate_xforward(delivery)
else:
validate_xclient(delivery, also_xforward=True)
finally:
restoration_errors: list[str] = []
if configuration_changed:
try:
replace_configuration("smtp", original_smtp)
except (OSError, SMTP22Error) as error:
restoration_errors.append(
f"SMTP configuration restore failed: {error}")
if server is not None:
try:
server.shutdown()
server.server_close()
if thread is not None:
thread.join(timeout=5)
except OSError as error:
restoration_errors.append(
f"SMTP fixture shutdown failed: {error}")
try:
restart_manager()
cleanup_queue()
except (OSError, SMTP22Error) as error:
restoration_errors.append(
f"service/Queue restore failed: {error}")
if restoration_errors:
raise SMTP22Error("; ".join(restoration_errors))
if read_configuration("smtp") != original_smtp:
raise SMTP22Error("SMTP configuration was not restored exactly")
print(
"SMTP-22-LIVE PASS metadata=xclient/xforward "
"trust=untrusted-blocked/ip/cidr attributes=advertised-only "
"xclient=220/fresh-ehlo xforward=250/in-transaction "
"combined=ordered duplicate=no queue-clean=yes "
"config-restored=yes manager-restored=active"
)
def main() -> int:
try:
perform_test()
return 0
except (
json.JSONDecodeError,
OSError,
SMTP22Error,
StopIteration,
ValueError,
) as error:
print(f"SMTP-22-LIVE FAIL: {error}", file=sys.stderr)
return 1
if __name__ == "__main__":
raise SystemExit(main())