tests: preserve MTA-STS failure diagnostics

This commit is contained in:
Mario Fetka
2026-07-28 15:25:55 +02:00
parent 2307fdd253
commit 853459766f
+59 -1
View File
@@ -1408,6 +1408,62 @@ def validate_tlsrpt_database(database: Path, start_day: int) -> None:
"TLSRPT did not aggregate both cached-policy successes")
def failure_diagnostics(
policy_server: PolicyServer | None,
cache_directory: Path | None,
) -> str:
"""Capture volatile MTA-STS evidence before the fixture is removed."""
details: list[str] = []
if policy_server is None:
details.append("policy-server=stopped")
else:
requests = list(policy_server.requests) # type: ignore[attr-defined]
details.append(
"policy-requests=" + (",".join(requests) if requests else "<none>"))
dig = shutil.which("dig")
if dig is None:
details.append("dig=unavailable")
else:
for route in ROUTES:
if route.sts_mode is None:
continue
completed = run(
[
dig,
f"@{DNS_RESOLVER}",
"+time=1",
"+tries=1",
"+short",
"TXT",
f"_mta-sts.{route.domain}",
],
check=False,
)
answer = completed.stdout.strip()
error = completed.stderr.strip()
details.append(
f"dns-txt[{route.domain}]="
+ (answer or error or f"<exit-{completed.returncode}>"))
if cache_directory is None or not cache_directory.exists():
details.append("policy-cache=<missing>")
else:
paths = sorted(
path for path in cache_directory.iterdir() if path.is_file())
if not paths:
details.append("policy-cache=<empty>")
for path in paths:
try:
content = path.read_text(
encoding="utf-8", errors="replace").strip()
except OSError as error:
content = f"<unreadable: {error}>"
details.append(f"policy-cache[{path.name}]={content}")
return "\n--- MTA-STS diagnostics ---\n" + "\n".join(details)
def append_trust_anchor(anchor: str) -> None:
if ANCHOR_BACKUP.exists():
raise SMTP27Error(
@@ -1479,6 +1535,7 @@ def perform_test(dns_stack: str) -> None:
resolver_process: subprocess.Popen | None = None
mail_servers: list[ThreadingTCPServer] = []
policy_server: PolicyServer | None = None
cache_directory: Path | None = None
primary_error: BaseException | None = None
def terminate(_signum: int, _frame) -> None:
@@ -1643,7 +1700,8 @@ def perform_test(dns_stack: str) -> None:
if tlsrpt_database is not None:
validate_tlsrpt_database(tlsrpt_database, start_day)
except BaseException as error:
primary_error = error
diagnostics = failure_diagnostics(policy_server, cache_directory)
primary_error = SMTP27Error(f"{error}{diagnostics}")
finally:
restoration_errors: list[str] = []
if policy_server is not None: