Skip to content
Open
Show file tree
Hide file tree
Changes from 2 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
66 changes: 66 additions & 0 deletions packages/notte-browser/src/notte_browser/controller.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
from notte_core.storage import BaseStorage
from notte_core.utils.code import text_contains_tabs
from notte_core.utils.platform import platform_control_key
from pydantic import BaseModel, Field
from typing_extensions import final

from notte_browser.captcha import CaptchaHandler
Expand All @@ -59,15 +60,70 @@
from notte_browser.window import BrowserWindow


class ActionBlocklist(BaseModel):
"""Blocklist policy for browser actions.

- disallow_types: exact action types to block (e.g., "goto", "click")
- keywords: element text keywords to block for interaction actions (case-insensitive)
"""

disallow_types: set[str] = Field(default_factory=set)
keywords: list[str] = Field(default_factory=list)

def is_type_blocked(self, action: BaseAction) -> bool:
return action.type in self.disallow_types

def is_keyword_blocked(self, action: BaseAction, prev_snapshot: BrowserSnapshot | None) -> bool:
if prev_snapshot is None:
return False
if not isinstance(action, InteractionAction):
return False
# id in InteractionAction is always a string; check emptiness only
if len(action.id) == 0:
return False
node = prev_snapshot.dom_node.find(action.id)
if node is None:
return False
# Aggregate potential text sources for matching
texts: list[str] = []
try:
texts.append(node.inner_text())
except Exception:
pass
try:
texts.append(node.text)
except Exception:
pass
try:
if node.attributes is not None:
for key in ("title", "aria_label", "name", "placeholder"):
val = getattr(node.attributes, key, None)
if isinstance(val, str):
texts.append(val)
except Exception:
pass
# `texts` is a list of strings; join directly
text = " ".join(texts).lower()
for kw in self.keywords:
if kw.lower() in text:
return True
return False

def is_blocked(self, action: BaseAction, prev_snapshot: BrowserSnapshot | None) -> bool:
return self.is_type_blocked(action) or self.is_keyword_blocked(action, prev_snapshot)


@final
class BrowserController:
def __init__(
self,
verbose: bool,
storage: BaseStorage | None = None,
blocklist: ActionBlocklist | None = None,
) -> None:
self.verbose: bool = verbose
self.storage: BaseStorage | None = storage
self.blocklist: ActionBlocklist | None = blocklist

async def switch_tab(self, window: BrowserWindow, tab_index: int) -> None:
context = window.page.context
Expand Down Expand Up @@ -373,6 +429,16 @@ async def execute(
context = window.page.context
num_pages = len(context.pages)
retval = True

# Enforce blocklist policy before execution
if self.blocklist is not None and self.blocklist.is_blocked(action, prev_snapshot):
# Raise an explicit error handled by session
raise ActionExecutionError(
action_id=getattr(action, "id", action.type),
url=window.page.url,
reason="Unauthorized action (blocked by policy)",
)

match action:
case InteractionAction():
retval = await self.execute_interaction_action(window, action, prev_snapshot)
Expand Down
7 changes: 5 additions & 2 deletions packages/notte-browser/src/notte_browser/session.py
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,7 @@

from notte_browser.action_selection.pipe import ActionSelectionPipe
from notte_browser.captcha import CaptchaHandler
from notte_browser.controller import BrowserController
from notte_browser.controller import ActionBlocklist, BrowserController
from notte_browser.dom.locate import locate_element
from notte_browser.errors import (
BrowserNotStartedError,
Expand Down Expand Up @@ -114,14 +114,17 @@ def __init__(
tools: list[BaseTool] | None = None,
window: BrowserWindow | None = None,
keep_alive: bool = False,
blocklist: ActionBlocklist | None = None,
**data: Unpack[SessionStartRequestDict],
) -> None:
self._request: SessionStartRequest = SessionStartRequest.model_validate(data)
if self._request.solve_captchas and not CaptchaHandler.is_available:
raise CaptchaSolverNotAvailableError()
self.screenshot_type: ScreenshotType = self._request.screenshot_type
self._window: BrowserWindow | None = window
self.controller: BrowserController = BrowserController(verbose=config.verbose, storage=storage)
self.controller: BrowserController = BrowserController(
verbose=config.verbose, storage=storage, blocklist=blocklist
)
self.storage: BaseStorage | None = storage
llmserve = LLMService.from_config(perception_type=perception_type)
self._action_space_pipe: MainActionSpacePipe = MainActionSpacePipe(llmserve=llmserve)
Expand Down
Loading
Loading