"""Browser state, stale-reference, and filesystem policy tests."""
from __future__ import annotations
import asyncio
from dataclasses import replace
import os
from pathlib import Path
import socket
import tempfile
import unittest
from src.browser_session import BrowserSession, SelectedTab
from src.config import Config
from src.errors import (
CancelledError,
JavaScriptError,
PageScriptError,
StaleElementError,
UploadFileInvalidError,
UploadTooLargeError,
)
from src.rdp_connection import TabRecord
def config(**changes) -> Config:
"""Create a compact deterministic test configuration."""
value = Config(
host="127.0.0.1",
port=6000,
max_file_bytes=1024,
max_upload_bytes=2048,
upload_chunk_bytes=4,
max_elements=20,
max_text_chars=1000,
request_timeout_ms=1000,
log_level="error",
)
return replace(value, **changes)
class FakeRdp:
"""Return queued JSON envelopes without a live Firefox."""
connected = True
generation = 1
target_generation = 1
def __init__(self, envelopes):
self.envelopes = list(envelopes)
self.evaluations = []
async def evaluateJson(self, actor, expression, timeout_ms=None):
"""Pop the next scripted page result."""
self.evaluations.append((actor, expression, timeout_ms))
result = self.envelopes.pop(0)
if isinstance(result, Exception):
raise result
return result
async def listTabs(self):
"""Return one stable fixture tab for selected-page checks."""
return [TabRecord(
actor_id="tab",
title="Fixture",
url="https://example.test/",
)]
class StaleReferenceTest(unittest.IsolatedAsyncioTestCase):
"""Verify fingerprints reject a selector's replacement node."""
async def test_changed_accessible_name_is_stale(self) -> None:
snapshot = {
"ok": True,
"value": {
"title": "Fixture",
"url": "https://example.test/",
"text": "",
"text_truncated": False,
"elements": [{
"selector": "#action",
"tag_name": "button",
"role": None,
"input_type": None,
"accessible_name": "Original",
"href": "",
"value": None,
}],
},
}
changed = {
"ok": True,
"value": {
"selector": "#action",
"tag_name": "button",
"role": None,
"input_type": None,
"accessible_name": "Replacement",
"href": "",
},
}
rdp = FakeRdp([snapshot, changed])
browser = BrowserSession(config(), rdp)
browser._generation = 1
browser.selected_tab = SelectedTab(
"tab", "console", 1, "https://example.test/",
"Fixture", 1, 1,
)
await browser.snapshot(
include_all_text=False,
max_elements=10,
max_text_chars=100,
)
with self.assertRaises(StaleElementError):
await browser._verify("e1")
async def test_waiting_for_operation_lock_is_cancellable(self) -> None:
browser = BrowserSession(config(), FakeRdp([]))
await browser._operation_lock.acquire()
cancelled = False
def isCancelled() -> bool:
return cancelled
task = asyncio.create_task(browser.status(isCancelled))
await asyncio.sleep(0.06)
cancelled = True
try:
with self.assertRaises(CancelledError):
await asyncio.wait_for(task, 0.2)
finally:
browser._operation_lock.release()
class JavaScriptTest(unittest.IsolatedAsyncioTestCase):
"""Verify arbitrary JavaScript uses the selected content actor."""
def browser(self, rdp: FakeRdp) -> BrowserSession:
"""Create a browser with one already-selected stable tab."""
browser = BrowserSession(config(), rdp)
browser._generation = 1
browser.selected_tab = SelectedTab(
"tab",
"console",
1,
"https://example.test/",
"Fixture",
1,
1,
)
return browser
async def test_returns_result_and_forwards_timeout(self) -> None:
envelope = {
"ok": True,
"value": {"result": 42, "result_type": "number"},
}
rdp = FakeRdp([envelope])
browser = self.browser(rdp)
result = await browser.runJavaScript("21 * 2", 4321)
self.assertEqual(
result,
{"result": 42, "result_type": "number"},
)
actor, expression, timeout = rdp.evaluations[0]
self.assertEqual(actor, "console")
self.assertIn("21 * 2", expression)
self.assertEqual(timeout, 4321)
async def test_reports_caller_exception(self) -> None:
envelope = {
"ok": False,
"error": {
"code": "javascript_error",
"message": "fixture failure",
},
}
browser = self.browser(FakeRdp([envelope]))
with self.assertRaisesRegex(
JavaScriptError,
"fixture failure",
):
await browser.runJavaScript("throw new Error()", 1000)
async def test_retries_a_statement_body_after_parse_failure(self) -> None:
envelope = {
"ok": True,
"value": {"result": "Fixture", "result_type": "string"},
}
rdp = FakeRdp([PageScriptError("syntax error"), envelope])
browser = self.browser(rdp)
result = await browser.runJavaScript(
"return document.title;",
1000,
)
self.assertEqual(result, envelope["value"])
self.assertEqual(len(rdp.evaluations), 2)
self.assertNotIn("async () => {\nreturn", rdp.evaluations[0][1])
self.assertIn("async () => {\nreturn", rdp.evaluations[1][1])
class FilesystemSecurityTest(unittest.TestCase):
"""Verify upload path, type, symlink, and size policy."""
def setUp(self) -> None:
self.temporary = tempfile.TemporaryDirectory()
self.base = Path(self.temporary.name)
self.root = self.base / "allowed"
self.root.mkdir()
self.browser = BrowserSession(config(), FakeRdp([]))
def tearDown(self) -> None:
self.temporary.cleanup()
def openFiles(self, paths):
"""Open and close streams returned by the private validation step."""
files = self.browser._openFiles(paths)
for item in files:
item.stream.close()
return files
def test_regular_file_succeeds(self) -> None:
path = self.root / "resume.pdf"
path.write_bytes(b"pdf")
files = self.openFiles([str(path)])
self.assertEqual(files[0].name, "resume.pdf")
def test_files_outside_previous_root_are_allowed(self) -> None:
sibling = self.base / "allowed-evil"
sibling.mkdir()
path = sibling / "secret"
path.write_bytes(b"secret")
traversal = self.root / ".." / sibling.name / path.name
for candidate in (str(path), str(traversal)):
with self.subTest(candidate=candidate):
files = self.openFiles([candidate])
self.assertEqual(files[0].name, "secret")
def test_symlink_policy_uses_resolved_target(self) -> None:
inside = self.root / "inside"
inside.write_bytes(b"ok")
inside_link = self.root / "inside-link"
inside_link.symlink_to(inside)
self.openFiles([str(inside_link)])
outside = self.base / "outside"
outside.write_bytes(b"no")
outside_link = self.root / "outside-link"
outside_link.symlink_to(outside)
files = self.openFiles([str(outside_link)])
self.assertEqual(files[0].name, "outside")
def test_directory_and_fifo_are_invalid(self) -> None:
with self.assertRaises(UploadFileInvalidError):
self.openFiles([str(self.root)])
fifo = self.root / "fifo"
os.mkfifo(fifo)
with self.assertRaises(UploadFileInvalidError):
self.openFiles([str(fifo)])
if hasattr(socket, "AF_UNIX"):
socket_path = self.root / "socket"
local_socket = socket.socket(socket.AF_UNIX)
try:
try:
local_socket.bind(str(socket_path))
except PermissionError:
return
with self.assertRaises(UploadFileInvalidError):
self.openFiles([str(socket_path)])
finally:
local_socket.close()
def test_per_file_and_aggregate_limits(self) -> None:
large_browser = BrowserSession(
config(max_file_bytes=2),
FakeRdp([]),
)
path = self.root / "large"
path.write_bytes(b"123")
with self.assertRaises(UploadTooLargeError):
large_browser._openFiles([str(path)])
first = self.root / "first"
second = self.root / "second"
first.write_bytes(b"12")
second.write_bytes(b"34")
aggregate_browser = BrowserSession(
config(max_upload_bytes=3),
FakeRdp([]),
)
with self.assertRaises(UploadTooLargeError):
aggregate_browser._openFiles([str(first), str(second)])
if __name__ == "__main__":
unittest.main()