From 8cf4b53dd49b17d8a9b6cc690a5ebdd87b3f5208 Mon Sep 17 00:00:00 2001 From: "tao.chen" <93983997+taochen-ct@users.noreply.github.com> Date: Fri, 21 Aug 2026 17:20:05 +0800 Subject: [PATCH] fix(scripts): workspace-wide parent_path listing + private visibility on reads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 同 workspace 互相可见(排除 private): - list_scripts / count_scripts 已 workspace-wide + visibility 过滤,但 单条读取(get/content/latest-version/versions)不校验 visibility,非 owner 猜 id 即可读他人 private 脚本。新增 script_can_view(与 data resources 的 can_view 对称)并在 get_script_row / latest_version 强制, private 对非 owner 返回 404。 2. parent_path 为空默认拉根路径文件: - 物理存储为 workspace/{user_id}/...,根 prefix 原来是 workspace/ + NOT LIKE workspace/%/%,所有文件都在两层被整体排除,list_scripts("") 恒空。改为 workspace/%/,配合 LIKE workspace/%/% AND NOT LIKE workspace/%/%/% 返回各 owner 根级文件。 3. 非空 parent_path 跨 owner 查询: - 原来 workspace/foo/ 永远匹配不到 workspace/{uid}/foo/...,子目录 懒加载返回空,其他用户目录点击无内容。改为 workspace/%/foo/(owner 段通配,与 list_resources 一致),_ / % 仍按字面转义。 测试:更新前缀契约断言,新增 SQLite 行为测试(跨 owner 根/子目录、转义) 与 script_can_view / get_script 权限测试,133 passed。 --- backend/src/backend/api/scripts.py | 84 +++-- backend/tests/test_count_scripts.py | 18 +- .../tests/test_list_scripts_parent_path.py | 311 +++++++++++++++++- 3 files changed, 362 insertions(+), 51 deletions(-) diff --git a/backend/src/backend/api/scripts.py b/backend/src/backend/api/scripts.py index 2ea5979..a4c31b4 100644 --- a/backend/src/backend/api/scripts.py +++ b/backend/src/backend/api/scripts.py @@ -170,23 +170,29 @@ def _build_list_scripts_workspace_descendant_prefix(parent_path: str) -> str: """Return the escaped materialized-path prefix for direct children of ``parent_path`` across **all owners** in the workspace. - Storage is still physically laid out as ``workspace/{user_id}/...``, but - listing is workspace-wide: the prefix starts at ``workspace/`` (no - embedded user_id) so the endpoint's ``LIKE '/%'`` walks every - owner's subtree. Non-admin scoping is handled separately in the SQL via + Storage is still physically laid out as ``workspace/{user_id}/...``, so + listing a subdirectory across every owner must match the owner segment + with an intentional ``%`` wildcard — ``workspace/%/foo`` — exactly like + list_resources does against ``object_key``. The endpoint's + ``LIKE '/%' AND NOT LIKE '/%/%'`` pair then grabs only + DIRECT children of ``parent_path`` per owner. Non-admin scoping is + handled separately in the SQL via ``owner_user_id = me OR visibility IN (workspace, public)``. - Empty ``parent_path`` returns ``"workspace/"`` (cross-owner root). - The prefix is run through ``_escape_like_pattern`` so folder names - containing ``_`` / ``%`` do not act as wildcards. The trailing ``/`` - is appended AFTER escaping so it remains a literal slash. + Empty ``parent_path`` returns ``"workspace/%/"``: the physical root + level is each owner's subtree (``workspace/{owner_id}/...``), so the + owner segment is wildcarded and the endpoint's direct-child pair + keeps every owner's root-level files (``workspace/%/%`` AND NOT + ``workspace/%/%/%``). The prefix is run through ``_escape_like_pattern`` + so folder names containing ``_`` / ``%`` do not act as wildcards. The + trailing ``/`` is appended AFTER escaping so it remains a literal slash. """ normalized_parent = normalize_user_path(parent_path) if normalized_parent: - target_prefix = f"workspace/{normalized_parent}" + target_prefix = f"workspace/%/{_escape_like_pattern(normalized_parent)}" else: - target_prefix = "workspace" - return f"{_escape_like_pattern(target_prefix)}/" + target_prefix = "workspace/%" + return f"{target_prefix}/" def safe_script_name(value: str, script_type: str) -> str: @@ -328,6 +334,22 @@ def version_payload(version: Versions) -> dict[str, Any]: } +def script_can_view(script: Scripts, context: RequestContext) -> bool: + """同一 workspace 内：owner 永远可见自己的脚本（含 private）； + 其他成员只见 visibility in {workspace, public} 的脚本； + admin 全部可见。与 data resources 的 can_view 完全对称。 + + 用于单脚本读取（get / content / latest-version / versions 列表）以及 + 所有写路径（update / delete / publish）的前置校验 —— 保证 A 的 private + 脚本对非 owner 不可见，而不仅是「列表里不出现」。 + """ + if script.owner_user_id == context.user.user_id: + return True + if script.visibility in {"workspace", "public"}: + return True + return context.is_admin + + async def get_script_row( script_id: str, context: RequestContext, @@ -392,6 +414,11 @@ async def get_script_row( if row is None: raise HTTPException(status.HTTP_404_NOT_FOUND, "script not found") script, storage_object = row + if not script_can_view(script, context): + # Private scripts are only visible to their owner (and admin); + # treat cross-owner access as not-found so the id cannot probe + # visibility, matching get_visible_resource for data resources. + raise HTTPException(status.HTTP_404_NOT_FOUND, "script not found") return script, storage_object @@ -1122,8 +1149,10 @@ async def delete_workspace_directory( # 列出当前工作区中用户有权查看的脚本与 Notebook 元数据。 # # 可选 ``parent_path`` Query 参数把返回范围限定为 *直接挂在该路径下的* 脚本 -# (不含更深的子目录)。空字符串等价于用户作用域根目录;这是前端按目录懒加载 -# 的关键端点,避免 10 万级脚本一次性返回。 +# (不含更深的子目录)。空字符串等价于工作区根目录(跨所有 owner 根级文件); +# 非空路径按 owner 段通配(``workspace/%/``)跨所有 owner 查询,与 +# list_resources 一致。这是前端按目录懒加载的关键端点,避免 10 万级脚本 +# 一次性返回。 # # 实现走 ``storage_objects.relative_path`` materialized path(``LIKE prefix%`` # + ``NOT LIKE prefix%/%`` 取直接子节点)。MySQL 优化器目前选 @@ -1140,10 +1169,12 @@ async def list_scripts( session: AsyncSession = Depends(database_session), ) -> dict[str, Any]: # Workspace-wide listing: storage is physically laid out as - # ``workspace/{user_id}/...``, but the prefix starts at ``workspace/`` - # (no embedded user_id) so the LIKE walks every owner's subtree. - # Non-admin scoping is applied below via visibility, matching - # list_resources (69a9a48). + # ``workspace/{user_id}/...``. Empty parent_path wildcards the owner + # segment (``workspace/%/``) so each owner's root files are returned; + # non-empty parent_path embeds the same owner wildcard + # (``workspace/%/foo``) so every owner's ``foo`` subtree matches, + # mirroring list_resources. Non-admin scoping is applied below via + # visibility, matching list_resources (69a9a48). descendant_prefix = _build_list_scripts_workspace_descendant_prefix(parent_path) statement = ( @@ -1638,18 +1669,15 @@ async def latest_version( published versions yet (so the frontend can render an empty label without a 404 round-trip). """ - script = await session.scalar( - select(Scripts.script_id).where( - Scripts.script_id == script_id, - Scripts.workspace_id == context.workspace.workspace_id, - Scripts.status == "active", - ) + # Route through get_script_row so the private-visibility guard applies + # here too: other members must not learn about a private script's + # versions by guessing its id. + _script, _ = await get_script_row( + script_id, + context, + session, + allow_missing_storage_object=True, ) - if script is None: - raise HTTPException( - status.HTTP_404_NOT_FOUND, - "script not found", - ) latest = await session.scalar( select(Versions) .where( diff --git a/backend/tests/test_count_scripts.py b/backend/tests/test_count_scripts.py index 7692c8a..7456ac1 100644 --- a/backend/tests/test_count_scripts.py +++ b/backend/tests/test_count_scripts.py @@ -68,10 +68,11 @@ async def test_count_scripts_returns_scalar_int() -> None: # JOIN to StorageObjects so orphaned scripts (no joinable row) are # excluded — matches list_scripts INNER JOIN behaviour. assert "inner join storage_objects" in sql - # Scope: workspace_id + active status + workspace-wide prefix. + # Scope: workspace_id + active status + workspace-wide prefix + # (owner segment wildcarded: workspace/%/%). assert "scripts.workspace_id" in sql assert "scripts.status" in sql - assert "like 'workspace/%%'" in sql + assert "like 'workspace/%%/%%'" in sql # Non-admin (default) narrows by visibility. assert "scripts.owner_user_id = 'u001'" in sql assert "scripts.visibility in ('workspace', 'public')" in sql @@ -87,9 +88,10 @@ async def test_count_scripts_handles_null_result() -> None: async def test_count_scripts_workspace_wide_not_user_scoped() -> None: - """The prefix is workspace-wide (``workspace/%`` — no embedded user_id), - so different users count the same physical tree; the only per-user - difference is the non-admin visibility predicate (owner_user_id = me).""" + """The prefix is workspace-wide (``workspace/%/%`` — owner segment + wildcarded, no embedded user_id), so different users count the same + physical tree; the only per-user difference is the non-admin + visibility predicate (owner_user_id = me).""" captured = [] mock_session = MagicMock() @@ -104,8 +106,8 @@ async def test_count_scripts_workspace_wide_not_user_scoped() -> None: sql_bob = _compile(captured[-1]).lower() # Both count the same workspace-wide subtree. - assert "like 'workspace/%%'" in sql_alice - assert "like 'workspace/%%'" in sql_bob + assert "like 'workspace/%%/%%'" in sql_alice + assert "like 'workspace/%%/%%'" in sql_bob # Neither embeds the user_id in the path prefix. assert "workspace/alice/%" not in sql_alice assert "workspace/bob/%" not in sql_bob @@ -129,7 +131,7 @@ async def test_count_scripts_admin_skips_visibility_filter() -> None: ) assert result["data"] == {"total": 42} sql = _compile(captured[0]).lower() - assert "like 'workspace/%%'" in sql + assert "like 'workspace/%%/%%'" in sql # visibility / owner_user_id still appear in the SELECT projection, but # the visibility WHERE predicate must be absent for admins. assert "scripts.visibility in ('workspace', 'public')" not in sql diff --git a/backend/tests/test_list_scripts_parent_path.py b/backend/tests/test_list_scripts_parent_path.py index 6f25f6f..feb0868 100644 --- a/backend/tests/test_list_scripts_parent_path.py +++ b/backend/tests/test_list_scripts_parent_path.py @@ -125,15 +125,23 @@ def test_normalize_user_path_strips() -> None: def test_workspace_descendant_prefix_root() -> None: - """Empty parent_path → workspace-wide root prefix (cross-owner).""" - assert _build_list_scripts_workspace_descendant_prefix("") == "workspace/" + """Empty parent_path → workspace-wide root prefix (cross-owner). + + Physical root level is each owner's subtree, so the prefix wildcards + the owner segment: ``workspace/%/`` (direct children ``workspace/%/%`` + minus ``workspace/%/%/%``).""" + assert _build_list_scripts_workspace_descendant_prefix("") == "workspace/%/" def test_workspace_descendant_prefix_subdir() -> None: - """Non-empty parent_path → appended under workspace root, no user_id.""" + """Non-empty parent_path → workspace-wide with owner-segment wildcard. + + Physical rows live under ``workspace/{owner_id}/...``, so a subdirectory + listing must match ANY owner: ``workspace/%/foo/bar`` (the ``%`` is the + intentional owner wildcard, exactly like list_resources).""" assert ( _build_list_scripts_workspace_descendant_prefix("foo/bar") - == "workspace/foo/bar/" + == "workspace/%/foo/bar/" ) @@ -142,21 +150,21 @@ def test_workspace_descendant_prefix_escapes_metachars() -> None: doesn't become 'match any single char'.""" assert ( _build_list_scripts_workspace_descendant_prefix("foo_bar") - == r"workspace/foo\_bar/" + == r"workspace/%/foo\_bar/" ) def test_workspace_descendant_prefix_escapes_percent() -> None: assert ( _build_list_scripts_workspace_descendant_prefix("100%match") - == r"workspace/100\%match/" + == r"workspace/%/100\%match/" ) def test_workspace_descendant_prefix_normalizes_leading_trailing_slashes() -> None: assert ( _build_list_scripts_workspace_descendant_prefix("/foo/bar/") - == "workspace/foo/bar/" + == "workspace/%/foo/bar/" ) @@ -198,8 +206,10 @@ async def test_list_scripts_where_clause_uses_like_prefix_and_excludes_deeper() assert len(captured_sql) == 1 sql = captured_sql[0].lower() - assert "like 'workspace/foo/bar/%%'" in sql - assert "not like 'workspace/foo/bar/%%/%%'" in sql + # 中间 ``%`` 是 owner 段通配符(跨所有 owner),父路径本身按字面匹配, + # 与 list_resources 对 object_key 的过滤一致。 + assert "like 'workspace/%%/foo/bar/%%'" in sql + assert "not like 'workspace/%%/foo/bar/%%/%%'" in sql async def test_list_scripts_where_clause_escapes_pattern_literal() -> None: @@ -228,9 +238,9 @@ async def test_list_scripts_where_clause_escapes_pattern_literal() -> None: # doubles the escape char inside the SQL string literal, so what # the helper emits as `foo\_bar` renders as `foo\\_bar` here # (2 backslash chars in the actual SQL string). - assert r"like 'workspace/foo\\_bar/%%'" in sql_lower + assert r"like 'workspace/%%/foo\\_bar/%%'" in sql_lower # NOT LIKE clause also escaped. - assert r"not like 'workspace/foo\\_bar/%%/%%'" in sql_lower + assert r"not like 'workspace/%%/foo\\_bar/%%/%%'" in sql_lower # And both declare ESCAPE '\\'. assert sql.count("ESCAPE '\\\\'") == 2, sql @@ -257,7 +267,7 @@ async def test_list_scripts_where_clause_escapes_percent_pattern() -> None: sql_lower = sql.lower() # SQLAlchemy doubles the escape char so `%` → `\%` becomes `\\%` # in the SQL string literal. - assert r"workspace/100\\%%match/%%" in sql_lower + assert r"workspace/%%/100\\%%match/%%" in sql_lower async def test_list_scripts_non_admin_adds_visibility_filter() -> None: @@ -281,7 +291,9 @@ async def test_list_scripts_non_admin_adds_visibility_filter() -> None: await list_scripts(parent_path="", context=_ctx("alice"), session=mock_session) sql = captured_sql[0].lower() - assert "like 'workspace/%%'" in sql + # Root listing wildcards the owner segment: LIKE workspace/%/% + # (each owner's root files), excluding 3+ segment descendants. + assert "like 'workspace/%%/%%'" in sql assert "scripts.owner_user_id = 'alice'" in sql assert "scripts.visibility in ('workspace', 'public')" in sql @@ -307,7 +319,7 @@ async def test_list_scripts_admin_skips_visibility_filter() -> None: parent_path="", context=_ctx("alice", is_admin=True), session=mock_session ) sql = captured_sql[0].lower() - assert "like 'workspace/%%'" in sql + assert "like 'workspace/%%/%%'" in sql # owner_user_id / visibility still appear in the SELECT projection; what # must be absent is the visibility WHERE predicate for non-admins. assert "scripts.visibility in ('workspace', 'public')" not in sql @@ -439,4 +451,273 @@ def test_sqlite_like_with_percent_in_name(sqlite_like_table): ) ).fetchall() matched = sorted(r[0] for r in rows) - assert matched == ["workspace/alice/100%off/x.py"], matched \ No newline at end of file + assert matched == ["workspace/alice/100%off/x.py"], matched + +# ─── layer 3.5: workspace-wide subdirectory semantics ────────────── + + +@pytest.fixture +def sqlite_workspace_table(): + """SQLite table of physical relative_path rows (workspace/{owner}/...).""" + engine = create_engine("sqlite:///:memory:") + metadata = MetaData() + table = Table( + "paths", + metadata, + Column("relative_path", String(1024), nullable=False), + ) + metadata.create_all(engine) + with engine.begin() as conn: + conn.execute( + table.insert(), + [ + # Direct children of `foo/bar` across two owners. + {"relative_path": "workspace/alice/foo/bar/a.py"}, + {"relative_path": "workspace/bob/foo/bar/b.py"}, + # Deeper than one level under foo/bar → must be excluded. + {"relative_path": "workspace/bob/foo/bar/nested/c.py"}, + # Sibling folder `barX` / `foobar` → must be excluded. + {"relative_path": "workspace/alice/foo/barX/decoy.py"}, + {"relative_path": "workspace/alice/foobar/x.py"}, + # Root-level files (not under any subdir). + {"relative_path": "workspace/alice/root.py"}, + ], + ) + yield engine, table + engine.dispose() + + +def _run_like(query_table, prefix: str, escape: str = "\\"): + """Run the endpoint's LIKE + NOT LIKE direct-child pair on SQLite. + + Mirrors list_scripts exactly: LIKE ``prefix + '%'`` and + NOT LIKE ``prefix + '%/%'``.""" + engine, table = query_table + stmt = select(table.c.relative_path).where( + table.c.relative_path.like(f"{prefix}%", escape=escape), + ~table.c.relative_path.like(f"{prefix}%/%", escape=escape), + ) + with engine.connect() as conn: + rows = conn.execute(stmt).fetchall() + return sorted(r[0] for r in rows) + + +def test_sqlite_workspace_subdir_prefix_matches_direct_children_across_owners( + sqlite_workspace_table, +) -> None: + """parent_path='foo/bar' must return each owner's DIRECT children of + foo/bar — the workspace-wide contract for lazy directory loading.""" + prefix = _build_list_scripts_workspace_descendant_prefix("foo/bar") + matched = _run_like(sqlite_workspace_table, prefix) + assert matched == [ + "workspace/alice/foo/bar/a.py", + "workspace/bob/foo/bar/b.py", + ], matched + + +def test_sqlite_workspace_root_prefix_matches_each_owner_root( + sqlite_workspace_table, +) -> None: + """parent_path='' must return root-level files across all owners — + requirement: empty parent_path defaults to pulling root-path files.""" + prefix = _build_list_scripts_workspace_descendant_prefix("") + matched = _run_like(sqlite_workspace_table, prefix) + # Direct children of each owner's root = that owner's root-level file. + assert matched == ["workspace/alice/root.py"], matched + + +def test_sqlite_workspace_subdir_escapes_underscore_across_owners() -> None: + """Escaping still works with the owner wildcard added: `foo_bar` must + not match `fooXbar` under ANY owner.""" + engine = create_engine("sqlite:///:memory:") + metadata = MetaData() + table = Table( + "paths", + metadata, + Column("relative_path", String(1024), nullable=False), + ) + metadata.create_all(engine) + with engine.begin() as conn: + conn.execute( + table.insert(), + [ + {"relative_path": "workspace/alice/foo_bar/inner.py"}, + {"relative_path": "workspace/bob/foo_bar/inner.py"}, + {"relative_path": "workspace/alice/fooXbar/decoy.py"}, + ], + ) + prefix = _build_list_scripts_workspace_descendant_prefix("foo_bar") + matched = _run_like((engine, table), prefix) + assert matched == [ + "workspace/alice/foo_bar/inner.py", + "workspace/bob/foo_bar/inner.py", + ], matched + engine.dispose() + + +# ─── layer 4: single-script visibility enforcement ──────────────── + + +def _make_script_for_can_view( + *, owner_user_id: str = "U001", visibility: str = "private" +) -> SimpleNamespace: + return SimpleNamespace( + script_id="S1", + workspace_id="W001", + current_object_id="O1", + owner_user_id=owner_user_id, + script_name="x.py", + script_type="python", + visibility=visibility, + status="active", + is_locked=0, + ) + + +class TestScriptCanView: + """同一 workspace 内：owner 永远可见自己的脚本（含 private）； + 其他成员只见 visibility in {workspace, public} 的脚本； + admin 全部可见——与 data resources 的 can_view 对称。""" + + @staticmethod + def _viewer() -> SimpleNamespace: + return _ctx("U002") # not the owner + + @staticmethod + def _owner() -> SimpleNamespace: + return _ctx("U001") + + @staticmethod + def _admin() -> SimpleNamespace: + return _ctx("U002", is_admin=True) + + def test_owner_can_view_own_private_script(self) -> None: + from backend.api.scripts import script_can_view + + assert ( + script_can_view( + _make_script_for_can_view(visibility="private"), + self._owner(), + ) + is True + ) + + def test_workspace_member_cannot_view_others_private_script(self) -> None: + from backend.api.scripts import script_can_view + + assert ( + script_can_view( + _make_script_for_can_view(visibility="private"), + self._viewer(), + ) + is False + ) + + def test_workspace_member_can_view_others_workspace_script(self) -> None: + from backend.api.scripts import script_can_view + + assert ( + script_can_view( + _make_script_for_can_view(visibility="workspace"), + self._viewer(), + ) + is True + ) + + def test_workspace_member_can_view_others_public_script(self) -> None: + from backend.api.scripts import script_can_view + + assert ( + script_can_view( + _make_script_for_can_view(visibility="public"), + self._viewer(), + ) + is True + ) + + def test_admin_can_view_others_private_script(self) -> None: + from backend.api.scripts import script_can_view + + assert ( + script_can_view( + _make_script_for_can_view(visibility="private"), + self._admin(), + ) + is True + ) + + +def _script_row_for_get(script: SimpleNamespace) -> SimpleNamespace: + """Build the (script, storage_object) tuple get_script_row returns.""" + storage = SimpleNamespace( + relative_path=f"workspace/{script.owner_user_id}/x.py", + object_key=f"W001/{script.owner_user_id}/x.py", + content_hash="h", + size_bytes=1, + ) + return SimpleNamespace(data=(script, storage)) + + +async def test_get_script_non_owner_private_returns_404() -> None: + """Non-owner must NOT read another member's private script by id + (mirrors get_visible_resource for data resources).""" + from datetime import datetime, timezone + + from backend.api.scripts import get_script + + script = _make_script_for_can_view(visibility="private") + script.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc) + script.updated_at = datetime(2026, 1, 1, tzinfo=timezone.utc) + mock_session = MagicMock() + mock_session.execute = AsyncMock( + return_value=MagicMock(one_or_none=MagicMock(return_value=(script, None))) + ) + with pytest.raises(HTTPException) as exc: + await get_script("S1", context=_ctx("U002"), session=mock_session) + assert exc.value.status_code == 404 + + +async def test_get_script_owner_can_read_own_private_script() -> None: + from datetime import datetime, timezone + + from backend.api.scripts import get_script + + script = _make_script_for_can_view(visibility="private") + script.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc) + script.updated_at = datetime(2026, 1, 1, tzinfo=timezone.utc) + storage = SimpleNamespace( + relative_path="workspace/U001/x.py", + object_key="W001/U001/x.py", + content_hash="h", + size_bytes=1, + ) + mock_session = MagicMock() + mock_session.execute = AsyncMock( + return_value=MagicMock(one_or_none=MagicMock(return_value=(script, storage))) + ) + result = await get_script("S1", context=_ctx("U001"), session=mock_session) + assert result["data"]["script_id"] == "S1" + assert result["data"]["visibility"] == "private" + + +async def test_get_script_non_owner_can_read_workspace_visible_script() -> None: + from datetime import datetime, timezone + + from backend.api.scripts import get_script + + script = _make_script_for_can_view(visibility="workspace") + script.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc) + script.updated_at = datetime(2026, 1, 1, tzinfo=timezone.utc) + storage = SimpleNamespace( + relative_path="workspace/U001/x.py", + object_key="W001/U001/x.py", + content_hash="h", + size_bytes=1, + ) + mock_session = MagicMock() + mock_session.execute = AsyncMock( + return_value=MagicMock(one_or_none=MagicMock(return_value=(script, storage))) + ) + result = await get_script("S1", context=_ctx("U002"), session=mock_session) + assert result["data"]["script_id"] == "S1" + assert result["data"]["visibility"] == "workspace"