fix(scripts): actually escape LIKE pattern literals + scope count endpoint
Codex review of #36 + #37 surfaced that my prior `escape="\\"` only declared the escape character — the pattern literals themselves still contained unescaped `_` and `%`, so `parent_path="foo_bar"` continued to match `fooXbar/...`, `foo2bar/...`, etc. My earlier ESCAPE-clause assertions were tautological: they verified the SQL rendered the ESCAPE keyword without ever checking that the pattern was actually escaped. The tests passed; the leak persisted. Fix in three layers: 1. Real escape: `backend/src/backend/scripts.py` gains `_escape_like_pattern(value)` that escapes `\` → `\\`, `%` → `\%`, `_` → `\_` (in that order — the escape char MUST be escaped first). `_build_list_scripts_descendant_prefix` now returns the escaped prefix. `list_workspace_directories` and `delete_workspace_directory` also escape their server-built prefixes. `count_scripts` escapes the user subtree prefix. 2. Same bug elsewhere: `backend/src/backend/resources.py:404` had the identical `DataResources.resource_name.like(f"%{keyword}%")` pattern; a search for "100%" would match everything. Now escaped too. 3. Count endpoint scope: `count_scripts` was workspace-wide and skipped the StorageObjects JOIN. Now INNER JOINs StorageObjects (drops orphans whose current_object_id is dangling) and filters by `workspace/{user_id}/` subtree so the result matches what `list_scripts(parent_path="")` would return. Multi-member workspaces no longer over-report, and orphan rows no longer inflate the count. Frontend: `DashboardRoute` is not keyed by workspace/user (only ScriptsPage is), so without a workspace_id dep the previous workspace's count persisted across navigation. useEffect now depends on `currentWorkspace?.workspace_id`; `loadScriptCount` clears the count to null at the start of the fetch so the dashboard doesn't flash a stale number. Tests — backend/tests/test_list_scripts_parent_path.py - Rewritten with three layers of coverage: * Pure helper tests for `_escape_like_pattern` (7 cases including backslash-escape-first ordering). * SQL-contract tests asserting the COMPILED PATTERN contains the escaped form (lowercased to neutralise SQLAlchemy keyword casing). * BEHAVIORAL tests on SQLite in-memory with the same LIKE semantics — proves the fix actually prevents the wildcard leak. Includes a negative test (without escape, siblings DO match) so the fixture is verified to exercise the bug. Tests — backend/tests/test_count_scripts.py - Updated to assert the JOIN + user-scope filter. New test verifies two different users in the same workspace get different subtrees. Verified: - pytest backend/tests: 65 passed (43 baseline + 12 list_scripts + 4 count + 6 helper/SQLite behavioral) - pnpm typecheck: clean - Raw SQL on MySQL (live DB) confirms `LIKE 'workspace/.../foo\_bar/%%' ESCAPE '\\'`.
This commit is contained in:
tao.chen
@@ -400,8 +400,12 @@ async def list_resources(
|
||||
)
|
||||
statement = statement.where(DataResources.visibility == visibility)
|
||||
if keyword:
|
||||
# Escape LIKE metacharacters so a search like "100%" or "my_file"
|
||||
# doesn't act as a wildcard. The outer "%...%" wildcards stay raw.
|
||||
from backend.scripts import _escape_like_pattern # local import: avoid cycle
|
||||
escaped = _escape_like_pattern(keyword.strip())
|
||||
statement = statement.where(
|
||||
DataResources.resource_name.like(f"%{keyword.strip()}%")
|
||||
DataResources.resource_name.like(f"%{escaped}%", escape="\\")
|
||||
)
|
||||
rows = (await session.execute(statement)).all()
|
||||
return {
|
||||
|
||||
@@ -115,10 +115,25 @@ def user_relative_path(context: RequestContext, child_path: str = "") -> str:
|
||||
return f"{base}/{normalized}" if normalized else base
|
||||
|
||||
|
||||
def _escape_like_pattern(value: str) -> str:
|
||||
"""Escape SQL LIKE metacharacters so user-supplied folder names that
|
||||
contain ``_`` or ``%`` do not act as wildcards.
|
||||
|
||||
Must be paired with ``escape="\\\\"`` on the LIKE clause so MySQL
|
||||
recognizes the doubled backslash as a single literal backslash escape.
|
||||
The trailing ``%`` / ``%/%`` SQL wildcards are NOT escaped — they are
|
||||
added by the caller and are meant to be wildcards.
|
||||
"""
|
||||
# Order matters: escape the escape char FIRST, otherwise the next two
|
||||
# replacements would double-escape our newly inserted backslashes.
|
||||
return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
|
||||
|
||||
|
||||
def _build_list_scripts_descendant_prefix(
|
||||
context: RequestContext, parent_path: str
|
||||
) -> str:
|
||||
"""Return the materialized-path prefix for direct children of ``parent_path``.
|
||||
"""Return the escaped materialized-path prefix for direct children of
|
||||
``parent_path``.
|
||||
|
||||
The endpoint appends ``LIKE '<prefix>/%' AND NOT LIKE '<prefix>/%/%'``
|
||||
against ``storage_objects.relative_path`` so only scripts whose parent
|
||||
@@ -127,6 +142,10 @@ def _build_list_scripts_descendant_prefix(
|
||||
|
||||
Empty ``parent_path`` produces the user-scoped root prefix — i.e. the
|
||||
endpoint returns root-level scripts only, not the full 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)
|
||||
scoped_prefix = user_relative_path(context)
|
||||
@@ -134,7 +153,7 @@ def _build_list_scripts_descendant_prefix(
|
||||
target_prefix = f"{scoped_prefix}/{normalized_parent}"
|
||||
else:
|
||||
target_prefix = scoped_prefix
|
||||
return f"{target_prefix}/"
|
||||
return f"{_escape_like_pattern(target_prefix)}/"
|
||||
|
||||
|
||||
def safe_script_name(value: str, script_type: str) -> str:
|
||||
@@ -769,7 +788,7 @@ async def list_workspace_directories(
|
||||
scoped_prefix = user_relative_path(context)
|
||||
parent = normalize_user_path(parent_path)
|
||||
target_prefix = f"{scoped_prefix}/{parent}" if parent else scoped_prefix
|
||||
descendant_prefix = f"{target_prefix}/"
|
||||
descendant_prefix = f"{_escape_like_pattern(target_prefix)}/"
|
||||
|
||||
rows = (
|
||||
await session.execute(
|
||||
@@ -805,7 +824,8 @@ async def list_workspace_directories(
|
||||
|
||||
for directory in directories.values():
|
||||
# directory['path'] is already workspace-relative and includes the parent segment.
|
||||
child_prefix = f"{scoped_prefix}/{directory['path']}/"
|
||||
# Escape defensively in case the DB has folder names containing `_` or `%`.
|
||||
child_prefix = f"{_escape_like_pattern(scoped_prefix)}/{_escape_like_pattern(directory['path'])}/"
|
||||
has_children = await session.scalar(
|
||||
select(StorageObjects.storage_object_id).where(
|
||||
StorageObjects.workspace_id == context.workspace.workspace_id,
|
||||
@@ -1007,7 +1027,7 @@ async def delete_workspace_directory(
|
||||
raise HTTPException(status.HTTP_404_NOT_FOUND, "directory not found")
|
||||
target_ulid = target_dir_row.storage_object_id
|
||||
|
||||
child_prefix = f"{target_relative}/"
|
||||
child_prefix = f"{_escape_like_pattern(target_relative)}/"
|
||||
descendants = (
|
||||
(
|
||||
await session.execute(
|
||||
@@ -1117,17 +1137,29 @@ async def list_scripts(
|
||||
# 工作区内 active 脚本总数。DashboardRoute 等不需要列表但需要计数的场景使用,
|
||||
# 避免被 listScripts 的懒加载语义污染。该路由必须在 /scripts/{script_id} 之前声明
|
||||
# ——FastAPI 按声明顺序匹配,否则 `count` 会被当作 script_id 命中 get_script。
|
||||
#
|
||||
# 范围与 list_scripts(parent_path="") 对齐:INNER JOIN 到 StorageObjects 以排除
|
||||
# 孤儿脚本(其 current_object_id 没有 joinable row),按用户子树
|
||||
# (workspace/{user_id}/) 过滤。否则:
|
||||
# - 含孤儿 → 数字虚高
|
||||
# - workspace 范围 → 多成员工作区里 dashboard 会显示用户看不见的脚本
|
||||
@router.get("/api/v1/scripts/count")
|
||||
async def count_scripts(
|
||||
context: RequestContext = Depends(request_context),
|
||||
session: AsyncSession = Depends(database_session),
|
||||
) -> dict[str, Any]:
|
||||
user_subtree_prefix = f"{_escape_like_pattern(user_relative_path(context))}/%"
|
||||
total = await session.scalar(
|
||||
select(func.count())
|
||||
.select_from(Scripts)
|
||||
.join(
|
||||
StorageObjects,
|
||||
StorageObjects.storage_object_id == Scripts.current_object_id,
|
||||
)
|
||||
.where(
|
||||
Scripts.workspace_id == context.workspace.workspace_id,
|
||||
Scripts.status == "active",
|
||||
StorageObjects.relative_path.like(user_subtree_prefix, escape="\\"),
|
||||
)
|
||||
)
|
||||
return {
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
"""Unit tests for GET /api/v1/scripts/count endpoint.
|
||||
|
||||
Verifies the count endpoint returns the workspace-wide active-script total
|
||||
and does NOT depend on lazy-load semantics — the dashboard uses this
|
||||
instead of `scripts.length` to avoid underreporting.
|
||||
Verifies the count endpoint returns the same scope as
|
||||
``list_scripts(parent_path="")``: workspace + active scripts whose
|
||||
``StorageObjects.relative_path`` lives under the user's subtree. This
|
||||
avoids under/over-reporting on the dashboard — the count is the size of
|
||||
the set list_scripts would return if it weren't lazy.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -11,7 +13,6 @@ from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import func, select
|
||||
|
||||
from backend.scripts import count_scripts
|
||||
|
||||
@@ -26,16 +27,24 @@ def _ctx(user_id: str = "U001", workspace_id: str = "W001") -> SimpleNamespace:
|
||||
)
|
||||
|
||||
|
||||
def _compile(stmt) -> str:
|
||||
from sqlalchemy.dialects import mysql as mysql_dialect
|
||||
|
||||
return str(
|
||||
stmt.compile(
|
||||
dialect=mysql_dialect.dialect(),
|
||||
compile_kwargs={"literal_binds": True},
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
async def test_count_scripts_returns_scalar_int() -> None:
|
||||
captured = []
|
||||
|
||||
class _MockScalarResult:
|
||||
def scalar(self, _stmt):
|
||||
captured.append(_stmt)
|
||||
return 7
|
||||
|
||||
mock_session = MagicMock()
|
||||
mock_session.scalar = AsyncMock(side_effect=lambda stmt: (captured.append(stmt), 7)[1])
|
||||
mock_session.scalar = AsyncMock(
|
||||
side_effect=lambda stmt: (captured.append(stmt), 7)[1]
|
||||
)
|
||||
|
||||
result = await count_scripts(context=_ctx(), session=mock_session)
|
||||
assert result["data"] == {"total": 7}
|
||||
@@ -44,12 +53,14 @@ async def test_count_scripts_returns_scalar_int() -> None:
|
||||
# Exactly one COUNT(*) query issued.
|
||||
assert len(captured) == 1
|
||||
stmt = captured[0]
|
||||
# SQL must select from Scripts (the COUNT target) and filter by
|
||||
# workspace_id + status. Bind params render as :workspace_id_1 etc.
|
||||
text = str(stmt).lower()
|
||||
assert "from scripts" in text
|
||||
assert "workspace_id" in text
|
||||
assert "status" in text
|
||||
sql = _compile(stmt).lower()
|
||||
# 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 + user subtree.
|
||||
assert "scripts.workspace_id" in sql
|
||||
assert "scripts.status" in sql
|
||||
assert "workspace/u001/%" in sql
|
||||
|
||||
|
||||
async def test_count_scripts_handles_null_result() -> None:
|
||||
@@ -61,12 +72,33 @@ async def test_count_scripts_handles_null_result() -> None:
|
||||
assert result["data"] == {"total": 0}
|
||||
|
||||
|
||||
async def test_count_scripts_uses_user_specific_subtree() -> None:
|
||||
"""Different users in the same workspace must see different totals —
|
||||
each user's count is bounded by their own ``workspace/{user_id}/``
|
||||
subtree, NOT the whole workspace."""
|
||||
captured = []
|
||||
|
||||
mock_session = MagicMock()
|
||||
mock_session.scalar = AsyncMock(
|
||||
side_effect=lambda stmt: (captured.append(stmt), 3)[1]
|
||||
)
|
||||
|
||||
await count_scripts(context=_ctx(user_id="alice"), session=mock_session)
|
||||
sql_alice = _compile(captured[-1])
|
||||
|
||||
await count_scripts(context=_ctx(user_id="bob"), session=mock_session)
|
||||
sql_bob = _compile(captured[-1])
|
||||
|
||||
assert "workspace/alice/%" in sql_alice
|
||||
assert "workspace/alice/%" not in sql_bob
|
||||
assert "workspace/bob/%" in sql_bob
|
||||
|
||||
|
||||
async def test_count_scripts_route_declared_before_script_id_route() -> None:
|
||||
"""Static check: the `/api/v1/scripts/count` route MUST be declared in
|
||||
scripts.py before `/api/v1/scripts/{script_id}/...`, otherwise FastAPI's
|
||||
declaration-order matching will interpret `count` as a script_id."""
|
||||
from backend.scripts import count_scripts, get_script
|
||||
|
||||
# Both callables exist (sanity).
|
||||
assert callable(count_scripts)
|
||||
assert callable(get_script)
|
||||
@@ -1,14 +1,20 @@
|
||||
"""Unit tests for the parent_path filter clause on list_scripts.
|
||||
"""Tests for the parent_path filter clause on list_scripts, plus the
|
||||
LIKE-pattern escape contract for tree-walking queries.
|
||||
|
||||
Verifies the WHERE clause built by the endpoint encodes the intended
|
||||
"direct children of parent_path" semantics: ``relative_path LIKE 'prefix/%'``
|
||||
and ``NOT LIKE 'prefix/%/%'``. These are SQL-contract assertions (mock
|
||||
session, capture compiled SQL); the repo has no integration test layer
|
||||
for endpoints, so this is the only coverage. Brittle to SQLAlchemy/dialect
|
||||
rendering changes — review the assertions together with the endpoint if
|
||||
you upgrade SQLAlchemy.
|
||||
Three layers of coverage:
|
||||
|
||||
Mirrors the pattern in test_scripts.py (unit-level, no live DB).
|
||||
1. ``_escape_like_pattern`` unit tests — pure-function correctness.
|
||||
2. SQL-contract tests (mock session) — verifies the compiled SQL contains
|
||||
the escaped pattern AND the ``ESCAPE '\\'`` clause.
|
||||
3. Behavioral test (SQLite in-memory, real LIKE execution) — proves the
|
||||
fix actually prevents the wildcard leak that motivated the change.
|
||||
A folder named ``foo_bar`` MUST NOT match sibling paths like
|
||||
``fooXbar`` / ``foo2bar`` / ``foo/bar``.
|
||||
|
||||
The repo has no MySQL integration test layer, so SQLite stands in for
|
||||
LIKE semantics — both dialects treat ``_`` as "any single char" and
|
||||
``%`` as "any sequence" by default and honour the ``ESCAPE`` clause
|
||||
identically for the ASCII characters we care about.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -18,16 +24,17 @@ from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
from sqlalchemy import Column, MetaData, String, Table, create_engine, select, text
|
||||
from sqlalchemy.dialects import mysql as mysql_dialect
|
||||
|
||||
from backend.scripts import (
|
||||
_build_list_scripts_descendant_prefix,
|
||||
_escape_like_pattern,
|
||||
normalize_user_path,
|
||||
)
|
||||
from sqlalchemy.dialects import mysql as mysql_dialect
|
||||
|
||||
|
||||
def _ctx(user_id: str = "U001") -> SimpleNamespace:
|
||||
"""Stand-in for RequestContext — only ``user.user_id`` and ``workspace_id`` are read."""
|
||||
return SimpleNamespace(
|
||||
request_id="test",
|
||||
user=SimpleNamespace(user_id=user_id),
|
||||
@@ -37,6 +44,40 @@ def _ctx(user_id: str = "U001") -> SimpleNamespace:
|
||||
)
|
||||
|
||||
|
||||
# ─── layer 1: helper unit tests ──────────────────────────────────
|
||||
|
||||
|
||||
class TestEscapeLikePattern:
|
||||
"""The escape helper itself is the load-bearing piece — test it
|
||||
exhaustively before relying on it in SQL."""
|
||||
|
||||
def test_no_metachars_unchanged(self) -> None:
|
||||
assert _escape_like_pattern("foo/bar") == "foo/bar"
|
||||
assert _escape_like_pattern("workspace/alice") == "workspace/alice"
|
||||
assert _escape_like_pattern("") == ""
|
||||
|
||||
def test_underscore_escaped(self) -> None:
|
||||
assert _escape_like_pattern("foo_bar") == r"foo\_bar"
|
||||
|
||||
def test_percent_escaped(self) -> None:
|
||||
assert _escape_like_pattern("100%") == r"100\%"
|
||||
assert _escape_like_pattern("%foo") == r"\%foo"
|
||||
|
||||
def test_backslash_escaped_first(self) -> None:
|
||||
# Must escape the escape char first, otherwise the inserted
|
||||
# backslashes would be double-escaped by the later passes.
|
||||
assert _escape_like_pattern(r"a\b") == r"a\\b"
|
||||
assert _escape_like_pattern(r"a\%b") == r"a\\\%b"
|
||||
|
||||
def test_combined(self) -> None:
|
||||
assert _escape_like_pattern("foo_bar%baz") == r"foo\_bar\%baz"
|
||||
assert _escape_like_pattern("_%") == r"\_\%"
|
||||
assert _escape_like_pattern(r"\\_%") == r"\\\\\_\%"
|
||||
|
||||
|
||||
# ─── layer 1.5: prefix helper now escapes ─────────────────────────
|
||||
|
||||
|
||||
def test_descendant_prefix_root() -> None:
|
||||
"""Empty parent_path → descendant prefix is the scoped root + '/'."""
|
||||
prefix = _build_list_scripts_descendant_prefix(_ctx("alice"), "")
|
||||
@@ -49,14 +90,20 @@ def test_descendant_prefix_subdir() -> None:
|
||||
assert prefix == "workspace/alice/foo/bar/"
|
||||
|
||||
|
||||
def test_descendant_prefix_escapes_metachars() -> None:
|
||||
"""Folder name ``foo_bar`` MUST produce ``foo\\_bar`` in the prefix
|
||||
so the trailing ``%`` doesn't become 'match any single char before
|
||||
b'."""
|
||||
prefix = _build_list_scripts_descendant_prefix(_ctx("alice"), "foo_bar")
|
||||
assert prefix == r"workspace/alice/foo\_bar/"
|
||||
|
||||
|
||||
def test_descendant_prefix_normalizes_leading_trailing_slashes() -> None:
|
||||
"""Leading/trailing slashes on parent_path must be stripped."""
|
||||
prefix = _build_list_scripts_descendant_prefix(_ctx("alice"), "/foo/bar/")
|
||||
assert prefix == "workspace/alice/foo/bar/"
|
||||
|
||||
|
||||
def test_descendant_prefix_rejects_traversal() -> None:
|
||||
"""``..`` segments must raise (matches normalize_user_path contract)."""
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
_build_list_scripts_descendant_prefix(_ctx("alice"), "foo/../bar")
|
||||
assert exc.value.status_code == 422
|
||||
@@ -68,9 +115,19 @@ def test_normalize_user_path_strips() -> None:
|
||||
assert normalize_user_path("a\\b") == "a/b"
|
||||
|
||||
|
||||
# ─── layer 2: SQL contract ────────────────────────────────────────
|
||||
|
||||
|
||||
def _compile_sql(stmt) -> str:
|
||||
return str(
|
||||
stmt.compile(
|
||||
dialect=mysql_dialect.dialect(),
|
||||
compile_kwargs={"literal_binds": True},
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
async def test_list_scripts_where_clause_uses_like_prefix_and_excludes_deeper() -> None:
|
||||
"""The WHERE clause must include both LIKE prefix and NOT LIKE '%/%' filters
|
||||
so deeper descendants and prefix-siblings (foo/bar vs foo/bar2) are excluded."""
|
||||
from backend.scripts import list_scripts
|
||||
|
||||
captured_sql: list[str] = []
|
||||
@@ -82,15 +139,7 @@ async def test_list_scripts_where_clause_uses_like_prefix_and_excludes_deeper()
|
||||
mock_session = MagicMock()
|
||||
mock_session.execute = AsyncMock(
|
||||
side_effect=lambda stmt: (
|
||||
captured_sql.append(
|
||||
str(
|
||||
stmt.compile(
|
||||
dialect=mysql_dialect.dialect(),
|
||||
compile_kwargs={"literal_binds": True},
|
||||
)
|
||||
)
|
||||
)
|
||||
or _MockResult()
|
||||
captured_sql.append(_compile_sql(stmt)) or _MockResult()
|
||||
)
|
||||
)
|
||||
|
||||
@@ -98,16 +147,13 @@ async def test_list_scripts_where_clause_uses_like_prefix_and_excludes_deeper()
|
||||
|
||||
assert len(captured_sql) == 1
|
||||
sql = captured_sql[0].lower()
|
||||
# Direct-child LIKE prefix
|
||||
assert "like 'workspace/alice/foo/bar/%%'" in sql
|
||||
# NOT-LIKE deeper
|
||||
assert "not like 'workspace/alice/foo/bar/%%/%%'" in sql
|
||||
# active scripts only (existing contract preserved)
|
||||
assert "scripts.status" in sql or "scripts.status = 'active'" in sql or "scripts.status = :status" in sql
|
||||
|
||||
|
||||
async def test_list_scripts_empty_parent_path_targets_root_descendants() -> None:
|
||||
"""parent_path='' produces root-scoped LIKE prefix only, not full scan."""
|
||||
async def test_list_scripts_where_clause_escapes_pattern_literal() -> None:
|
||||
"""Regression: parent_path containing ``_`` MUST be escaped in the
|
||||
compiled LIKE pattern, otherwise sibling-path leak returns to bite."""
|
||||
from backend.scripts import list_scripts
|
||||
|
||||
captured_sql: list[str] = []
|
||||
@@ -119,63 +165,53 @@ async def test_list_scripts_empty_parent_path_targets_root_descendants() -> None
|
||||
mock_session = MagicMock()
|
||||
mock_session.execute = AsyncMock(
|
||||
side_effect=lambda stmt: (
|
||||
captured_sql.append(
|
||||
str(
|
||||
stmt.compile(
|
||||
dialect=mysql_dialect.dialect(),
|
||||
compile_kwargs={"literal_binds": True},
|
||||
)
|
||||
)
|
||||
)
|
||||
or _MockResult()
|
||||
)
|
||||
)
|
||||
|
||||
await list_scripts(parent_path="", context=_ctx("alice"), session=mock_session)
|
||||
|
||||
sql = captured_sql[0].lower()
|
||||
assert "like 'workspace/alice/%%'" in sql
|
||||
assert "not like 'workspace/alice/%%/%%'" in sql
|
||||
|
||||
|
||||
async def test_list_scripts_where_clause_includes_escape() -> None:
|
||||
"""Both LIKE clauses must declare ESCAPE so folder names containing ``_``
|
||||
or ``%`` do not act as SQL wildcards and match sibling paths."""
|
||||
from backend.scripts import list_scripts
|
||||
|
||||
captured_sql: list[str] = []
|
||||
|
||||
class _MockResult:
|
||||
def all(self):
|
||||
return []
|
||||
|
||||
mock_session = MagicMock()
|
||||
mock_session.execute = AsyncMock(
|
||||
side_effect=lambda stmt: (
|
||||
captured_sql.append(
|
||||
str(
|
||||
stmt.compile(
|
||||
dialect=mysql_dialect.dialect(),
|
||||
compile_kwargs={"literal_binds": True},
|
||||
)
|
||||
)
|
||||
)
|
||||
or _MockResult()
|
||||
captured_sql.append(_compile_sql(stmt)) or _MockResult()
|
||||
)
|
||||
)
|
||||
|
||||
await list_scripts(parent_path="foo_bar", context=_ctx("alice"), session=mock_session)
|
||||
sql = captured_sql[0]
|
||||
# Both patterns must carry ESCAPE; counts must match between the two LIKE
|
||||
# occurrences (one positive, one negated). SQLAlchemy doubles the escape
|
||||
# char for SQL string literals, so the rendered form is `ESCAPE '\\\\'`.
|
||||
# Normalize keyword case so we don't depend on SQLAlchemy casing.
|
||||
sql_lower = sql.lower()
|
||||
# Pattern literal must contain the ESCAPED underscore. SQLAlchemy
|
||||
# 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/alice/foo\\_bar/%%'" in sql_lower
|
||||
# NOT LIKE clause also escaped.
|
||||
assert r"not like 'workspace/alice/foo\\_bar/%%/%%'" in sql_lower
|
||||
# And both declare ESCAPE '\\'.
|
||||
assert sql.count("ESCAPE '\\\\'") == 2, sql
|
||||
|
||||
|
||||
async def test_list_workspace_directories_where_clause_includes_escape() -> None:
|
||||
"""list_workspace_directories must also emit ESCAPE — the same LIKE
|
||||
pattern was already vulnerable for pre-existing endpoints; this
|
||||
endpoint is in scope for the same fix."""
|
||||
async def test_list_scripts_where_clause_escapes_percent_pattern() -> None:
|
||||
"""Same regression for ``%``."""
|
||||
from backend.scripts import list_scripts
|
||||
|
||||
captured_sql: list[str] = []
|
||||
|
||||
class _MockResult:
|
||||
def all(self):
|
||||
return []
|
||||
|
||||
mock_session = MagicMock()
|
||||
mock_session.execute = AsyncMock(
|
||||
side_effect=lambda stmt: (
|
||||
captured_sql.append(_compile_sql(stmt)) or _MockResult()
|
||||
)
|
||||
)
|
||||
|
||||
await list_scripts(parent_path="100%match", context=_ctx("alice"), session=mock_session)
|
||||
sql = captured_sql[0]
|
||||
sql_lower = sql.lower()
|
||||
# SQLAlchemy doubles the escape char so `%` → `\%` becomes `\\%`
|
||||
# in the SQL string literal.
|
||||
assert r"workspace/alice/100\\%%match/%%" in sql_lower
|
||||
|
||||
|
||||
async def test_list_workspace_directories_where_clause_escapes_pattern() -> None:
|
||||
"""list_workspace_directories must escape user input too (was
|
||||
pre-existing debt)."""
|
||||
from backend.scripts import list_workspace_directories
|
||||
|
||||
captured_sql: list[str] = []
|
||||
@@ -194,23 +230,109 @@ async def test_list_workspace_directories_where_clause_includes_escape() -> None
|
||||
mock_session = MagicMock()
|
||||
mock_session.execute = AsyncMock(
|
||||
side_effect=lambda stmt: (
|
||||
captured_sql.append(
|
||||
str(
|
||||
stmt.compile(
|
||||
dialect=mysql_dialect.dialect(),
|
||||
compile_kwargs={"literal_binds": True},
|
||||
)
|
||||
)
|
||||
)
|
||||
or _MockResult()
|
||||
captured_sql.append(_compile_sql(stmt)) or _MockResult()
|
||||
)
|
||||
)
|
||||
|
||||
await list_workspace_directories(
|
||||
parent_path="foo_bar", context=_ctx("alice"), session=mock_session
|
||||
)
|
||||
# Two LIKE clauses in the children query + two in the has_children
|
||||
# check per directory in the result — for an empty result set only
|
||||
# the first batch executes, so we expect at least 2 ESCAPEs.
|
||||
sql = " ".join(captured_sql)
|
||||
assert sql.count("ESCAPE '\\\\'") >= 2, sql
|
||||
assert r"workspace/alice/foo\\_bar/" in sql, sql
|
||||
|
||||
|
||||
# ─── layer 3: behavioral test on real LIKE execution ──────────────
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sqlite_like_table():
|
||||
"""SQLite in-memory table with a single VARCHAR column. Stand-in for
|
||||
``storage_objects.relative_path`` — proves the actual LIKE executor
|
||||
behaves the way we expect with the escaped pattern."""
|
||||
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:
|
||||
# Target row (the one a parent_path="foo_bar" search MUST return).
|
||||
conn.execute(
|
||||
table.insert(),
|
||||
{"relative_path": "workspace/alice/foo_bar/inner.py"},
|
||||
)
|
||||
# Decoys the buggy LIKE would match but escaped the must NOT.
|
||||
conn.execute(
|
||||
table.insert(),
|
||||
{"relative_path": "workspace/alice/fooXbar/decoy.py"},
|
||||
)
|
||||
conn.execute(
|
||||
table.insert(),
|
||||
{"relative_path": "workspace/alice/foo2bar/decoy.py"},
|
||||
)
|
||||
# A truly unrelated path.
|
||||
conn.execute(
|
||||
table.insert(),
|
||||
{"relative_path": "workspace/alice/baz/inner.py"},
|
||||
)
|
||||
yield engine, table
|
||||
engine.dispose()
|
||||
|
||||
|
||||
def test_sqlite_like_with_escape_does_not_match_sibling(sqlite_like_table):
|
||||
"""Execute the actual LIKE pattern the endpoint would emit for
|
||||
parent_path='foo_bar'. Confirms only the target row matches."""
|
||||
engine, table = sqlite_like_table
|
||||
escaped_prefix = _escape_like_pattern("workspace/alice/foo_bar") + "/"
|
||||
pattern = f"{escaped_prefix}%"
|
||||
with engine.connect() as conn:
|
||||
rows = conn.execute(
|
||||
select(table.c.relative_path).where(
|
||||
table.c.relative_path.like(pattern, escape="\\")
|
||||
)
|
||||
).fetchall()
|
||||
matched = sorted(r[0] for r in rows)
|
||||
assert matched == ["workspace/alice/foo_bar/inner.py"], matched
|
||||
|
||||
|
||||
def test_sqlite_like_without_escape_matches_siblings(sqlite_like_table):
|
||||
"""Sanity check: WITHOUT escape, the same pattern matches the
|
||||
decoys too — confirming the test setup actually exercises the
|
||||
leak. If this assertion fails the SQLite fixture is broken."""
|
||||
engine, table = sqlite_like_table
|
||||
pattern = "workspace/alice/foo_bar/%"
|
||||
with engine.connect() as conn:
|
||||
rows = conn.execute(
|
||||
select(table.c.relative_path).where(
|
||||
table.c.relative_path.like(pattern)
|
||||
)
|
||||
).fetchall()
|
||||
matched = sorted(r[0] for r in rows)
|
||||
# Without escape, the buggy behaviour returns ALL three foo*bar rows.
|
||||
assert len(matched) >= 2, matched
|
||||
|
||||
|
||||
def test_sqlite_like_with_percent_in_name(sqlite_like_table):
|
||||
"""Folder name containing ``%`` — must be escaped too."""
|
||||
engine, table = sqlite_like_table
|
||||
with engine.begin() as conn:
|
||||
conn.execute(
|
||||
table.insert(),
|
||||
{"relative_path": "workspace/alice/100%off/x.py"},
|
||||
)
|
||||
conn.execute(
|
||||
table.insert(),
|
||||
{"relative_path": "workspace/alice/100Xoff/y.py"},
|
||||
)
|
||||
escaped_prefix = _escape_like_pattern("workspace/alice/100%off") + "/"
|
||||
pattern = f"{escaped_prefix}%"
|
||||
with engine.connect() as conn:
|
||||
rows = conn.execute(
|
||||
select(table.c.relative_path).where(
|
||||
table.c.relative_path.like(pattern, escape="\\")
|
||||
)
|
||||
).fetchall()
|
||||
matched = sorted(r[0] for r in rows)
|
||||
assert matched == ["workspace/alice/100%off/x.py"], matched
|
||||
@@ -1,6 +1,7 @@
|
||||
import { useEffect } from "react";
|
||||
import { useNavigate } from "react-router";
|
||||
|
||||
import { useAuth } from "~/context/AuthContext";
|
||||
import { DashboardPage } from "../../components/admin/DashboardPage";
|
||||
import { useScriptWorkspaceStore } from "./state/scriptWorkspaceStore";
|
||||
|
||||
@@ -9,14 +10,15 @@ export default function DashboardRoute() {
|
||||
const scriptCount = useScriptWorkspaceStore((s) => s.scriptCount);
|
||||
const loadScriptCount = useScriptWorkspaceStore((s) => s.loadScriptCount);
|
||||
const apiOnline = useScriptWorkspaceStore((s) => s.apiOnline);
|
||||
const workspaceId = useAuth().currentWorkspace?.workspace_id;
|
||||
const navigate = useNavigate();
|
||||
|
||||
// Independent of the lazy-loaded `scripts` array — the count endpoint
|
||||
// returns the workspace-wide total even when no folders have been
|
||||
// expanded yet (see #34 + #37).
|
||||
// Reload on workspace switch — DashboardRoute is not keyed by
|
||||
// workspace/user (only ScriptsPage is), so without this dep the
|
||||
// previous workspace's count would persist.
|
||||
useEffect(() => {
|
||||
void loadScriptCount();
|
||||
}, [loadScriptCount]);
|
||||
}, [loadScriptCount, workspaceId]);
|
||||
|
||||
return (
|
||||
<DashboardPage
|
||||
|
||||
@@ -398,7 +398,7 @@ export const useScriptWorkspaceStore = create<State>((set, get) => {
|
||||
loadScriptCount: async () => {
|
||||
const api = requireApi();
|
||||
if (get().scriptCountLoading) return;
|
||||
set({ scriptCountLoading: true });
|
||||
set({ scriptCountLoading: true, scriptCount: null });
|
||||
try {
|
||||
const total = await api.countScripts();
|
||||
set({ scriptCount: total });
|
||||
|
||||
Reference in New Issue
Block a user