fix: bind prompt history/version ops to the authorized prompt (#25056)

The history diff, delete, and version-restore routes authorize the URL
prompt_id but then act on a caller-supplied history/version id without
checking it belongs to that prompt (IDOR). Filter by prompt_id in
compute_diff and delete_history_entry, and reject a cross-prompt version_id
in update_prompt_version.

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Classic298andClaude Opus 4.7 authored and GitHub committed 2026-06-01 12:07:52 -07:00
1 parent e623081b2b
commit 9a3eea6448
3 files changed
+15 -6

No files matched your search

+11 -3
View File
@@ -151,13 +151,19 @@ class PromptHistoryTable:
self, self,
from_id: str, from_id: str,
to_id: str, to_id: str,
prompt_id: str,
db: Optional[AsyncSession] = None, db: Optional[AsyncSession] = None,
) -> Optional[dict]: ) -> Optional[dict]:
"""Compute diff between two history entries.""" """Compute diff between two history entries."""
async with get_async_db_context(db) as db: async with get_async_db_context(db) as db:
result_from = await db.execute(select(PromptHistory).filter(PromptHistory.id == from_id)) # Bind both entries to the authorized prompt; an unbound id reads another prompt's snapshot.
result_from = await db.execute(
select(PromptHistory).filter(PromptHistory.id == from_id, PromptHistory.prompt_id == prompt_id)
)
from_entry = result_from.scalars().first() from_entry = result_from.scalars().first()
result_to = await db.execute(select(PromptHistory).filter(PromptHistory.id == to_id)) result_to = await db.execute(
select(PromptHistory).filter(PromptHistory.id == to_id, PromptHistory.prompt_id == prompt_id)
)
to_entry = result_to.scalars().first() to_entry = result_to.scalars().first()
if not from_entry or not to_entry: if not from_entry or not to_entry:
@@ -203,11 +209,13 @@ class PromptHistoryTable:
async def delete_history_entry( async def delete_history_entry(
self, self,
history_id: str, history_id: str,
prompt_id: str,
db: Optional[AsyncSession] = None, db: Optional[AsyncSession] = None,
) -> bool: ) -> bool:
"""Delete a history entry and reparent its children to grandparent.""" """Delete a history entry and reparent its children to grandparent."""
async with get_async_db_context(db) as db: async with get_async_db_context(db) as db:
result = await db.execute(select(PromptHistory).filter_by(id=history_id)) # Bind to the authorized prompt; an unbound id deletes another prompt's history.
result = await db.execute(select(PromptHistory).filter_by(id=history_id, prompt_id=prompt_id))
entry = result.scalars().first() entry = result.scalars().first()
if not entry: if not entry:
return False return False
+2 -1
View File
@@ -580,7 +580,8 @@ class PromptsTable:
history_entry = await PromptHistories.get_history_entry_by_id(version_id, db=session) history_entry = await PromptHistories.get_history_entry_by_id(version_id, db=session)
if not history_entry: # Reject a version_id from another prompt; restoring it would copy a foreign snapshot in.
if not history_entry or history_entry.prompt_id != prompt_id:
return None return None
# Restore prompt content from the snapshot # Restore prompt content from the snapshot
+2 -2
View File
@@ -653,7 +653,7 @@ async def delete_prompt_history_entry(
detail='Cannot delete the active production version', detail='Cannot delete the active production version',
) )
success = await PromptHistories.delete_history_entry(history_id, db=db) success = await PromptHistories.delete_history_entry(history_id, prompt.id, db=db)
if not success: if not success:
raise HTTPException( raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, status_code=status.HTTP_404_NOT_FOUND,
@@ -697,7 +697,7 @@ async def get_prompt_diff(
detail=ERROR_MESSAGES.ACCESS_PROHIBITED, detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
) )
diff = await PromptHistories.compute_diff(from_id, to_id, db=db) diff = await PromptHistories.compute_diff(from_id, to_id, prompt.id, db=db)
if not diff: if not diff:
raise HTTPException( raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, status_code=status.HTTP_404_NOT_FOUND,