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-08-21 10:52:54 +08:00
parent a804a2f441
commit 9233d99237
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())
)