diff --git a/CHANGELOG.md b/CHANGELOG.md index 6e4bd39ca..b52bada3e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,6 +31,8 @@ All notable changes to OriginWeave are documented in this file. The format follo ### Changed +- Classified Manifest V3 real-click post-condition failures as a fixed mismatch token so page-controlled WebDriver text cannot enter runner exception text. +- Recorded the current Chrome Extensions `chrome.downloads` primary reference in APA 7th form and stated that the active downloads lane proves one controlled loopback payload in pinned Chromium, not Agent filesystem authority. - Separated logical origin authority from resolved network destination authority; an origin grant no longer implies permission to connect to every resolver result. - Separated resolved-address authorization from direct transport evidence; an approved IP now becomes a usable stream only after the operating system reports the exact requested IP and port. - Separated exact TCP peer proof from authenticated TLS service identity; an observed peer becomes an authenticated HTTPS stream only after explicit-root, fixed-time, SAN-bound WebPKI verification over that same stream. diff --git a/docs/doctoring.md b/docs/doctoring.md index 75c107ef0..8030e1156 100644 --- a/docs/doctoring.md +++ b/docs/doctoring.md @@ -8,6 +8,14 @@ This document records external evidence that changes OriginWeave architecture, t The 1 June 2026 WebDriver BiDi Working Draft defines a bidirectional remote-control protocol, events, commands, and user contexts. Because it remains a W3C Working Draft, OriginWeave places BiDi behind a versioned adapter and Web Platform Tests-derived contract tests rather than make it the internal authority model. +### Manifest V3 downloads compatibility + +The current Chrome Extensions Downloads API documents the `downloads` manifest permission and `chrome.downloads` methods that initiate, monitor, search, and inspect downloads. That living vendor reference is API semantics only. OriginWeave treats a successful controlled loopback download in pinned Chromium as compatibility evidence for one declared surface, not as Agent filesystem authority, general download persistence, or a claim that every Downloads method is supported. + +### Manifest V3 click post-condition diagnostics + +W3C WebDriver Get Element Text returns the rendered text content of a located element. That value is page-controlled data, not a trusted diagnostic token. The Manifest V3 compatibility runner therefore compares the fixture output against the exact expected `clicked` token and, on mismatch, raises only the classified message `real click post-condition mismatch`. Raw element text must not enter exception text, trial evidence, or logs. + ### Browser origin equivalence The WHATWG URL host parser and Chromium canonicalizer classify shortened decimal, integer, hexadecimal, legacy octal-looking, and mixed-component numeric hosts as IPv4 or broken IPv4 candidates rather than ordinary DNS names. Chromium's regression suite includes values such as `192`, `0xC0a80001`, `030052000001`, and mixed hexadecimal components. A non-final empty `0x` component can participate in Chromium's multi-part IPv4 truncation behavior, but a final `0x` label does not produce an IPv4 number because stripping its prefix leaves no digits; it remains a domain label. Chromium also warns that broken IP-like hosts must not be connected because another resolver could accept them. OriginWeave therefore admits only canonical dotted-decimal IPv4 into its policy origin type, rejects browser-special numeric spellings before DNS validation, and preserves final non-numeric DNS labels such as `0x`. @@ -98,6 +106,8 @@ Autio, C., Schwartz, R., Dunietz, J., Jain, S., Stanley, M., Tabassi, E., Hall, Bonica, R., Cotton, M., Haberman, B., & Vegoda, L. (2017). *Updates to the special-purpose IP address registries* (RFC 8190). Internet Engineering Task Force. https://doi.org/10.17487/RFC8190 +Chrome for Developers. (n.d.). *chrome.downloads*. Google. Retrieved August 16, 2026, from https://developer.chrome.com/docs/extensions/reference/api/downloads + Chromium Authors. (n.d.). *Proxy support in Chrome* [Source documentation]. Chromium. https://chromium.googlesource.com/chromium/src/+/a3e71ebfa307d8760eb68b777e2998a869940092/net/docs/proxy.md Chromium Authors. (2026). *URL canonicalizer unit tests* [Source code]. Chromium. https://chromium.googlesource.com/chromium/src/+/446d05d21720f0b3505ec21057b3e9f909784262/url/url_canon_unittest.cc @@ -154,6 +164,8 @@ Web Hypertext Application Technology Working Group. (2026). *URL standard*. http World Wide Web Consortium. (2013). *PROV-O: The PROV ontology*. https://www.w3.org/TR/prov-o/ +World Wide Web Consortium. (2018, June 5). *WebDriver* (W3C Recommendation). https://www.w3.org/TR/2018/REC-webdriver1-20180605/ + World Wide Web Consortium. (2026, June 1). *WebDriver BiDi* (W3C Working Draft). https://www.w3.org/TR/2026/WD-webdriver-bidi-20260601/ Xu, J., Sun, Q., Schwendeman, P., Nielsen, S., Cetin, E., & Tang, Y. (2025). *TRINITY: An evolved LLM coordinator* [Preprint]. arXiv. https://doi.org/10.48550/arXiv.2512.04695 diff --git a/docs/doctoring/mv3-compatibility.md b/docs/doctoring/mv3-compatibility.md index 571c49329..7fe360bec 100644 --- a/docs/doctoring/mv3-compatibility.md +++ b/docs/doctoring/mv3-compatibility.md @@ -1,7 +1,7 @@ # Manifest V3 compatibility evidence baseline - **Status:** Active implementation evidence for issue #27 -- **Reviewed:** 2026-08-11 +- **Reviewed:** 2026-08-16 - **Pinned browser:** Chrome for Testing `150.0.7871.129`, Chromium revision `r1639810` OriginWeave uses Chromium as its compatibility kernel, so browser-extension compatibility must be demonstrated with executable Chromium evidence rather than inferred from architecture alone. The protected-main lane exercises a controlled unpacked Manifest V3 extension against one exact Chrome for Testing build and proves service-worker, content-script, storage, declarative-network-request, tabs, windows, scripting, commands, side-panel, bookmarks/history read compatibility, restart persistence, repeatability, and one real WebDriver click/post-condition. Active stacked compatibility work adds downloads, bounded bookmark/history mutation, profile isolation, explicit extension update/version-migration evidence, and an exact content-script isolated-world check. OriginWeave does **not claim 100% Chrome extension compatibility**. @@ -38,6 +38,14 @@ The release-quality capability matrix must remain coupled to executable evidence For history compatibility specifically, the current official Chrome Extensions API documents the `history` manifest permission and Promise-returning `chrome.history.addUrl`, `chrome.history.search`, and `chrome.history.deleteUrl` methods. This living vendor reference establishes API semantics only. OriginWeave release evidence continues to depend on the exact pinned Chromium fixture and exact-head CI result rather than inferring compatibility from documentation. +## Downloads API primary evidence + +For downloads compatibility specifically, the current official Chrome Extensions API documents the `downloads` manifest permission and the `chrome.downloads` methods that initiate, monitor, search, and inspect downloads. This living vendor reference establishes API semantics only. Active PR #43 exercises one controlled loopback payload through pinned Chromium and retains only allow-listed stage diagnostics. That proof is not Agent filesystem authority, general download persistence, unsafe-filename handling, or a release claim that every `chrome.downloads` method works. + +## Click post-condition diagnostic boundary + +W3C WebDriver Get Element Text returns rendered element text. That value is page-controlled data. The compatibility runner compares the fixture output against the exact expected `clicked` token and, on mismatch, retains only the classified message `real click post-condition mismatch`. Raw element text must not enter exception text or trial evidence. + ## Update-migration evidence boundary Restart persistence and extension update migration are separate compatibility claims. A successful restart proves only that state survives a new browser process. The active update-migration lane additionally uses a trial-local copy of the checked-in fixture, preserves the same extension path and ephemeral profile across passes, changes only the controlled manifest version from `1.0.0` to `1.0.1`, observes `chrome.runtime.getManifest().version`, and requires the fixture schema marker to migrate from version 1 to version 2. The checked-in fixture is not rewritten by the test. This establishes one deterministic unpacked-extension version transition; it does not establish Chrome Web Store update behavior, enterprise rollout semantics, downgrade behavior, or arbitrary third-party extension migration safety. @@ -60,6 +68,8 @@ Chrome for Developers. (2023, May 2). *The extension service worker lifecycle*. Chrome for Developers. (n.d.). *chrome.declarativeNetRequest*. Google. Retrieved August 9, 2026, from https://developer.chrome.com/docs/extensions/reference/api/declarativeNetRequest +Chrome for Developers. (n.d.). *chrome.downloads*. Google. Retrieved August 16, 2026, from https://developer.chrome.com/docs/extensions/reference/api/downloads + Chrome for Developers. (n.d.). *chrome.history*. Google. Retrieved August 11, 2026, from https://developer.chrome.com/docs/extensions/reference/api/history Chrome for Developers. (n.d.). *Manifest file format*. Google. Retrieved August 9, 2026, from https://developer.chrome.com/docs/extensions/reference/manifest @@ -67,3 +77,5 @@ Chrome for Developers. (n.d.). *Manifest file format*. Google. Retrieved August Bynens, M. (2023, June 12). *Chrome for Testing*. Chrome for Developers. https://developer.chrome.com/docs/automation-and-testing/chrome-for-testing Google Chrome Labs. (2026, July 21). *Chrome for Testing availability*. https://googlechromelabs.github.io/chrome-for-testing/ + +World Wide Web Consortium. (2018, June 5). *WebDriver* (W3C Recommendation). https://www.w3.org/TR/2018/REC-webdriver1-20180605/ diff --git a/scripts/ci/run_mv3_compatibility.py b/scripts/ci/run_mv3_compatibility.py old mode 100755 new mode 100644 index 28a3fb1e2..614c0e736 --- a/scripts/ci/run_mv3_compatibility.py +++ b/scripts/ci/run_mv3_compatibility.py @@ -5,13 +5,12 @@ W3C WebDriver HTTP protocol only to prove that a real Chrome for Testing build can load the controlled MV3 fixture and repeatedly exercise service-worker, content-script, storage, declarative-net-request, tabs, windows, scripting, -commands, side-panel, bookmarks, history, real browser-click, and +commands, side-panel, bookmarks, history, downloads, real browser-click, and restart-persistence behavior. """ from __future__ import annotations -import contextlib import http.client import http.server import json @@ -29,6 +28,12 @@ FIXTURE = ROOT / "tests" / "fixtures" / "mv3_basic" PINNED_CHROME_VERSION = "150.0.7871.129" PINNED_CHROME_REVISION = "r1639810" +PINNED_CHROME_RELATIVE_PATH = pathlib.PurePosixPath( + ".mv3-browser/chrome-linux64/chrome" +) +PINNED_CHROMEDRIVER_RELATIVE_PATH = pathlib.PurePosixPath( + ".mv3-browser/chromedriver-linux64/chromedriver" +) REPEATABILITY_TRIALS = 3 REQUEST_TIMEOUT_SECONDS = 5.0 STARTUP_TIMEOUT_SECONDS = 20.0 @@ -36,6 +41,58 @@ MAX_WEBDRIVER_RESPONSE_BYTES = 1_048_576 W3C_ELEMENT_KEY = "element-6066-11e4-a52e-4f735466cecf" PATH_TOKEN_CHARACTERS = frozenset(string.ascii_letters + string.digits + "-_.") +SURFACE_EVIDENCE_KEYS = ( + "content", + "storage", + "storagePersistence", + "workerReply", + "workerState", + "workerStartCount", + "dnr", + "tabs", + "windows", + "scripting", + "scriptingExecuted", + "commands", + "sidePanel", + "bookmarks", + "history", + "downloads", + "downloadsDiagnostic", +) +SURFACE_EVIDENCE_VALUES = frozenset( + {"ready", "missing", "initialized", "persisted", "pong", "installed", "blocked"} +) +DOWNLOAD_DIAGNOSTIC_VALUES = frozenset( + { + "download-source-rejected", + "download-start-rejected", + "download-search-missing", + "download-interrupted", + "download-url-mismatch", + "download-byte-count-mismatch", + "download-exists-false", + "download-timeout", + "download-complete-ready", + "download-not-evaluated", + } +) + + +class CompatibilitySurfaceError(RuntimeError): + """Report only bounded fixture-surface state when real-browser evidence does not converge.""" + + def __init__(self, observed: dict[str, str]) -> None: + self.observed = { + key: _safe_surface_value(key, observed[key]) + for key in SURFACE_EVIDENCE_KEYS + if key in observed + } + super().__init__("Manifest V3 fixture surfaces did not converge") + + +class WebDriverSessionCleanupError(RuntimeError): + """Report a reviewed WebDriver session-delete failure after process teardown.""" class QuietFixtureHandler(http.server.SimpleHTTPRequestHandler): @@ -45,6 +102,30 @@ def log_message(self, _format: str, *args: object) -> None: """Suppress request logs because the fixture contains no diagnostic value.""" +def _safe_surface_value(key: str, value: str) -> str: + """Reduce one controlled DOM evidence value to a non-sensitive diagnostic token.""" + + if key == "workerStartCount": + return value if value.isdecimal() and len(value) <= 20 else "invalid" + if key == "downloadsDiagnostic": + return value if value in DOWNLOAD_DIAGNOSTIC_VALUES else "unexpected" + return value if value in SURFACE_EVIDENCE_VALUES else "unexpected" + + +def _failure_evidence(error: BaseException) -> dict[str, Any]: + """Classify one browser-trial failure without retaining raw exception text.""" + + if isinstance(error, CompatibilitySurfaceError): + return {"failure_kind": "surface_mismatch", "observed": error.observed} + if isinstance(error, json.JSONDecodeError): + return {"failure_kind": "json_decode_error"} + if isinstance(error, OSError): + return {"failure_kind": "io_error"} + if isinstance(error, ValueError): + return {"failure_kind": "value_error"} + return {"failure_kind": "runtime_error"} + + def _free_loopback_port() -> int: """Reserve and release one loopback TCP port for a short-lived local service.""" @@ -108,8 +189,7 @@ def _json_request( if len(raw) > MAX_WEBDRIVER_RESPONSE_BYTES: raise RuntimeError("WebDriver response exceeded the bounded JSON limit") if response.status >= 400: - detail = raw.decode("utf-8", errors="replace") - raise RuntimeError(f"WebDriver HTTP {response.status}: {detail}") + raise RuntimeError(f"WebDriver HTTP {response.status} error") finally: connection.close() @@ -118,24 +198,26 @@ def _json_request( raise RuntimeError("WebDriver returned a non-object JSON payload") value = decoded.get("value") if isinstance(value, dict) and value.get("error"): - raise RuntimeError(f"WebDriver error: {value.get('error')}: {value.get('message')}") + raise RuntimeError("WebDriver returned a protocol error") return decoded def _wait_for_driver(driver_port: int) -> None: - """Wait for the exact local ChromeDriver process to become ready.""" + """Wait for local ChromeDriver readiness while retaining only a safe failure class.""" deadline = time.monotonic() + STARTUP_TIMEOUT_SECONDS - last_error: Exception | None = None + last_failure_kind = "not_observed" while time.monotonic() < deadline: try: status = _json_request(driver_port, "GET", "/status", timeout=1.0) if status.get("value", {}).get("ready") is True: return except (OSError, ValueError, RuntimeError, json.JSONDecodeError) as exc: - last_error = exc + last_failure_kind = str(_failure_evidence(exc)["failure_kind"]) time.sleep(0.1) - raise RuntimeError(f"ChromeDriver did not become ready: {last_error}") + raise RuntimeError( + f"ChromeDriver did not become ready ({last_failure_kind})" + ) def _execute(driver_port: int, session_id: str, script: str) -> Any: @@ -178,7 +260,10 @@ def _wait_for_extension_evidence( commands: document.documentElement.dataset.originweaveCommands || "missing", sidePanel: document.documentElement.dataset.originweaveSidePanel || "missing", bookmarks: document.documentElement.dataset.originweaveBookmarks || "missing", - history: document.documentElement.dataset.originweaveHistory || "missing" + history: document.documentElement.dataset.originweaveHistory || "missing", + downloads: document.documentElement.dataset.originweaveDownloads || "missing", + downloadsDiagnostic: + document.documentElement.dataset.originweaveDownloadsDiagnostic || "download-not-evaluated" }; """ expected = { @@ -196,6 +281,8 @@ def _wait_for_extension_evidence( "sidePanel": "ready", "bookmarks": "ready", "history": "ready", + "downloads": "ready", + "downloadsDiagnostic": "download-complete-ready", } deadline = time.monotonic() + FIXTURE_TIMEOUT_SECONDS latest: dict[str, str] = {} @@ -212,13 +299,11 @@ def _wait_for_extension_evidence( ): return latest time.sleep(0.1) - raise RuntimeError( - f"MV3 fixture did not converge: expected={expected!r}, observed={latest!r}" - ) + raise CompatibilitySurfaceError(latest) def _exercise_real_click(driver_port: int, session_id: str) -> str: - """Use the WebDriver element-click command and verify the DOM post-condition.""" + """Use the WebDriver element-click command and classify DOM post-condition mismatches.""" found = _json_request( driver_port, @@ -253,10 +338,38 @@ def _exercise_real_click(driver_port: int, session_id: str) -> str: _webdriver_path(session_id, f"/element/{safe_output}/text"), ).get("value") if text != "clicked": - raise RuntimeError(f"real click post-condition failed: {text!r}") + raise RuntimeError("real click post-condition mismatch") return str(text) +def _teardown_driver_process(driver: subprocess.Popen[str]) -> Exception | None: + """Best-effort reap ChromeDriver while preserving reviewed process failures.""" + + try: + driver.terminate() + except OSError as terminate_error: + try: + driver.kill() + driver.wait(timeout=5) + except (OSError, subprocess.TimeoutExpired) as fallback_error: + terminate_error.add_note( + "bounded ChromeDriver kill fallback also failed: " + f"{type(fallback_error).__name__}" + ) + return terminate_error + + try: + driver.wait(timeout=5) + return None + except subprocess.TimeoutExpired: + try: + driver.kill() + driver.wait(timeout=5) + except (OSError, subprocess.TimeoutExpired) as fallback_error: + return fallback_error + return None + + def _run_browser_pass( chrome_bin: pathlib.Path, chromedriver_bin: pathlib.Path, @@ -268,6 +381,8 @@ def _run_browser_pass( driver_port = _free_loopback_port() session_id: str | None = None + download_dir = pathlib.Path(profile_dir) / "downloads" + download_dir.mkdir(mode=0o700, parents=True, exist_ok=True) driver = subprocess.Popen( [str(chromedriver_bin), f"--port={driver_port}", "--allowed-ips=127.0.0.1"], stdout=subprocess.DEVNULL, @@ -298,6 +413,11 @@ def _run_browser_pass( f"--disable-extensions-except={FIXTURE}", f"--load-extension={FIXTURE}", ], + "prefs": { + "download.default_directory": str(download_dir), + "download.prompt_for_download": False, + "download.directory_upgrade": True, + }, }, } } @@ -349,24 +469,31 @@ def _run_browser_pass( "side-panel": surfaces["sidePanel"] == "ready", "bookmarks": surfaces["bookmarks"] == "ready", "history": surfaces["history"] == "ready", + "downloads": surfaces["downloads"] == "ready", "real-browser-click": click_result == "clicked", }, } finally: - if session_id is not None: - with contextlib.suppress(Exception): - _json_request( - driver_port, - "DELETE", - _webdriver_path(session_id, ""), - {}, - ) - driver.terminate() + cleanup_error: Exception | None = None try: - driver.wait(timeout=5) - except subprocess.TimeoutExpired: - driver.kill() - driver.wait(timeout=5) + if session_id is not None: + try: + _json_request( + driver_port, + "DELETE", + _webdriver_path(session_id, ""), + {}, + ) + except (OSError, ValueError, RuntimeError, json.JSONDecodeError) as error: + cleanup_error = error + finally: + teardown_error = _teardown_driver_process(driver) + if cleanup_error is not None: + raise WebDriverSessionCleanupError( + "WebDriver session cleanup failed after bounded process teardown" + ) from cleanup_error + if teardown_error is not None: + raise teardown_error def _run_restart_trial( @@ -433,15 +560,66 @@ def _run_restart_trial( } +def _pinned_workspace_binary( + env_name: str, + relative_path: pathlib.PurePosixPath, + label: str, + *, + root: pathlib.Path = ROOT, +) -> pathlib.Path: + """Authorize only the exact non-symlink executable provisioned under the workspace. + + Environment variables remain compatibility inputs for the workflow, but they + cannot redirect execution. The release lane has one reviewed path for each + pinned Chrome-for-Testing artifact, and any other executable fails closed. + """ + + if relative_path.is_absolute() or ".." in relative_path.parts: + raise SystemExit(f"{label} pinned workspace path is invalid") + + trusted_root = pathlib.Path(os.path.abspath(root)) + expected = pathlib.Path(os.path.abspath(trusted_root.joinpath(*relative_path.parts))) + configured = os.environ.get(env_name) + if configured: + configured_path = pathlib.Path(configured) + if not configured_path.is_absolute(): + raise SystemExit(f"{env_name} must name the pinned workspace executable") + if pathlib.Path(os.path.abspath(configured_path)) != expected: + raise SystemExit(f"{env_name} must name the pinned workspace executable") + + current = expected + while current != trusted_root: + if current.is_symlink(): + raise SystemExit(f"{label} pinned workspace executable path contains a symlink") + parent = current.parent + if parent == current: + raise SystemExit(f"{label} pinned workspace executable escaped the workspace") + current = parent + + try: + expected.relative_to(trusted_root) + except ValueError as exc: + raise SystemExit(f"{label} pinned workspace executable escaped the workspace") from exc + if not expected.is_file(): + raise SystemExit(f"{label} pinned workspace executable is missing") + if not os.access(expected, os.X_OK): + raise SystemExit(f"{label} pinned workspace executable is not executable") + return expected + + def main() -> int: """Run three independent restart trials and emit bounded repeatability evidence.""" - chrome_bin = pathlib.Path(os.environ.get("CHROME_BIN", "")) - chromedriver_bin = pathlib.Path(os.environ.get("CHROMEDRIVER_BIN", "")) - if not chrome_bin.is_file(): - raise SystemExit("CHROME_BIN must point to the pinned Chrome for Testing executable") - if not chromedriver_bin.is_file(): - raise SystemExit("CHROMEDRIVER_BIN must point to the matching pinned ChromeDriver") + chrome_bin = _pinned_workspace_binary( + "CHROME_BIN", + PINNED_CHROME_RELATIVE_PATH, + "Chrome for Testing", + ) + chromedriver_bin = _pinned_workspace_binary( + "CHROMEDRIVER_BIN", + PINNED_CHROMEDRIVER_RELATIVE_PATH, + "ChromeDriver", + ) if not (FIXTURE / "manifest.json").is_file(): raise SystemExit("MV3 fixture manifest is missing") @@ -468,13 +646,13 @@ def main() -> int: trial_number, ) ) - except (OSError, ValueError, RuntimeError, json.JSONDecodeError): - trial_results.append( - { - "trial_number": trial_number, - "passed": False, - } - ) + except (OSError, ValueError, RuntimeError, json.JSONDecodeError) as exc: + failed_trial: dict[str, Any] = { + "trial_number": trial_number, + "passed": False, + } + failed_trial.update(_failure_evidence(exc)) + trial_results.append(failed_trial) successful_trials = sum( 1 for trial in trial_results if trial.get("passed") is True diff --git a/tests/fixtures/mv3_basic/content_script.js b/tests/fixtures/mv3_basic/content_script.js index 8b6af5314..b70d1a27f 100644 --- a/tests/fixtures/mv3_basic/content_script.js +++ b/tests/fixtures/mv3_basic/content_script.js @@ -27,4 +27,7 @@ document.documentElement.dataset.originweaveSidePanel = response?.sidePanel ?? "missing"; document.documentElement.dataset.originweaveBookmarks = response?.bookmarks ?? "missing"; document.documentElement.dataset.originweaveHistory = response?.history ?? "missing"; + document.documentElement.dataset.originweaveDownloads = response?.downloads ?? "missing"; + document.documentElement.dataset.originweaveDownloadsDiagnostic = + response?.downloadsDiagnostic ?? "download-not-evaluated"; })(); diff --git a/tests/fixtures/mv3_basic/download.txt b/tests/fixtures/mv3_basic/download.txt new file mode 100644 index 000000000..c6cde1c6a --- /dev/null +++ b/tests/fixtures/mv3_basic/download.txt @@ -0,0 +1 @@ +OriginWeave deterministic MV3 download fixture. diff --git a/tests/fixtures/mv3_basic/manifest.json b/tests/fixtures/mv3_basic/manifest.json index f366329ac..960fee780 100644 --- a/tests/fixtures/mv3_basic/manifest.json +++ b/tests/fixtures/mv3_basic/manifest.json @@ -11,7 +11,8 @@ "scripting", "sidePanel", "bookmarks", - "history" + "history", + "downloads" ], "host_permissions": [ "http://127.0.0.1/*" diff --git a/tests/fixtures/mv3_basic/service_worker.js b/tests/fixtures/mv3_basic/service_worker.js index 70687838a..33ac59efd 100644 --- a/tests/fixtures/mv3_basic/service_worker.js +++ b/tests/fixtures/mv3_basic/service_worker.js @@ -1,5 +1,9 @@ "use strict"; +const DOWNLOAD_PAYLOAD = "OriginWeave deterministic MV3 download fixture.\n"; +const DOWNLOAD_POLL_ATTEMPTS = 100; +const DOWNLOAD_POLL_INTERVAL_MS = 50; + const workerStartPromise = (async () => { const values = await chrome.storage.local.get("originweave_worker_start_count"); const previous = Number(values.originweave_worker_start_count ?? 0); @@ -16,6 +20,85 @@ async function ensureWorkerState() { return "installed"; } +async function waitForDownload(downloadId, expectedUrl) { + const expectedBytes = new TextEncoder().encode(DOWNLOAD_PAYLOAD).byteLength; + let observedDownload = false; + for (let attempt = 0; attempt < DOWNLOAD_POLL_ATTEMPTS; attempt += 1) { + let items; + try { + items = await chrome.downloads.search({ id: downloadId, limit: 1 }); + } catch (_error) { + return { ready: false, diagnostic: "download-search-missing" }; + } + if (!Array.isArray(items) || items.length !== 1) { + await new Promise((resolve) => setTimeout(resolve, DOWNLOAD_POLL_INTERVAL_MS)); + continue; + } + observedDownload = true; + const item = items[0]; + if (item.state === "interrupted") { + return { ready: false, diagnostic: "download-interrupted" }; + } + if (item.state === "complete") { + if (item.url !== expectedUrl) { + return { ready: false, diagnostic: "download-url-mismatch" }; + } + if (item.bytesReceived !== expectedBytes || item.totalBytes !== expectedBytes) { + return { ready: false, diagnostic: "download-byte-count-mismatch" }; + } + if (item.exists === false) { + return { ready: false, diagnostic: "download-exists-false" }; + } + return { ready: true, diagnostic: "download-complete-ready" }; + } + await new Promise((resolve) => setTimeout(resolve, DOWNLOAD_POLL_INTERVAL_MS)); + } + return { + ready: false, + diagnostic: observedDownload ? "download-timeout" : "download-search-missing", + }; +} + +async function exerciseDownload(sender) { + const sourceUrl = sender?.tab?.url; + if (typeof sourceUrl !== "string") { + return { ready: false, diagnostic: "download-source-rejected" }; + } + + let parsed; + try { + parsed = new URL(sourceUrl); + } catch (_error) { + return { ready: false, diagnostic: "download-source-rejected" }; + } + if ( + parsed.protocol !== "http:" || + parsed.hostname !== "127.0.0.1" || + parsed.pathname !== "/page.html" || + parsed.username !== "" || + parsed.password !== "" + ) { + return { ready: false, diagnostic: "download-source-rejected" }; + } + + const url = new URL("download.txt", sourceUrl).href; + let downloadId; + try { + downloadId = await chrome.downloads.download({ + url, + filename: "originweave-mv3/download.txt", + conflictAction: "uniquify", + saveAs: false, + }); + } catch (_error) { + return { ready: false, diagnostic: "download-start-rejected" }; + } + if (!Number.isInteger(downloadId)) { + return { ready: false, diagnostic: "download-start-rejected" }; + } + return waitForDownload(downloadId, url); +} + async function exerciseCoreApis(sender) { const tabId = sender?.tab?.id; if (!Number.isInteger(tabId)) { @@ -56,6 +139,9 @@ async function exerciseCoreApis(sender) { }); const historyReady = Array.isArray(historyItems); + const downloadResult = await exerciseDownload(sender); + const downloadsReady = downloadResult.ready; + return { tabs: tabReady ? "ready" : "missing", windows: windowReady ? "ready" : "missing", @@ -64,6 +150,8 @@ async function exerciseCoreApis(sender) { sidePanel: sidePanelReady ? "ready" : "missing", bookmarks: bookmarksReady ? "ready" : "missing", history: historyReady ? "ready" : "missing", + downloads: downloadsReady ? "ready" : "missing", + downloadsDiagnostic: downloadResult.diagnostic, }; } @@ -91,6 +179,8 @@ chrome.runtime.onMessage.addListener((message, sender, sendResponse) => { sidePanel: "missing", bookmarks: "missing", history: "missing", + downloads: "missing", + downloadsDiagnostic: "download-not-evaluated", }); } ); diff --git a/tests/test_mv3_binary_authority_contract.py b/tests/test_mv3_binary_authority_contract.py new file mode 100644 index 000000000..68fedbf9f --- /dev/null +++ b/tests/test_mv3_binary_authority_contract.py @@ -0,0 +1,108 @@ +"""Security contract for pinned Manifest V3 browser executable authority.""" + +from __future__ import annotations + +import os +import pathlib +import runpy +import tempfile +import unittest +import unittest.mock + +ROOT = pathlib.Path(__file__).resolve().parents[1] +RUNNER = ROOT / "scripts" / "ci" / "run_mv3_compatibility.py" + + +class ManifestV3BinaryAuthorityContractTests(unittest.TestCase): + """Prevent environment variables from selecting arbitrary executable code.""" + + def setUp(self) -> None: + """Load the production runner without executing its command-line entrypoint.""" + + self.namespace = runpy.run_path(str(RUNNER), run_name="mv3_binary_authority") + self.validate = self.namespace["_pinned_workspace_binary"] + + @staticmethod + def _make_executable(path: pathlib.Path) -> None: + """Create one inert executable fixture without ever executing it.""" + + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") + path.chmod(0o755) + + def test_untrusted_environment_override_is_rejected_before_execution(self) -> None: + """An existing executable outside the pinned workspace path must fail closed.""" + + with tempfile.TemporaryDirectory(prefix="originweave-binary-authority-") as temp_dir: + root = pathlib.Path(temp_dir) + expected = root / ".mv3-browser" / "chromedriver-linux64" / "chromedriver" + attacker = root / "attacker-controlled" / "chromedriver" + self._make_executable(expected) + self._make_executable(attacker) + + with unittest.mock.patch.dict( + os.environ, + {"CHROMEDRIVER_BIN": str(attacker)}, + clear=False, + ): + with self.assertRaisesRegex(SystemExit, "pinned workspace executable"): + self.validate( + "CHROMEDRIVER_BIN", + pathlib.PurePosixPath( + ".mv3-browser/chromedriver-linux64/chromedriver" + ), + "ChromeDriver", + root=root, + ) + + def test_exact_pinned_workspace_executable_is_accepted(self) -> None: + """The exact executable provisioned by the pinned workflow remains usable.""" + + with tempfile.TemporaryDirectory(prefix="originweave-binary-authority-") as temp_dir: + root = pathlib.Path(temp_dir) + expected = root / ".mv3-browser" / "chrome-linux64" / "chrome" + self._make_executable(expected) + + with unittest.mock.patch.dict( + os.environ, + {"CHROME_BIN": str(expected)}, + clear=False, + ): + actual = self.validate( + "CHROME_BIN", + pathlib.PurePosixPath(".mv3-browser/chrome-linux64/chrome"), + "Chrome for Testing", + root=root, + ) + + self.assertEqual(actual, expected) + + def test_symlink_at_pinned_executable_path_is_rejected(self) -> None: + """A matching pathname must not authorize a symlink to foreign executable code.""" + + with tempfile.TemporaryDirectory(prefix="originweave-binary-authority-") as temp_dir: + root = pathlib.Path(temp_dir) + expected = root / ".mv3-browser" / "chromedriver-linux64" / "chromedriver" + attacker = root / "attacker-controlled" / "chromedriver" + self._make_executable(attacker) + expected.parent.mkdir(parents=True, exist_ok=True) + expected.symlink_to(attacker) + + with unittest.mock.patch.dict( + os.environ, + {"CHROMEDRIVER_BIN": str(expected)}, + clear=False, + ): + with self.assertRaisesRegex(SystemExit, "symlink"): + self.validate( + "CHROMEDRIVER_BIN", + pathlib.PurePosixPath( + ".mv3-browser/chromedriver-linux64/chromedriver" + ), + "ChromeDriver", + root=root, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_mv3_click_diagnostic_contract.py b/tests/test_mv3_click_diagnostic_contract.py new file mode 100644 index 000000000..e0e28931a --- /dev/null +++ b/tests/test_mv3_click_diagnostic_contract.py @@ -0,0 +1,48 @@ +"""Fail-closed contract for real-click diagnostic handling in the MV3 runner.""" + +from __future__ import annotations + +import pathlib +import runpy +import unittest +import unittest.mock + +ROOT = pathlib.Path(__file__).resolve().parents[1] +RUNNER = ROOT / "scripts" / "ci" / "run_mv3_compatibility.py" + + +class ManifestV3ClickDiagnosticContractTests(unittest.TestCase): + """Keep browser-controlled click postconditions out of exception text.""" + + def test_click_mismatch_does_not_retain_raw_browser_text(self) -> None: + """A failed click must classify the mismatch without copying page-controlled text.""" + + namespace = runpy.run_path(str(RUNNER), run_name="mv3_click_contract") + exercise = namespace["_exercise_real_click"] + element_key = namespace["W3C_ELEMENT_KEY"] + raw_text = "secret-token /home/runner/private https://example.invalid" + responses = iter( + ( + {"value": {element_key: "f.1.d.2.e.3"}}, + {"value": {}}, + {"value": {element_key: "f.4.d.5.e.6"}}, + {"value": raw_text}, + ) + ) + + with unittest.mock.patch.dict( + exercise.__globals__, + {"_json_request": unittest.mock.Mock(side_effect=lambda *_a, **_k: next(responses))}, + ): + with self.assertRaises(RuntimeError) as raised: + exercise(9515, "session.1") + + rendered = str(raised.exception) + self.assertEqual(rendered, "real click post-condition mismatch") + self.assertNotIn("secret-token", rendered) + self.assertNotIn("/home/runner/private", rendered) + self.assertNotIn("example.invalid", rendered) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_mv3_compatibility_contract.py b/tests/test_mv3_compatibility_contract.py index 10872ddac..afa0d8511 100644 --- a/tests/test_mv3_compatibility_contract.py +++ b/tests/test_mv3_compatibility_contract.py @@ -6,6 +6,7 @@ import pathlib import runpy import unittest +import unittest.mock ROOT = pathlib.Path(__file__).resolve().parents[1] FIXTURE = ROOT / "tests" / "fixtures" / "mv3_basic" @@ -162,6 +163,110 @@ def test_runner_reports_repeated_trial_pass_rate(self) -> None: with self.subTest(expected=expected): self.assertIn(expected, runner) + def test_runner_preserves_safe_surface_failure_evidence(self) -> None: + """A failed trial must identify the bounded fixture surface without leaking raw errors.""" + + namespace = runpy.run_path(str(RUNNER), run_name="mv3_contract") + surface_error = namespace["CompatibilitySurfaceError"] + failure_evidence = namespace["_failure_evidence"] + + observed = {"downloads": "missing", "storage": "ready"} + diagnostic = failure_evidence(surface_error(observed)) + self.assertEqual(diagnostic["failure_kind"], "surface_mismatch") + self.assertEqual(diagnostic["observed"], observed) + + generic = failure_evidence( + RuntimeError("secret-token https://example.invalid /home/runner/private") + ) + self.assertEqual(generic, {"failure_kind": "runtime_error"}) + + def test_webdriver_errors_do_not_retain_raw_response_payloads(self) -> None: + """WebDriver protocol failures must stay useful without copying raw browser text.""" + + namespace = runpy.run_path(str(RUNNER), run_name="mv3_contract") + json_request = namespace["_json_request"] + http_module = namespace["http"] + + class FakeResponse: + def __init__(self, status: int, body: bytes) -> None: + self.status = status + self.body = body + + def read(self, _limit: int) -> bytes: + return self.body + + class FakeConnection: + def __init__(self, response: FakeResponse) -> None: + self.response = response + + def request(self, *_args: object, **_kwargs: object) -> None: + return None + + def getresponse(self) -> FakeResponse: + return self.response + + def close(self) -> None: + return None + + raw_secret = "secret-token /home/runner/private https://example.invalid" + cases = ( + FakeResponse(500, raw_secret.encode("utf-8")), + FakeResponse( + 200, + json.dumps( + { + "value": { + "error": "unknown error", + "message": raw_secret, + } + } + ).encode("utf-8"), + ), + ) + for response in cases: + with self.subTest(status=response.status): + with unittest.mock.patch.object( + http_module.client, + "HTTPConnection", + return_value=FakeConnection(response), + ): + with self.assertRaises(RuntimeError) as raised: + json_request(9515, "GET", "/status") + rendered = str(raised.exception) + self.assertNotIn("secret-token", rendered) + self.assertNotIn("/home/runner/private", rendered) + self.assertNotIn("example.invalid", rendered) + + def test_chromedriver_startup_timeout_does_not_retain_raw_last_error(self) -> None: + """Startup timeout diagnostics must classify transient errors without copying raw text.""" + + namespace = runpy.run_path(str(RUNNER), run_name="mv3_contract") + wait_for_driver = namespace["_wait_for_driver"] + time_module = namespace["time"] + raw_error = "secret-token /home/runner/private https://example.invalid" + + with ( + unittest.mock.patch.dict( + wait_for_driver.__globals__, + {"_json_request": unittest.mock.Mock(side_effect=OSError(raw_error))}, + ), + unittest.mock.patch.object( + time_module, + "monotonic", + side_effect=(0.0, 0.0, 99.0), + ), + unittest.mock.patch.object(time_module, "sleep", return_value=None), + ): + with self.assertRaises(RuntimeError) as raised: + wait_for_driver(9515) + + rendered = str(raised.exception) + self.assertIn("ChromeDriver did not become ready", rendered) + self.assertIn("io_error", rendered) + self.assertNotIn("secret-token", rendered) + self.assertNotIn("/home/runner/private", rendered) + self.assertNotIn("example.invalid", rendered) + def test_workflow_runs_the_real_browser_lane_without_model_credentials(self) -> None: """Compatibility evidence must execute Chromium and never require LLM secrets.""" @@ -193,6 +298,8 @@ def test_doctoring_records_primary_chromium_evidence(self) -> None: "not claim 100% Chrome extension compatibility", "Chrome for Developers", "Google Chrome Labs", + "chrome.downloads", + "https://developer.chrome.com/docs/extensions/reference/api/downloads", ): with self.subTest(expected=expected): self.assertIn(expected, doctoring) diff --git a/tests/test_mv3_downloads_contract.py b/tests/test_mv3_downloads_contract.py new file mode 100644 index 000000000..53218b965 --- /dev/null +++ b/tests/test_mv3_downloads_contract.py @@ -0,0 +1,155 @@ +"""Fail-first contract for real Manifest V3 downloads compatibility.""" + +from __future__ import annotations + +import importlib.util +import json +import pathlib +import unittest + +ROOT = pathlib.Path(__file__).resolve().parents[1] +FIXTURE = ROOT / "tests" / "fixtures" / "mv3_basic" +RUNNER = ROOT / "scripts" / "ci" / "run_mv3_compatibility.py" + + +def _load_runner_module(): + """Load the compatibility runner without invoking its command-line entry point.""" + + spec = importlib.util.spec_from_file_location("originweave_mv3_runner", RUNNER) + if spec is None or spec.loader is None: + raise AssertionError("unable to load the MV3 compatibility runner") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +class ManifestV3DownloadsContractTests(unittest.TestCase): + """Require the real Chrome downloads API in every pinned-browser trial.""" + + def test_fixture_declares_downloads_permission_and_local_resource(self) -> None: + """The controlled extension must request downloads and serve its test payload locally.""" + + manifest = json.loads((FIXTURE / "manifest.json").read_text(encoding="utf-8")) + self.assertIn("downloads", manifest["permissions"]) + payload = (FIXTURE / "download.txt").read_bytes() + self.assertEqual(payload, b"OriginWeave deterministic MV3 download fixture.\n") + + def test_service_worker_executes_and_verifies_a_real_loopback_download(self) -> None: + """Evidence must originate from the controlled fixture origin and bounded inspection.""" + + worker = (FIXTURE / "service_worker.js").read_text(encoding="utf-8") + for expected in ( + "chrome.downloads.download", + "chrome.downloads.search", + 'new URL("download.txt", sourceUrl).href', + 'parsed.hostname !== "127.0.0.1"', + 'parsed.protocol !== "http:"', + "downloadsReady", + ): + with self.subTest(expected=expected): + self.assertIn(expected, worker) + self.assertNotIn('chrome.runtime.getURL("download.txt")', worker) + + def test_restart_pair_never_overwrites_the_previous_controlled_download(self) -> None: + """Restart evidence must not race Chrome while replacing the first pass's file.""" + + worker = (FIXTURE / "service_worker.js").read_text(encoding="utf-8") + self.assertIn('conflictAction: "uniquify"', worker) + self.assertNotIn('conflictAction: "overwrite"', worker) + + def test_download_failures_emit_only_bounded_stage_diagnostics(self) -> None: + """Fixture diagnostics must name a reviewed stage without retaining raw browser errors.""" + + worker = (FIXTURE / "service_worker.js").read_text(encoding="utf-8") + content = (FIXTURE / "content_script.js").read_text(encoding="utf-8") + for expected in ( + "download-source-rejected", + "download-start-rejected", + "download-search-missing", + "download-interrupted", + "download-url-mismatch", + "download-byte-count-mismatch", + "download-exists-false", + "download-timeout", + "download-complete-ready", + "downloadsDiagnostic", + ): + with self.subTest(expected=expected): + self.assertIn(expected, worker) + self.assertIn("originweaveDownloadsDiagnostic", content) + self.assertNotIn("download.default_directory", worker) + self.assertNotIn("item.filename", worker) + self.assertNotIn("_error.message", worker) + self.assertNotIn("String(_error)", worker) + + def test_content_script_and_runner_require_downloads_on_every_pass(self) -> None: + """The compatibility report must fail closed when downloads evidence is missing.""" + + content = (FIXTURE / "content_script.js").read_text(encoding="utf-8") + runner = RUNNER.read_text(encoding="utf-8") + self.assertIn("originweaveDownloads", content) + self.assertIn('"downloads": surfaces["downloads"] == "ready"', runner) + self.assertIn('"downloads": "ready"', runner) + + def test_runner_preserves_only_reviewed_download_diagnostic_tokens(self) -> None: + """Runner failure evidence must retain stage tokens while rejecting raw diagnostics.""" + + runner = _load_runner_module() + approved = { + "download-source-rejected", + "download-start-rejected", + "download-search-missing", + "download-interrupted", + "download-url-mismatch", + "download-byte-count-mismatch", + "download-exists-false", + "download-timeout", + "download-complete-ready", + "download-not-evaluated", + } + self.assertIn("downloadsDiagnostic", runner.SURFACE_EVIDENCE_KEYS) + self.assertEqual(runner.DOWNLOAD_DIAGNOSTIC_VALUES, frozenset(approved)) + for token in approved: + with self.subTest(token=token): + self.assertEqual( + runner._safe_surface_value("downloadsDiagnostic", token), token + ) + + approved_error = runner.CompatibilitySurfaceError( + { + "downloads": "missing", + "downloadsDiagnostic": "download-source-rejected", + } + ) + approved_evidence = runner._failure_evidence(approved_error) + self.assertEqual( + approved_evidence["observed"]["downloadsDiagnostic"], + "download-source-rejected", + ) + + raw_download_path = str(ROOT / "private" / "download.txt") + raw_browser_error = "Error: secret browser failure" + for raw in (raw_download_path, raw_browser_error): + with self.subTest(raw=raw): + error = runner.CompatibilitySurfaceError( + { + "downloads": "missing", + "downloadsDiagnostic": raw, + } + ) + evidence = runner._failure_evidence(error) + self.assertEqual( + evidence["observed"]["downloadsDiagnostic"], "unexpected" + ) + self.assertNotIn(raw, repr(evidence)) + + def test_runner_collects_download_diagnostic_from_fixture_dataset(self) -> None: + """The WebDriver evidence script must collect the bounded fixture diagnostic field.""" + + runner = RUNNER.read_text(encoding="utf-8") + self.assertIn("originweaveDownloadsDiagnostic", runner) + self.assertIn('"downloadsDiagnostic": "download-complete-ready"', runner) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_mv3_session_cleanup_exception_contract.py b/tests/test_mv3_session_cleanup_exception_contract.py new file mode 100644 index 000000000..6ba9e3301 --- /dev/null +++ b/tests/test_mv3_session_cleanup_exception_contract.py @@ -0,0 +1,210 @@ +"""Regression contract for fail-closed WebDriver session cleanup.""" + +from __future__ import annotations + +import pathlib +import runpy +import subprocess +import tempfile +import unittest +import unittest.mock + +ROOT = pathlib.Path(__file__).resolve().parents[1] +RUNNER = ROOT / "scripts" / "ci" / "run_mv3_compatibility.py" + + +class _UnexpectedCleanupFailure(Exception): + """Model an unreviewed programming/integration failure during session deletion.""" + + +class _FakeDriver: + """Record process cleanup without launching ChromeDriver.""" + + def __init__( + self, + *, + terminate_error: OSError | None = None, + kill_error: OSError | None = None, + wait_timeout_once: bool = False, + ) -> None: + self.terminated = False + self.killed = False + self.terminate_error = terminate_error + self.kill_error = kill_error + self.wait_timeout_once = wait_timeout_once + self.wait_calls = 0 + + def terminate(self) -> None: + """Record the graceful process-termination fallback.""" + + self.terminated = True + if self.terminate_error is not None: + raise self.terminate_error + + def kill(self) -> None: + """Record the bounded hard-kill fallback when requested.""" + + self.killed = True + if self.kill_error is not None: + raise self.kill_error + + def wait(self, timeout: float) -> int: + """Model either an immediately reaped process or one bounded timeout.""" + + if timeout <= 0: + raise AssertionError("timeout must remain positive") + self.wait_calls += 1 + if self.wait_timeout_once and self.wait_calls == 1: + raise subprocess.TimeoutExpired("controlled-chromedriver", timeout) + return 0 + + +class ManifestV3SessionCleanupExceptionTests(unittest.TestCase): + """Unexpected cleanup failures must remain visible after process teardown.""" + + @staticmethod + def _surfaces() -> dict[str, str]: + """Return one fully passing controlled compatibility surface set.""" + + return { + "workerStartCount": "1", + "storagePersistence": "initialized", + "workerReply": "pong", + "content": "ready", + "storage": "ready", + "dnr": "blocked", + "tabs": "ready", + "windows": "ready", + "scripting": "ready", + "scriptingExecuted": "ready", + "commands": "ready", + "sidePanel": "ready", + "bookmarks": "ready", + "history": "ready", + "downloads": "ready", + } + + def _run_with_cleanup_failure( + self, + cleanup_failure: Exception, + fake_driver: _FakeDriver, + ) -> tuple[object, object]: + """Run the production browser-pass boundary with controlled cleanup failures.""" + + namespace = runpy.run_path(str(RUNNER), run_name="mv3_cleanup_contract") + run_browser_pass = namespace["_run_browser_pass"] + globals_ = run_browser_pass.__globals__ + + def fake_json_request( + _driver_port: int, + method: str, + path: str, + _payload=None, + *, + timeout: float = 5.0, + ): + if timeout <= 0: + raise AssertionError("timeout must remain positive") + if method == "POST" and path == "/session": + return { + "value": { + "sessionId": "session-1", + "capabilities": { + "browserVersion": namespace["PINNED_CHROME_VERSION"] + }, + } + } + if method == "POST" and path.endswith("/url"): + return {"value": None} + if method == "DELETE" and path.endswith("/session/session-1"): + raise cleanup_failure + raise AssertionError(f"unexpected WebDriver request: {method} {path}") + + with tempfile.TemporaryDirectory(prefix="originweave-cleanup-contract-") as profile_dir: + with ( + unittest.mock.patch.object( + globals_["subprocess"], "Popen", return_value=fake_driver + ), + unittest.mock.patch.dict( + globals_, + { + "_free_loopback_port": lambda: 43123, + "_wait_for_driver": lambda _port: None, + "_json_request": fake_json_request, + "_wait_for_extension_evidence": ( + lambda _port, _session, _expected: self._surfaces() + ), + "_exercise_real_click": lambda _port, _session: "clicked", + }, + ), + ): + try: + run_browser_pass( + pathlib.Path("/controlled/chrome"), + pathlib.Path("/controlled/chromedriver"), + "http://127.0.0.1:8080/page.html", + profile_dir, + "initialized", + ) + except Exception as error: # noqa: BLE001 - return exact boundary error. + return namespace, error + self.fail("cleanup failure unexpectedly became success") + + def test_unreviewed_session_cleanup_exception_is_not_silently_suppressed(self) -> None: + """A new exception class must propagate while ChromeDriver is still terminated.""" + + fake_driver = _FakeDriver() + expected = _UnexpectedCleanupFailure("must not be normalized") + _namespace, error = self._run_with_cleanup_failure(expected, fake_driver) + + self.assertIs(error, expected) + self.assertTrue(fake_driver.terminated) + self.assertFalse(fake_driver.killed) + + def test_reviewed_session_cleanup_error_survives_teardown_failure(self) -> None: + """The causal session failure must not be replaced by a later terminate error.""" + + fake_driver = _FakeDriver(terminate_error=OSError("terminate failed")) + session_error = RuntimeError("session delete failed") + namespace, error = self._run_with_cleanup_failure(session_error, fake_driver) + + self.assertIsInstance(error, namespace["WebDriverSessionCleanupError"]) + self.assertIs(error.__cause__, session_error) + self.assertTrue(fake_driver.terminated) + self.assertTrue(fake_driver.killed) + + def test_successful_kill_after_wait_timeout_is_normal_cleanup(self) -> None: + """A bounded wait timeout must remain a successful fallback when kill reaps the process.""" + + namespace = runpy.run_path(str(RUNNER), run_name="mv3_teardown_contract") + fake_driver = _FakeDriver(wait_timeout_once=True) + + error = namespace["_teardown_driver_process"](fake_driver) + + self.assertIsNone(error) + self.assertTrue(fake_driver.terminated) + self.assertTrue(fake_driver.killed) + self.assertEqual(fake_driver.wait_calls, 2) + + def test_failed_kill_fallback_is_recorded_on_the_primary_teardown_error(self) -> None: + """A secondary fallback failure must not disappear while the first error stays causal.""" + + namespace = runpy.run_path(str(RUNNER), run_name="mv3_teardown_contract") + terminate_error = OSError("terminate failed") + fake_driver = _FakeDriver( + terminate_error=terminate_error, + kill_error=PermissionError("kill denied"), + ) + + error = namespace["_teardown_driver_process"](fake_driver) + + self.assertIs(error, terminate_error) + self.assertTrue(fake_driver.killed) + self.assertIn( + "bounded ChromeDriver kill fallback also failed: PermissionError", + getattr(error, "__notes__", []), + ) + + +if __name__ == "__main__": + unittest.main()