diff --git a/scripts/ci/run_mv3_compatibility.py b/scripts/ci/run_mv3_compatibility.py index 28a3fb1e..131edfd0 100755 --- a/scripts/ci/run_mv3_compatibility.py +++ b/scripts/ci/run_mv3_compatibility.py @@ -1,12 +1,15 @@ #!/usr/bin/env python3 -"""Run bounded repeatable Manifest V3 compatibility evidence against pinned Chromium. +"""Run bounded repeatable real-browser evidence against pinned Chromium. This is a release/CI evidence runner, not a product browser adapter. It uses the 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 -restart-persistence behavior. +restart-persistence behavior. It also executes the controlled Agent Task fixture +with extensions disabled in a fresh profile, performs real WebDriver input and +click operations, verifies the observable post-condition, and proves profile +cleanup without treating page content as instruction or authority. """ from __future__ import annotations @@ -27,9 +30,12 @@ ROOT = pathlib.Path(__file__).resolve().parents[2] FIXTURE = ROOT / "tests" / "fixtures" / "mv3_basic" +AGENT_TASK_FIXTURE = ROOT / "tests/fixtures/agent_task_basic" PINNED_CHROME_VERSION = "150.0.7871.129" PINNED_CHROME_REVISION = "r1639810" REPEATABILITY_TRIALS = 3 +AGENT_TASK_REPEATABILITY_TRIALS = 3 +AGENT_TASK_INPUT_VALUE = "originweave controlled input" REQUEST_TIMEOUT_SECONDS = 5.0 STARTUP_TIMEOUT_SECONDS = 20.0 FIXTURE_TIMEOUT_SECONDS = 20.0 @@ -150,6 +156,29 @@ def _execute(driver_port: int, session_id: str, script: str) -> Any: return response.get("value") +def _find_element(driver_port: int, session_id: str, selector: str) -> str: + """Find one fixture element and return its validated ChromeDriver identifier.""" + + found = _json_request( + driver_port, + "POST", + _webdriver_path(session_id, "/element"), + {"using": "css selector", "value": selector}, + ) + element = found.get("value", {}) + element_id = element.get(W3C_ELEMENT_KEY) if isinstance(element, dict) else None + if not isinstance(element_id, str): + raise RuntimeError("WebDriver did not return a W3C element identifier") + return _path_token(element_id, "element identifier") + + +def _element_command_path(session_id: str, element_id: str, suffix: str) -> str: + """Build a bounded WebDriver element command path from validated identifiers.""" + + safe_element = _path_token(element_id, "element identifier") + return _webdriver_path(session_id, f"/element/{safe_element}{suffix}") + + def _wait_for_extension_evidence( driver_port: int, session_id: str, @@ -220,37 +249,18 @@ def _wait_for_extension_evidence( def _exercise_real_click(driver_port: int, session_id: str) -> str: """Use the WebDriver element-click command and verify the DOM post-condition.""" - found = _json_request( - driver_port, - "POST", - _webdriver_path(session_id, "/element"), - {"using": "css selector", "value": "#fixture-button"}, - ) - element = found.get("value", {}) - element_id = element.get(W3C_ELEMENT_KEY) if isinstance(element, dict) else None - if not isinstance(element_id, str): - raise RuntimeError("WebDriver did not return a W3C element identifier") - safe_element = _path_token(element_id, "element identifier") + safe_element = _find_element(driver_port, session_id, "#fixture-button") _json_request( driver_port, "POST", - _webdriver_path(session_id, f"/element/{safe_element}/click"), + _element_command_path(session_id, safe_element, "/click"), {}, ) - output = _json_request( - driver_port, - "POST", - _webdriver_path(session_id, "/element"), - {"using": "css selector", "value": "#fixture-output"}, - ).get("value", {}) - output_id = output.get(W3C_ELEMENT_KEY) if isinstance(output, dict) else None - if not isinstance(output_id, str): - raise RuntimeError("WebDriver did not return the fixture output element") - safe_output = _path_token(output_id, "element identifier") + safe_output = _find_element(driver_port, session_id, "#fixture-output") text = _json_request( driver_port, "GET", - _webdriver_path(session_id, f"/element/{safe_output}/text"), + _element_command_path(session_id, safe_output, "/text"), ).get("value") if text != "clicked": raise RuntimeError(f"real click post-condition failed: {text!r}") @@ -433,8 +443,220 @@ def _run_restart_trial( } +def _run_agent_task_browser_pass( + chrome_bin: pathlib.Path, + chromedriver_bin: pathlib.Path, + fixture_url: str, + profile_dir: str, +) -> dict[str, Any]: + """Execute one synthetic Agent Task through real WebDriver input in pinned Chrome.""" + + started = time.monotonic() + driver_port = _free_loopback_port() + session_id: str | None = None + driver = subprocess.Popen( + [str(chromedriver_bin), f"--port={driver_port}", "--allowed-ips=127.0.0.1"], + stdout=subprocess.DEVNULL, + stderr=subprocess.STDOUT, + text=True, + ) + try: + _wait_for_driver(driver_port) + session = _json_request( + driver_port, + "POST", + "/session", + { + "capabilities": { + "alwaysMatch": { + "browserName": "chrome", + "goog:chromeOptions": { + "binary": str(chrome_bin), + "args": [ + "--headless=new", + "--no-first-run", + "--disable-default-apps", + "--disable-component-update", + "--disable-sync", + "--disable-dev-shm-usage", + "--no-sandbox", + "--disable-extensions", + f"--user-data-dir={profile_dir}", + ], + }, + } + } + }, + ).get("value", {}) + if not isinstance(session, dict): + raise RuntimeError("ChromeDriver Agent Task session response is malformed") + raw_session_id = session.get("sessionId") + capabilities = session.get("capabilities", {}) + if not isinstance(raw_session_id, str): + raise RuntimeError("ChromeDriver did not return an Agent Task session id") + session_id = _path_token(raw_session_id, "session identifier") + browser_version = ( + capabilities.get("browserVersion") if isinstance(capabilities, dict) else None + ) + if browser_version != PINNED_CHROME_VERSION: + raise RuntimeError( + f"unexpected Agent Task Chrome version: expected {PINNED_CHROME_VERSION}, " + f"got {browser_version!r}" + ) + + _json_request( + driver_port, + "POST", + _webdriver_path(session_id, "/url"), + {"url": fixture_url}, + ) + initial_url = _json_request( + driver_port, + "GET", + _webdriver_path(session_id, "/url"), + ).get("value") + if initial_url != fixture_url: + raise RuntimeError( + f"Agent Task initial URL mismatch: expected {fixture_url!r}, got {initial_url!r}" + ) + input_element = _find_element(driver_port, session_id, "#task-text") + _json_request( + driver_port, + "POST", + _element_command_path(session_id, input_element, "/clear"), + {}, + ) + _json_request( + driver_port, + "POST", + _element_command_path(session_id, input_element, "/value"), + {"text": AGENT_TASK_INPUT_VALUE, "value": list(AGENT_TASK_INPUT_VALUE)}, + ) + submit_element = _find_element( + driver_port, + session_id, + "#agent-task-form button[type=submit]", + ) + _json_request( + driver_port, + "POST", + _element_command_path(session_id, submit_element, "/click"), + {}, + ) + post_submit_url = _json_request( + driver_port, + "GET", + _webdriver_path(session_id, "/url"), + ).get("value") + url_unchanged = post_submit_url == initial_url + if not url_unchanged: + raise RuntimeError("Agent Task URL changed during submission") + result_element = _find_element(driver_port, session_id, "#task-result") + state = _json_request( + driver_port, + "GET", + _element_command_path(session_id, result_element, "/attribute/data-state"), + ).get("value") + text = _json_request( + driver_port, + "GET", + _element_command_path(session_id, result_element, "/text"), + ).get("value") + if state != "submitted": + raise RuntimeError(f"Agent Task state post-condition failed: {state!r}") + if text != AGENT_TASK_INPUT_VALUE: + raise RuntimeError("Agent Task result did not match the synthetic typed value") + return { + "browser_version": browser_version, + "post_condition": True, + "input_echo_verified": True, + "url_unchanged": url_unchanged, + "extensions_disabled": True, + "duration_ms": round((time.monotonic() - started) * 1000), + } + finally: + if session_id is not None: + with contextlib.suppress(Exception): + _json_request( + driver_port, + "DELETE", + _webdriver_path(session_id, ""), + {}, + ) + driver.terminate() + try: + driver.wait(timeout=5) + except subprocess.TimeoutExpired: + driver.kill() + driver.wait(timeout=5) + + +def _run_agent_task_trial( + chrome_bin: pathlib.Path, + chromedriver_bin: pathlib.Path, + fixture_url: str, + trial_number: int, +) -> dict[str, Any]: + """Run one isolated Agent Task browser trial and prove its profile is removed.""" + + trial_started = time.monotonic() + profile_path: pathlib.Path + with tempfile.TemporaryDirectory( + prefix=f"originweave-agent-task-trial-{trial_number}-" + ) as profile_dir: + profile_path = pathlib.Path(profile_dir) + result = _run_agent_task_browser_pass( + chrome_bin, + chromedriver_bin, + fixture_url, + profile_dir, + ) + profile_cleaned = not profile_path.exists() + if not profile_cleaned: + raise RuntimeError(f"Agent Task profile cleanup failed in trial {trial_number}") + + return { + "trial_number": trial_number, + "passed": True, + "browser_version": result["browser_version"], + "post_condition": result["post_condition"], + "input_echo_verified": result["input_echo_verified"], + "url_unchanged": result["url_unchanged"], + "extensions_disabled": result["extensions_disabled"], + "profile_cleaned": profile_cleaned, + "duration_ms": round((time.monotonic() - trial_started) * 1000), + } + + +def _start_fixture_server(directory: pathlib.Path) -> tuple[http.server.ThreadingHTTPServer, threading.Thread]: + """Start one loopback-only static fixture server for a bounded browser lane.""" + + server = http.server.ThreadingHTTPServer( + ("127.0.0.1", 0), + lambda *args, **kwargs: QuietFixtureHandler( + *args, + directory=str(directory), + **kwargs, + ), + ) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + return server, thread + + +def _stop_fixture_server( + server: http.server.ThreadingHTTPServer, + thread: threading.Thread, +) -> None: + """Stop one bounded fixture server and join its helper thread.""" + + server.shutdown() + server.server_close() + thread.join(timeout=5) + + def main() -> int: - """Run three independent restart trials and emit bounded repeatability evidence.""" + """Run bounded MV3 and Agent Task trials and emit credential-free evidence.""" chrome_bin = pathlib.Path(os.environ.get("CHROME_BIN", "")) chromedriver_bin = pathlib.Path(os.environ.get("CHROMEDRIVER_BIN", "")) @@ -444,15 +666,11 @@ def main() -> int: raise SystemExit("CHROMEDRIVER_BIN must point to the matching pinned ChromeDriver") if not (FIXTURE / "manifest.json").is_file(): raise SystemExit("MV3 fixture manifest is missing") + if not (AGENT_TASK_FIXTURE / "index.html").is_file(): + raise SystemExit("Agent Task fixture is missing") - fixture_server = http.server.ThreadingHTTPServer( - ("127.0.0.1", 0), - lambda *args, **kwargs: QuietFixtureHandler( - *args, directory=str(FIXTURE), **kwargs - ), - ) - fixture_thread = threading.Thread(target=fixture_server.serve_forever, daemon=True) - fixture_thread.start() + fixture_server, fixture_thread = _start_fixture_server(FIXTURE) + agent_task_server, agent_task_thread = _start_fixture_server(AGENT_TASK_FIXTURE) started = time.monotonic() try: @@ -496,6 +714,44 @@ def main() -> int: for name in first_surfaces } + agent_task_url = ( + f"http://127.0.0.1:{agent_task_server.server_port}/index.html" + ) + agent_task_trials: list[dict[str, Any]] = [] + for trial_number in range(1, AGENT_TASK_REPEATABILITY_TRIALS + 1): + try: + agent_task_trials.append( + _run_agent_task_trial( + chrome_bin, + chromedriver_bin, + agent_task_url, + trial_number, + ) + ) + except (OSError, ValueError, RuntimeError, json.JSONDecodeError): + agent_task_trials.append( + { + "trial_number": trial_number, + "passed": False, + } + ) + + agent_task_successful_trials = sum( + 1 for trial in agent_task_trials if trial.get("passed") is True + ) + agent_task_trial_pass_rate = ( + agent_task_successful_trials / AGENT_TASK_REPEATABILITY_TRIALS + ) + agent_task_surfaces_complete = all( + trial.get("post_condition") is True + and trial.get("input_echo_verified") is True + and trial.get("url_unchanged") is True + and trial.get("extensions_disabled") is True + and trial.get("profile_cleaned") is True + for trial in agent_task_trials + if trial.get("passed") is True + ) + evidence = { "chrome_version": PINNED_CHROME_VERSION, "chrome_revision": PINNED_CHROME_REVISION, @@ -509,6 +765,12 @@ def main() -> int: if successful_results else [] ), + "agent_task": { + "repeatability_trials": AGENT_TASK_REPEATABILITY_TRIALS, + "successful_trials": agent_task_successful_trials, + "trial_pass_rate": agent_task_trial_pass_rate, + "trial_results": agent_task_trials, + }, "duration_ms": round((time.monotonic() - started) * 1000), } print(json.dumps(evidence, sort_keys=True)) @@ -519,11 +781,18 @@ def main() -> int: ) if not common_surfaces or not all(common_surfaces.values()): raise RuntimeError("Manifest V3 repeatability surfaces were incomplete") + if agent_task_successful_trials != AGENT_TASK_REPEATABILITY_TRIALS: + raise RuntimeError( + "Agent Task repeatability gate failed: " + f"{agent_task_successful_trials}/{AGENT_TASK_REPEATABILITY_TRIALS} " + "trials passed" + ) + if not agent_task_surfaces_complete: + raise RuntimeError("Agent Task repeatability surfaces were incomplete") return 0 finally: - fixture_server.shutdown() - fixture_server.server_close() - fixture_thread.join(timeout=5) + _stop_fixture_server(agent_task_server, agent_task_thread) + _stop_fixture_server(fixture_server, fixture_thread) if __name__ == "__main__": diff --git a/tests/test_agent_task_fixture_contract.py b/tests/test_agent_task_fixture_contract.py index 2a1465b2..2565a35c 100644 --- a/tests/test_agent_task_fixture_contract.py +++ b/tests/test_agent_task_fixture_contract.py @@ -10,6 +10,21 @@ FIXTURE = ROOT / "tests" / "fixtures" / "agent_task_basic" / "index.html" +def _is_credential_input(attributes: dict[str, str | None]) -> bool: + """Return whether parsed input attributes describe a credential surface.""" + + input_type = (attributes.get("type") or "").strip().lower() + if input_type == "password": + return True + + autocomplete = (attributes.get("autocomplete") or "").strip().lower() + autocomplete_tokens = autocomplete.split() + return any( + token == "one-time-code" or "password" in token + for token in autocomplete_tokens + ) + + class _FixtureParser(HTMLParser): """Collect the small semantic surface required by the deterministic fixture.""" @@ -18,6 +33,7 @@ def __init__(self) -> None: self.ids: set[str] = set() self.labels_for: set[str] = set() self.input_names: set[str] = set() + self.input_attributes: list[dict[str, str | None]] = [] self.button_types: set[str] = set() self.hidden_injection_markers = 0 @@ -30,12 +46,15 @@ def handle_starttag( self.ids.add(element_id) if tag == "label" and attributes.get("for"): self.labels_for.add(attributes["for"]) - if tag == "input" and attributes.get("name"): - self.input_names.add(attributes["name"]) + if tag == "input": + self.input_attributes.append(attributes) + if attributes.get("name"): + self.input_names.add(attributes["name"]) if tag == "button" and attributes.get("type"): self.button_types.add(attributes["type"]) if ( attributes.get("data-originweave-untrusted") == "prompt-injection" + and "hidden" in attributes and attributes.get("aria-hidden") == "true" ): self.hidden_injection_markers += 1 @@ -70,20 +89,49 @@ def test_fixture_contains_explicit_untrusted_hidden_prompt_injection(self) -> No self.assertIn("UNTRUSTED_PAGE_INSTRUCTION", self.html) self.assertIn("request new browser capabilities", self.html) + def test_hidden_injection_requires_the_actual_hidden_attribute(self) -> None: + """ARIA metadata alone must not satisfy the hidden-injection fixture contract.""" + + parser = _FixtureParser() + parser.feed( + "" + "" + ) + self.assertEqual(parser.hidden_injection_markers, 1) + def test_fixture_is_synthetic_and_has_no_credential_fields(self) -> None: """The controlled workflow must not require or imitate real secret collection.""" + for attributes in self.parser.input_attributes: + with self.subTest(attributes=attributes): + self.assertFalse(_is_credential_input(attributes)) + lowered = self.html.lower() - for forbidden in ( - 'type="password"', - 'autocomplete="current-password"', - 'autocomplete="one-time-code"', - "api_key", - "secret_key", - ): + for forbidden in ("api_key", "secret_key"): with self.subTest(forbidden=forbidden): self.assertNotIn(forbidden, lowered) + def test_credential_detection_is_quote_independent(self) -> None: + """Parsed credential semantics must reject single-quoted and tokenized forms.""" + + for html in ( + "", + "", + "", + "", + "", + ): + with self.subTest(html=html): + parser = _FixtureParser() + parser.feed(html) + self.assertEqual(len(parser.input_attributes), 1) + self.assertTrue(_is_credential_input(parser.input_attributes[0])) + + parser = _FixtureParser() + parser.feed("") + self.assertEqual(len(parser.input_attributes), 1) + self.assertFalse(_is_credential_input(parser.input_attributes[0])) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_agent_task_pinned_chrome_contract.py b/tests/test_agent_task_pinned_chrome_contract.py new file mode 100644 index 00000000..bf084090 --- /dev/null +++ b/tests/test_agent_task_pinned_chrome_contract.py @@ -0,0 +1,77 @@ +"""Contract for executing the controlled Agent Task fixture on pinned Chrome.""" + +from __future__ import annotations + +import pathlib +import runpy +import unittest + +ROOT = pathlib.Path(__file__).resolve().parents[1] +RUNNER = ROOT / "scripts" / "ci" / "run_mv3_compatibility.py" +FIXTURE = ROOT / "tests" / "fixtures" / "agent_task_basic" / "index.html" +WORKFLOW = ROOT / ".github" / "workflows" / "mv3-compatibility.yml" + + +class AgentTaskPinnedChromeContractTests(unittest.TestCase): + """Keep the first real-browser Agent Task evidence bounded and reproducible.""" + + def test_runner_exposes_a_separate_agent_task_browser_boundary(self) -> None: + """The pinned-browser runner must execute the controlled Agent Task fixture.""" + + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_contract") + for expected in ( + "AGENT_TASK_FIXTURE", + "AGENT_TASK_REPEATABILITY_TRIALS", + "_run_agent_task_browser_pass", + "_run_agent_task_trial", + ): + with self.subTest(expected=expected): + self.assertIn(expected, namespace) + + def test_agent_task_pass_uses_real_webdriver_input_and_post_condition(self) -> None: + """The evidence lane must type, click, and verify fixture state in real Chrome.""" + + runner = RUNNER.read_text(encoding="utf-8") + for expected in ( + "tests/fixtures/agent_task_basic", + '"--disable-extensions"', + '"/value"', + '"/click"', + '"/attribute/data-state"', + '"submitted"', + '"profile_cleaned"', + '"agent_task"', + '"trial_pass_rate"', + ): + with self.subTest(expected=expected): + self.assertIn(expected, runner) + + def test_agent_task_submission_preserves_the_loaded_url(self) -> None: + """Submission must prove that the controlled action did not navigate away.""" + + runner = RUNNER.read_text(encoding="utf-8") + for expected in ( + "initial_url", + "post_submit_url", + "url_unchanged", + ): + with self.subTest(expected=expected): + self.assertIn(expected, runner) + self.assertIn("Agent Task URL changed during submission", runner) + + def test_agent_task_fixture_runs_under_the_existing_pinned_chrome_job(self) -> None: + """No floating browser or second workflow may be introduced for this slice.""" + + workflow = WORKFLOW.read_text(encoding="utf-8") + runner = RUNNER.read_text(encoding="utf-8") + self.assertTrue(FIXTURE.is_file()) + self.assertIn('CHROME_VERSION: "150.0.7871.129"', workflow) + self.assertIn("run_mv3_compatibility.py", workflow) + self.assertIn("150.0.7871.129", runner) + self.assertNotIn("google-chrome-stable", runner.lower()) + self.assertNotIn("COPILOT_GITHUB_TOKEN", runner) + self.assertNotIn("NVIDIA_NIM_API_KEY", runner) + + +if __name__ == "__main__": + unittest.main()