Mortdecai
65622501fe
The LAN-bound HTTP server joined the raw URL path onto the project dir, so /../../.ssh/id_ed25519 (and %2e%2e variants) served any file the user could read. Resolve and require the path stay inside the project. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
127 lines
5.2 KiB
Python
127 lines
5.2 KiB
Python
import asyncio
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from unittest.mock import patch, AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from workbench.server import WorkbenchServer
|
|
|
|
|
|
@pytest.fixture
|
|
def server(tmp_workbench):
|
|
return WorkbenchServer(workbench_dir=tmp_workbench)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scaffold_creates_project(server, tmp_workbench):
|
|
with patch.object(server, "_start_http_server", new_callable=AsyncMock, return_value=8070):
|
|
result = await server.workbench_scaffold("test-proj", "Test Project")
|
|
data = json.loads(result)
|
|
assert "url" in data
|
|
assert (tmp_workbench / "test-proj" / "index.html").exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scaffold_reattaches_to_running_server(server, tmp_workbench):
|
|
from workbench.project import create_project, write_server_info
|
|
create_project("test-proj", "Test", workbench_dir=tmp_workbench)
|
|
write_server_info("test-proj", pid=os.getpid(), port=9999, workbench_dir=tmp_workbench)
|
|
with patch.object(server, "_is_server_alive", return_value=True):
|
|
with patch.object(server, "_start_http_server", new_callable=AsyncMock) as mock_start:
|
|
result = await server.workbench_scaffold("test-proj", "Test")
|
|
mock_start.assert_not_called()
|
|
data = json.loads(result)
|
|
assert "9999" in data["url"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scaffold_replaces_dead_server(server, tmp_workbench):
|
|
from workbench.project import create_project, write_server_info
|
|
create_project("test-proj", "Test", workbench_dir=tmp_workbench)
|
|
write_server_info("test-proj", pid=99999, port=9999, workbench_dir=tmp_workbench)
|
|
with patch.object(server, "_is_server_alive", return_value=False):
|
|
with patch.object(server, "_start_http_server", new_callable=AsyncMock, return_value=8070):
|
|
result = await server.workbench_scaffold("test-proj", "Test")
|
|
data = json.loads(result)
|
|
assert "8070" in data["url"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_empty(server):
|
|
result = await server.workbench_list()
|
|
data = json.loads(result)
|
|
assert data["projects"] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_log_writes_to_disk(server, tmp_workbench):
|
|
with patch.object(server, "_start_http_server", new_callable=AsyncMock, return_value=8070):
|
|
await server.workbench_scaffold("test-proj", "Test")
|
|
result = await server.workbench_log("test-proj", "R412 measured 1.05M")
|
|
data = json.loads(result)
|
|
assert data["ok"] is True
|
|
jsonl = (tmp_workbench / "test-proj" / "session.jsonl").read_text().strip()
|
|
assert "R412 measured 1.05M" in jsonl
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_state_saves_to_disk(server, tmp_workbench):
|
|
with patch.object(server, "_start_http_server", new_callable=AsyncMock, return_value=8070):
|
|
await server.workbench_scaffold("test-proj", "Test")
|
|
state = json.dumps({"template": "<h1>Hello</h1>"})
|
|
result = await server.workbench_state("test-proj", state)
|
|
data = json.loads(result)
|
|
assert data["ok"] is True
|
|
saved = json.loads((tmp_workbench / "test-proj" / "state.json").read_text())
|
|
assert saved["template"] == "<h1>Hello</h1>"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_cleans_up(server, tmp_workbench):
|
|
with patch.object(server, "_start_http_server", new_callable=AsyncMock, return_value=8070):
|
|
await server.workbench_scaffold("test-proj", "Test")
|
|
server._runners["test-proj"] = AsyncMock()
|
|
result = await server.workbench_stop("test-proj")
|
|
data = json.loads(result)
|
|
assert data["ok"] is True
|
|
assert not (tmp_workbench / "test-proj" / ".server.json").exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reconnect_on_startup(tmp_workbench):
|
|
from workbench.project import create_project, write_server_info
|
|
create_project("live-proj", "Live", workbench_dir=tmp_workbench)
|
|
write_server_info("live-proj", pid=os.getpid(), port=8070, workbench_dir=tmp_workbench)
|
|
create_project("dead-proj", "Dead", workbench_dir=tmp_workbench)
|
|
write_server_info("dead-proj", pid=99999, port=8071, workbench_dir=tmp_workbench)
|
|
srv = WorkbenchServer(workbench_dir=tmp_workbench)
|
|
with patch.object(srv, "_is_server_alive", side_effect=lambda port: port == 8070):
|
|
await srv.reconnect_existing_servers()
|
|
assert "live-proj" in srv._active
|
|
assert srv._active["live-proj"]["port"] == 8070
|
|
assert "dead-proj" not in srv._active
|
|
assert not (tmp_workbench / "dead-proj" / ".server.json").exists()
|
|
|
|
|
|
async def _live(server, name="p"):
|
|
import aiohttp
|
|
from workbench.project import create_project
|
|
create_project(name, "T", workbench_dir=server.workbench_dir)
|
|
port = await server._start_http_server(name)
|
|
return aiohttp.ClientSession(), f"http://127.0.0.1:{port}"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_static_blocks_path_traversal(server, tmp_workbench):
|
|
(tmp_workbench / "secret.txt").write_text("nope")
|
|
session, base = await _live(server)
|
|
async with session:
|
|
for evil in ["/../secret.txt", "/%2e%2e/secret.txt", "/..%2fsecret.txt"]:
|
|
async with session.get(base + evil, allow_redirects=False) as r:
|
|
assert r.status == 404, evil
|
|
async with session.get(base + "/") as r:
|
|
assert r.status == 200
|
|
await server.workbench_stop("p")
|