diff --git a/contrib/testing/smtp-dane-mta-sts-check.py b/contrib/testing/smtp-dane-mta-sts-check.py index f841d93..0968bcb 100755 --- a/contrib/testing/smtp-dane-mta-sts-check.py +++ b/contrib/testing/smtp-dane-mta-sts-check.py @@ -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 "")) + + 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"")) + + if cache_directory is None or not cache_directory.exists(): + details.append("policy-cache=") + else: + paths = sorted( + path for path in cache_directory.iterdir() if path.is_file()) + if not paths: + details.append("policy-cache=") + for path in paths: + try: + content = path.read_text( + encoding="utf-8", errors="replace").strip() + except OSError as error: + content = f"" + 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: