Is browser-use/browser-use safe?

exec(_NAMESPACE_IMPORTS, ns)

exec(code, ns)

exec('from browser_harness.helpers import *', ns)

tree = compile(code, f'<cell {n}>', 'exec', flags=ast.PyCF_ONLY_AST | flags)

return compile(tree, f'<cell {n}>', 'exec', flags=flags)

return loop.run_until_complete(eval(code_obj, ns))

return exec(code_obj, ns)

obj = compile(msg['code'], '<finish>', 'eval', flags=ast.PyCF_ALLOW_TOP_LEVEL_AWAIT)

value = _run(obj, ns, loop) if obj.co_flags & inspect.CO_COROUTINE else eval(obj, ns)

types.Tool(
			name='browser_exec',

types.Tool(
				name='browser_exec',

types.Tool(
				name='browser_screenshot',

types.Tool(
				name='browser_login',

curl -LsSf https://astral.sh/uv/install.sh | sh

self.proc = subprocess.Popen(
					[binary, *args],
					stdin=subprocess.DEVNULL,
					stdout=subprocess.DEVNULL,
					stderr=err,
					start_new_session=sys.platform != 'win32',  # its own process group: _kill reaches every helper
					cr …

return subprocess.Popen(
			[sys.executable, '-P', '-c', WATCHDOG, *args],
			env={**os.environ, 'BU_CLEANUP_BROWSER_ID': cloud_id},
			stdin=subprocess.PIPE,
			stdout=subprocess.DEVNULL,
			stderr=subprocess.DEVNULL,
			start_new_session= …

result = subprocess.run(cmd)

r = subprocess.run(
				cmd,
				stdin=subprocess.DEVNULL,
				capture_output=True,
				text=True,
				errors='replace',
				env=env,
				timeout=INSTALL_TIMEOUT_S,
			)

subprocess.Popen(command, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

subprocess.Popen(
					args,
					env=env,
					stdin=subprocess.PIPE,
					stdout=subprocess.DEVNULL,
					stderr=subprocess.DEVNULL,
					**detach,
				)

subprocess.run(['taskkill', '/F', '/T', '/PID', str(pid)], capture_output=True)

self.proc = subprocess.Popen(  # noqa: ASYNC220 - asyncio forks synchronously too; see self.proc
				# -P: the workspace is not sys.path[0]: a json.py there must not shadow the stdlib (worker.py appends it)
				[sys.executable, '-P', '-m', …

probe = subprocess.run([uv, 'tool', 'dir'], capture_output=True, text=True)

result = subprocess.run([uv, *args])

subprocess.Popen([sys.executable, '-I', '-c', EXIT_SENDER, f'{POSTHOG_HOST}/batch/', *files], **kw)

value = json.loads((self.repl.state / 'partial.json').read_text(encoding='utf-8'))['value']

log = download_records(log.read_text(encoding='utf-8').splitlines())

self._agent_tabs.update(Path(self._tabs_file).read_text().split())

self._agent_tabs.update(Path(tabs_file).read_text().split())  # a follow-up run's sweep still knows them

lines = log.read_text(encoding='utf-8').split('\n')[:-1]

port, path = (base / 'DevToolsActivePort').read_text(encoding='utf-8', errors='replace').split()[:2]

port = (base / 'DevToolsActivePort').read_text(errors='replace').split()[0]

with open(real, 'rb') as f:

cache = json.loads((root / 'Local State').read_text(encoding='utf-8', errors='replace'))['profile']['info_cache']

lines = (user_data_dir / 'DevToolsActivePort').read_text(encoding='utf-8', errors='replace').splitlines()

if m := re.search(rb'DevTools listening on (ws://\S+)', log.read_bytes()):

tail = log.read_bytes()[-600:].decode(errors='replace').strip()

profiles = list((json.loads(path.read_text(encoding='utf-8')).get('browser_profile') or {}).values())

or 'docker' in Path('/proc/1/cgroup').read_text(encoding='utf-8', errors='replace').lower()

return _marker(name).read_text().strip()

return open(file, mode, buffering, encoding, *args, **kwargs)

with open(path, 'rb') as f:

files = [Path(p).read_bytes() for p in paths]

saved = json.loads(self.path.read_text(encoding='utf-8'))

with open(path, encoding='utf-8', errors='replace') as f:

for line in (STATE / 'downloads.jsonl').read_text(encoding='utf-8').splitlines():

return _add_image(p.read_bytes(), f'image/{ext}')

with open(path, mode, encoding='utf-8', newline='') as out:  # Windows: the cell already wrote \r\n; no \r\r\n

with open(out_path, encoding='utf-8', errors='replace') as f:

return skill_path.read_text(encoding='utf-8')

path.write_text(json.dumps(redact(archive, s.repl.secrets), ensure_ascii=False), encoding='utf-8')

f = open(path, 'xb')  # - exclusive create: two sessions never write the same file

os.unlink(tabs_file)

shutil.copy2(root / 'Local State', dest / 'Local State')

shutil.copytree(root / directory, dest / directory, ignore=SKIP_COPY, symlinks=True, dirs_exist_ok=True)

with open(d / OWNER, 'r+b') as f:

shutil.rmtree(d, ignore_errors=True)

self._owner = open(self.tmp / OWNER, 'wb')  # locked while this process lives: sweep_stale_dirs skips the dir

with open(log, 'wb') as err:

shutil.rmtree(self.tmp, ignore_errors=True)

shutil.rmtree(udd, ignore_errors=True)

shutil.rmtree(_own_chromium_dir(), ignore_errors=True)

tmp.write_text(json.dumps(data), encoding='utf-8')

log = open(self.state / 'worker.log', 'ab')  # noqa: ASYNC230 - handed to the child process

with contextlib.suppress(OSError), open(STATE / 'downloads.jsonl', 'a', encoding='utf-8') as f:

with contextlib.suppress(OSError), open(tabs_file, 'a') as f:

path.write_bytes(base64.b64decode(r['data']))

tmp.write_text(text, encoding='utf-8')

output_path.write_text(text, encoding='utf-8')

tmp.write_text(new_device_id, encoding='utf-8')

path.write_bytes(data)

tmp.write_text(json.dumps({**marker, 'props': props()}, default=str), encoding='utf-8')

cache.write_text(json.dumps({'latest': latest, 'checked_at': time.time()}), encoding='utf-8')

output_path.write_text(expected, encoding='utf-8')

import httpx

import urllib.request

req = urllib.request.Request(
		f'{CLOUD_API}{path}',
		method=method,
		data=json.dumps(body).encode() if body is not None else None,
		headers={'X-Browser-Use-API-Key': browser_use_api_key(), 'Content-Type': 'application/json'},
	)

with urllib.request.urlopen(url, timeout=min(30.0, left), context=tls_context()) as r:  # the timeout bounds each read

return urllib.parse.urlsplit(url)._replace(query='', fragment='').geturl()

items = _cloud(f'/profiles?pageSize=100&query={urllib.parse.quote(profile)}').get('items', [])

u = urllib.parse.urlsplit(cdp_url)  # a token in the query stays after /json/version

with urllib.request.urlopen(
		u._replace(path=u.path.rstrip('/') + '/json/version').geturl(), timeout=30, context=tls_context()
	) as r:

u = urllib.parse.urlsplit(cdp_url)

import urllib.request

local = urllib.request.build_opener(urllib.request.ProxyHandler({}))  # 127.0.0.1 itself, never an HTTP proxy

return urllib.parse.urlsplit(ws).path == path

import urllib.request

request = urllib.request.Request(f'{auth.auth_base()}/api/v4/browsers?pageSize=1', headers={'X-Browser-Use-API-Key': key})

import urllib.request

opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))  # an HTTP_PROXY never sees localhost

import urllib.request

u = urllib.parse.urlsplit(url)

if urllib.request.getproxies().get(u.scheme) and not _loopback(host) and not urllib.request.proxy_bypass(host):

import urllib.request

request = urllib.request.Request(url, headers={'Authorization': f'Bearer {key}'})

with urlopen_keyed(urllib.request.Request(url, headers=headers), PREFLIGHT_TIMEOUT_S):

import urllib.request

self.json_pages.add(urllib.parse.urlsplit(params['frame'].get('url') or '')._replace(query='').geturl())

return urllib.request.build_opener(CheckedRedirects, TLSHandler)

allowed-tools: Bash, Read, Task

command -v uv || curl -LsSf https://astral.sh/uv/install.sh | sh   # the uv installer, only if missing

{

server.add_request_handler('tools/list', types.PaginatedRequestParams, list_tools)

server.add_request_handler('tools/call', types.CallToolRequestParams, call_tool)

self.server.add_request_handler('tools/list', types.PaginatedRequestParams, handle_list_tools)

self.server.add_request_handler('tools/call', types.CallToolRequestParams, handle_call_tool)

types.Tool(
			name='browser_exec',

types.Tool(
				name='browser_exec',