mirror of
https://github.com/TencentCloud/Octop.git
synced 2026-10-20 06:19:17 +08:00
智浪淘沙
fb35284327
The picker derived a single browse root from `Path.home().anchor`, so a user whose profile lives on `C:` could not reach `D:` at all. That was structural, not a policy filter: `host_fs_tree_root()` returns one path, `/api/filesystem/defaults` hands it over as `tree_root`, and `rootDirTree.sanitizeTree` keeps exactly one tree and returns `[cleaned]` - every sibling of that root was dropped on each update, so even a hand-injected `D:/` node could not survive. `/filesystem/dirs` only ever listed children of `C:/`, and the router exposed no endpoint that enumerates drives. Enumerate ready drive roots instead: * `host_dirs.host_browse_roots()` returns `/` on POSIX and every ready drive on Windows, falling back to the home anchor when none respond so the picker never renders an empty tree. Probing uses `os.path.isdir` rather than `GetLogicalDrives` so mapped network drives are covered too, with no ctypes call. * `GET /api/filesystem/roots` exposes that list, and `/api/filesystem/defaults` now also returns `browse_roots`. A `workspace_root_dir` policy jail still collapses both to that one path, which is the point of the policy. * `RootDirSelect` takes `treeRoots` and renders a forest. `sanitizeTree` keeps one tree per root while still deduping globally - that set is what fixes antd#37228 duplicate rows, and drive roots are distinct paths, so global uniqueness is still the right rule. `ancestorDirPaths` now takes the longest containing root, so `D:/x/y` expands under `D:/` even though `C:/` is listed first. Single-string roots keep working, so the existing tests stayed valid. Second, the hint text promised a sandbox that does not exist off Linux: `ensure_bubblewrap` returns `not_linux` with "bubblewrap jail is Linux-only; execute uses plain local_shell", so a non-root `root_dir` only bounds the agent's *tool* paths while the agent process keeps running with the server account's own filesystem access. That is a weaker guarantee than the copy claimed, and worth correcting on its own. `/api/filesystem/defaults` now reports `jail_enforced`, and `backendRootDirJailHint` splits into a real-sandbox variant and `backendRootDirPathLimitHint` for hosts without one. `backendRootDirDesc` no longer says `/` unconditionally, which was only true on POSIX. Drive enumeration and `shutil.which` sit behind `_ready_drive_roots()` / `_bwrap_on_path()` seams: patching `os.path.isdir` or `shutil.which` globally breaks pytest's own path handling mid-test.
416 lines
13 KiB
Python
416 lines
13 KiB
Python
"""Integration tests for /api/filesystem (host root_dir pickers)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
posix_only = pytest.mark.skipif(os.name != "posix", reason="POSIX root '/' probe behavior")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_host_dirs_requires_auth(
|
|
env: tuple[httpx.AsyncClient, Any, dict[str, str]],
|
|
) -> None:
|
|
client, _srv, _auth = env
|
|
r = await client.get("/api/filesystem/dirs")
|
|
assert r.status_code == 401, r.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_host_dirs_lists_children(
|
|
env_admin_client: tuple[httpx.AsyncClient, dict[str, str]],
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
client, auth = env_admin_client
|
|
(tmp_path / "alpha").mkdir()
|
|
(tmp_path / "beta").mkdir()
|
|
(tmp_path / "notes.txt").write_text("x", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(
|
|
"octop.infra.utils.host_dirs.normalize_host_path",
|
|
lambda path: Path(path).resolve(),
|
|
)
|
|
|
|
r = await client.get(
|
|
f"/api/filesystem/dirs?path={tmp_path.as_posix()}",
|
|
headers=auth,
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["path"] == tmp_path.resolve().as_posix()
|
|
names = [entry["name"] for entry in body["entries"]]
|
|
assert {"alpha", "beta"}.issubset(names)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_host_dirs_rejects_proc(
|
|
env_admin_client: tuple[httpx.AsyncClient, dict[str, str]],
|
|
) -> None:
|
|
client, auth = env_admin_client
|
|
r = await client.get("/api/filesystem/dirs?path=/proc", headers=auth)
|
|
assert r.status_code == 400, r.text
|
|
assert r.json()["error"]["code"] == "WORKSPACE_OP_UNSUPPORTED"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_probe_host_dir_requires_auth(
|
|
env: tuple[httpx.AsyncClient, Any, dict[str, str]],
|
|
) -> None:
|
|
client, _srv, _auth = env
|
|
r = await client.post("/api/filesystem/probe", json={"path": "/"})
|
|
assert r.status_code == 401, r.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_bwrap_requires_auth(
|
|
env: tuple[httpx.AsyncClient, Any, dict[str, str]],
|
|
) -> None:
|
|
client, _srv, _auth = env
|
|
r = await client.post("/api/filesystem/ensure-bwrap")
|
|
assert r.status_code == 401, r.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_bwrap_returns_status_shape(
|
|
env_admin_client: tuple[httpx.AsyncClient, dict[str, str]],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
client, auth = env_admin_client
|
|
monkeypatch.setattr(
|
|
"octop.api.routers.filesystem.ensure_bubblewrap",
|
|
lambda: {"status": "skipped", "reason": "not_linux", "detail": "test"},
|
|
)
|
|
r = await client.post("/api/filesystem/ensure-bwrap", headers=auth)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["status"] == "skipped"
|
|
assert body["reason"] == "not_linux"
|
|
assert "detail" in body
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_filesystem_defaults_for_admin(
|
|
env_admin_client: tuple[httpx.AsyncClient, dict[str, str]],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from octop.infra.utils.host_dirs import (
|
|
host_browse_roots,
|
|
host_fs_tree_root,
|
|
host_jail_enforced,
|
|
)
|
|
|
|
client, auth = env_admin_client
|
|
home = tmp_path / "os_home"
|
|
home.mkdir()
|
|
monkeypatch.setattr("octop.infra.utils.host_dirs.Path.home", lambda: home)
|
|
monkeypatch.setenv("OCTOP_IN_CONTAINER", "0")
|
|
|
|
r = await client.get("/api/filesystem/defaults", headers=auth)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["default_root_dir"] == host_fs_tree_root()
|
|
assert body["tree_root"] == host_fs_tree_root()
|
|
assert body["browse_roots"] == host_browse_roots()
|
|
assert body["jail_enforced"] == host_jail_enforced()
|
|
assert body["in_container"] is False
|
|
assert "home" not in body
|
|
assert "allow_outside_home" not in body
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_filesystem_roots_lists_every_drive_not_just_home_drive(
|
|
env_admin_client: tuple[httpx.AsyncClient, dict[str, str]],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Regression: the picker used to be pinned to the home drive's root.
|
|
|
|
``host_fs_tree_root`` derives a single root from ``Path.home()``, so a user
|
|
whose profile lives on ``C:`` could not browse ``D:`` at all. The roots list
|
|
must enumerate drives independently of where home happens to be.
|
|
"""
|
|
client, auth = env_admin_client
|
|
home = tmp_path / "os_home"
|
|
home.mkdir()
|
|
monkeypatch.setattr("octop.infra.utils.host_dirs.Path.home", lambda: home)
|
|
monkeypatch.setenv("OCTOP_IN_CONTAINER", "0")
|
|
|
|
r = await client.get("/api/filesystem/roots", headers=auth)
|
|
assert r.status_code == 200, r.text
|
|
roots = r.json()["roots"]
|
|
assert roots, "roots must never be empty — the picker needs a usable root"
|
|
from octop.infra.utils.host_dirs import host_browse_roots
|
|
|
|
assert roots == host_browse_roots()
|
|
# Every entry is an absolute, POSIX-serialized path the /dirs API accepts.
|
|
for root in roots:
|
|
assert root.startswith("/") or root.endswith(":/")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_filesystem_roots_requires_auth(
|
|
env: tuple[httpx.AsyncClient, Any, dict[str, str]],
|
|
) -> None:
|
|
client, _srv, _auth = env
|
|
r = await client.get("/api/filesystem/roots")
|
|
assert r.status_code == 401, r.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_filesystem_defaults_in_container(
|
|
env_admin_client: tuple[httpx.AsyncClient, dict[str, str]],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from octop.infra.utils.host_dirs import host_fs_tree_root
|
|
|
|
client, auth = env_admin_client
|
|
home = tmp_path / "os_home"
|
|
home.mkdir()
|
|
monkeypatch.setattr("octop.infra.utils.host_dirs.Path.home", lambda: home)
|
|
monkeypatch.setenv("OCTOP_IN_CONTAINER", "1")
|
|
|
|
r = await client.get("/api/filesystem/defaults", headers=auth)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["default_root_dir"] == host_fs_tree_root()
|
|
assert body["tree_root"] == host_fs_tree_root()
|
|
assert body["in_container"] is True
|
|
assert "home" not in body
|
|
assert "allow_outside_home" not in body
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_admin_can_list_outside_home(
|
|
env: tuple[httpx.AsyncClient, Any, dict[str, str]],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from octop.infra.utils.host_dirs import host_fs_tree_root
|
|
from tests.support.auth import create_user
|
|
|
|
client, _srv, admin_auth = env
|
|
home = tmp_path / "os_home"
|
|
home.mkdir()
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
monkeypatch.setattr("octop.infra.utils.host_dirs.Path.home", lambda: home)
|
|
monkeypatch.setenv("OCTOP_IN_CONTAINER", "0")
|
|
|
|
user_auth = await create_user(client, admin_auth, username="alice", password="TestPass12")
|
|
|
|
defaults = await client.get("/api/filesystem/defaults", headers=user_auth)
|
|
assert defaults.status_code == 200, defaults.text
|
|
body = defaults.json()
|
|
assert body["default_root_dir"] == host_fs_tree_root()
|
|
assert body["tree_root"] == host_fs_tree_root()
|
|
assert body["in_container"] is False
|
|
assert "home" not in body
|
|
assert "allow_outside_home" not in body
|
|
|
|
listed = await client.get(
|
|
f"/api/filesystem/dirs?path={outside.as_posix()}",
|
|
headers=user_auth,
|
|
)
|
|
assert listed.status_code == 200, listed.text
|
|
|
|
ok = await client.get(
|
|
f"/api/filesystem/dirs?path={home.as_posix()}",
|
|
headers=user_auth,
|
|
)
|
|
assert ok.status_code == 200, ok.text
|
|
|
|
probe = await client.post(
|
|
"/api/filesystem/probe",
|
|
headers=user_auth,
|
|
json={"path": outside.as_posix()},
|
|
)
|
|
assert probe.status_code == 200, probe.text
|
|
assert probe.json()["ok"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@posix_only
|
|
async def test_probe_host_dir_ok_for_slash(
|
|
env_admin_client: tuple[httpx.AsyncClient, dict[str, str]],
|
|
) -> None:
|
|
client, auth = env_admin_client
|
|
r = await client.post("/api/filesystem/probe", headers=auth, json={"path": "/"})
|
|
assert r.status_code == 200, r.text
|
|
assert r.json() == {"ok": True, "path": "/"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_probe_host_dir_ok_for_writable_dir(
|
|
env_admin_client: tuple[httpx.AsyncClient, dict[str, str]],
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
client, auth = env_admin_client
|
|
monkeypatch.setattr(
|
|
"octop.infra.utils.host_dirs.normalize_host_path",
|
|
lambda path: Path(path).resolve(),
|
|
)
|
|
|
|
r = await client.post(
|
|
"/api/filesystem/probe",
|
|
headers=auth,
|
|
json={"path": str(tmp_path)},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
assert r.json() == {"ok": True, "path": tmp_path.resolve().as_posix()}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_probe_host_dir_rejects_file(
|
|
env_admin_client: tuple[httpx.AsyncClient, dict[str, str]],
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
client, auth = env_admin_client
|
|
file_path = tmp_path / "notes.txt"
|
|
file_path.write_text("x", encoding="utf-8")
|
|
monkeypatch.setattr(
|
|
"octop.infra.utils.host_dirs.normalize_host_path",
|
|
lambda path: Path(path).resolve(),
|
|
)
|
|
|
|
r = await client.post(
|
|
"/api/filesystem/probe",
|
|
headers=auth,
|
|
json={"path": str(file_path)},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["ok"] is False
|
|
assert body["code"] == "not_directory"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mkdir_host_dir_requires_auth(
|
|
env: tuple[httpx.AsyncClient, Any, dict[str, str]],
|
|
) -> None:
|
|
client, _srv, _auth = env
|
|
r = await client.post("/api/filesystem/mkdir", json={"path": "/"})
|
|
assert r.status_code == 401, r.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mkdir_host_dir_creates_child(
|
|
env_admin_client: tuple[httpx.AsyncClient, dict[str, str]],
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
client, auth = env_admin_client
|
|
monkeypatch.setattr(
|
|
"octop.infra.utils.host_dirs.normalize_host_path",
|
|
lambda path: Path(path).resolve(),
|
|
)
|
|
|
|
r = await client.post(
|
|
"/api/filesystem/mkdir",
|
|
headers=auth,
|
|
json={"path": str(tmp_path), "base_name": "New Folder"},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["name"] == "New Folder"
|
|
assert (tmp_path / "New Folder").is_dir()
|
|
assert Path(body["path"]).resolve() == (tmp_path / "New Folder").resolve()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rename_host_dir_renames_child(
|
|
env_admin_client: tuple[httpx.AsyncClient, dict[str, str]],
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
client, auth = env_admin_client
|
|
target = tmp_path / "New Folder"
|
|
target.mkdir()
|
|
monkeypatch.setattr(
|
|
"octop.infra.utils.host_dirs.normalize_host_path",
|
|
lambda path: Path(path).resolve(),
|
|
)
|
|
|
|
r = await client.post(
|
|
"/api/filesystem/rename",
|
|
headers=auth,
|
|
json={"path": str(target), "new_name": "workspace"},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["name"] == "workspace"
|
|
assert (tmp_path / "workspace").is_dir()
|
|
assert not target.exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_filesystem_respects_user_workspace_root(
|
|
env: tuple[httpx.AsyncClient, Any, dict[str, str]],
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setenv("OCTOP_IN_CONTAINER", "0")
|
|
from tests.support.auth import TEST_PASSWORD, create_user
|
|
|
|
client, _srv, admin_auth = env
|
|
jail = tmp_path / "jail"
|
|
nested = jail / "ok"
|
|
outside = tmp_path / "outside"
|
|
jail.mkdir()
|
|
nested.mkdir()
|
|
outside.mkdir()
|
|
|
|
user_auth = await create_user(client, admin_auth, username="fs_policy", password=TEST_PASSWORD)
|
|
listed = (await client.get("/api/users", headers=admin_auth)).json()
|
|
uid = next(u["id"] for u in listed if u["username"] == "fs_policy")
|
|
patched = await client.patch(
|
|
f"/api/users/{uid}",
|
|
headers=admin_auth,
|
|
json={"workspace_root_dir": jail.as_posix()},
|
|
)
|
|
assert patched.status_code == 200, patched.text
|
|
|
|
defaults = await client.get("/api/filesystem/defaults", headers=user_auth)
|
|
assert defaults.status_code == 200, defaults.text
|
|
body = defaults.json()
|
|
assert body["tree_root"] == jail.resolve().as_posix()
|
|
assert body["default_root_dir"] == jail.resolve().as_posix()
|
|
# A policy jail collapses the picker to exactly that one path.
|
|
assert body["browse_roots"] == [jail.resolve().as_posix()]
|
|
assert "home" not in body
|
|
assert "allow_outside_home" not in body
|
|
|
|
jailed_roots = await client.get("/api/filesystem/roots", headers=user_auth)
|
|
assert jailed_roots.status_code == 200, jailed_roots.text
|
|
assert jailed_roots.json()["roots"] == [jail.resolve().as_posix()]
|
|
|
|
inside = await client.get(
|
|
f"/api/filesystem/dirs?path={nested.as_posix()}",
|
|
headers=user_auth,
|
|
)
|
|
assert inside.status_code == 200, inside.text
|
|
|
|
listed_out = await client.get(
|
|
f"/api/filesystem/dirs?path={outside.as_posix()}",
|
|
headers=user_auth,
|
|
)
|
|
assert listed_out.status_code == 400, listed_out.text
|
|
|
|
probe = await client.post(
|
|
"/api/filesystem/probe",
|
|
headers=user_auth,
|
|
json={"path": outside.as_posix()},
|
|
)
|
|
assert probe.status_code == 200, probe.text
|
|
assert probe.json()["ok"] is False
|
|
assert probe.json()["code"] == "outside_root"
|