Spaces:

ifroyou
/

chat-browser-bridge

Running

Add isolated browser JavaScript, keyboard and pointer tools (#1)

6e4eb30 about 1 month ago

9.72 kB

from __future__ import annotations

import asyncio
from dataclasses import dataclass, field
from typing import Any

from browser_core import BrowserBridgeError, BrowserManager

POINTER_BACKEND = "playwright-page-mouse"
_ALLOWED_BUTTONS = {"left", "middle", "right"}
_INTERACTIVE_SELECTOR = 'a,button,input,textarea,select,[role="button"],[role="link"]'


@dataclass
class PointerState:
    x: int = 0
    y: int = 0
    buttons: set[str] = field(default_factory=set)


_POINTER_STATES: dict[tuple[str, str], PointerState] = {}


def _button(value: str) -> str:
    selected = str(value or "left").strip().casefold()
    if selected not in _ALLOWED_BUTTONS:
        raise BrowserBridgeError("POINTER_BUTTON_INVALID", "button must be left, middle, or right.")
    return selected


def _state(owner: str, session_id: str) -> PointerState:
    return _POINTER_STATES.setdefault((owner.casefold(), session_id), PointerState())


async def _point(manager: BrowserManager, session: Any, x: int, y: int) -> tuple[Any, int, int]:
    page = await manager._sync_page(session)
    viewport = page.viewport_size or {"width": 1280, "height": 900}
    px, py = int(x), int(y)
    width = int(viewport.get("width") or 0)
    height = int(viewport.get("height") or 0)
    if px < 0 or py < 0 or px >= width or py >= height:
        raise BrowserBridgeError(
            "POINTER_COORDINATE_OUT_OF_VIEWPORT",
            "Pointer coordinates must be inside the current isolated cloud-browser viewport.",
            details={"x": px, "y": py, "viewport_width": width, "viewport_height": height},
        )
    return page, px, py


async def _guard_raw_pointer_target(page: Any, x: int, y: int) -> None:
    """Fail closed on normal interactive controls; raw pointer is for canvas/custom surfaces only."""
    controls = page.locator(_INTERACTIVE_SELECTOR)
    try:
        count = min(await controls.count(), 200)
    except Exception as exc:
        raise BrowserBridgeError(
            "POINTER_TARGET_GUARD_UNAVAILABLE",
            "The browser could not verify the raw-pointer target safely.",
            details={"error_type": type(exc).__name__},
        ) from exc
    for index in range(count):
        control = controls.nth(index)
        try:
            box = await control.bounding_box(timeout=400)
        except Exception:
            continue
        if not box:
            continue
        left = float(box.get("x") or 0)
        top = float(box.get("y") or 0)
        right = left + float(box.get("width") or 0)
        bottom = top + float(box.get("height") or 0)
        if not (left <= x < right and top <= y < bottom):
            continue
        attrs: list[str] = []
        for name in ("type", "role", "name", "aria-label", "title", "placeholder", "autocomplete"):
            try:
                value = await control.get_attribute(name, timeout=300)
            except Exception:
                value = None
            if value:
                attrs.append(str(value)[:200])
        try:
            text = (await control.inner_text(timeout=300))[:200]
        except Exception:
            text = ""
        raise BrowserBridgeError(
            "POINTER_INTERACTIVE_CONTROL_BLOCKED",
            "Raw pointer presses/drags are restricted to non-standard browser surfaces such as canvas. Use reviewed DOM tools for links, buttons, form fields, uploads, and submit controls.",
            details={"target_signature": " ".join([*attrs, text]).strip()[:500]},
        )


def _result(session: Any, state: PointerState, *, status: str, button: str | None = None,
            duration_ms: int | None = None, steps: int | None = None) -> dict[str, Any]:
    return {
        "ok": True,
        "status": status,
        "session_id": session.session_id,
        "url": session.page.url,
        "x": state.x,
        "y": state.y,
        "button": button,
        "buttons_down": sorted(state.buttons),
        "duration_ms": duration_ms,
        "steps": steps,
        "pointer_backend": POINTER_BACKEND,
        "desktop_input_used": False,
        "submitted": False,
    }


async def mouse_move(manager: BrowserManager, owner: str, session_id: str, x: int, y: int) -> dict[str, Any]:
    session = await manager.get(owner, session_id, touch=False)
    async with session.lock:
        await manager.revalidate_locked_session(owner, session_id, session)
        page, px, py = await _point(manager, session, x, y)
        await page.mouse.move(px, py)
        state = _state(owner, session_id)
        state.x, state.y = px, py
        return _result(session, state, status="BROWSER_POINTER_MOVED")


async def mouse_down(manager: BrowserManager, owner: str, session_id: str, button: str = "left",
                     *, user_confirmed_pointer_action: bool = False) -> dict[str, Any]:
    selected = _button(button)
    session = await manager.get(owner, session_id, touch=False)
    async with session.lock:
        await manager.revalidate_locked_session(owner, session_id, session)
        state = _state(owner, session_id)
        page, px, py = await _point(manager, session, state.x, state.y)
        if selected in state.buttons:
            raise BrowserBridgeError("POINTER_BUTTON_ALREADY_DOWN", "This pointer button is already held down.")
        await _guard_raw_pointer_target(page, px, py)
        await page.mouse.down(button=selected)
        state.buttons.add(selected)
        return _result(session, state, status="BROWSER_POINTER_DOWN", button=selected)


async def mouse_up(manager: BrowserManager, owner: str, session_id: str, button: str = "left") -> dict[str, Any]:
    selected = _button(button)
    session = await manager.get(owner, session_id, touch=False)
    async with session.lock:
        await manager.revalidate_locked_session(owner, session_id, session)
        state = _state(owner, session_id)
        page, _px, _py = await _point(manager, session, state.x, state.y)
        if selected not in state.buttons:
            raise BrowserBridgeError("POINTER_BUTTON_NOT_DOWN", "This pointer button is not currently held down.")
        await page.mouse.up(button=selected)
        state.buttons.discard(selected)
        return _result(session, state, status="BROWSER_POINTER_UP", button=selected)


async def mouse_click(manager: BrowserManager, owner: str, session_id: str, x: int, y: int,
                      *, button: str = "left", click_count: int = 1) -> dict[str, Any]:
    """Atomically click a canvas/custom surface in the isolated cloud browser.
    Standard browser controls remain guarded so ordinary pages retain explicit semantic
    controls for submit, authentication and sensitive fields.  This is deliberately
    a browser-page mouse operation, never Windows desktop input.
    """
    selected = _button(button)
    count = max(1, min(2, int(click_count)))
    session = await manager.get(owner, session_id, touch=False)
    async with session.lock:
        await manager.revalidate_locked_session(owner, session_id, session)
        state = _state(owner, session_id)
        page, px, py = await _point(manager, session, x, y)
        if state.buttons:
            raise BrowserBridgeError(
                "POINTER_BUTTON_STATE_BUSY",
                "Release held pointer buttons before starting an atomic pointer click.",
            )
        await _guard_raw_pointer_target(page, px, py)
        await page.mouse.click(px, py, button=selected, click_count=count)
        state.x, state.y = px, py
        return _result(
            session,
            state,
            status="BROWSER_POINTER_CLICKED",
            button=selected,
            steps=count,
        )


async def drag(manager: BrowserManager, owner: str, session_id: str,
               start_x: int, start_y: int, end_x: int, end_y: int, *, duration_ms: int = 400,
               button: str = "left", user_confirmed_pointer_action: bool = False) -> dict[str, Any]:
    selected = _button(button)
    duration = max(0, min(5000, int(duration_ms)))
    steps = max(1, min(120, round(duration / 16) if duration else 1))
    session = await manager.get(owner, session_id, touch=False)
    async with session.lock:
        await manager.revalidate_locked_session(owner, session_id, session)
        state = _state(owner, session_id)
        page, sx, sy = await _point(manager, session, start_x, start_y)
        _page2, ex, ey = await _point(manager, session, end_x, end_y)
        if state.buttons:
            raise BrowserBridgeError(
                "POINTER_BUTTON_STATE_BUSY",
                "Release held pointer buttons before starting an atomic drag.",
            )
        await _guard_raw_pointer_target(page, sx, sy)
        await _guard_raw_pointer_target(page, ex, ey)
        await page.mouse.move(sx, sy)
        state.x, state.y = sx, sy
        down = False
        try:
            await page.mouse.down(button=selected)
            down = True
            state.buttons.add(selected)
            pause = duration / steps / 1000 if duration else 0
            for index in range(1, steps + 1):
                px = round(sx + (ex - sx) * index / steps)
                py = round(sy + (ey - sy) * index / steps)
                await page.mouse.move(px, py)
                state.x, state.y = px, py
                if pause and index < steps:
                    await asyncio.sleep(pause)
        finally:
            if down:
                await page.mouse.up(button=selected)
                state.buttons.discard(selected)
        return _result(
            session,
            state,
            status="BROWSER_POINTER_DRAGGED",
            button=selected,
            duration_ms=duration,
            steps=steps,
        )

	from __future__ import annotations

	import asyncio
	from dataclasses import dataclass, field
	from typing import Any

	from browser_core import BrowserBridgeError, BrowserManager

	POINTER_BACKEND = "playwright-page-mouse"
	_ALLOWED_BUTTONS = {"left", "middle", "right"}
	_INTERACTIVE_SELECTOR = 'a,button,input,textarea,select,[role="button"],[role="link"]'


	@dataclass
	class PointerState:
	x: int = 0
	y: int = 0
	buttons: set[str] = field(default_factory=set)


	_POINTER_STATES: dict[tuple[str, str], PointerState] = {}


	def _button(value: str) -> str:
	selected = str(value or "left").strip().casefold()
	if selected not in _ALLOWED_BUTTONS:
	raise BrowserBridgeError("POINTER_BUTTON_INVALID", "button must be left, middle, or right.")
	return selected


	def _state(owner: str, session_id: str) -> PointerState:
	return _POINTER_STATES.setdefault((owner.casefold(), session_id), PointerState())


	async def _point(manager: BrowserManager, session: Any, x: int, y: int) -> tuple[Any, int, int]:
	page = await manager._sync_page(session)
	viewport = page.viewport_size or {"width": 1280, "height": 900}
	px, py = int(x), int(y)
	width = int(viewport.get("width") or 0)
	height = int(viewport.get("height") or 0)
	if px < 0 or py < 0 or px >= width or py >= height:
	raise BrowserBridgeError(
	"POINTER_COORDINATE_OUT_OF_VIEWPORT",
	"Pointer coordinates must be inside the current isolated cloud-browser viewport.",
	details={"x": px, "y": py, "viewport_width": width, "viewport_height": height},
	)
	return page, px, py


	async def _guard_raw_pointer_target(page: Any, x: int, y: int) -> None:
	"""Fail closed on normal interactive controls; raw pointer is for canvas/custom surfaces only."""
	controls = page.locator(_INTERACTIVE_SELECTOR)
	try:
	count = min(await controls.count(), 200)
	except Exception as exc:
	raise BrowserBridgeError(
	"POINTER_TARGET_GUARD_UNAVAILABLE",
	"The browser could not verify the raw-pointer target safely.",
	details={"error_type": type(exc).__name__},
	) from exc
	for index in range(count):
	control = controls.nth(index)
	try:
	box = await control.bounding_box(timeout=400)
	except Exception:
	continue
	if not box:
	continue
	left = float(box.get("x") or 0)
	top = float(box.get("y") or 0)
	right = left + float(box.get("width") or 0)
	bottom = top + float(box.get("height") or 0)
	if not (left <= x < right and top <= y < bottom):
	continue
	attrs: list[str] = []
	for name in ("type", "role", "name", "aria-label", "title", "placeholder", "autocomplete"):
	try:
	value = await control.get_attribute(name, timeout=300)
	except Exception:
	value = None
	if value:
	attrs.append(str(value)[:200])
	try:
	text = (await control.inner_text(timeout=300))[:200]
	except Exception:
	text = ""
	raise BrowserBridgeError(
	"POINTER_INTERACTIVE_CONTROL_BLOCKED",
	"Raw pointer presses/drags are restricted to non-standard browser surfaces such as canvas. Use reviewed DOM tools for links, buttons, form fields, uploads, and submit controls.",
	details={"target_signature": " ".join([*attrs, text]).strip()[:500]},
	)


	def _result(session: Any, state: PointerState, *, status: str, button: str \| None = None,
	duration_ms: int \| None = None, steps: int \| None = None) -> dict[str, Any]:
	return {
	"ok": True,
	"status": status,
	"session_id": session.session_id,
	"url": session.page.url,
	"x": state.x,
	"y": state.y,
	"button": button,
	"buttons_down": sorted(state.buttons),
	"duration_ms": duration_ms,
	"steps": steps,
	"pointer_backend": POINTER_BACKEND,
	"desktop_input_used": False,
	"submitted": False,
	}


	async def mouse_move(manager: BrowserManager, owner: str, session_id: str, x: int, y: int) -> dict[str, Any]:
	session = await manager.get(owner, session_id, touch=False)
	async with session.lock:
	await manager.revalidate_locked_session(owner, session_id, session)
	page, px, py = await _point(manager, session, x, y)
	await page.mouse.move(px, py)
	state = _state(owner, session_id)
	state.x, state.y = px, py
	return _result(session, state, status="BROWSER_POINTER_MOVED")


	async def mouse_down(manager: BrowserManager, owner: str, session_id: str, button: str = "left",
	*, user_confirmed_pointer_action: bool = False) -> dict[str, Any]:
	selected = _button(button)
	session = await manager.get(owner, session_id, touch=False)
	async with session.lock:
	await manager.revalidate_locked_session(owner, session_id, session)
	state = _state(owner, session_id)
	page, px, py = await _point(manager, session, state.x, state.y)
	if selected in state.buttons:
	raise BrowserBridgeError("POINTER_BUTTON_ALREADY_DOWN", "This pointer button is already held down.")
	await _guard_raw_pointer_target(page, px, py)
	await page.mouse.down(button=selected)
	state.buttons.add(selected)
	return _result(session, state, status="BROWSER_POINTER_DOWN", button=selected)


	async def mouse_up(manager: BrowserManager, owner: str, session_id: str, button: str = "left") -> dict[str, Any]:
	selected = _button(button)
	session = await manager.get(owner, session_id, touch=False)
	async with session.lock:
	await manager.revalidate_locked_session(owner, session_id, session)
	state = _state(owner, session_id)
	page, _px, _py = await _point(manager, session, state.x, state.y)
	if selected not in state.buttons:
	raise BrowserBridgeError("POINTER_BUTTON_NOT_DOWN", "This pointer button is not currently held down.")
	await page.mouse.up(button=selected)
	state.buttons.discard(selected)
	return _result(session, state, status="BROWSER_POINTER_UP", button=selected)


	async def mouse_click(manager: BrowserManager, owner: str, session_id: str, x: int, y: int,
	*, button: str = "left", click_count: int = 1) -> dict[str, Any]:
	"""Atomically click a canvas/custom surface in the isolated cloud browser.

	Standard browser controls remain guarded so ordinary pages retain explicit semantic
	controls for submit, authentication and sensitive fields. This is deliberately
	a browser-page mouse operation, never Windows desktop input.
	"""
	selected = _button(button)
	count = max(1, min(2, int(click_count)))
	session = await manager.get(owner, session_id, touch=False)
	async with session.lock:
	await manager.revalidate_locked_session(owner, session_id, session)
	state = _state(owner, session_id)
	page, px, py = await _point(manager, session, x, y)
	if state.buttons:
	raise BrowserBridgeError(
	"POINTER_BUTTON_STATE_BUSY",
	"Release held pointer buttons before starting an atomic pointer click.",
	)
	await _guard_raw_pointer_target(page, px, py)
	await page.mouse.click(px, py, button=selected, click_count=count)
	state.x, state.y = px, py
	return _result(
	session,
	state,
	status="BROWSER_POINTER_CLICKED",
	button=selected,
	steps=count,
	)


	async def drag(manager: BrowserManager, owner: str, session_id: str,
	start_x: int, start_y: int, end_x: int, end_y: int, *, duration_ms: int = 400,
	button: str = "left", user_confirmed_pointer_action: bool = False) -> dict[str, Any]:
	selected = _button(button)
	duration = max(0, min(5000, int(duration_ms)))
	steps = max(1, min(120, round(duration / 16) if duration else 1))
	session = await manager.get(owner, session_id, touch=False)
	async with session.lock:
	await manager.revalidate_locked_session(owner, session_id, session)
	state = _state(owner, session_id)
	page, sx, sy = await _point(manager, session, start_x, start_y)
	_page2, ex, ey = await _point(manager, session, end_x, end_y)
	if state.buttons:
	raise BrowserBridgeError(
	"POINTER_BUTTON_STATE_BUSY",
	"Release held pointer buttons before starting an atomic drag.",
	)
	await _guard_raw_pointer_target(page, sx, sy)
	await _guard_raw_pointer_target(page, ex, ey)
	await page.mouse.move(sx, sy)
	state.x, state.y = sx, sy
	down = False
	try:
	await page.mouse.down(button=selected)
	down = True
	state.buttons.add(selected)
	pause = duration / steps / 1000 if duration else 0
	for index in range(1, steps + 1):
	px = round(sx + (ex - sx) * index / steps)
	py = round(sy + (ey - sy) * index / steps)
	await page.mouse.move(px, py)
	state.x, state.y = px, py
	if pause and index < steps:
	await asyncio.sleep(pause)
	finally:
	if down:
	await page.mouse.up(button=selected)
	state.buttons.discard(selected)
	return _result(
	session,
	state,
	status="BROWSER_POINTER_DRAGGED",
	button=selected,
	duration_ms=duration,
	steps=steps,
	)