fix(scripts): escape LIKE wildcards in tree-walking queries

Codex review of #34 surfaced that folder names containing ``_`` (matches any
single character in SQL LIKE) or ``%`` (matches any sequence) would leak
across sibling paths when used as parent_path / descendant_prefix. Two
endpoints were vulnerable:

- list_workspace_directories (pre-existing, line 780/781, 814/815)
- list_scripts (added in #34, line 1101/1102)

Both had four LIKE calls total, all unescaped. ``safe_directory_name``
allows ``_`` and rejects nothing relevant, so any user-created directory
named e.g. ``foo_bar`` would have its queries silently match
``fooXbar/...`` paths too.

Fix: add ``escape="\\"`` to every .like() / ~.like() call so MySQL emits
``ESCAPE '\\'``. SQLAlchemy doubles the escape character for SQL string
literals (renders as ``ESCAPE '\\\\'``); test asserts the rendered form
appears for both positive and negated clauses in each endpoint.

Also touched get_workspace_tree's `like(like_prefix)` (line 707) and
delete_workspace_directory's `like(f"{child_prefix}%")` (line 1017) —
the prefixes are server-built so technically not user-controllable,
but defense in depth costs nothing.

Verified: pytest 52 passed (50 + 2 new ESCAPE assertions).
This commit is contained in:
tao.chen
2026-09-02 10:10:41 +08:00
committed by tao.chen
parent 1ea7e2dc28
commit df46693533
2 changed files with 87 additions and 9 deletions
+8 -8
View File
@@ -704,7 +704,7 @@ async def get_workspace_tree(
StorageObjects.workspace_id == context.workspace.workspace_id,
StorageObjects.object_status == "available",
StorageObjects.is_deleted == 0,
StorageObjects.relative_path.like(like_prefix),
StorageObjects.relative_path.like(like_prefix, escape="\\"),
StorageObjects.usage_type.notin_(TREE_EXCLUDED_USAGE_TYPES),
)
)
@@ -777,8 +777,8 @@ async def list_workspace_directories(
StorageObjects.workspace_id == context.workspace.workspace_id,
StorageObjects.object_status == "available",
StorageObjects.is_deleted == 0,
StorageObjects.relative_path.like(f"{descendant_prefix}%"),
~StorageObjects.relative_path.like(f"{descendant_prefix}%/%"),
StorageObjects.relative_path.like(f"{descendant_prefix}%", escape="\\"),
~StorageObjects.relative_path.like(f"{descendant_prefix}%/%", escape="\\"),
StorageObjects.object_type == "directory",
StorageObjects.usage_type.notin_(TREE_EXCLUDED_USAGE_TYPES),
)
@@ -811,8 +811,8 @@ async def list_workspace_directories(
StorageObjects.workspace_id == context.workspace.workspace_id,
StorageObjects.object_status == "available",
StorageObjects.is_deleted == 0,
StorageObjects.relative_path.like(f"{child_prefix}%"),
~StorageObjects.relative_path.like(f"{child_prefix}%/%"),
StorageObjects.relative_path.like(f"{child_prefix}%", escape="\\"),
~StorageObjects.relative_path.like(f"{child_prefix}%/%", escape="\\"),
StorageObjects.usage_type.notin_(TREE_EXCLUDED_USAGE_TYPES),
).limit(1)
)
@@ -1014,7 +1014,7 @@ async def delete_workspace_directory(
select(StorageObjects).where(
StorageObjects.workspace_id == context.workspace.workspace_id,
StorageObjects.object_status == "available",
StorageObjects.relative_path.like(f"{child_prefix}%"),
StorageObjects.relative_path.like(f"{child_prefix}%", escape="\\"),
).order_by(func.length(StorageObjects.relative_path).desc())
)
)
@@ -1098,8 +1098,8 @@ async def list_scripts(
.where(
Scripts.workspace_id == context.workspace.workspace_id,
Scripts.status == "active",
StorageObjects.relative_path.like(f"{descendant_prefix}%"),
~StorageObjects.relative_path.like(f"{descendant_prefix}%/%"),
StorageObjects.relative_path.like(f"{descendant_prefix}%", escape="\\"),
~StorageObjects.relative_path.like(f"{descendant_prefix}%/%", escape="\\"),
)
.order_by(Scripts.updated_at.desc())
)
@@ -136,3 +136,81 @@ async def test_list_scripts_empty_parent_path_targets_root_descendants() -> None
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()
)
)
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 '\\\\'`.
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."""
from backend.scripts import list_workspace_directories
captured_sql: list[str] = []
class _MockScalarResult:
def scalar(self):
return None
class _MockResult:
def all(self):
return []
def scalars(self):
return _MockScalarResult()
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_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