fix(scripts): workspace-wide parent_path listing + private visibility on reads

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。
This commit is contained in:
tao.chen
2026-08-21 17:20:05 +08:00
parent 5045f0ad8c
commit 8cf4b53dd4
3 changed files with 362 additions and 51 deletions
+56 -28
View File
@@ -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 '<prefix>/%'`` 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 '<prefix>/%' AND NOT LIKE '<prefix>/%/%'`` 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/%/<parent>``)跨所有 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(
+10 -8
View File
@@ -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
+296 -15
View File
@@ -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
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"