From 3ca724e73e684dffe7598e7f7c46521f72b807a0 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Sun, 18 Oct 2026 02:27:11 -0700 Subject: [PATCH] Preserve native USB failures when endpoint cleanup also fails --- test/test_official_console_acceptance.py | 127 +++++++++++++++++++++++ tools/hil/official_console_acceptance.py | 95 +++++++++++++++-- 2 files changed, 215 insertions(+), 7 deletions(-) diff --git a/test/test_official_console_acceptance.py b/test/test_official_console_acceptance.py index 84c0d26ca..4b2b9d2aa 100644 --- a/test/test_official_console_acceptance.py +++ b/test/test_official_console_acceptance.py @@ -1,4 +1,5 @@ """Offline owned-CDP startup regressions; never open a browser or serial port.""" +from contextlib import contextmanager import importlib.util import io import json @@ -260,6 +261,132 @@ class OwnedCleanupTests(unittest.TestCase): self.assertIs(signal.getsignal(signal.SIGTERM), previous) +class PrimaryFailureCleanupTests(unittest.TestCase): + def main_with(self, execute): + with tempfile.TemporaryDirectory() as directory: + work = Path(directory) + work.chmod(0o700) + output = work / "result.json" + stdout = io.StringIO() + with patch.object(acceptance, "execute", side_effect=execute), \ + patch("sys.stdout", new=stdout): + status = acceptance.main(["--serial", "exact", "--expected-board", "board", + "--expected-version", "version", "--logging", "on", "--console-source", "/unused", + "--artifact-manifest", "/unused", "--output", str(output), + "--work-dir", str(work), "--execute"]) + self.assertEqual(output.stat().st_mode & 0o777, 0o600) + return status, json.loads(output.read_text()), stdout.getvalue() + + @staticmethod + @contextmanager + def disappeared_tty(_port): + try: + yield {"vmin_during_test": 1, "vtime_during_test": 0} + finally: + raise FileNotFoundError("private missing /dev/ttyACM-secret") + + def test_idle_disconnect_survives_missing_tty_cleanup_and_main_report(self): + original = acceptance.NativeConsoleJavaScriptError({"exception": { + "className": "Error", "description": "Error: console_disconnected_while_idle\nprivate stack"}}) + + def disconnected(_args, report): + report["phase"] = "uninterrupted_idle" + with acceptance.recorded_linux_web_serial("/dev/inert", report): + raise original + + with patch.object(acceptance, "prepared_linux_web_serial", self.disappeared_tty): + status, report, stdout = self.main_with(disconnected) + self.assertEqual(status, 1) + self.assertFalse(report["ok"]) + self.assertEqual(report["error_type"], "NativeConsoleJavaScriptError") + self.assertEqual(report["error"], "native_console_javascript_failed:console_disconnected_while_idle") + self.assertEqual(report["primary_failure"]["phase"], "uninterrupted_idle") + self.assertEqual(report["TTY_cleanup_failure"]["class"], "FileNotFoundError") + self.assertIn("message_sha256", report["TTY_cleanup_failure"]) + self.assertEqual(report["TTY_cleanup_endpoint"], "missing") + self.assertFalse(report["original_TTY_restored"]) + self.assertNotIn("private", json.dumps(report) + stdout) + + def test_cleanup_failure_after_successful_body_cannot_pass(self): + def completed(_args, report): + report["phase"] = "complete" + with acceptance.recorded_linux_web_serial("/dev/inert", report): + report["native_streams_closed"] = True + + with patch.object(acceptance, "prepared_linux_web_serial", self.disappeared_tty): + status, report, _stdout = self.main_with(completed) + self.assertEqual(status, 1) + self.assertFalse(report["ok"]) + self.assertEqual(report["error_type"], "FileNotFoundError") + self.assertNotIn("primary_failure", report) + self.assertFalse(report["original_TTY_restored"]) + + def test_browser_and_tty_cleanup_failures_cannot_replace_original_transport_error(self): + original = TimeoutError("private original transport deadline") + report = {"phase": "uninterrupted_idle"} + + def browser_cleanup(): + raise OSError("private browser cleanup error") + + with patch.object(acceptance, "prepared_linux_web_serial", self.disappeared_tty), \ + self.assertRaises(TimeoutError) as caught: + with acceptance.recorded_linux_web_serial("/dev/inert", report): + with acceptance.preserving_cleanup_failure(report, "browser_cleanup_failure", browser_cleanup): + raise original + self.assertIs(caught.exception, original) + self.assertEqual(report["primary_failure"]["class"], "TimeoutError") + self.assertEqual(report["browser_cleanup_failure"]["class"], "OSError") + self.assertEqual(report["TTY_cleanup_failure"]["class"], "FileNotFoundError") + self.assertNotIn("private", json.dumps(report)) + + def test_replaced_tty_is_not_modified_or_reported_as_restored(self): + # Real Linux pseudoterminals exercise the unchanged identity guard; + # no physical serial device or browser is opened. + import os + import pty + import termios + import linux_web_serial as tty_helper + master, slave = pty.openpty() + other_master, other_slave = pty.openpty() + try: + path, other_path = os.ttyname(slave), os.ttyname(other_slave) + attributes = termios.tcgetattr(other_slave) + attributes[6][termios.VTIME] = b"\x07" + termios.tcsetattr(other_slave, termios.TCSANOW, attributes) + original_other = termios.tcgetattr(other_slave) + original_open = tty_helper._open_tty + calls = [] + + def replaced_open(selected): + calls.append(selected) + return original_open(selected if len(calls) == 1 else other_path) + + report = {"phase": "complete"} + with patch.object(tty_helper, "_open_tty", side_effect=replaced_open), \ + self.assertRaisesRegex(RuntimeError, "endpoint changed"): + with acceptance.recorded_linux_web_serial(path, report): + pass + self.assertEqual(termios.tcgetattr(other_slave), original_other) + self.assertEqual(report["TTY_cleanup_endpoint"], "identity_changed") + self.assertFalse(report["original_TTY_restored"]) + self.assertEqual(report["TTY_cleanup_failure"]["class"], "RuntimeError") + self.assertNotIn("primary_failure", report) + finally: + for fd in (master, slave, other_master, other_slave): + os.close(fd) + + def test_unknown_runtime_error_text_is_hashed_in_final_report(self): + def failed(_args, _report): + raise RuntimeError("private password from an external dependency") + + status, report, stdout = self.main_with(failed) + self.assertEqual(status, 1) + self.assertFalse(report["ok"]) + self.assertEqual(report["error"], "hardware_acceptance_interrupted_or_failed") + self.assertIn("message_sha256", report["failure"]) + self.assertNotIn("private", json.dumps(report) + stdout) + + class SafeJavaScriptFailureTests(unittest.TestCase): def evaluate_error(self, details): page = object.__new__(acceptance.CDP) diff --git a/tools/hil/official_console_acceptance.py b/tools/hil/official_console_acceptance.py index cd675607b..9f3c82c83 100755 --- a/tools/hil/official_console_acceptance.py +++ b/tools/hil/official_console_acceptance.py @@ -180,6 +180,87 @@ class NativeConsoleTerminated(KeyboardInterrupt): """Unwind the owned resources and report a normal SIGTERM interruption.""" +NATIVE_ERROR_CODES = frozenset({ + "CDP_browser_does_not_match_owned_profile", "CDP_request_failed", "CDP_request_deadline", + "USB_endpoint_changed_before_browser", "USB_identity_or_ownership_changed_after_cleanup", + "USB_vendor_product_ID_required", "complete_CRLF_reply_required", + "exact_USB_serial_requires_one_primary_interface", "existing_private_work_directory_required", + "firmware_command_rejected", "logging_mode_changed", "logging_mode_must_be_preconfigured", + "native_stream_cleanup_failed", "official_console_source_pin_mismatch", + "official_module_did_not_load", "owned_Chromium_exited_before_ready", + "owned_Chromium_startup_deadline", "owned_browser_requires_one_initial_blank_page", + "reopen_reset_or_uptime_regression", "runtime_board_mismatch", + "runtime_identity_changed_on_reopen", "runtime_version_mismatch", + "selected_serial_port_must_be_unowned", "uninterrupted_no_transmit_idle_required", + "uptime_delta_indicates_reset_or_stall", "usb_logging_reply_format_unrecognized", + "work_directory_must_be_private_and_owned", +}) +NATIVE_ERROR_CLASSES = frozenset({ + "RuntimeError", "OSError", "FileNotFoundError", "PermissionError", "TimeoutError", + "ConnectionError", "ConnectionResetError", "BrokenPipeError", "ValueError", + "KeyboardInterrupt", "NativeConsoleTerminated", "NativeConsoleJavaScriptError", +}) + + +def safe_native_failure(error, phase): + """Keep controlled codes; hash arbitrary transport, TTY and exception text.""" + name = type(error).__name__ + safe = {"class": name if name in NATIVE_ERROR_CLASSES else "unrecognized", + "phase": phase, "code": "hardware_acceptance_interrupted_or_failed"} + if isinstance(error, NativeConsoleJavaScriptError): + safe["code"] = "native_console_javascript_failed:" + error.safe_details["code"] + safe["javascript_exception"] = error.safe_details + elif type(error) is RuntimeError and str(error) in NATIVE_ERROR_CODES: + safe["code"] = str(error) + else: + safe["message_sha256"] = hashlib.sha256( + str(error).encode("utf-8", errors="replace")).hexdigest() + return safe + + +@contextmanager +def preserving_cleanup_failure(report, label, cleanup): + """A cleanup failure must not replace the failed native trial's evidence.""" + primary = None + try: + yield + except BaseException as error: + primary = error + report.setdefault("primary_failure", safe_native_failure(error, report.get("phase"))) + raise + finally: + try: + cleanup() + except BaseException as error: + report[label] = safe_native_failure(error, report.get("phase")) + if primary is None: + raise + + +@contextmanager +def recorded_linux_web_serial(port, report): + manager = prepared_linux_web_serial(port) + tty = manager.__enter__() + report["original_TTY_restored"] = False + + def restore(): + try: + # This helper's exit only restores termios; the primary trial + # exception is handled by the surrounding cleanup boundary. + manager.__exit__(None, None, None) + except FileNotFoundError: + report["TTY_cleanup_endpoint"] = "missing" + raise + except RuntimeError as error: + if str(error) == "TTY endpoint changed; original attributes were not applied": + report["TTY_cleanup_endpoint"] = "identity_changed" + raise + report["original_TTY_restored"] = True + + with preserving_cleanup_failure(report, "TTY_cleanup_failure", restore): + yield tty + + @contextmanager def cleanup_on_sigterm(): previous = signal.getsignal(signal.SIGTERM) @@ -419,9 +500,10 @@ def execute(args, report): browser = page = None browser_log = (work / "chromium-private.log").open("xb") try: - with prepared_linux_web_serial(device.device) as tty: + with recorded_linux_web_serial(device.device, report) as tty: report["TTY_preflight"] = tty - try: + with preserving_cleanup_failure(report, "browser_cleanup_failure", + lambda: close_owned_browser(page, browser, report)): # Re-enumerate the exact UID and ownership after TTY preparation. require(select_device(args.serial).device == device.device, "USB_endpoint_changed_before_browser") browser = subprocess.Popen([str(args.chromium), "--headless=new", "--no-sandbox", @@ -495,9 +577,6 @@ def execute(args, report): setup_writes=page.evaluate("hil.setup_writes")) report["native_streams_closed"] = True report["phase"] = "complete" - finally: - close_owned_browser(page, browser, report) - report["original_TTY_restored"] = True require(select_device(args.serial).device == device.device, "USB_identity_or_ownership_changed_after_cleanup") require(not (report.get("browser_stream_cleanup_error") or report.get("browser_socket_cleanup_error")), "native_stream_cleanup_failed") @@ -556,8 +635,10 @@ def main(argv=None): status = 0 except BaseException as error: # Preserve only controlled errors, never arbitrary native console output. - report["error_type"] = type(error).__name__ - report["error"] = str(error) if isinstance(error, RuntimeError) else "hardware_acceptance_interrupted_or_failed" + failure = safe_native_failure(error, report.get("phase")) + report["failure"] = failure + report["error_type"] = failure["class"] + report["error"] = failure["code"] if isinstance(error, NativeConsoleJavaScriptError): report["javascript_exception"] = error.safe_details if isinstance(error, NativeConsoleTerminated):