import hashlib
import socket
import time
import uuid
import pytest
from hayva_computer.config import Settings
from hayva_computer.errors import ComputerRuntimeError
from hayva_computer.network import validate_public_url
from hayva_computer.runtime import ComputerRuntime
from hayva_computer.schemas import (
BrowserActionInput,
HumanKeyboardInput,
StartSessionInput,
)
@pytest.mark.asyncio
async def test_url_guard_blocks_local_file_credentials_and_dns_rebinding(monkeypatch):
blocked = [
"http://127.0.0.1/private",
"http://[::1]/private",
"http://169.254.169.254/latest/meta-data",
"file:///etc/passwd",
"https://user:password@example.com/",
"http://service.internal/",
]
for url in blocked:
with pytest.raises(ComputerRuntimeError):
await validate_public_url(url)
def rebound(*_args, **_kwargs):
return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.0.8", 443))]
monkeypatch.setattr(socket, "getaddrinfo", rebound)
with pytest.raises(ComputerRuntimeError) as failure:
await validate_public_url("https://public-name.example/")
assert failure.value.code == "BROWSER_DESTINATION_BLOCKED"
def runtime_settings(tmp_path):
return Settings(
computer_service_token="runtime-test-token-that-is-longer-than-32-characters",
browser_profile_root=str(tmp_path / "profiles"),
computer_artifact_root=str(tmp_path / "artifacts"),
browser_headless=True,
browser_action_timeout_ms=10_000,
browser_max_sessions=2,
)
def session_payload(session_id, workspace_id, profile_id, token=1):
return StartSessionInput(
session_id=session_id,
workspace_id=workspace_id,
profile_id=profile_id,
profile_storage_key="a" * 32,
fencing_token=token,
)
@pytest.mark.asyncio
async def test_real_chromium_dom_tools_takeover_fencing_and_persistent_profile(tmp_path):
session_id, workspace_id, profile_id = uuid.uuid4(), uuid.uuid4(), uuid.uuid4()
settings = runtime_settings(tmp_path)
runtime = ComputerRuntime(settings)
await runtime.startup()
try:
started = await runtime.start_session(
session_payload(session_id, workspace_id, profile_id), request_id="start-1"
)
assert started.status == "ai_controlled"
current = runtime.sessions[session_id]
page = await runtime._page(current)
await page.set_content("""
Computer Runtime Test
""")
observed = await runtime.action(
session_id,
BrowserActionInput(fencing_token=1, tool="observe", arguments={}),
request_id="observe-1",
)
assert observed.result["title"] == "Computer Runtime Test"
assert any(item["tag"] == "button" for item in observed.result["interactive_elements"])
typed = await runtime.action(
session_id,
BrowserActionInput(
fencing_token=1, tool="type",
arguments={"selector": "#name", "text": "private value"},
),
request_id="type-1",
)
assert typed.result == {"typed": True, "character_count": 13}
assert "private value" not in str(typed.model_dump())
clicked = await runtime.action(
session_id,
BrowserActionInput(
fencing_token=1, tool="click", arguments={"selector": "#save"}
),
request_id="click-1",
)
assert clicked.result["clicked"] is True
extracted = await runtime.action(
session_id,
BrowserActionInput(
fencing_token=1, tool="extract", arguments={"selector": "#save"}
),
request_id="extract-1",
)
assert extracted.result["value"] == "Saved"
screenshot = await runtime.action(
session_id,
BrowserActionInput(fencing_token=1, tool="screenshot", arguments={}),
request_id="screenshot-1",
)
artifact = screenshot.result["artifact"]
assert "image_base64" not in screenshot.result
content = runtime.read_artifact(
workspace_id, session_id, uuid.UUID(artifact["id"])
)
assert len(content) == artifact["size_bytes"]
assert hashlib.sha256(content).hexdigest() == artifact["sha256"]
artifact_id = uuid.UUID(artifact["id"])
assert runtime.delete_artifact(workspace_id, session_id, artifact_id) is True
assert runtime.delete_artifact(workspace_id, session_id, artifact_id) is False
with pytest.raises(ComputerRuntimeError) as deleted:
runtime.read_artifact(workspace_id, session_id, artifact_id)
assert deleted.value.code == "BROWSER_ARTIFACT_NOT_FOUND"
await current.context.add_cookies([{
"name": "persistent", "value": "yes", "domain": "example.com", "path": "/",
"expires": time.time() + 3600,
}])
paused = await runtime.command(session_id, "pause", 1, request_id="pause-1")
assert paused.status == "paused"
await page.locator("#name").focus()
human = await runtime.human_keyboard(
session_id,
HumanKeyboardInput(fencing_token=2, text="human typed"),
request_id="human-keyboard-1",
)
assert human.status == "human_controlled"
with pytest.raises(ComputerRuntimeError) as stale:
await runtime.snapshot(session_id, 1, request_id="stale-snapshot")
assert stale.value.code == "BROWSER_FENCING_TOKEN_STALE"
resumed = await runtime.command(session_id, "resume", 3, request_id="resume-1")
assert resumed.status == "ai_controlled"
finally:
await runtime.shutdown()
recovered_runtime = ComputerRuntime(settings)
await recovered_runtime.startup()
try:
recovered = await recovered_runtime.start_session(
session_payload(session_id, workspace_id, profile_id, token=3),
request_id="recover-1",
)
assert recovered.status == "ai_controlled"
recovered_session = recovered_runtime.sessions[session_id]
cookies = await recovered_session.context.cookies("https://example.com")
assert any(cookie["name"] == "persistent" and cookie["value"] == "yes"
for cookie in cookies)
stopped = await recovered_runtime.command(
session_id, "stop", 4, request_id="stop-1"
)
assert stopped.status == "stopped"
assert recovered_runtime._read_manifest(recovered_session.profile_path)["status"] == (
"stopped"
)
finally:
await recovered_runtime.shutdown()