From 9233d99237563204da64a80af50a2ba50daa5aea Mon Sep 17 00:00:00 2001 From: "tao.chen" <93983997+taochen-ct@users.noreply.github.com> Date: Fri, 21 Aug 2026 10:52:54 +0800 Subject: [PATCH] fix(scripts): escape LIKE wildcards in tree-walking queries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). --- backend/src/backend/scripts.py | 16 ++-- .../tests/test_list_scripts_parent_path.py | 80 ++++++++++++++++++- 2 files changed, 87 insertions(+), 9 deletions(-) diff --git a/backend/src/backend/scripts.py b/backend/src/backend/scripts.py index 6ea82d0..35ac45c 100644 --- a/backend/src/backend/scripts.py +++ b/backend/src/backend/scripts.py @@ -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()) ) diff --git a/backend/tests/test_list_scripts_parent_path.py b/backend/tests/test_list_scripts_parent_path.py index 77d7ac5..8901491 100644 --- a/backend/tests/test_list_scripts_parent_path.py +++ b/backend/tests/test_list_scripts_parent_path.py @@ -135,4 +135,82 @@ 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 \ No newline at end of file + 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 \ No newline at end of file