diff --git a/scripts/ci/run_mv3_compatibility.py b/scripts/ci/run_mv3_compatibility.py index b7d3cc4d1..2b1c6320d 100755 --- a/scripts/ci/run_mv3_compatibility.py +++ b/scripts/ci/run_mv3_compatibility.py @@ -7,15 +7,17 @@ content-script, storage, declarative-net-request, tabs, windows, scripting, commands, side-panel, bookmarks, history, real browser-click, and restart-persistence behavior. It also executes the controlled Agent Task fixture -with extensions disabled in a fresh profile, verifies browser-computed role/name -for the controlled action targets, performs real WebDriver input and click -operations, verifies the observable post-condition, proves the controlled action -preserves its loaded URL, and records bounded runtime resource evidence without -treating page content as instruction or authority. +with extensions disabled in a fresh profile, locates the controlled action +targets by exact browser-computed role/name evidence, performs real WebDriver +input and click operations, verifies the observable post-condition, proves the +controlled action preserves its loaded URL, and records bounded runtime resource +evidence without treating page content as instruction or authority. """ from __future__ import annotations +import contextlib +import hashlib import http.client import http.server import json @@ -45,6 +47,8 @@ MAX_PROC_STATUS_CHARACTERS = 65_536 MAX_BROWSER_PROCESS_TREE_SIZE = 256 MAX_PROC_PROCESS_SCAN_SIZE = 32_768 +MAX_SEMANTIC_LOCATOR_CANDIDATES = 128 +MAX_AGENT_TASK_STRUCTURED_VALUE_BYTES = 4_096 MAX_U64 = (1 << 64) - 1 W3C_ELEMENT_KEY = "element-6066-11e4-a52e-4f735466cecf" PATH_TOKEN_CHARACTERS = frozenset(string.ascii_letters + string.digits + "-_.") @@ -239,6 +243,60 @@ def _get_element_semantics( return role, label +def _find_element_by_accessible_role_name( + driver_port: int, + session_id: str, + role: str, + accessible_name: str, +) -> str: + """Find exactly one controlled element by browser-computed role and name.""" + + found = _json_request( + driver_port, + "POST", + _webdriver_path(session_id, "/elements"), + {"using": "css selector", "value": "*"}, + ) + elements = found.get("value") + if not isinstance(elements, list): + raise RuntimeError("WebDriver did not return a semantic locator candidate list") + if len(elements) > MAX_SEMANTIC_LOCATOR_CANDIDATES: + raise RuntimeError("semantic locator exceeded bounded candidate limit") + + matches: list[str] = [] + for element in elements: + element_id = element.get(W3C_ELEMENT_KEY) if isinstance(element, dict) else None + if not isinstance(element_id, str): + raise RuntimeError("WebDriver returned malformed semantic locator candidate") + safe_element = _path_token(element_id, "element identifier") + candidate_role, candidate_name = _get_element_semantics( + driver_port, + session_id, + safe_element, + ) + if candidate_role == role and candidate_name == accessible_name: + matches.append(safe_element) + if len(matches) > 1: + raise RuntimeError("semantic locator returned multiple exact matches") + + if not matches: + raise RuntimeError("semantic locator returned no exact match") + return matches[0] + + +def _hash_agent_task_structured_value(value: str) -> str: + """Hash one bounded extracted text value without retaining the raw value in evidence.""" + + if not isinstance(value, str): + raise TypeError("Agent Task structured value must be text") + encoded = value.encode("utf-8") + if not encoded: + raise ValueError("Agent Task structured value must not be empty") + if len(encoded) > MAX_AGENT_TASK_STRUCTURED_VALUE_BYTES: + raise ValueError("Agent Task structured value exceeded the bounded text contract") + return "sha256:" + hashlib.sha256(encoded).hexdigest() + + def _parse_linux_proc_status_rss_bytes(status_text: str) -> int: """Parse exactly one positive Linux ``VmRSS`` kB field into bounded bytes.""" @@ -833,7 +891,12 @@ def _run_agent_task_browser_pass( if initial_url != fixture_url: raise RuntimeError("Agent Task did not load the requested fixture URL") - input_element = _find_element(driver_port, session_id, "#task-text") + input_element = _find_element_by_accessible_role_name( + driver_port, + session_id, + "textbox", + "Task text", + ) input_role, input_name = _get_element_semantics( driver_port, session_id, @@ -841,10 +904,11 @@ def _run_agent_task_browser_pass( ) if input_role != "textbox" or input_name != "Task text": raise RuntimeError("Agent Task input semantic evidence mismatch") - submit_element = _find_element( + submit_element = _find_element_by_accessible_role_name( driver_port, session_id, - "#agent-task-form button[type=submit]", + "button", + "Submit task", ) submit_role, submit_name = _get_element_semantics( driver_port, @@ -900,7 +964,19 @@ def _run_agent_task_browser_pass( if not url_unchanged: raise RuntimeError("Agent Task URL changed during submission") - result_element = _find_element(driver_port, session_id, "#task-result") + result_element = _find_element_by_accessible_role_name( + driver_port, + session_id, + "status", + "Task result", + ) + result_role, result_name = _get_element_semantics( + driver_port, + session_id, + result_element, + ) + if result_role != "status" or result_name != "Task result": + raise RuntimeError("Agent Task result semantic evidence mismatch") state = _json_request( driver_port, "GET", @@ -914,6 +990,7 @@ def _run_agent_task_browser_pass( _validate_agent_task_submitted_state(state) if text != AGENT_TASK_INPUT_VALUE: raise RuntimeError("Agent Task result did not match the synthetic typed value") + structured_value_sha256 = _hash_agent_task_structured_value(text) process_evidence = _snapshot_linux_process_evidence() chromium_process_ids = _discover_linux_process_tree_ids( browser_process_id, @@ -937,6 +1014,9 @@ def _run_agent_task_browser_pass( "url_unchanged": url_unchanged, "input_semantics_verified": True, "submit_semantics_verified": True, + "result_semantics_verified": True, + "structured_value_field": "task_result", + "structured_value_sha256": structured_value_sha256, "extensions_disabled": True, "browser_process_rss_bytes": browser_process_rss_bytes, "chromium_process_count": len(chromium_process_ids), @@ -1011,6 +1091,9 @@ def _run_agent_task_trial( "url_unchanged": result["url_unchanged"], "input_semantics_verified": result["input_semantics_verified"], "submit_semantics_verified": result["submit_semantics_verified"], + "result_semantics_verified": result["result_semantics_verified"], + "structured_value_field": result["structured_value_field"], + "structured_value_sha256": result["structured_value_sha256"], "extensions_disabled": result["extensions_disabled"], "browser_process_rss_bytes": result["browser_process_rss_bytes"], "chromium_process_count": result["chromium_process_count"], @@ -1035,6 +1118,11 @@ def _agent_task_surfaces_complete(agent_task_trials: list[dict[str, Any]]) -> bo and trial.get("url_unchanged") is True and trial.get("input_semantics_verified") is True and trial.get("submit_semantics_verified") is True + and trial.get("result_semantics_verified") is True + and trial.get("structured_value_field") == "task_result" + and isinstance(trial.get("structured_value_sha256"), str) + and len(trial["structured_value_sha256"]) == len("sha256:") + 64 + and trial["structured_value_sha256"].startswith("sha256:") and trial.get("extensions_disabled") is True and trial.get("profile_cleaned") is True and isinstance(trial.get("browser_process_rss_bytes"), int) diff --git a/tests/fixtures/agent_task_basic/index.html b/tests/fixtures/agent_task_basic/index.html index 510b239f1..d97f2046e 100644 --- a/tests/fixtures/agent_task_basic/index.html +++ b/tests/fixtures/agent_task_basic/index.html @@ -16,7 +16,12 @@

Controlled Agent Task

- idle + idle

None: "url_unchanged": True, "input_semantics_verified": True, "submit_semantics_verified": True, + "result_semantics_verified": True, + "structured_value_field": "task_result", + "structured_value_sha256": "sha256:" + "0" * 64, "extensions_disabled": True, "profile_cleaned": True, "browser_process_rss_bytes": 1, @@ -317,6 +320,9 @@ def successful_agent_task_trial(*_args: object, **_kwargs: object) -> dict[str, "url_unchanged": True, "input_semantics_verified": True, "submit_semantics_verified": True, + "result_semantics_verified": True, + "structured_value_field": "task_result", + "structured_value_sha256": "sha256:" + "0" * 64, "extensions_disabled": True, "browser_process_rss_bytes": 1, "chromium_process_count": 1, diff --git a/tests/test_agent_task_structured_value_contract.py b/tests/test_agent_task_structured_value_contract.py new file mode 100644 index 000000000..1ab0c8eb1 --- /dev/null +++ b/tests/test_agent_task_structured_value_contract.py @@ -0,0 +1,66 @@ +"""Contract for credential-safe structured extraction in the controlled Agent Task.""" + +from __future__ import annotations + +import hashlib +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" + + +class AgentTaskStructuredValueContractTests(unittest.TestCase): + """Require bounded semantic extraction without retaining the extracted value.""" + + def test_result_is_discovered_by_exact_browser_semantics(self) -> None: + """The result node must be located by browser-computed role/name, not fixture CSS.""" + + runner = RUNNER.read_text(encoding="utf-8") + fixture = FIXTURE.read_text(encoding="utf-8") + self.assertIn('aria-label="Task result"', fixture) + self.assertIn('"status"', runner) + self.assertIn('"Task result"', runner) + self.assertIn('"result_semantics_verified"', runner) + self.assertNotIn('_find_element(driver_port, session_id, "#task-result")', runner) + + def test_structured_value_hash_is_bounded_and_canonical(self) -> None: + """Only a canonical SHA-256 digest may leave the controlled extraction boundary.""" + + namespace = runpy.run_path(str(RUNNER), run_name="agent_task_structured_value_contract") + self.assertIn("MAX_AGENT_TASK_STRUCTURED_VALUE_BYTES", namespace) + self.assertIn("_hash_agent_task_structured_value", namespace) + helper = namespace["_hash_agent_task_structured_value"] + maximum = namespace["MAX_AGENT_TASK_STRUCTURED_VALUE_BYTES"] + + value = "synthetic structured value" + expected = "sha256:" + hashlib.sha256(value.encode("utf-8")).hexdigest() + digest = helper(value) + self.assertEqual(digest, expected) + self.assertNotIn(value, digest) + self.assertEqual(len(digest), len("sha256:") + 64) + + with self.assertRaises(ValueError): + helper("") + with self.assertRaises(ValueError): + helper("x" * (maximum + 1)) + with self.assertRaises(TypeError): + helper(42) + + def test_agent_task_evidence_reports_field_and_digest_not_raw_result(self) -> None: + """Trial evidence must expose a field identifier and digest, not extracted text.""" + + runner = RUNNER.read_text(encoding="utf-8") + for expected in ( + '"structured_value_field"', + '"structured_value_sha256"', + '"task_result"', + ): + with self.subTest(expected=expected): + self.assertIn(expected, runner) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_mv3_compatibility_contract.py b/tests/test_mv3_compatibility_contract.py index 10872ddac..8b7e0de98 100644 --- a/tests/test_mv3_compatibility_contract.py +++ b/tests/test_mv3_compatibility_contract.py @@ -2,9 +2,11 @@ from __future__ import annotations +import http.client import json import pathlib import runpy +import tempfile import unittest ROOT = pathlib.Path(__file__).resolve().parents[1] @@ -120,6 +122,32 @@ def test_runner_transport_cannot_follow_dynamic_url_schemes(self) -> None: self.assertNotIn("urllib.request", runner) self.assertNotIn("urllib.error", runner) + def test_fixture_handler_cannot_follow_symlinks_outside_its_root(self) -> None: + """The loopback fixture server must not expose files outside its configured root.""" + + namespace = runpy.run_path(str(RUNNER), run_name="mv3_contract") + start_server = namespace["_start_fixture_server"] + stop_server = namespace["_stop_fixture_server"] + with tempfile.TemporaryDirectory() as temporary_directory: + temporary_root = pathlib.Path(temporary_directory) + fixture_root = temporary_root / "fixture" + outside_root = temporary_root / "outside" + fixture_root.mkdir() + outside_root.mkdir() + (outside_root / "secret.txt").write_text("outside", encoding="utf-8") + (fixture_root / "escape").symlink_to(outside_root, target_is_directory=True) + + server, thread = start_server(fixture_root) + connection = http.client.HTTPConnection(*server.server_address, timeout=1) + try: + connection.request("GET", "/escape/secret.txt") + response = connection.getresponse() + self.assertEqual(response.status, 404) + self.assertNotIn(b"outside", response.read()) + finally: + connection.close() + stop_server(server, thread) + def test_runner_accepts_real_chromedriver_element_ids_without_path_injection(self) -> None: """ChromeDriver dotted element IDs must work while path syntax stays fail-closed."""