Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 49 additions & 4 deletions scripts/ci/run_mv3_compatibility.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,10 @@
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, performs real WebDriver input and
click operations, verifies the observable post-condition, and proves profile
cleanup without treating page content as instruction or authority.
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, and proves profile cleanup
without treating page content as instruction or authority.
"""

from __future__ import annotations
Expand Down Expand Up @@ -179,6 +180,28 @@ def _element_command_path(session_id: str, element_id: str, suffix: str) -> str:
return _webdriver_path(session_id, f"/element/{safe_element}{suffix}")


def _get_element_semantics(
driver_port: int,
session_id: str,
element_id: str,
) -> tuple[str, str]:
"""Read one controlled element's browser-computed role and accessible name."""

role = _json_request(
driver_port,
"GET",
_element_command_path(session_id, element_id, "/computedrole"),
).get("value")
label = _json_request(
driver_port,
"GET",
_element_command_path(session_id, element_id, "/computedlabel"),
).get("value")
if not isinstance(role, str) or not isinstance(label, str):
raise RuntimeError("WebDriver returned malformed element semantics")
return role, label


def _wait_for_extension_evidence(
driver_port: int,
session_id: str,
Expand Down Expand Up @@ -511,6 +534,13 @@ def _run_agent_task_browser_pass(
{"url": fixture_url},
)
input_element = _find_element(driver_port, session_id, "#task-text")
input_role, input_name = _get_element_semantics(
driver_port,
session_id,
input_element,
)
if input_role != "textbox" or input_name != "Task text":
raise RuntimeError("Agent Task input semantic evidence mismatch")
_json_request(
driver_port,
"POST",
Expand All @@ -528,6 +558,13 @@ def _run_agent_task_browser_pass(
session_id,
"#agent-task-form button[type=submit]",
)
submit_role, submit_name = _get_element_semantics(
driver_port,
session_id,
submit_element,
)
if submit_role != "button" or submit_name != "Submit task":
raise RuntimeError("Agent Task submit semantic evidence mismatch")
_json_request(
driver_port,
"POST",
Expand All @@ -553,6 +590,8 @@ def _run_agent_task_browser_pass(
"browser_version": browser_version,
"post_condition": True,
"input_echo_verified": True,
"input_semantics_verified": True,
"submit_semantics_verified": True,
"extensions_disabled": True,
"duration_ms": round((time.monotonic() - started) * 1000),
}
Expand Down Expand Up @@ -603,13 +642,17 @@ def _run_agent_task_trial(
"browser_version": result["browser_version"],
"post_condition": result["post_condition"],
"input_echo_verified": result["input_echo_verified"],
"input_semantics_verified": result["input_semantics_verified"],
"submit_semantics_verified": result["submit_semantics_verified"],
"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]:
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(
Expand Down Expand Up @@ -726,6 +769,8 @@ def main() -> int:
agent_task_surfaces_complete = all(
trial.get("post_condition") is True
and trial.get("input_echo_verified") is True
and trial.get("input_semantics_verified") is True
and trial.get("submit_semantics_verified") is True
and trial.get("extensions_disabled") is True
and trial.get("profile_cleaned") is True
for trial in agent_task_trials
Expand Down
66 changes: 57 additions & 9 deletions tests/test_agent_task_fixture_contract.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""

Expand All @@ -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

Expand All @@ -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
Expand Down Expand Up @@ -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(
"<p data-originweave-untrusted='prompt-injection' aria-hidden='true'>visible</p>"
"<p data-originweave-untrusted='prompt-injection' aria-hidden='true' hidden>hidden</p>"
)
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 (
"<input type='password'>",
"<input autocomplete='current-password'>",
"<input autocomplete='new-password'>",
"<input autocomplete='section-login username current-password'>",
"<input autocomplete='one-time-code'>",
):
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("<input type='text' autocomplete='username'>")
self.assertEqual(len(parser.input_attributes), 1)
self.assertFalse(_is_credential_input(parser.input_attributes[0]))


if __name__ == "__main__":
unittest.main()
32 changes: 32 additions & 0 deletions tests/test_agent_task_pinned_chrome_contract.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,38 @@ def test_agent_task_pass_uses_real_webdriver_input_and_post_condition(self) -> N
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_observes_computed_role_and_name_before_action(self) -> None:
"""Real-browser evidence must bind the controlled targets to semantic role/name."""

namespace = runpy.run_path(str(RUNNER), run_name="agent_task_semantics_contract")
runner = RUNNER.read_text(encoding="utf-8")
self.assertIn("_get_element_semantics", namespace)
for expected in (
'"/computedrole"',
'"/computedlabel"',
'"textbox"',
'"Task text"',
'"button"',
'"Submit task"',
'"input_semantics_verified"',
'"submit_semantics_verified"',
):
with self.subTest(expected=expected):
self.assertIn(expected, 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."""

Expand Down