feat: OPTIC-1938: Proxy for storages when presigned urls are off

label_studio/core/all_urls.json

@@ -275,18 +275,6 @@
         "name": "data_import:storage-data-upload",
         "decorators": ""
     },
-    {
-        "url": "/tasks/<int:task_id>/presign/",
-        "module": "data_import.api.TaskPresignStorageData",
-        "name": "data_import:task-storage-data-presign",
-        "decorators": ""
-    },
-    {
-        "url": "/projects/<int:project_id>/presign/",
-        "module": "data_import.api.ProjectPresignStorageData",
-        "name": "data_import:project-storage-data-presign",
-        "decorators": ""
-    },
     {
         "url": "/api/dm/views/",
         "module": "data_manager.api.ViewAPI",
@@ -947,6 +935,30 @@
         "name": "storages:api:export-storage-localfiles-form",
         "decorators": ""
     },
+    {
+        "url": "/tasks/<int:task_id>/resolve/",
+        "module": "io_storages.proxy_api.TaskResolveStorageUri",
+        "name": "storages:task-storage-data-resolve",
+        "decorators": ""
+    },
+    {
+        "url": "/projects/<int:project_id>/resolve/",
+        "module": "io_storages.proxy_api.ProjectResolveStorageUri",
+        "name": "storages:project-storage-data-resolve",
+        "decorators": ""
+    },
+    {
+        "url": "/tasks/<int:task_id>/presign/",
+        "module": "io_storages.proxy_api.TaskResolveStorageUri",
+        "name": "storages:task-storage-data-presign",
+        "decorators": ""
+    },
+    {
+        "url": "/projects/<int:project_id>/presign/",
+        "module": "io_storages.proxy_api.ProjectResolveStorageUri",
+        "name": "storages:project-storage-data-presign",
+        "decorators": ""
+    },
     {
         "url": "/api/ml/",
         "module": "ml.api.MLBackendListAPI",

label_studio/core/settings/base.py

@@ -820,3 +820,14 @@ def collect_versions_dummy(**kwargs):
     }
 
 LOGOUT_REDIRECT_URL = get_env('LOGOUT_REDIRECT_URL', None)
+
+RESOLVER_PROXY_BUFFER_SIZE = int(get_env('RESOLVER_PROXY_BUFFER_SIZE', 512 * 1024))
+RESOLVER_PROXY_TIMEOUT = int(get_env('RESOLVER_PROXY_TIMEOUT', 20))
+RESOLVER_PROXY_MAX_RANGE_SIZE = int(get_env('RESOLVER_PROXY_MAX_RANGE_SIZE', 8 * 1024 * 1024))
+RESOLVER_PROXY_GCS_DOWNLOAD_URL = get_env(
+    'RESOLVER_PROXY_GCS_DOWNLOAD_URL',
+    'https://storage.googleapis.com/download/storage/v1/b/{bucket_name}/o/{blob_name}?alt=media',
+)
+RESOLVER_PROXY_GCS_HTTP_TIMEOUT = int(get_env('RESOLVER_PROXY_GCS_HTTP_TIMEOUT', 5))
+RESOLVER_PROXY_ENABLE_ETAG_CACHE = get_bool_env('RESOLVER_PROXY_ENABLE_ETAG_CACHE', True)
+RESOLVER_PROXY_CACHE_TIMEOUT = int(get_env('RESOLVER_PROXY_CACHE_TIMEOUT', 3600))

label_studio/core/static/js/sw.js

@@ -43,19 +43,20 @@ async function handlePresignedUrl(event) {
   // even when it is not expired.
   // This is so the server can just naively proxy the presign request
   // and the performance is not degraded.
-  const requestPathToCache = "/presign/";
+  const requestsPathToCache = /\/presign\/|\/resolve\//;
 
   // Check if the request URL doesn't match the specified path part
   // or if it is not a request to the same origin we will just fetch it
   if (
     !event.request.url.startsWith(self.location.origin) ||
-    !event.request.url.includes(requestPathToCache) ||
+    !requestsPathToCache.test(event.request.url) ||
     // This is to avoid an error trying to load a direct presign URL in a new tab
     !event.request.referrer ||
     // Easier to leave this uncached as if we were to handle this caching
     // it would be more complex and not really worth the effort as it is not the most repeated
     // request and it is not a big deal if it is not cached.
     // If this was to be cached it forces a full resource request instead of a chunked ranged one.
+    // This also avoids caching the data directly when resolving uri -> data, which is not the goal of this service worker
     event.request.headers.get("range")
   ) {
     // For other requests, just allow the network to handle it

label_studio/data_import/api.py

@@ -1,11 +1,9 @@
 """This file and its contents are licensed under the Apache License 2.0. Please see the included NOTICE for copyright information and LICENSE for a copy of the license.
 """
-import base64
 import json
 import logging
 import mimetypes
 import time
-from typing import Union
 from urllib.parse import unquote, urlparse
 
 import drf_yasg.openapi as openapi
@@ -18,7 +16,7 @@
 from csp.decorators import csp
 from django.conf import settings
 from django.db import transaction
-from django.http import HttpRequest, HttpResponse, HttpResponseRedirect
+from django.http import HttpResponse
 from django.utils.decorators import method_decorator
 from drf_yasg.utils import swagger_auto_schema
 from projects.models import Project, ProjectImport, ProjectReimport
@@ -748,89 +746,3 @@ def get(self, request, *args, **kwargs):
         response['X-Accel-Redirect'] = redirect
         response['Content-Disposition'] = 'attachment; filename="{}"'.format(filepath)
         return response
-
-
-class PresignAPIMixin:
-    def handle_presign(self, request: HttpRequest, fileuri: str, instance: Union[Task, Project]) -> Response:
-        model_name = type(instance).__name__
-
-        if not instance.has_permission(request.user):
-            return Response(status=status.HTTP_403_FORBIDDEN)
-
-        # Attempt to base64 decode the fileuri
-        try:
-            fileuri = base64.urlsafe_b64decode(fileuri.encode()).decode()
-        # For backwards compatibility, try unquote if this fails
-        except Exception as exc:
-            logger.debug(
-                f'Failed to decode base64 {fileuri} for {model_name} {instance.id}: {exc} falling back to unquote'
-            )
-            fileuri = unquote(fileuri)
-
-        try:
-            resolved = instance.resolve_storage_uri(fileuri)
-        except Exception as exc:
-            logger.error(f'Failed to resolve storage uri {fileuri} for {model_name} {instance.id}: {exc}')
-            return Response(status=status.HTTP_404_NOT_FOUND)
-
-        if resolved is None or resolved.get('url') is None:
-            return Response(status=status.HTTP_404_NOT_FOUND)
-
-        url = resolved['url']
-        max_age = 0
-        if resolved.get('presign_ttl'):
-            max_age = resolved.get('presign_ttl') * 60
-
-        # Proxy to presigned url
-        response = HttpResponseRedirect(redirect_to=url, status=status.HTTP_303_SEE_OTHER)
-        response.headers['Cache-Control'] = f'no-store, max-age={max_age}'
-
-        return response
-
-
-class TaskPresignStorageData(PresignAPIMixin, APIView):
-    """A file proxy to presign storage urls at the task level."""
-
-    swagger_schema = None
-    http_method_names = ['get']
-    permission_classes = (IsAuthenticated,)
-
-    def get(self, request, *args, **kwargs):
-        """Get the presigned url for a given fileuri"""
-        request = self.request
-        task_id = kwargs.get('task_id')
-        fileuri = request.GET.get('fileuri')
-
-        if fileuri is None or task_id is None:
-            return Response(status=status.HTTP_400_BAD_REQUEST)
-
-        try:
-            task = Task.objects.get(pk=task_id)
-        except Task.DoesNotExist:
-            return Response(status=status.HTTP_404_NOT_FOUND)
-
-        return self.handle_presign(request, fileuri, task)
-
-
-class ProjectPresignStorageData(PresignAPIMixin, APIView):
-    """A file proxy to presign storage urls at the project level."""
-
-    swagger_schema = None
-    http_method_names = ['get']
-    permission_classes = (IsAuthenticated,)
-
-    def get(self, request, *args, **kwargs):
-        """Get the presigned url for a given fileuri"""
-        request = self.request
-        project_id = kwargs.get('project_id')
-        fileuri = request.GET.get('fileuri')
-
-        if fileuri is None or project_id is None:
-            return Response(status=status.HTTP_400_BAD_REQUEST)
-
-        try:
-            project = Project.objects.get(pk=project_id)
-        except Project.DoesNotExist:
-            return Response(status=status.HTTP_404_NOT_FOUND)
-
-        return self.handle_presign(request, fileuri, project)

label_studio/data_import/urls.py

@@ -23,10 +23,4 @@
     # special endpoints for serving imported files
     path('data/upload/<path:filename>', api.UploadedFileResponse.as_view(), name='data-upload'),
     path('storage-data/uploaded/', api.DownloadStorageData.as_view(), name='storage-data-upload'),
-    path('tasks/<int:task_id>/presign/', api.TaskPresignStorageData.as_view(), name='task-storage-data-presign'),
-    path(
-        'projects/<int:project_id>/presign/',
-        api.ProjectPresignStorageData.as_view(),
-        name='project-storage-data-presign',
-    ),
 ]

label_studio/io_storages/README.md

@@ -115,4 +115,73 @@ All these states are present in both the open-source and enterprise editions for
 4. RQ workers were killed manually => `storage_background_failure` wasn't called.
 5. Job was removed from RQ Queue => it's not a failure, but we need to update storage status somehow. 
 
-To handle these cases correctly, all these conditions must be checked in ensure_storage_status when the Storage List API is retrieved.
+To handle these cases correctly, all these conditions must be checked in ensure_storage_status when the Storage List API is retrieved.
+
+## Storage Proxy API
+
+The Storage Proxy API is a critical component that handles access to files stored in cloud storages (S3, GCS, Azure, etc.). It serves two main purposes:
+
+1. **Security & Access Control**: It acts as a secure gateway to cloud storage resources, enforcing Label Studio's permission model and preventing direct exposure of cloud credentials to the client.
+
+2. **Flexible Content Delivery**: It supports two modes of operation based on the storage configuration:
+   - **Redirect Mode** (`presign=True`): Generates pre-signed URLs with temporary access and redirects the client to them. This is efficient as content flows directly from the storage to the client.
+   - **Proxy Mode** (`presign=False`): Streams content through the Label Studio server. This provides additional security and is useful when storage providers don't support pre-signed URLs or when administrators want to enforce stricter access control.
+
+### How It Works
+
+1. When tasks contain references to cloud storage URIs (e.g., `s3://bucket/file.jpg`), these are converted to proxy URLs (`/tasks/{task_id}/resolve/?fileuri=base64encodeduri`).
+
+2. When a client requests this URL, the Proxy API:
+   - Decodes the URI and locates the appropriate storage connection
+   - Validates user permissions for the task/project
+   - Either redirects to a pre-signed URL or streams the content directly, based on the storage's `presign` setting
+
+3. The API handles both task-level and project-level resources through dedicated endpoints:
+   - `/tasks/<task_id>/resolve/` - for resolving files referenced in tasks
+   - `/projects/<project_id>/resolve/` - for resolving project-level resources
+
+This architecture ensures secure, controlled access to cloud storage resources while maintaining flexibility for different deployment scenarios and security requirements.
+
+### Proxy Mode Optimizations*
+
+The Proxy Mode has been optimized with several mechanisms to improve performance, reliability, and resource utilization:
+
+* *Range Header Processing*
+
+The `override_range_header` function processes and intelligently modifies Range headers to limit stream sizes:
+
+- It enforces a maximum size for range requests (controlled by `RESOLVER_PROXY_MAX_RANGE_SIZE`)
+- Converts unbounded range requests (`bytes=123456-`) to bounded ones
+- Handles various range request formats including header probes (`bytes=0-`)
+- Prevents worker exhaustion by chunking large file transfers
+
+* *Time-Limited Streaming*
+
+The `time_limited_chunker` generator provides controlled streaming with timeout protection:
+
+- Stops yielding chunks after a configurable timeout period (`RESOLVER_PROXY_TIMEOUT`)
+- Uses buffer-sized chunks (`RESOLVER_PROXY_BUFFER_SIZE`) for efficient memory usage
+- Tracks statistics about stream performance and reports on timeouts and print it as debug info
+- Properly closes streams to prevent resource leaks
+
+* *Response Header Management*
+
+The `prepare_headers` function manages HTTP response headers for optimal client handling:
+
+- Forwards important headers from storage providers (Content-Length, Content-Range, Last-Modified)
+- Enables range requests with Accept-Ranges header
+- Implements cache control with configurable TTL (`RESOLVER_PROXY_CACHE_TIMEOUT`)
+- Generates ETags based on user permissions to invalidate cache when access changes
+
+### *Environment Variables*
+
+The Storage Proxy API behavior can be configured using the following environment variables:
+
+| Variable | Description | Default |
+|----------|-------------|---------|
+| `RESOLVER_PROXY_BUFFER_SIZE` | Size in bytes of each chunk when streaming data | 64*1024 |
+| `RESOLVER_PROXY_TIMEOUT` | Maximum time in seconds a streaming connection can remain open | 10 |
+| `RESOLVER_PROXY_MAX_RANGE_SIZE` | Maximum size in bytes for a single range request | 7*1024*1024 |
+| `RESOLVER_PROXY_CACHE_TIMEOUT` | Cache TTL in seconds for proxy responses | 3600 |
+
+These optimizations ensure that the Proxy API remains responsive and resource-efficient, even when handling large files or many concurrent requests.

label_studio/io_storages/all_api.py

@@ -127,8 +127,8 @@
             response = view(request._request, *args, **kwargs)
             payload = response.data
             if not isinstance(payload, list):
-                raise ValueError('Response is not list')
-            return response.data
+                raise ValueError(f'Response is not list: {payload}')
+            return payload
         except Exception:
             logger.error(f"Can't process {api.__class__.__name__}", exc_info=True)
             return []
@@ -158,9 +158,16 @@
     permission_required = all_permissions.projects_change
 
     def _get_response(self, api, request, *args, **kwargs):
-        view = api.as_view()
-        response = view(request._request, *args, **kwargs)
-        return response.data
+        try:
+            view = api.as_view()
+            response = view(request._request, *args, **kwargs)
+            payload = response.data
+            if not isinstance(payload, list):
+                raise ValueError(f'Response is not list: {payload}')
+            return payload
+        except Exception:
+            logger.error(f"Can't process {api.__class__.__name__}", exc_info=True)
+            return []
 
     def list(self, request, *args, **kwargs):
         list_responses = sum(