test: harden webui and gateway checks

.github/workflows/ci.yml

@@ -44,10 +44,36 @@ jobs:
         run: sudo apt-get update && sudo apt-get install -y libolm-dev build-essential
 
       - name: Install dependencies
-        run: uv sync --all-extras
+        run: uv sync --all-extras --dev
 
       - name: Lint with ruff
         run: uv run ruff check nanobot --select F
 
       - name: Run tests
-        run: uv run pytest tests/
+        run: uv run python -m pytest tests/ --cov=nanobot --cov-report=term-missing:skip-covered
+
+  webui:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Bun
+        uses: oven-sh/setup-bun@v2
+
+      - name: Install WebUI dependencies
+        working-directory: webui
+        run: bun install --frozen-lockfile
+
+      - name: Lint WebUI
+        working-directory: webui
+        run: bun run lint
+
+      - name: Test WebUI
+        working-directory: webui
+        run: bun run test
+
+      - name: Build WebUI
+        working-directory: webui
+        run: bun run build

pyproject.toml

@@ -182,6 +182,7 @@ source = ["nanobot"]
 omit = ["tests/*", "**/tests/*"]
 
 [tool.coverage.report]
+fail_under = 75
 exclude_lines = [
     "pragma: no cover",
     "def __repr__",

tests/channels/test_websocket_protocol_boundaries.py

@@ -0,0 +1,77 @@
+"""Boundary tests for pure WebSocket protocol helpers."""
+
+from __future__ import annotations
+
+import pytest
+
+from nanobot.channels.websocket import (
+    _extract_data_url_mime,
+    _is_valid_chat_id,
+    _parse_envelope,
+)
+
+
+def test_chat_id_validator_accepts_only_compact_capability_keys() -> None:
+    valid = [
+        "a",
+        "A-Z_09:chat-id",
+        "x" * 64,
+    ]
+    invalid = [
+        "",
+        "x" * 65,
+        "../escape",
+        "chat/id",
+        "chat id",
+        "chat\nid",
+        None,
+        123,
+    ]
+
+    for value in valid:
+        assert _is_valid_chat_id(value), value
+    for value in invalid:
+        assert not _is_valid_chat_id(value), repr(value)
+
+
+@pytest.mark.parametrize(
+    ("raw", "expected_type"),
+    [
+        ("plain text", None),
+        ("{not json", None),
+        ("[]", None),
+        ("{}", None),
+        ('{"type": 42}', None),
+        ('{"type": "message", "content": "hi"}', "message"),
+        ('  {"type": "new_chat"}  ', "new_chat"),
+    ],
+)
+def test_parse_envelope_only_accepts_typed_json_objects(
+    raw: str,
+    expected_type: str | None,
+) -> None:
+    parsed = _parse_envelope(raw)
+    if expected_type is None:
+        assert parsed is None
+    else:
+        assert parsed is not None
+        assert parsed["type"] == expected_type
+
+
+@pytest.mark.parametrize(
+    ("url", "expected"),
+    [
+        ("data:image/png;base64,AAAA", "image/png"),
+        ("data:IMAGE/JPEG;charset=utf-8;base64,AAAA", "image/jpeg"),
+        ("data:video/webm;codecs=vp9;base64,AAAA", "video/webm"),
+        ("data:image/svg+xml;base64,AAAA", "image/svg+xml"),
+        ("data:image/png,AAAA", None),
+        ("data:;base64,AAAA", None),
+        ("https://example.invalid/image.png", None),
+    ],
+)
+def test_extract_data_url_mime_normalizes_only_base64_data_urls(
+    url: str,
+    expected: str | None,
+) -> None:
+    assert _extract_data_url_mime(url) == expected

tests/security/test_security_network.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import ipaddress
 import socket
 from unittest.mock import patch
 
@@ -107,6 +108,38 @@ def test_blocks_ipv6_mapped_rfc1918():
         assert not ok
 
 
+def test_blocks_sampled_addresses_from_internal_networks():
+    """Property-style guard: sampled blocked CIDRs must all fail closed."""
+    configure_ssrf_whitelist([])
+    blocked_networks = [
+        "0.0.0.0/8",
+        "10.0.0.0/8",
+        "100.64.0.0/10",
+        "127.0.0.0/8",
+        "169.254.0.0/16",
+        "172.16.0.0/12",
+        "192.168.0.0/16",
+        "::1/128",
+        "fc00::/7",
+        "fe80::/10",
+    ]
+    samples: list[str] = []
+    for cidr in blocked_networks:
+        network = ipaddress.ip_network(cidr)
+        samples.append(str(network.network_address))
+        if network.num_addresses > 2:
+            samples.append(str(network.network_address + 1))
+            samples.append(str(network[-2]))
+
+    for idx, ip in enumerate(samples):
+        host = f"internal-{idx}.example"
+        resolver = _fake_resolve_v6 if ":" in ip else _fake_resolve
+        with patch("nanobot.security.network.socket.getaddrinfo", resolver(host, [ip])):
+            ok, err = validate_url_target(f"http://{host}/")
+        assert not ok, f"expected {ip} to be blocked"
+        assert "blocked" in err.lower() or "private" in err.lower()
+
+
 def test_allows_public_ipv6():
     """Public IPv6 addresses must still be allowed."""
     with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve_v6("example.com", ["2606:4700::6810:84e5"])):

tests/security/test_workspace_policy.py

@@ -53,6 +53,38 @@ def test_resolve_allowed_path_blocks_parent_traversal(tmp_path: Path) -> None:
         resolve_allowed_path("../secret.txt", workspace=workspace, allowed_root=workspace)
 
 
+def test_resolve_allowed_path_blocks_traversal_shapes(tmp_path: Path) -> None:
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    outside = tmp_path / "secret.txt"
+    outside.write_text("secret", encoding="utf-8")
+
+    traversal_shapes: list[str | Path] = [
+        "../secret.txt",
+        "src/../../secret.txt",
+        Path("..") / "secret.txt",
+        workspace / "src" / ".." / ".." / "secret.txt",
+    ]
+    if os.name == "nt":
+        traversal_shapes.append("src\\..\\..\\secret.txt")
+
+    for candidate in traversal_shapes:
+        with pytest.raises(WorkspaceBoundaryError, match="outside allowed directory"):
+            resolve_allowed_path(candidate, workspace=workspace, allowed_root=workspace)
+
+
+def test_resolve_allowed_path_blocks_prefix_sibling(tmp_path: Path) -> None:
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    sibling = tmp_path / "workspace-other"
+    sibling.mkdir()
+    secret = sibling / "secret.txt"
+    secret.write_text("secret", encoding="utf-8")
+
+    with pytest.raises(WorkspaceBoundaryError, match="outside allowed directory"):
+        resolve_allowed_path(secret, workspace=workspace, allowed_root=workspace)
+
+
 def test_resolve_allowed_path_blocks_symlink_escape(tmp_path: Path) -> None:
     workspace = tmp_path / "workspace"
     workspace.mkdir()

tests/webui/test_gateway_webui_smoke.py

@@ -0,0 +1,179 @@
+"""Black-box smoke test for the real gateway WebUI transport."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import socket
+import subprocess
+import sys
+import time
+from pathlib import Path
+from urllib.parse import quote
+
+import httpx
+import pytest
+import websockets
+
+
+def _free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind(("127.0.0.1", 0))
+        return int(sock.getsockname()[1])
+
+
+def _write_smoke_config(path: Path, *, workspace: Path, ws_port: int, gateway_port: int) -> None:
+    config = {
+        "agents": {
+            "defaults": {
+                "workspace": str(workspace),
+                "provider": "custom",
+                "model": "custom/smoke-model",
+                "maxToolIterations": 1,
+                "dream": {"enabled": False},
+            }
+        },
+        "providers": {
+            "custom": {
+                "apiKey": "smoke-no-external-call",
+                "apiBase": "http://127.0.0.1:9/v1",
+            }
+        },
+        "channels": {
+            "websocket": {
+                "enabled": True,
+                "host": "127.0.0.1",
+                "port": ws_port,
+                "allowFrom": ["*"],
+            }
+        },
+        "gateway": {
+            "host": "127.0.0.1",
+            "port": gateway_port,
+            "heartbeat": {"enabled": False},
+        },
+    }
+    path.write_text(json.dumps(config), encoding="utf-8")
+
+
+def _start_gateway(config_path: Path, log_path: Path) -> subprocess.Popen[bytes]:
+    log_file = log_path.open("wb")
+    try:
+        process = subprocess.Popen(
+            [
+                sys.executable,
+                "-m",
+                "nanobot",
+                "gateway",
+                "--config",
+                str(config_path),
+            ],
+            cwd=Path(__file__).resolve().parents[2],
+            stdout=log_file,
+            stderr=subprocess.STDOUT,
+        )
+    finally:
+        log_file.close()
+    return process
+
+
+def _stop_gateway(process: subprocess.Popen[bytes]) -> None:
+    if process.poll() is not None:
+        return
+    process.terminate()
+    try:
+        process.wait(timeout=10)
+    except subprocess.TimeoutExpired:
+        process.kill()
+        process.wait(timeout=10)
+
+
+def _get_json(url: str, *, token: str | None = None) -> dict:
+    headers = {"Authorization": f"Bearer {token}"} if token else {}
+    response = httpx.get(url, headers=headers, timeout=5.0, trust_env=False)
+    response.raise_for_status()
+    return response.json()
+
+
+def _wait_for_bootstrap(base_url: str, process: subprocess.Popen[bytes], log_path: Path) -> dict:
+    deadline = time.monotonic() + 20
+    last_error: Exception | None = None
+    while time.monotonic() < deadline:
+        if process.poll() is not None:
+            break
+        try:
+            return _get_json(f"{base_url}/webui/bootstrap")
+        except (httpx.HTTPError, OSError) as exc:
+            last_error = exc
+            time.sleep(0.2)
+    logs = log_path.read_text(encoding="utf-8", errors="replace")
+    raise AssertionError(f"gateway did not start; last_error={last_error!r}\n{logs}")
+
+
+async def _recv_until(ws: websockets.WebSocketClientProtocol, event: str) -> dict:
+    deadline = time.monotonic() + 20
+    while time.monotonic() < deadline:
+        raw = await asyncio.wait_for(ws.recv(), timeout=5)
+        payload = json.loads(raw)
+        if payload.get("event") == event:
+            return payload
+    raise AssertionError(f"websocket event {event!r} was not received")
+
+
+@pytest.mark.asyncio
+async def test_gateway_webui_bootstrap_message_and_thread_hydration(tmp_path: Path) -> None:
+    ws_port = _free_port()
+    gateway_port = _free_port()
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    config_path = tmp_path / "config.json"
+    log_path = tmp_path / "gateway.log"
+    _write_smoke_config(
+        config_path,
+        workspace=workspace,
+        ws_port=ws_port,
+        gateway_port=gateway_port,
+    )
+
+    process = _start_gateway(config_path, log_path)
+    base_url = f"http://127.0.0.1:{ws_port}"
+    try:
+        bootstrap = _wait_for_bootstrap(base_url, process, log_path)
+        assert bootstrap["model_name"] == "custom/smoke-model"
+
+        ws_url = f'{bootstrap["ws_url"]}?token={bootstrap["token"]}&client_id=smoke'
+        async with websockets.connect(ws_url) as ws:
+            ready = await _recv_until(ws, "ready")
+            assert ready["client_id"] == "smoke"
+
+            await ws.send(json.dumps({"type": "new_chat"}))
+            attached = await _recv_until(ws, "attached")
+            chat_id = attached["chat_id"]
+            await _recv_until(ws, "session_updated")
+
+            await ws.send(json.dumps({
+                "type": "message",
+                "chat_id": chat_id,
+                "content": "/model",
+                "webui": True,
+                "turn_id": "smoke-turn",
+            }))
+            answer = await _recv_until(ws, "message")
+            assert "Current model: `custom/smoke-model`" in answer["text"]
+            await _recv_until(ws, "turn_end")
+
+        api_token = _wait_for_bootstrap(base_url, process, log_path)["token"]
+        sessions = _get_json(f"{base_url}/api/sessions", token=api_token)
+        key = f"websocket:{chat_id}"
+        assert key in {row["key"] for row in sessions["sessions"]}
+
+        encoded_key = quote(key, safe="")
+        thread = _get_json(
+            f"{base_url}/api/sessions/{encoded_key}/webui-thread",
+            token=api_token,
+        )
+        contents = [str(message.get("content") or "") for message in thread["messages"]]
+        assert "/model" in contents
+        assert any("Current model: `custom/smoke-model`" in text for text in contents)
+    finally:
+        _stop_gateway(process)

webui/src/components/thread/ThreadShell.tsx

@@ -494,6 +494,7 @@ export function ThreadShell({
     return client.onSessionUpdate((updatedChatId, scope) => {
       if (updatedChatId !== chatId) return;
       if (scope === "metadata") return;
+      viewportRef.current?.cancelAutoScroll();
       pendingCanonicalHydrateRef.current.add(chatId);
       refreshHistory();
     });

webui/src/components/thread/ThreadViewport.tsx

@@ -26,6 +26,7 @@ import type { CliAppInfo, McpPresetInfo, UIMessage } from "@/lib/types";
 
 export interface ThreadViewportHandle {
   jumpToUserPrompt: (promptId: string) => void;
+  cancelAutoScroll: () => void;
 }
 
 interface ThreadViewportProps {
@@ -290,7 +291,14 @@ export const ThreadViewport = forwardRef<ThreadViewportHandle, ThreadViewportPro
     setVisibleMessageCount((count) => Math.max(count, messages.length - index));
   }, [messages]);
 
-  useImperativeHandle(ref, () => ({ jumpToUserPrompt }), [jumpToUserPrompt]);
+  useImperativeHandle(
+    ref,
+    () => ({
+      jumpToUserPrompt,
+      cancelAutoScroll: cancelScheduledBottomScroll,
+    }),
+    [cancelScheduledBottomScroll, jumpToUserPrompt],
+  );
 
   const measureComposerDock = useCallback(() => {
     const el = composerDockRef.current;