fix: parameterize SQL queries in search repository type filters

Replace f-string interpolation with parameterized queries for note_types,
search_item_types, and metadata filter paths to prevent SQL injection.

Fixes #591

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: phernandez <paul@basicmachines.co>

Author: phernandez

src/basic_memory/repository/postgres_search_repository.py

@@ -16,10 +16,7 @@
 from basic_memory.repository.embedding_provider_factory import create_embedding_provider
 from basic_memory.repository.search_index_row import SearchIndexRow
 from basic_memory.repository.search_repository_base import SearchRepositoryBase
-from basic_memory.repository.metadata_filters import (
-    parse_metadata_filters,
-    build_postgres_json_path,
-)
+from basic_memory.repository.metadata_filters import parse_metadata_filters
 from basic_memory.repository.semantic_errors import SemanticDependenciesMissingError
 from basic_memory.schemas.search import SearchItemType, SearchRetrievalMode
 
@@ -677,19 +674,23 @@ async def search(
             else:
                 conditions.append("search_index.permalink = :permalink")
 
-        # Handle search item type filter
+        # Handle search item type filter (parameterized for defense-in-depth)
         if search_item_types:
-            type_list = ", ".join(f"'{t.value}'" for t in search_item_types)
-            conditions.append(f"search_index.type IN ({type_list})")
-
-        # Handle note type filter using JSONB containment (frontmatter type field)
+            type_placeholders = []
+            for idx, t in enumerate(search_item_types):
+                param_name = f"search_type_{idx}"
+                params[param_name] = t.value
+                type_placeholders.append(f":{param_name}")
+            conditions.append(f"search_index.type IN ({', '.join(type_placeholders)})")
+
+        # Handle note type filter using JSONB containment (parameterized)
         if note_types:
-            # Use JSONB @> operator for efficient containment queries
             type_conditions = []
-            for note_type in note_types:
-                # Create JSONB containment condition for each note type
+            for idx, note_type in enumerate(note_types):
+                param_name = f"note_type_{idx}"
+                params[param_name] = json.dumps({"note_type": note_type})
                 type_conditions.append(
-                    f'search_index.metadata @> \'{{"note_type": "{note_type}"}}\''
+                    f"search_index.metadata @> CAST(:{param_name} AS jsonb)"
                 )
             conditions.append(f"({' OR '.join(type_conditions)})")
 
@@ -701,15 +702,23 @@ async def search(
             order_by_clause = ", search_index.updated_at DESC"
 
         # Handle structured metadata filters (frontmatter)
+        # Uses jsonb_extract_path_text() / jsonb_extract_path() with parameterized
+        # path parts instead of #>> / #> with interpolated paths.
         if metadata_filters:
             parsed_filters = parse_metadata_filters(metadata_filters)
             from_clause = "search_index JOIN entity ON search_index.entity_id = entity.id"
             metadata_expr = "entity.entity_metadata::jsonb"
 
             for idx, filt in enumerate(parsed_filters):
-                path = build_postgres_json_path(filt.path_parts)
-                text_expr = f"({metadata_expr} #>> '{path}')"
-                json_expr = f"({metadata_expr} #> '{path}')"
+                # Parameterize each JSON path part individually
+                path_param_names = []
+                for j, part in enumerate(filt.path_parts):
+                    path_param = f"meta_path_{idx}_{j}"
+                    params[path_param] = part
+                    path_param_names.append(f":{path_param}")
+                path_args = ", ".join(path_param_names)
+                text_expr = f"jsonb_extract_path_text({metadata_expr}, {path_args})"
+                json_expr = f"jsonb_extract_path({metadata_expr}, {path_args})"
 
                 if filt.op == "eq":
                     value_param = f"meta_val_{idx}"
@@ -727,14 +736,12 @@ async def search(
                     continue
 
                 if filt.op == "contains":
-                    import json as _json
-
                     base_param = f"meta_val_{idx}"
                     tag_conditions = []
                     # Require all values to be present
                     for j, val in enumerate(filt.value):
                         tag_param = f"{base_param}_{j}"
-                        params[tag_param] = _json.dumps([val])
+                        params[tag_param] = json.dumps([val])
                         like_param = f"{base_param}_{j}_like"
                         params[like_param] = f'%"{val}"%'
                         like_param_single = f"{base_param}_{j}_like_single"
@@ -749,7 +756,7 @@ async def search(
 
                 if filt.op in {"gt", "gte", "lt", "lte", "between"}:
                     compare_expr = (
-                        f"({metadata_expr} #>> '{path}')::double precision"
+                        f"{text_expr}::double precision"
                         if filt.comparison == "numeric"
                         else text_expr
                     )

src/basic_memory/repository/sqlite_search_repository.py

@@ -666,16 +666,24 @@ async def search(
                     params["permalink"] = permalink_text
                     match_conditions.append("search_index.permalink MATCH :permalink")
 
-        # Handle entity type filter
+        # Handle entity type filter (parameterized for defense-in-depth)
         if search_item_types:
-            type_list = ", ".join(f"'{t.value}'" for t in search_item_types)
-            conditions.append(f"search_index.type IN ({type_list})")
-
-        # Handle note type filter (frontmatter type field)
+            type_placeholders = []
+            for idx, t in enumerate(search_item_types):
+                param_name = f"search_type_{idx}"
+                params[param_name] = t.value
+                type_placeholders.append(f":{param_name}")
+            conditions.append(f"search_index.type IN ({', '.join(type_placeholders)})")
+
+        # Handle note type filter (frontmatter type field, parameterized)
         if note_types:
-            type_list = ", ".join(f"'{t}'" for t in note_types)
+            type_placeholders = []
+            for idx, t in enumerate(note_types):
+                param_name = f"note_type_{idx}"
+                params[param_name] = t
+                type_placeholders.append(f":{param_name}")
             conditions.append(
-                f"json_extract(search_index.metadata, '$.note_type') IN ({type_list})"
+                f"json_extract(search_index.metadata, '$.note_type') IN ({', '.join(type_placeholders)})"
             )
 
         # Handle date filter using datetime() for proper comparison

tests/repository/test_postgres_search_repository.py

@@ -572,3 +572,28 @@ async def test_postgres_dimension_mismatch_triggers_table_recreation(session_mak
         row = result.fetchone()
         assert row is not None
         assert int(row[0]) == 8
+
+
+@pytest.mark.asyncio
+async def test_postgres_note_types_sql_injection_returns_empty(session_maker, test_project):
+    """Postgres JSONB containment with SQL injection payload must not alter query."""
+    repo = PostgresSearchRepository(session_maker, project_id=test_project.id)
+
+    malicious_payloads = [
+        'note"}}\' OR \'1\'=\'1',
+        "note\"; DROP TABLE search_index;--",
+        'note"}} UNION SELECT * FROM entity--',
+    ]
+    for payload in malicious_payloads:
+        results = await repo.search(note_types=[payload])
+        assert results == [], f"Injection payload should not match: {payload}"
+
+
+@pytest.mark.asyncio
+async def test_postgres_metadata_filters_path_parameterized(session_maker, test_project):
+    """Metadata filter paths use jsonb_extract_path_text with parameterized parts."""
+    repo = PostgresSearchRepository(session_maker, project_id=test_project.id)
+
+    # Nested path should work without SQL injection risk
+    results = await repo.search(metadata_filters={"schema.confidence": {"$gt": 0.5}})
+    assert isinstance(results, list)

tests/repository/test_search_repository.py

@@ -941,3 +941,34 @@ async def test_search_metadata_filters_numeric_comparisons(search_repository, se
         metadata_filters={"schema.confidence": {"$between": [0.3, 0.6]}}
     )
     assert {result.id for result in results} == {entity_low.id}
+
+
+# --- SQL injection safety tests ---
+# These tests verify that user-supplied filter values are parameterized and cannot
+# alter query structure. Each test passes a malicious payload and asserts the query
+# completes safely (returning empty results) rather than causing a SQL error or
+# data exfiltration.
+
+
+@pytest.mark.asyncio
+async def test_note_types_sql_injection_returns_empty(search_repository):
+    """note_types with SQL injection payload must not alter query structure."""
+    malicious_payloads = [
+        "note' OR '1'='1",
+        "note'; DROP TABLE search_index;--",
+        'note" OR 1=1--',
+        "note') UNION SELECT * FROM entity--",
+    ]
+    for payload in malicious_payloads:
+        results = await search_repository.search(note_types=[payload])
+        # Injection should be treated as a literal string value, not executed as SQL
+        assert results == [], f"Injection payload should not match: {payload}"
+
+
+@pytest.mark.asyncio
+async def test_search_item_types_parameterized(search_repository):
+    """search_item_types enum values are parameterized, not interpolated."""
+    # Normal enum usage still works
+    results = await search_repository.search(search_item_types=[SearchItemType.ENTITY])
+    # Should not raise — parameterized query handles enum values safely
+    assert isinstance(results, list)