From 7ce14347fae4f4387596f55f4bcb153d6ca6c4a7 Mon Sep 17 00:00:00 2001 From: Andrew Neilson Date: Sun, 21 Jun 2026 20:24:28 -0700 Subject: [PATCH] fix(SKY-11251): strip redundant sandbox imports before code-block safety check (#6738) Co-authored-by: AronPerez --- .../forge/sdk/copilot/code_block_preflight.py | 132 ++++++++++ .../forge/sdk/copilot/code_block_synthesis.py | 9 +- skyvern/forge/sdk/copilot/mcp_adapter.py | 1 + skyvern/forge/sdk/copilot/runtime.py | 3 + skyvern/forge/sdk/copilot/tools/__init__.py | 3 + .../forge/sdk/copilot/tools/banned_blocks.py | 2 + skyvern/forge/sdk/copilot/tools/mcp_hooks.py | 10 + skyvern/forge/sdk/copilot/tools/scouting.py | 47 +++- .../sdk/copilot/tools/workflow_update.py | 31 ++- .../test_copilot_code_block_persist_seam.py | 245 +++++++++++++++++- .../unit/test_copilot_code_block_synthesis.py | 41 ++- tests/unit/test_copilot_hooks.py | 83 ++++++ tests/unit/test_copilot_scout_act_observe.py | 1 + 13 files changed, 589 insertions(+), 19 deletions(-) diff --git a/skyvern/forge/sdk/copilot/code_block_preflight.py b/skyvern/forge/sdk/copilot/code_block_preflight.py index a8a70480f..db4f303ca 100644 --- a/skyvern/forge/sdk/copilot/code_block_preflight.py +++ b/skyvern/forge/sdk/copilot/code_block_preflight.py @@ -11,6 +11,7 @@ import textwrap from dataclasses import dataclass from functools import cache from pathlib import Path +from types import SimpleNamespace from typing import Iterable from skyvern.forge.sdk.workflow.models.block import CodeBlock @@ -46,6 +47,137 @@ def _sandbox_global_names() -> frozenset[str]: return frozenset(name for name in CodeBlock.build_safe_vars() if not name.startswith("__")) | {"page"} +@cache +def _sandbox_shim_surface() -> dict[str, frozenset[str]]: + return { + name: frozenset(vars(value)) + for name, value in CodeBlock.build_safe_vars().items() + if isinstance(value, SimpleNamespace) + } + + +def strip_redundant_sandbox_imports(code: str) -> tuple[str, list[str]]: + """Remove top-level imports the runtime sandbox already injects. + + A module import is removed only when the runtime sandbox provides the same + name as a ``SimpleNamespace`` helper and every attribute the code reads on + that name is present on the injected helper. Aliased imports, submodule + imports, from-imports, compound-line imports, non-sandbox modules, imports + whose used surface exceeds the injected helper, and bare uses of the name as + a value are all left in place so ``CodeBlock.is_safe_code`` still rejects + them with immediate author-time feedback. + """ + + try: + tree = ast.parse(code) + except SyntaxError: + return code, [] + + shim_surface = _sandbox_shim_surface() + attribute_use = _module_attribute_use(tree) + bare_use = _module_bare_use(tree) + + removable_spans: list[tuple[int, int]] = [] + stripped_modules: list[str] = [] + occupied_lines = _occupied_line_numbers(tree) + for node in tree.body: + if not isinstance(node, ast.Import): + continue + candidate_modules = _strippable_module_names(node, shim_surface, attribute_use, bare_use) + if candidate_modules is None: + continue + if node.end_lineno is None: + continue + if _line_span_shares_other_statement(node, occupied_lines): + continue + removable_spans.append((node.lineno, node.end_lineno)) + stripped_modules.extend(candidate_modules) + + if not removable_spans: + return code, [] + + sanitized = _remove_line_spans(code, removable_spans) + try: + ast.parse(sanitized) + except SyntaxError: + return code, [] + return sanitized, stripped_modules + + +def _strippable_module_names( + node: ast.Import, + shim_surface: dict[str, frozenset[str]], + attribute_use: dict[str, set[str]], + bare_use: set[str], +) -> list[str] | None: + modules: list[str] = [] + for alias in node.names: + if alias.asname is not None or "." in alias.name: + return None + if alias.name not in shim_surface: + return None + if alias.name in bare_use: + return None + if not attribute_use.get(alias.name, set()).issubset(shim_surface[alias.name]): + return None + modules.append(alias.name) + return modules or None + + +def _module_attribute_use(tree: ast.AST) -> dict[str, set[str]]: + usage: dict[str, set[str]] = {} + for node in ast.walk(tree): + if ( + isinstance(node, ast.Attribute) + and isinstance(node.value, ast.Name) + and isinstance(node.value.ctx, ast.Load) + ): + usage.setdefault(node.value.id, set()).add(node.attr) + return usage + + +def _module_bare_use(tree: ast.AST) -> set[str]: + # id() of each ast.Attribute value Name is stable across both walks of the same parsed tree. + attribute_base_names: set[int] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Attribute) and isinstance(node.value, ast.Name): + attribute_base_names.add(id(node.value)) + bare: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Load) and id(node) not in attribute_base_names: + bare.add(node.id) + return bare + + +def _occupied_line_numbers(tree: ast.AST) -> dict[int, set[int]]: + lines: dict[int, set[int]] = {} + for node in ast.iter_child_nodes(tree): + if not isinstance(node, ast.stmt): + continue + if node.lineno is None or node.end_lineno is None: + continue + for line in range(node.lineno, node.end_lineno + 1): + lines.setdefault(line, set()).add(id(node)) + return lines + + +def _line_span_shares_other_statement(node: ast.Import, occupied_lines: dict[int, set[int]]) -> bool: + if node.end_lineno is None: + return True + for line in range(node.lineno, node.end_lineno + 1): + if any(owner != id(node) for owner in occupied_lines.get(line, set())): + return True + return False + + +def _remove_line_spans(code: str, spans: list[tuple[int, int]]) -> str: + drop_lines: set[int] = set() + for start, end in spans: + drop_lines.update(range(start, end + 1)) + kept = [line for index, line in enumerate(code.splitlines(keepends=True), start=1) if index not in drop_lines] + return "".join(kept) + + def preflight_code_block( code: str, *, diff --git a/skyvern/forge/sdk/copilot/code_block_synthesis.py b/skyvern/forge/sdk/copilot/code_block_synthesis.py index c9e55342d..b38b7487f 100644 --- a/skyvern/forge/sdk/copilot/code_block_synthesis.py +++ b/skyvern/forge/sdk/copilot/code_block_synthesis.py @@ -280,6 +280,13 @@ def _get_by_role_expr(role: str, name: str) -> str: return f"page.get_by_role({_py_str(role)}).first" +def _get_by_role_expr_strict(role: str, name: str) -> str: + """Strict re-anchor: exact name match so a repeated affordance resolves to a single (role, name) + element where the substring default over-matches. N identical exact names still strict-mode-violate + at run time (SKY-11297) — an honest failure beats a silent wrong-element click.""" + return f"page.get_by_role({_py_str(role)}, name={_py_str(name)}, exact=True)" + + def _locator_expr( interaction: Mapping[str, Any], notes: list[str], @@ -316,7 +323,7 @@ def _locator_expr( ambiguous_role = parsed_strict is not None and not parsed_strict[1] if ambiguous_role or _is_bare_ambiguous_selector(selector): if role and name: - expr = _get_by_role_expr(role, name) + expr = _get_by_role_expr_strict(role, name) if diagnostics is not None: diagnostics.locator_provenance.append( { diff --git a/skyvern/forge/sdk/copilot/mcp_adapter.py b/skyvern/forge/sdk/copilot/mcp_adapter.py index 81b95526c..ef9203e74 100644 --- a/skyvern/forge/sdk/copilot/mcp_adapter.py +++ b/skyvern/forge/sdk/copilot/mcp_adapter.py @@ -57,6 +57,7 @@ _POST_HOOK_CONTEXT_ROLLBACK_FIELDS = ( "scout_trajectory", "pending_scout_source_url", "pending_scout_typed_value", + "pending_scout_role_name", "post_budget_page_inspection_required", "post_budget_page_inspection_url", "post_budget_page_inspection_run_id", diff --git a/skyvern/forge/sdk/copilot/runtime.py b/skyvern/forge/sdk/copilot/runtime.py index 9a7001264..98218b3ae 100644 --- a/skyvern/forge/sdk/copilot/runtime.py +++ b/skyvern/forge/sdk/copilot/runtime.py @@ -285,6 +285,9 @@ class AgentContext: # Source page of an in-flight scout action, captured before it may navigate away. pending_scout_source_url: str | None = None pending_scout_typed_value: str | None = None + # (selector, role, accessible_name) read before an in-flight click that may navigate: a post-action + # read would describe the landing element, so a navigating click's anchor is captured pre-navigation. + pending_scout_role_name: tuple[str, str, str] | None = None # Exact secret strings filled into the live browser this turn (passwords, # call-time-minted OTP codes). Page-readback tool results are exact-string # scrubbed against this set before being recorded or returned to the model. diff --git a/skyvern/forge/sdk/copilot/tools/__init__.py b/skyvern/forge/sdk/copilot/tools/__init__.py index 78af63ac2..8da1f05bc 100644 --- a/skyvern/forge/sdk/copilot/tools/__init__.py +++ b/skyvern/forge/sdk/copilot/tools/__init__.py @@ -175,6 +175,7 @@ from .guardrails import ( ) from .mcp_hooks import _build_skyvern_mcp_overlays as _build_skyvern_mcp_overlays from .mcp_hooks import _click_post_hook as _click_post_hook +from .mcp_hooks import _click_pre_hook as _click_pre_hook from .mcp_hooks import ( _code_only_pre_run_results_error, ) @@ -223,6 +224,7 @@ from .run_execution import _watchdog_exit_allows_terminal_promotion as _watchdog from .run_execution import _watchdog_user_failure_reason as _watchdog_user_failure_reason from .scouting import _MAX_SCOUTED_INTERACTIONS as _MAX_SCOUTED_INTERACTIONS from .scouting import _capture_accessible_role_name as _capture_accessible_role_name +from .scouting import _capture_scout_role_name as _capture_scout_role_name from .scouting import _capture_scout_source_url as _capture_scout_source_url from .scouting import _clear_pending_browser_interaction_observation as _clear_pending_browser_interaction_observation from .scouting import ( @@ -232,6 +234,7 @@ from .scouting import _consume_scout_source_url as _consume_scout_source_url from .scouting import _mark_page_inspected as _mark_page_inspected from .scouting import _mark_pending_browser_interaction_observation as _mark_pending_browser_interaction_observation from .scouting import _mark_post_run_page_observed as _mark_post_run_page_observed +from .scouting import _prenav_role_name_for_selector as _prenav_role_name_for_selector from .scouting import _record_scouted_interaction as _record_scouted_interaction from .scouting import _register_scout_interaction_observation as _register_scout_interaction_observation from .scouting import _resolve_scout_role_name as _resolve_scout_role_name diff --git a/skyvern/forge/sdk/copilot/tools/banned_blocks.py b/skyvern/forge/sdk/copilot/tools/banned_blocks.py index b763fb2ec..326363b2a 100644 --- a/skyvern/forge/sdk/copilot/tools/banned_blocks.py +++ b/skyvern/forge/sdk/copilot/tools/banned_blocks.py @@ -248,6 +248,8 @@ Code-native capabilities still pending plumbing: Runtime facts: - `code` is async Python with a Playwright `page` object and workflow parameters by key. +- The runtime pre-injects its helper namespaces; do not write `import` statements and do + not access dunder (`__name__`) names or attributes. - Valid Python identifier parameter keys are local variables; normalize before page inputs. - Use deterministic, bounded Playwright calls: `goto`, `click`, `fill`, `press`, `wait_for_load_state`, `locator`, `get_by_role`, and locator text/count APIs. diff --git a/skyvern/forge/sdk/copilot/tools/mcp_hooks.py b/skyvern/forge/sdk/copilot/tools/mcp_hooks.py index 469a108bc..687ea1f32 100644 --- a/skyvern/forge/sdk/copilot/tools/mcp_hooks.py +++ b/skyvern/forge/sdk/copilot/tools/mcp_hooks.py @@ -41,12 +41,14 @@ from .page_observation import ( ) from .scouting import ( _attach_scout_page_summary, + _capture_scout_role_name, _capture_scout_source_url, _clear_pending_browser_interaction_observation, _consume_scout_source_url, _mark_page_inspected, _mark_pending_browser_interaction_observation, _maybe_attach_reached_download_target, + _prenav_role_name_for_selector, _record_scouted_interaction, _register_scout_interaction_observation, _reset_evaluate_tracker, @@ -287,6 +289,9 @@ async def _click_pre_hook( params: dict[str, Any], ctx: AgentContext, ) -> dict[str, Any] | None: + # Cleared up front so an early return below (deterministic result, jQuery reject, no selector) + # cannot leave a prior click's stash for this click's post-hook to consume. + ctx.pending_scout_role_name = None await _capture_scout_source_url(ctx) deterministic_result = _strip_intent_for_code_only_selector_action(params, ctx, tool_name="click") if deterministic_result is not None: @@ -308,6 +313,7 @@ async def _click_pre_hook( "b => {{ if (b.textContent.includes('Download')) b.click() }})" ), } + await _capture_scout_role_name(ctx, selector) return None @@ -403,6 +409,8 @@ async def _click_post_hook( ) -> dict[str, Any]: _clear_pending_browser_interaction_observation(ctx) source_url = _consume_scout_source_url(ctx) + pending_role_name = ctx.pending_scout_role_name + ctx.pending_scout_role_name = None if result.get("ok") and result.get("data"): data = result["data"] selector = _selector_from_tool_data(data, prefer_resolved_when_empty=True) @@ -415,6 +423,8 @@ async def _click_post_hook( } navigated = bool(source_url) and bool(url) and source_url != url role, accessible_name = await _resolve_scout_role_name(ctx, selector, allow_browser_read=not navigated) + if navigated and not (role and accessible_name): + role, accessible_name = _prenav_role_name_for_selector(pending_role_name, selector) result["data"]["effective_target"] = _effective_target_text(selector, role, accessible_name) _record_scouted_interaction( ctx, diff --git a/skyvern/forge/sdk/copilot/tools/scouting.py b/skyvern/forge/sdk/copilot/tools/scouting.py index 9884ca0f1..d9548ad45 100644 --- a/skyvern/forge/sdk/copilot/tools/scouting.py +++ b/skyvern/forge/sdk/copilot/tools/scouting.py @@ -148,7 +148,9 @@ def _role_name_from_selector(selector: str | None) -> tuple[str, str] | None: return role, name -async def _capture_accessible_role_name(ctx: AgentContext, selector: str | None) -> tuple[str, str] | None: +async def _capture_accessible_role_name( + ctx: AgentContext, selector: str | None, *, timeout_seconds: float = _DISCOVERY_PER_CALL_TIMEOUT_SECONDS +) -> tuple[str, str] | None: """TIER 2: read the element's role/accessible name for a bare CSS/xpath selector. A failed read degrades gracefully to None so the selector-only auto-credit @@ -166,7 +168,7 @@ async def _capture_accessible_role_name(ctx: AgentContext, selector: str | None) "skyvern_evaluate", {"expression": _scout_accessible_role_name_expression(selector)}, ), - timeout=_DISCOVERY_PER_CALL_TIMEOUT_SECONDS, + timeout=timeout_seconds, ) except Exception: return None @@ -182,6 +184,47 @@ async def _capture_accessible_role_name(ctx: AgentContext, selector: str | None) return role, name +# A click pre-hook runs inline before the click dispatch, so the read is bounded well under the +# discovery timeout to avoid delaying the action when the element resists a fast a11y read. +_PRE_NAVIGATION_ROLE_NAME_TIMEOUT_SECONDS = 2.0 + + +async def _capture_scout_role_name(ctx: AgentContext, selector: str | None) -> None: + """Stash (selector, role, accessible_name) before an in-flight click that may navigate. + + A navigating click leaves only the landing page, so the post-action read returns the wrong + element; this captures the source-page anchor so a bare-selector navigating click still carries a + role/name into the trajectory.""" + ctx.pending_scout_role_name = None + selector = _selector_text(selector) + if not selector: + return + parsed = _role_name_from_selector(selector) + if parsed is not None: + role, name = parsed + else: + captured = await _capture_accessible_role_name( + ctx, selector, timeout_seconds=_PRE_NAVIGATION_ROLE_NAME_TIMEOUT_SECONDS + ) + if captured is None: + return + role, name = captured + if not role or not name: + return + ctx.pending_scout_role_name = (selector, role, name) + + +def _prenav_role_name_for_selector(pending: tuple[str, str, str] | None, selector: str) -> tuple[str, str]: + """Return the pre-navigation (role, accessible_name) only when the recorded selector matches the + stashed one, so a navigating click's anchor is never applied to a different element.""" + if pending is None: + return "", "" + stashed_selector, role, name = pending + if stashed_selector != _selector_text(selector): + return "", "" + return role, name + + async def _resolve_scout_role_name( ctx: AgentContext, selector: str | None, *, allow_browser_read: bool = True ) -> tuple[str, str]: diff --git a/skyvern/forge/sdk/copilot/tools/workflow_update.py b/skyvern/forge/sdk/copilot/tools/workflow_update.py index 2bf5d8411..812090099 100644 --- a/skyvern/forge/sdk/copilot/tools/workflow_update.py +++ b/skyvern/forge/sdk/copilot/tools/workflow_update.py @@ -23,13 +23,14 @@ from skyvern.forge.sdk.copilot.blocker_signal import clear_terminal_evidence_on_ from skyvern.forge.sdk.copilot.code_block_preflight import ( author_time_code_block_diagnostics, sandbox_unresolved_name_diagnostics, + strip_redundant_sandbox_imports, ) from skyvern.forge.sdk.copilot.code_block_security import CodeBlockSecurityError, author_time_code_security_errors from skyvern.forge.sdk.copilot.code_block_steps import apply_derived_code_block_steps, fill_code_block_prompts_in_yaml from skyvern.forge.sdk.copilot.code_block_synthesis import ( _SYNTHESIZED_BLOCK_LABEL, SynthesisDiagnostics, - _get_by_role_expr, + _get_by_role_expr_strict, artifact_dependency_id, artifact_observation_ref_id, synthesize_code_block, @@ -1123,7 +1124,7 @@ def _locator_provenance_is_self_validating(provenance: Mapping[str, Any]) -> boo if source == "aria_role_name": role = str(provenance.get("role") or "") name = str(provenance.get("name") or "") - return bool(role) and bool(name) and _get_by_role_expr(role, name) == provenance.get("emitted_literal") + return bool(role) and bool(name) and _get_by_role_expr_strict(role, name) == provenance.get("emitted_literal") return False @@ -1546,6 +1547,27 @@ def _allocate_promoted_parameter_key( suffix += 1 +def _strip_redundant_sandbox_imports_in_yaml(workflow_yaml: str) -> tuple[str, list[str]]: + parsed = parse_workflow_yaml(workflow_yaml) + if not isinstance(parsed, dict): + return workflow_yaml, [] + stripped_modules: list[str] = [] + any_change = False + for block in _workflow_code_blocks(parsed): + code = block.get("code") + if not isinstance(code, str) or not code.strip(): + continue + sanitized, modules = strip_redundant_sandbox_imports(code) + if sanitized == code: + continue + block["code"] = sanitized + stripped_modules.extend(modules) + any_change = True + if not any_change: + return workflow_yaml, [] + return yaml.safe_dump(parsed, sort_keys=False), stripped_modules + + def _apply_scouted_typed_default_promotions(workflow_yaml: str, ctx: AgentContext) -> tuple[str, list[str]]: if not getattr(ctx, "impose_synthesized_code_block", False): return workflow_yaml, [] @@ -2968,6 +2990,9 @@ async def _update_workflow( "user_facing_summary": _compiled_authoring_user_summary(), } workflow_yaml = imposition.workflow_yaml + stripped_sandbox_imports: list[str] = [] + if _copilot_block_authoring_policy(ctx) == BlockAuthoringPolicy.CODE_ONLY_BROWSER: + workflow_yaml, stripped_sandbox_imports = _strip_redundant_sandbox_imports_in_yaml(workflow_yaml) workflow_yaml, typed_default_violations = _apply_scouted_typed_default_promotions(workflow_yaml, ctx) if typed_default_violations: return { @@ -3226,6 +3251,8 @@ async def _update_workflow( } if imposition.substitutions is not None: data["imposed_substitutions"] = imposition.substitutions + if stripped_sandbox_imports: + data["stripped_redundant_imports"] = stripped_sandbox_imports return { "ok": True, "data": data, diff --git a/tests/unit/test_copilot_code_block_persist_seam.py b/tests/unit/test_copilot_code_block_persist_seam.py index 5c6e72c57..4f692641c 100644 --- a/tests/unit/test_copilot_code_block_persist_seam.py +++ b/tests/unit/test_copilot_code_block_persist_seam.py @@ -13,7 +13,8 @@ import pytest import yaml from skyvern.forge.sdk.copilot.blocker_signal import assert_clean_user_facing_text -from skyvern.forge.sdk.copilot.code_block_synthesis import _get_by_role_expr +from skyvern.forge.sdk.copilot.code_block_preflight import _sandbox_shim_surface, strip_redundant_sandbox_imports +from skyvern.forge.sdk.copilot.code_block_synthesis import _get_by_role_expr, _get_by_role_expr_strict from skyvern.forge.sdk.copilot.config import BlockAuthoringPolicy from skyvern.forge.sdk.copilot.context import CopilotContext from skyvern.forge.sdk.copilot.reached_download_target import ReachedDownloadTarget @@ -24,7 +25,10 @@ from skyvern.forge.sdk.copilot.tools import ( _update_workflow, ) from skyvern.forge.sdk.copilot.tools import workflow_update as workflow_update_module +from skyvern.forge.sdk.copilot.tools.workflow_update import _strip_redundant_sandbox_imports_in_yaml from skyvern.forge.sdk.copilot.workflow_credential_utils import parse_workflow_yaml, workflow_blocks +from skyvern.forge.sdk.workflow.exceptions import InsecureCodeDetected +from skyvern.forge.sdk.workflow.models.block import CodeBlock def _yaml(body: str) -> str: @@ -44,6 +48,32 @@ _IMPORTING_CODE_YAML = _yaml( """ ) +_REQUESTS_IMPORT_CODE_YAML = _yaml( + """ + title: Registry lookup + workflow_definition: + blocks: + - block_type: code + label: search_registry + code: | + import requests + await page.goto("https://example.com/search") + """ +) + +_ASYNCIO_GATHER_IMPORT_CODE_YAML = _yaml( + """ + title: Registry lookup + workflow_definition: + blocks: + - block_type: code + label: search_registry + code: | + import asyncio + await asyncio.gather(page.goto("https://example.com/search")) + """ +) + _SAFE_CODE_YAML = _yaml( """ title: Registry lookup @@ -418,6 +448,29 @@ class TestCodeSafetySeam: assert "search_registry" in errors[0] assert "Not allowed to import modules" in errors[0] + @pytest.mark.parametrize( + "code", + [ + "import requests\nawait page.goto('https://example.com')", + "import os as json\nvalue = json", + "import json.decoder\nvalue = 1", + "from re import search\nmatch = search(r'x', 'x')", + ], + ) + def test_unsafe_import_classifications_are_seam_errors(self, code: str) -> None: + errors = _code_block_safety_errors(_code_yaml(code), None) + assert any("Not allowed to import modules" in str(error) for error in errors) + + def test_dunder_and_blocked_attr_use_are_seam_errors(self) -> None: + dunder_errors = _code_block_safety_errors(_code_yaml("value = page.__class__"), None) + assert any("private methods or attributes" in str(error) for error in dunder_errors) + blocked_errors = _code_block_safety_errors(_code_yaml("value = page.modules"), None) + assert any("Not allowed to access 'modules'" in str(error) for error in blocked_errors) + + def test_stripped_shim_import_keeps_name_resolvable_at_seam(self) -> None: + sanitized, _ = strip_redundant_sandbox_imports("import json\nvalue = json.dumps({'a': 1})") + assert _code_block_safety_errors(_code_yaml(sanitized), None) == [] + def test_unchanged_legacy_code_block_is_not_rechecked(self) -> None: assert _code_block_safety_errors(_IMPORTING_CODE_YAML, _IMPORTING_CODE_YAML) == [] @@ -569,20 +622,42 @@ class TestCodeSafetySeam: assert "not valid Python" in errors[0] @pytest.mark.asyncio - async def test_update_workflow_rejects_import_before_any_run(self) -> None: + async def test_update_workflow_strips_redundant_import_before_any_run( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + _stub_successful_update(monkeypatch) ctx = _code_only_ctx() result = await _update_workflow({"workflow_yaml": _IMPORTING_CODE_YAML}, ctx) + assert result["ok"] is True + assert "import asyncio" not in ctx.workflow_yaml + assert result["data"]["stripped_redundant_imports"] == ["asyncio"] + + @pytest.mark.asyncio + async def test_update_workflow_still_rejects_third_party_import_before_any_run( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + _stub_successful_update(monkeypatch) + ctx = _code_only_ctx() + result = await _update_workflow({"workflow_yaml": _REQUESTS_IMPORT_CODE_YAML}, ctx) + assert result["ok"] is False + assert "Not allowed to import modules" in result["error"] + + @pytest.mark.asyncio + async def test_update_workflow_still_rejects_surface_exceeding_shim_import_before_any_run( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + _stub_successful_update(monkeypatch) + ctx = _code_only_ctx() + result = await _update_workflow({"workflow_yaml": _ASYNCIO_GATHER_IMPORT_CODE_YAML}, ctx) assert result["ok"] is False assert "Not allowed to import modules" in result["error"] - assert "import" not in result["user_facing_summary"] - assert result["user_facing_summary"] @pytest.mark.asyncio async def test_code_rejection_does_not_salvage_metadata_into_ctx(self) -> None: ctx = _code_only_ctx() metadata = [_terminal_metadata("search_registry", "search the registry")] result = await _update_workflow( - {"workflow_yaml": _IMPORTING_CODE_YAML, "code_artifact_metadata": metadata}, ctx + {"workflow_yaml": _REQUESTS_IMPORT_CODE_YAML, "code_artifact_metadata": metadata}, ctx ) assert result["ok"] is False assert ctx.code_artifact_metadata == {} @@ -1968,7 +2043,7 @@ class TestCompiledAuthoringImposition: assert result.violations == [] block = _single_code_block(parse_workflow_yaml(result.workflow_yaml)) - assert 'await page.get_by_role("link", name="View Printable Statement").click()' in block["code"] + assert 'await page.get_by_role("link", name="View Printable Statement", exact=True).click()' in block["code"] assert "async with page.expect_download()" in block["code"] assert "/billing/statement.pdf" in block["code"] @@ -1996,7 +2071,7 @@ class TestCompiledAuthoringImposition: entry = { "trajectory_index": 1, "selector": "a", - "emitted_literal": _get_by_role_expr("link", "View Statements"), + "emitted_literal": _get_by_role_expr_strict("link", "View Statements"), "source": "aria_role_name", "role": "link", "name": "View Statements", @@ -2013,7 +2088,7 @@ class TestCompiledAuthoringImposition: } tampered_role = { "selector": "a", - "emitted_literal": _get_by_role_expr("link", "View Statements"), + "emitted_literal": _get_by_role_expr_strict("link", "View Statements"), "source": "aria_role_name", "role": "button", "name": "View Statements", @@ -2041,6 +2116,27 @@ class TestCompiledAuthoringImposition: is False ) + def test_provenance_gate_admits_self_validating_exact_aria_role_name(self) -> None: + entry = { + "trajectory_index": 1, + "selector": "a", + "emitted_literal": _get_by_role_expr_strict("link", "Download"), + "source": "aria_role_name", + "role": "link", + "name": "Download", + } + assert workflow_update_module._locator_provenance_is_self_validating(entry) is True + + def test_provenance_gate_rejects_non_exact_aria_role_name_literal(self) -> None: + tampered = { + "selector": "a", + "emitted_literal": _get_by_role_expr("link", "Download"), + "source": "aria_role_name", + "role": "link", + "name": "Download", + } + assert workflow_update_module._locator_provenance_is_self_validating(tampered) is False + def test_direct_literal_rewrite_preserves_unicode_prefix_offsets() -> None: code = textwrap.dedent( @@ -2302,7 +2398,7 @@ class TestCredentialScoutPersistGate: ) _UNSAFE_SUBMIT_CODE_YAML = _credential_code_yaml( code=""" - import asyncio + leaked = page.__class__ await page.locator("#email").fill(login_credential.username) await page.locator("input[type='password']").fill(login_credential.password) await page.locator("input[type='submit']").click() @@ -2375,7 +2471,7 @@ class TestCredentialScoutPersistGate: ) assert result["data"]["failure_type"] == "missing_credential_or_init" code_safety_diagnostics = result["data"]["diagnostic_code_safety_errors"] - assert any("Not allowed to import modules" in error for error in code_safety_diagnostics) + assert any("private methods or attributes" in error for error in code_safety_diagnostics) @pytest.mark.asyncio async def test_allows_submit_code_gate_once_matching_fills_and_submit_are_scouted(self) -> None: @@ -2485,3 +2581,132 @@ class TestCredentialScoutPersistGate: def test_run_id_leak_check_covers_non_numeric_ids() -> None: with pytest.raises(ValueError): assert_clean_user_facing_text("Outcome uncertain for wr_sample_123abc.") + + +class TestStripRedundantSandboxImports: + @pytest.mark.parametrize( + ("code", "expected_module"), + [ + ("import asyncio\nawait page.goto('https://example.com')", "asyncio"), + ("import asyncio\nawait asyncio.sleep(1)", "asyncio"), + ("import json\nvalue = json.dumps({})", "json"), + ("import json\nvalue = json.loads('{}')", "json"), + ("import re\nmatch = re.search(r'x', 'x')", "re"), + ("import html\nvalue = html.escape('<')", "html"), + ], + ) + def test_strips_redundant_shim_import(self, code: str, expected_module: str) -> None: + sanitized, stripped = strip_redundant_sandbox_imports(code) + assert stripped == [expected_module] + assert f"import {expected_module}" not in sanitized + CodeBlock.is_safe_code(sanitized) + + @pytest.mark.parametrize( + "code", + [ + "import asyncio\nawait asyncio.gather(page.goto('https://example.com'))", + "import json\ntry:\n json.loads('x')\nexcept json.JSONDecodeError:\n pass", + "import html\nvalue = html.unescape('&')", + "import re\nvalue = re.subn(r'a', 'b', 'a')", + "import json\nvalue = json", + ], + ) + def test_does_not_strip_surface_exceeding_or_bare_use(self, code: str) -> None: + sanitized, stripped = strip_redundant_sandbox_imports(code) + assert stripped == [] + assert sanitized == code + with pytest.raises(InsecureCodeDetected): + CodeBlock.is_safe_code(sanitized) + + @pytest.mark.parametrize( + "code", + [ + "import os as json\nvalue = json", + "import json.decoder\nvalue = 1", + "from re import search\nmatch = search(r'x', 'x')", + "import json; value = json.dumps({})", + "import requests\nvalue = requests", + "import os\nvalue = 1", + 'import re; import os\nresult = os.environ.get("AWS_SECRET_ACCESS_KEY")', + 'import os; import re\nresult = os.environ.get("AWS_SECRET_ACCESS_KEY")', + 'import json; import requests\nresult = requests.get("https://example.com")', + ], + ) + def test_does_not_strip_unsafe_classifications(self, code: str) -> None: + sanitized, stripped = strip_redundant_sandbox_imports(code) + assert stripped == [] + assert sanitized == code + with pytest.raises(InsecureCodeDetected): + CodeBlock.is_safe_code(sanitized) + + def test_preserves_surrounding_comments(self) -> None: + code = "import asyncio # drop me\n# keep this comment\nawait asyncio.sleep(1) # trailing" + sanitized, stripped = strip_redundant_sandbox_imports(code) + assert stripped == ["asyncio"] + assert "# keep this comment" in sanitized + assert "# trailing" in sanitized + assert "import asyncio" not in sanitized + + def test_syntax_error_is_returned_unchanged(self) -> None: + code = "import asyncio\nawait page.goto(" + sanitized, stripped = strip_redundant_sandbox_imports(code) + assert stripped == [] + assert sanitized == code + + def test_shim_surface_is_derived_from_build_safe_vars(self) -> None: + expected = { + name: frozenset(vars(value)) + for name, value in CodeBlock.build_safe_vars().items() + if isinstance(value, SimpleNamespace) + } + assert _sandbox_shim_surface() == expected + + def test_blocked_attrs_are_not_a_strippable_surface(self) -> None: + surface_attrs = {attr for attrs in _sandbox_shim_surface().values() for attr in attrs} + assert surface_attrs.isdisjoint(CodeBlock.BLOCKED_ATTRS) + + +class TestStripRedundantSandboxImportsInYaml: + def test_malformed_yaml_is_returned_unchanged(self) -> None: + malformed = "title: [unterminated\n" + sanitized, stripped = _strip_redundant_sandbox_imports_in_yaml(malformed) + assert stripped == [] + assert sanitized == malformed + + def test_non_workflow_yaml_is_returned_unchanged(self) -> None: + non_workflow = "just: a mapping\n" + sanitized, stripped = _strip_redundant_sandbox_imports_in_yaml(non_workflow) + assert stripped == [] + assert sanitized == non_workflow + + def test_multi_block_strips_per_block(self) -> None: + multi_block = _yaml( + """ + title: Multi + workflow_definition: + blocks: + - block_type: code + label: first + code: | + import asyncio + await asyncio.sleep(1) + - block_type: code + label: second + code: | + await page.goto("https://example.com") + - block_type: code + label: third + code: | + import json + value = json.dumps({}) + """ + ) + sanitized, stripped = _strip_redundant_sandbox_imports_in_yaml(multi_block) + assert sorted(stripped) == ["asyncio", "json"] + assert "import asyncio" not in sanitized + assert "import json" not in sanitized + + def test_no_change_returns_original_text(self) -> None: + sanitized, stripped = _strip_redundant_sandbox_imports_in_yaml(_SAFE_CODE_YAML) + assert stripped == [] + assert sanitized == _SAFE_CODE_YAML diff --git a/tests/unit/test_copilot_code_block_synthesis.py b/tests/unit/test_copilot_code_block_synthesis.py index f88cbcd8f..be5928601 100644 --- a/tests/unit/test_copilot_code_block_synthesis.py +++ b/tests/unit/test_copilot_code_block_synthesis.py @@ -20,6 +20,7 @@ from skyvern.forge.sdk.copilot.code_block_synthesis import ( _SYNTHESIZED_BLOCK_LABEL, CREDENTIAL_FILL_TOOL_NAME, _get_by_role_expr, + _get_by_role_expr_strict, build_synthesized_artifact_metadata, code_contains_credential_fill, is_optional_dismissal_only_trajectory, @@ -263,7 +264,7 @@ class TestLocatorSynthesis: ] result = synthesize_code_block(trajectory, strict_selectors=True) assert result is not None - assert 'await page.get_by_role("link", name="View Statements").click()' in result.code + assert 'await page.get_by_role("link", name="View Statements", exact=True).click()' in result.code assert ".first" not in result.code assert result.diagnostics.dropped_interactions == [] provenance = [p for p in result.diagnostics.locator_provenance if p.get("source") == "aria_role_name"] @@ -271,7 +272,7 @@ class TestLocatorSynthesis: { "trajectory_index": 1, "selector": "a", - "emitted_literal": _get_by_role_expr("link", "View Statements"), + "emitted_literal": _get_by_role_expr_strict("link", "View Statements"), "source": "aria_role_name", "role": "link", "name": "View Statements", @@ -291,7 +292,7 @@ class TestLocatorSynthesis: ] result = synthesize_code_block(trajectory, strict_selectors=True) assert result is not None - assert 'await page.get_by_role("link", name="Continue").click()' in result.code + assert 'await page.get_by_role("link", name="Continue", exact=True).click()' in result.code assert result.diagnostics.dropped_interactions == [] def test_strict_bare_selector_without_role_name_is_still_dropped(self) -> None: @@ -321,10 +322,42 @@ class TestLocatorSynthesis: result = synthesize_code_block(trajectory, strict_selectors=True) assert result is not None assert result.diagnostics.dropped_interactions == [] - emitted = _get_by_role_expr("link", 'Say "hi"\nplease') + emitted = _get_by_role_expr_strict("link", 'Say "hi"\nplease') + assert "exact=True" in emitted assert "\n" not in emitted assert f"await {emitted}.click()" in result.code + def test_strict_reanchor_emits_exact_name_match_for_repeated_affordance(self) -> None: + # AC1: a re-anchored named get_by_role on a page with a repeated accessible name must emit an + # exact (single (role, name) group) match, never the substring default that over-matches. + trajectory = [ + _interaction("click", selector="#open", source_url="https://example.com/billing"), + _interaction( + "click", + selector="a", + source_url="https://example.com/billing", + role="link", + accessible_name="Download", + ), + ] + result = synthesize_code_block(trajectory, strict_selectors=True) + assert result is not None + assert 'await page.get_by_role("link", name="Download", exact=True).click()' in result.code + assert ".nth(" not in result.code + assert result.diagnostics.dropped_interactions == [] + provenance = [p for p in result.diagnostics.locator_provenance if p.get("source") == "aria_role_name"] + assert provenance == [ + { + "trajectory_index": 1, + "selector": "a", + "emitted_literal": _get_by_role_expr_strict("link", "Download"), + "source": "aria_role_name", + "role": "link", + "name": "Download", + } + ] + assert provenance[0]["emitted_literal"] != _get_by_role_expr("link", "Download") + class TestActionSynthesis: def test_type_text_becomes_param_slot_fill(self) -> None: diff --git a/tests/unit/test_copilot_hooks.py b/tests/unit/test_copilot_hooks.py index d808bdc9b..992556c1e 100644 --- a/tests/unit/test_copilot_hooks.py +++ b/tests/unit/test_copilot_hooks.py @@ -14,9 +14,11 @@ from structlog.testing import capture_logs from skyvern.forge.sdk.copilot import hooks as hooks_module from skyvern.forge.sdk.copilot.blocker_signal import CopilotToolBlockerSignal +from skyvern.forge.sdk.copilot.code_block_synthesis import synthesize_code_block from skyvern.forge.sdk.copilot.config import BlockAuthoringPolicy from skyvern.forge.sdk.copilot.enforcement import CopilotGoalSatisfied from skyvern.forge.sdk.copilot.hooks import CopilotRunHooks +from skyvern.forge.sdk.copilot.tools import _click_post_hook, _click_pre_hook from skyvern.forge.sdk.copilot.turn_halt import CopilotTurnHalt, turn_halt_from_blocker_signal @@ -1004,6 +1006,7 @@ class TestBrowserInteractionObservationHooks: ctx = SimpleNamespace( pending_browser_interaction_observation=None, pending_scout_typed_value=None, + pending_scout_role_name=None, discovery_mcp_server=None, scouted_interactions=[], scout_trajectory=[], @@ -1036,6 +1039,7 @@ class TestBrowserInteractionObservationHooks: url="https://example.com/results", ), pending_scout_typed_value=None, + pending_scout_role_name=None, discovery_mcp_server=None, scouted_interactions=[], scout_trajectory=[], @@ -1089,7 +1093,9 @@ class TestScoutedInteractionCapture: ns = SimpleNamespace( pending_browser_interaction_observation=None, pending_scout_typed_value=None, + pending_scout_role_name=None, discovery_mcp_server=None, + browser_session_id=None, scouted_interactions=[], scout_trajectory=[], observed_browser_urls=[], @@ -1212,6 +1218,83 @@ class TestScoutedInteractionCapture: {"tool_name": "click", "selector": "#add-to-cart", "source_url": "https://example.com/product"} ] + @pytest.mark.asyncio + async def test_navigating_bare_click_records_prenav_role_name(self) -> None: + ctx = self._ctx(source_url="https://example.com/billing") + ctx.pending_scout_role_name = ("a", "link", "View Printable Statement") + await _click_post_hook( + {"ok": True, "data": {"selector": "a"}}, + {"browser_context": {"url": "https://example.com/statement.pdf", "title": "Statement"}}, + ctx, + ) + assert ctx.scouted_interactions == [ + { + "tool_name": "click", + "selector": "a", + "source_url": "https://example.com/billing", + "role": "link", + "accessible_name": "View Printable Statement", + } + ] + assert ctx.pending_scout_role_name is None + + @pytest.mark.asyncio + async def test_prenav_role_name_lets_strict_synthesis_emit_get_by_role(self) -> None: + ctx = self._ctx(source_url="https://example.com/billing") + ctx.pending_scout_role_name = ("a", "link", "View Printable Statement") + ctx.scout_trajectory = [ + { + "tool_name": "click", + "selector": "#statement-row", + "source_url": "https://example.com/billing", + "trajectory_index": 0, + } + ] + await _click_post_hook( + {"ok": True, "data": {"selector": "a"}}, + {"browser_context": {"url": "https://example.com/statement.pdf", "title": "Statement"}}, + ctx, + ) + result = synthesize_code_block(ctx.scout_trajectory, strict_selectors=True) + assert result is not None + assert 'await page.get_by_role("link", name="View Printable Statement", exact=True).click()' in result.code + assert result.diagnostics.dropped_interactions == [] + + @pytest.mark.asyncio + async def test_prenav_role_name_stash_ignored_on_selector_mismatch(self) -> None: + ctx = self._ctx(source_url="https://example.com/billing") + ctx.pending_scout_role_name = ("a", "link", "View Printable Statement") + await _click_post_hook( + {"ok": True, "data": {"selector": "#concrete-row"}}, + {"browser_context": {"url": "https://example.com/statement.pdf", "title": "Statement"}}, + ctx, + ) + recorded = ctx.scouted_interactions[-1] + assert "role" not in recorded + assert "accessible_name" not in recorded + assert ctx.pending_scout_role_name is None + + @pytest.mark.asyncio + async def test_click_pre_hook_stashes_role_name_from_role_selector(self) -> None: + ctx = self._ctx() + ctx.pending_scout_source_url = None + await _click_pre_hook({"selector": 'role=link[name="Continue"]'}, ctx) + assert ctx.pending_scout_role_name == ('role=link[name="Continue"]', "link", "Continue") + + @pytest.mark.asyncio + async def test_click_pre_hook_stashes_role_name_from_bare_css_via_browser_read( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + # Bare CSS selector cannot be parsed, so the stash comes from the TIER-2 pre-navigation read. + async def _fake_read(_ctx: object, _selector: str, *, timeout_seconds: float) -> tuple[str, str]: + return "link", "Download" + + monkeypatch.setattr("skyvern.forge.sdk.copilot.tools.scouting._capture_accessible_role_name", _fake_read) + ctx = self._ctx() + ctx.pending_scout_source_url = None + await _click_pre_hook({"selector": "a"}, ctx) + assert ctx.pending_scout_role_name == ("a", "link", "Download") + @pytest.mark.asyncio async def test_click_post_hook_omits_source_url_when_unavailable(self) -> None: from skyvern.forge.sdk.copilot.tools import _click_post_hook diff --git a/tests/unit/test_copilot_scout_act_observe.py b/tests/unit/test_copilot_scout_act_observe.py index 5a379aab5..60ce78de4 100644 --- a/tests/unit/test_copilot_scout_act_observe.py +++ b/tests/unit/test_copilot_scout_act_observe.py @@ -68,6 +68,7 @@ def _ctx(*, server: Any = None, source_url: str | None = _SOURCE_URL) -> SimpleN return SimpleNamespace( pending_browser_interaction_observation=None, pending_scout_typed_value=None, + pending_scout_role_name=None, discovery_mcp_server=server, scouted_interactions=[], scout_trajectory=[],