Spaces:
Sleeping
Sleeping
File size: 7,000 Bytes
6e4eb30 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 | from __future__ import annotations
import asyncio
import json
import re
from typing import Any
from browser_core import BrowserBridgeError, BrowserManager
JAVASCRIPT_BACKEND = "playwright-page-evaluate-anonymous-cloud"
MAX_JAVASCRIPT_CHARS = 24_000
MAX_JAVASCRIPT_RESULT_BYTES = 128 * 1024
MAX_JAVASCRIPT_TIMEOUT_MS = 5_000
_FORBIDDEN_SOURCE = re.compile(
r"(?:document\s*\.\s*cookie|localStorage|sessionStorage|indexedDB|\bcaches\b|"
r"navigator\s*\.\s*credentials|\bpassword\b|\bpasscode\b|\botp\b|"
r"one[\s_-]*time[\s_-]*code|\bauthorization\b|\bbearer\b|\baccess[\s_-]*token\b|"
r"\brefresh[\s_-]*token\b|\bwebsocket\b|\beventsource\b|\bxmlhttprequest\b|"
r"\bnavigator\s*\.\s*sendBeacon\b)",
re.IGNORECASE,
)
def _source(value: str) -> str:
source = str(value or "")
if not source.strip():
raise BrowserBridgeError("JAVASCRIPT_EMPTY", "A JavaScript snippet is required.")
if len(source) > MAX_JAVASCRIPT_CHARS:
raise BrowserBridgeError(
"JAVASCRIPT_TOO_LARGE",
"JavaScript is limited to 24,000 characters in the isolated browser session.",
details={"max_characters": MAX_JAVASCRIPT_CHARS},
)
blocked = _FORBIDDEN_SOURCE.search(source)
if blocked:
raise BrowserBridgeError(
"JAVASCRIPT_SECRET_OR_NETWORK_API_BLOCKED",
"The JavaScript tool never reads browser storage, cookies, credentials, password/OTP fields, or direct network APIs.",
details={"blocked_token": blocked.group(0)[:80]},
)
return source
async def run_javascript(
manager: BrowserManager,
owner: str,
session_id: str,
script: str,
*,
user_confirmed_javascript_action: bool,
) -> dict[str, Any]:
if user_confirmed_javascript_action is not True:
raise BrowserBridgeError(
"JAVASCRIPT_ACTION_CONFIRMATION_REQUIRED",
"Confirm the exact JavaScript action on the current anonymous public cloud page before execution.",
)
source = _source(script)
session = await manager.get(owner, session_id, touch=False)
async with session.lock:
await manager.revalidate_locked_session(owner, session_id, session)
page = await manager._sync_page(session)
try:
raw = await asyncio.wait_for(
page.evaluate(
"""
async ({ source, maxBytes }) => {
const AsyncFunction = Object.getPrototypeOf(async function () {}).constructor;
const toSafeJson = (value) => {
const seen = new WeakSet();
const json = JSON.stringify(value === undefined ? null : value, (_key, item) => {
if (typeof item === 'bigint') return item.toString();
if (typeof item === 'function' || typeof item === 'symbol') return undefined;
if (item && typeof item === 'object') {
if (seen.has(item)) return '[Circular]';
seen.add(item);
}
return item;
});
return json === undefined ? 'null' : json;
};
const originalFetch = globalThis.fetch;
const originalOpen = globalThis.open;
const originalWebSocket = globalThis.WebSocket;
const originalXhr = globalThis.XMLHttpRequest;
const originalEventSource = globalThis.EventSource;
const originalBeacon = navigator.sendBeacon;
const deny = (name) => () => { throw new Error(`${name} is unavailable in this browser JavaScript tool`); };
try {
globalThis.fetch = deny('fetch');
globalThis.open = deny('window.open');
globalThis.WebSocket = undefined;
globalThis.XMLHttpRequest = undefined;
globalThis.EventSource = undefined;
navigator.sendBeacon = deny('navigator.sendBeacon');
const run = new AsyncFunction(`"use strict"; return await (async () => { ${source}\n })();`);
const value = await run();
const resultJson = toSafeJson(value);
if (new TextEncoder().encode(resultJson).length > maxBytes) {
throw new Error(`JavaScript result exceeds ${maxBytes} bytes`);
}
return { ok: true, result_json: resultJson, result_bytes: new TextEncoder().encode(resultJson).length };
} finally {
globalThis.fetch = originalFetch;
globalThis.open = originalOpen;
globalThis.WebSocket = originalWebSocket;
globalThis.XMLHttpRequest = originalXhr;
globalThis.EventSource = originalEventSource;
navigator.sendBeacon = originalBeacon;
}
}
""",
{"source": source, "maxBytes": MAX_JAVASCRIPT_RESULT_BYTES},
),
timeout=MAX_JAVASCRIPT_TIMEOUT_MS / 1000,
)
except asyncio.TimeoutError as exc:
raise BrowserBridgeError(
"JAVASCRIPT_TIMEOUT",
"JavaScript exceeded the five-second cloud-browser limit.",
details={"timeout_ms": MAX_JAVASCRIPT_TIMEOUT_MS},
) from exc
except BrowserBridgeError:
raise
except Exception as exc:
raise BrowserBridgeError(
"JAVASCRIPT_EXECUTION_FAILED",
"JavaScript could not be completed in the isolated browser session.",
details={"error_type": type(exc).__name__},
) from exc
try:
result = json.loads(str(raw.get("result_json") or "null"))
except Exception as exc:
raise BrowserBridgeError(
"JAVASCRIPT_RESULT_INVALID",
"JavaScript returned a result that could not be safely serialized.",
) from exc
return {
"ok": True,
"status": "BROWSER_JAVASCRIPT_COMPLETED",
"session_id": session.session_id,
"url": page.url,
"result": result,
"result_bytes": int(raw.get("result_bytes") or 0),
"script_characters": len(source),
"javascript_backend": JAVASCRIPT_BACKEND,
"execution_scope": "anonymous-public-cloud-browser-session",
"desktop_input_used": False,
"user_edge_profile_used": False,
"cookies_or_profile_exposed": False,
"submitted": False,
}
|