fix: OCI26ai sql query patches (#5046)

# What does this PR do?
<!-- Provide a short summary of what this PR does and why. Link to
relevant issues if applicable. -->
Parameterizes SQL queries to prevent injection in the sqlstore and OCI
providers.

- Replace manual string escaping (`replace("'", "''")`) with
parameterized bind variables in the authorized sqlstore's access control
WHERE
  clause builder
- Fix SQL injection in OCI 26AI vector_io provider where `chunk_id`
values were directly concatenated into a `DELETE` query
- Add `where_sql_params` support to the `SqlStore` protocol and
SQLAlchemy implementation so raw SQL clauses can carry bound parameters
- Add a pre-commit hook to detect f-string interpolation in SQL
statements and prevent regressions

<!-- If resolving an issue, uncomment and update the line below -->
<!-- Closes #[issue-number] -->
Closes RHAIENG-3254

## Test Plan
<!-- Describe the tests you ran to verify your changes with result
summaries. *Provide clear instructions so the plan can be easily
re-executed.* -->
Existing unit tests pass (`test_authorized_sqlstore.py`,
`test_sqlstore.py` — 15/15)
Pre-commit hooks pass on all modified files

<!-- For API changes, include:
1. A testing script (Python, curl, etc.) that exercises the new/modified
endpoints
2. The output from running your script

Example:
```python
...
...
```

Output:
```
<paste actual output here>
```
-->

Signed-off-by: Doug Edgar <dedgar@redhat.com>
Co-authored-by: Charlie Doern <cdoern@redhat.com>

Author: rhdedgar

.pre-commit-config.yaml

@@ -221,6 +221,33 @@ repos:
               echo;
               exit 1;
             } || true
+      - id: no-sql-string-interpolation
+        name: Block f-string SQL construction (SQL injection risk)
+        entry: bash
+        language: system
+        types: [python]
+        pass_filenames: true
+        args:
+          - -c
+          - |
+            grep -EnH 'f"""[^"]*\b(DELETE|INSERT|UPDATE|SELECT|MERGE)\b.*\{[^}]*\}' "$@" \
+            | grep -v 'self\.table_name' \
+            | grep -v 'self\.dimensions' \
+            | grep -v 'self\.vector_datatype' \
+            | grep -v '#\s*nosec' \
+            && {
+              echo;
+              echo "SQL injection risk: f-string interpolation in SQL query detected."
+              echo "Use parameterized bind variables instead:"
+              echo "  oracledb:    :param_name"
+              echo "  psycopg2:    %s or %(name)s"
+              echo "  aiosqlite:   ? or :name"
+              echo "  SQLAlchemy:  text().bindparams()"
+              echo "If the interpolated value is a safe schema identifier (not user data),"
+              echo "add '# nosec' to the line to suppress this check."
+              echo;
+              exit 1;
+            } || true
       - id: check-api-independence
         name: Ensure llama_stack_api does not import llama_stack
         entry: bash

src/llama_stack/core/storage/sqlstore/authorized_sqlstore.py

@@ -4,6 +4,7 @@
 # This source code is licensed under the terms described in the LICENSE file in
 # the root directory of this source tree.
 
+import re
 from collections.abc import Mapping, Sequence
 from typing import Any, Literal
 
@@ -165,11 +166,12 @@ async def fetch_all(
         action: Action = Action.READ,
     ) -> PaginatedResponse:
         """Fetch all rows with automatic access control filtering."""
-        access_where = self._build_access_control_where_clause(self.policy)
+        access_where, access_params = self._build_access_control_where_clause(self.policy)
         rows = await self.sql_store.fetch_all(
             table=table,
             where=where,
             where_sql=access_where,
+            where_sql_params=access_params,
             limit=limit,
             order_by=order_by,
             cursor=cursor,
@@ -236,9 +238,10 @@ async def delete(self, table: str, where: Mapping[str, Any]) -> None:
         """Delete rows with automatic access control filtering."""
         await self.sql_store.delete(table, where)
 
-    def _build_access_control_where_clause(self, policy: list[AccessRule]) -> str:
+    def _build_access_control_where_clause(self, policy: list[AccessRule]) -> tuple[str, dict[str, Any]]:
         """Build SQL WHERE clause for access control filtering.
 
+        Returns a tuple of (sql_clause, bind_params) using parameterized queries.
         Only applies SQL filtering for the default policy to ensure correctness.
         For custom policies, uses conservative filtering to avoid blocking legitimate access.
         """
@@ -294,46 +297,46 @@ def _get_public_access_conditions(self) -> list[str]:
         """
         return ["owner_principal = ''"]
 
-    def _build_default_policy_where_clause(self, current_user: User | None) -> str:
+    def _build_default_policy_where_clause(self, current_user: User | None) -> tuple[str, dict[str, Any]]:
         """Build SQL WHERE clause for the default policy.
 
+        Returns a tuple of (sql_clause, bind_params) using parameterized queries.
         Default policy: permit all actions when user in owners [roles, teams, projects, namespaces]
         This means user must match ANY attribute category that exists in the resource (OR logic).
         """
         base_conditions = self._get_public_access_conditions()
+        params: dict[str, Any] = {}
 
         if current_user:
-            # Add "user is owner" condition - user's principal matches owner_principal
-            escaped_principal = current_user.principal.replace("'", "''")
-            base_conditions.append(f"owner_principal = '{escaped_principal}'")
+            params["owner_principal_match"] = current_user.principal
+            base_conditions.append("owner_principal = :owner_principal_match")
 
-            # Add "user in owners" conditions for attribute matching
             if current_user.attributes:
                 for attr_key, user_values in current_user.attributes.items():
                     if user_values:
                         value_conditions = []
-                        for value in user_values:
-                            # Check if JSON array contains the value
-                            escaped_value = value.replace("'", "''")
+                        safe_key = re.sub(r"[^a-zA-Z0-9_]", "_", attr_key)
+                        for j, value in enumerate(user_values):
+                            param_name = f"attr_{safe_key}_{j}"
                             json_text = self._json_extract_text("access_attributes", attr_key)
-                            value_conditions.append(f"({json_text} LIKE '%\"{escaped_value}\"%')")
+                            value_conditions.append(f"({json_text} LIKE :{param_name})")
+                            params[param_name] = f'%"{value}"%'
 
                         if value_conditions:
-                            # User matches this category if any of their values match
-                            user_matches_category = f"({' OR '.join(value_conditions)})"
-                            base_conditions.append(user_matches_category)
-        return f"({' OR '.join(base_conditions)})"
+                            base_conditions.append(f"({' OR '.join(value_conditions)})")
 
-    def _build_conservative_where_clause(self) -> str:
+        return f"({' OR '.join(base_conditions)})", params
+
+    def _build_conservative_where_clause(self) -> tuple[str, dict[str, Any]]:
         """Conservative SQL filtering for custom policies.
 
+        Returns a tuple of (sql_clause, bind_params) using parameterized queries.
         Only filters records we're 100% certain would be denied by any reasonable policy.
         """
         current_user = get_authenticated_user()
 
         if not current_user:
-            # Only allow public records
             base_conditions = self._get_public_access_conditions()
-            return f"({' OR '.join(base_conditions)})"
+            return f"({' OR '.join(base_conditions)})", {}
 
-        return "1=1"
+        return "1=1", {}

src/llama_stack/core/storage/sqlstore/sqlalchemy_sqlstore.py

@@ -187,6 +187,7 @@ async def fetch_all(
         table: str,
         where: Mapping[str, Any] | None = None,
         where_sql: str | None = None,
+        where_sql_params: Mapping[str, Any] | None = None,
         limit: int | None = None,
         order_by: list[tuple[str, Literal["asc", "desc"]]] | None = None,
         cursor: tuple[str, str] | None = None,
@@ -200,7 +201,10 @@ async def fetch_all(
                     query = query.where(_build_where_expr(table_obj.c[key], value))
 
             if where_sql:
-                query = query.where(text(where_sql))
+                clause = text(where_sql)
+                if where_sql_params:
+                    clause = clause.bindparams(**where_sql_params)
+                query = query.where(clause)
 
             # Handle cursor-based pagination
             if cursor:
@@ -287,9 +291,10 @@ async def fetch_one(
         table: str,
         where: Mapping[str, Any] | None = None,
         where_sql: str | None = None,
+        where_sql_params: Mapping[str, Any] | None = None,
         order_by: list[tuple[str, Literal["asc", "desc"]]] | None = None,
     ) -> dict[str, Any] | None:
-        result = await self.fetch_all(table, where, where_sql, limit=1, order_by=order_by)
+        result = await self.fetch_all(table, where, where_sql, where_sql_params, limit=1, order_by=order_by)
         if not result.data:
             return None
         return result.data[0]

src/llama_stack/providers/remote/vector_io/oci/oci26ai.py

@@ -425,13 +425,15 @@ async def delete(self):
 
     async def delete_chunks(self, chunks_for_deletion: list[ChunkForDeletion]) -> None:
         chunk_ids = [c.chunk_id for c in chunks_for_deletion]
+        if not chunk_ids:
+            return
+        placeholders = [f":id_{i}" for i in range(len(chunk_ids))]
+        params = {f"id_{i}": cid for i, cid in enumerate(chunk_ids)}
         cursor = self.connection.cursor()
         try:
             cursor.execute(
-                f"""
-                DELETE FROM {self.table_name}
-                WHERE chunk_id IN ({", ".join([f"'{chunk_id}'" for chunk_id in chunk_ids])})
-                """
+                f"DELETE FROM {self.table_name} WHERE chunk_id IN ({', '.join(placeholders)})",
+                params,
             )
         except Exception as e:
             logger.error(f"Error deleting chunks from Oracle 26AI table {self.table_name}: {e}")

src/llama_stack_api/internal/sqlstore.py

@@ -50,6 +50,7 @@ async def fetch_all(
         table: str,
         where: Mapping[str, Any] | None = None,
         where_sql: str | None = None,
+        where_sql_params: Mapping[str, Any] | None = None,
         limit: int | None = None,
         order_by: list[tuple[str, Literal["asc", "desc"]]] | None = None,
         cursor: tuple[str, str] | None = None,
@@ -60,6 +61,7 @@ async def fetch_one(
         table: str,
         where: Mapping[str, Any] | None = None,
         where_sql: str | None = None,
+        where_sql_params: Mapping[str, Any] | None = None,
         order_by: list[tuple[str, Literal["asc", "desc"]]] | None = None,
     ) -> dict[str, Any] | None: ...