UN-3487 [FIX] Restrict the S3/MinIO connector to one bucket

backend/connector_v2/tests/test_connector_register.py

@@ -24,6 +24,7 @@
 from connector_v2.models import ConnectorInstance
 
 MINIO_CONNECTOR_ID = "minio|c799f6e3-2b57-434e-aaac-b5daa415da19"
+_BUCKET = "connector-register-test"
 
 pytestmark = pytest.mark.skipif(
     not all(
@@ -45,12 +46,28 @@ def _credentials(secret: str | None = None) -> dict:
         "secret": secret or os.environ["MINIO_SECRET_ACCESS_KEY"],
         "endpoint_url": os.environ["MINIO_ENDPOINT_URL"],
         "region_name": "",
+        "bucket": _BUCKET,
         "path": "/",
     }
 
 
+def _ensure_bucket_exists() -> None:
+    # UN-3487: MinioFS now requires a bucket, and `test_credentials` probes
+    # it directly — it must actually exist in the rig's MinIO.
+    from s3fs.core import S3FileSystem
+
+    fs = S3FileSystem(
+        key=os.environ["MINIO_ACCESS_KEY_ID"],
+        secret=os.environ["MINIO_SECRET_ACCESS_KEY"],
+        client_kwargs={"endpoint_url": os.environ["MINIO_ENDPOINT_URL"]},
+    )
+    if not fs.exists(_BUCKET):
+        fs.mkdir(_BUCKET)
+
+
 class ConnectorRegisterTest(TestCase):
     def setUp(self) -> None:
+        _ensure_bucket_exists()
         self.org = Organization.objects.create(
             name="org-conn", display_name="Org Conn", organization_id="org-conn"
         )

backend/file_management/file_management_helper.py

@@ -11,6 +11,7 @@
 from django.conf import settings
 from django.http import StreamingHttpResponse
 from fsspec import AbstractFileSystem
+from fsspec.implementations.dirfs import DirFileSystem
 from pydrive2.files import ApiRequestError
 
 from file_management.exceptions import (
@@ -31,6 +32,27 @@
 logger = logging.getLogger(__name__)
 
 
+def _default_root_path(
+    file_system: UnstractFileSystem, fs: AbstractFileSystem, path: str
+) -> str | None:
+    """Some connectors restrict browsing to their own default root (e.g. a
+    configured `path`) when the caller didn't ask for a specific one.
+
+    `DirFileSystem.path` (used by a bucket-scoped MinioFS) is excluded: it's
+    the wrapped bucket prefix, not a default root, and every operation
+    already resolves relative to it — applying it again here would
+    double-prefix the path (UN-3487).
+    """
+    if not path or path == "/":
+        try:
+            if file_system.path:
+                return file_system.path
+        except AttributeError:
+            if hasattr(fs, "path") and fs.path and not isinstance(fs, DirFileSystem):
+                return fs.path
+    return None
+
+
 class FileManagerHelper:
     @staticmethod
     def get_file_system(connector: ConnectorInstance) -> UnstractFileSystem:
@@ -47,14 +69,8 @@ def get_file_system(connector: ConnectorInstance) -> UnstractFileSystem:
     @staticmethod
     def list_files(file_system: UnstractFileSystem, path: str) -> list[FileInformation]:
         fs = file_system.get_fsspec_fs()
-        file_path = f"{path}"
-        # TODO: Add below logic by checking each connector?
         try:
-            if file_system.path and (not path or path == "/"):
-                file_path = file_system.path
-        except AttributeError:
-            if hasattr(fs, "path") and fs.path and (not path or path == "/"):
-                file_path = fs.path
+            file_path = _default_root_path(file_system, fs, path) or path
         except Exception:
             logger.error(f"Missing path Atribute in {fs}")
             raise MissingConnectorParams()
@@ -135,13 +151,8 @@ def upload_file(
     ) -> None:
         fs = file_system.get_fsspec_fs()
 
-        file_path = f"{path}"
-        try:
-            if file_system.path and (not path or path == "/"):
-                file_path = f"{file_system.path}/"
-        except AttributeError:
-            if fs.path and (not path or path == "/"):
-                file_path = f"{fs.path}/"
+        root = _default_root_path(file_system, fs, path)
+        file_path = f"{root}/" if root else path
 
         file_path = file_path + "/" if not file_path.endswith("/") else file_path
 

backend/file_management/tests.py

@@ -1 +1,101 @@
-# Create your tests here.
+"""UN-3487: `file/`, `file/download` and `file/upload` are scoped to the
+caller, not just the org.
+
+Before this fix, `FileManagementViewSet` resolved a connector by a raw
+`ConnectorInstance.objects.get(pk=id)` — any authenticated org member who
+knew (or guessed) another user's connector id could browse, download from,
+or upload to it, sharing settings aside. These drive the real views through
+DRF's request factory, so a gate that exists only in the queryset — and
+never reaches the route — is still caught.
+"""
+
+import unittest
+from unittest.mock import mock_open, patch
+
+from connector_v2.models import ConnectorInstance
+from django.test import TestCase
+from permissions.roles import ResourceRole
+from permissions.tests.base import CoOwnerOrgTestMixin
+from rest_framework import status
+from rest_framework.response import Response
+from rest_framework.test import APIRequestFactory, force_authenticate
+
+from file_management.file_management_helper import FileManagerHelper
+from file_management.views import FileManagementViewSet
+from unstract.connectors.filesystems.minio.minio import MinioFS
+
+
+class FileManagementAccessScopeTest(CoOwnerOrgTestMixin, TestCase):
+    def setUp(self) -> None:
+        self._seed_org()
+        self.connector = ConnectorInstance.objects.create(
+            connector_name="team-a-s3",
+            connector_id="minio|c799f6e3-2b57-434e-aaac-b5daa415da19",
+            connector_metadata={"bucket": "team-a-data"},
+            organization=self.org,
+            created_by=self.owner,
+        )
+        # `created_by` is audit-only — access runs through the membership
+        # table, same as every other shareable resource in this codebase.
+        self.connector.memberships.create(user=self.owner, role=ResourceRole.OWNER)
+        self.connector.memberships.create(user=self.viewer, role=ResourceRole.VIEWER)
+        self.factory = APIRequestFactory()
+
+    def _list(self, actor) -> Response:
+        view = FileManagementViewSet.as_view({"get": "list"})
+        request = self.factory.get(
+            "/file", {"connector_id": str(self.connector.pk), "path": "/"}
+        )
+        force_authenticate(request, user=actor)
+        with (
+            patch(
+                "file_management.views.FileManagerHelper.get_file_system",
+                return_value=None,
+            ),
+            patch("file_management.views.FileManagerHelper.list_files", return_value=[]),
+        ):
+            return view(request)
+
+    def test_owner_can_list_their_own_connector(self) -> None:
+        self.assertEqual(self._list(self.owner).status_code, status.HTTP_200_OK)
+
+    def test_org_admin_can_list_any_connector(self) -> None:
+        self.assertEqual(self._list(self.admin).status_code, status.HTTP_200_OK)
+
+    def test_shared_viewer_can_list_the_connector(self) -> None:
+        self.assertEqual(self._list(self.viewer).status_code, status.HTTP_200_OK)
+
+    def test_outsider_org_member_cannot_list_an_unshared_connector(self) -> None:
+        # In scope (org member), but the connector was never shared with them.
+        self.assertEqual(self._list(self.outsider).status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_non_org_member_cannot_list_the_connector(self) -> None:
+        self.assertEqual(self._list(self.stranger).status_code, status.HTTP_404_NOT_FOUND)
+
+
+class BucketScopedRootPathTest(unittest.TestCase):
+    """Regression for a Greptile finding on this PR: a bucket-scoped
+    connector's own `DirFileSystem.path` is its wrapped bucket prefix, not a
+    connector-level default root. Treating it as one double-prefixes every
+    root-level operation (`<bucket>/<bucket>/...`) instead of resolving
+    within the bucket.
+    """
+
+    def _minio_fs(self) -> MinioFS:
+        return MinioFS({"bucket": "team-a-data", "key": "k", "secret": "s"})
+
+    def test_list_files_at_root_is_not_double_prefixed(self) -> None:
+        minio_fs = self._minio_fs()
+        with patch.object(minio_fs.s3, "ls", return_value=[]) as mock_ls:
+            FileManagerHelper.list_files(minio_fs, "/")
+        mock_ls.assert_called_once_with("team-a-data/", detail=True)
+
+    def test_upload_file_at_root_is_not_double_prefixed(self) -> None:
+        minio_fs = self._minio_fs()
+        with patch.object(minio_fs.s3, "open", mock_open()) as mock_open_call:
+            FileManagerHelper.upload_file(minio_fs, "/", b"data", "report.pdf")
+        (called_path,), kwargs = mock_open_call.call_args
+        self.assertEqual(kwargs, {"mode": "wb"})
+        self.assertNotIn("team-a-data/team-a-data", called_path)
+        self.assertTrue(called_path.startswith("team-a-data/"))
+        self.assertTrue(called_path.endswith("report.pdf"))

backend/file_management/views.py

@@ -2,6 +2,7 @@
 from typing import Any
 
 from connector_v2.models import ConnectorInstance
+from django.db.models import QuerySet
 from django.http import HttpRequest
 from oauth2client.client import HttpAccessTokenRefreshError
 from rest_framework import serializers, viewsets
@@ -30,8 +31,22 @@ class FileManagementViewSet(viewsets.ModelViewSet):
 
     versioning_class = URLPathVersioning
 
-    def get_queryset(self):
-        return ConnectorInstance.objects.all()
+    def get_queryset(self) -> QuerySet[ConnectorInstance]:
+        # Org-scoped alone isn't enough: this must also respect ownership /
+        # sharing, or any org member could browse another user's connector
+        # by guessing its id.
+        return ConnectorInstance.objects.for_user(self.request.user)
+
+    def _get_connector_or_404(self, id: str) -> ConnectorInstance:
+        """Resolve a connector within the caller's own access scope.
+
+        Raises the same not-found error whether the id is unknown or simply
+        outside `get_queryset()` — the caller can't tell those apart.
+        """
+        try:
+            return self.get_queryset().get(pk=id)
+        except ConnectorInstance.DoesNotExist:
+            raise ConnectorInstanceNotFound()
 
     def get_serializer_class(self) -> serializers.Serializer:
         if self.action == "upload":
@@ -50,13 +65,11 @@ def list(self, request: HttpRequest) -> Response:
         id: str = serializer.validated_data.get("connector_id")
         path: str = serializer.validated_data.get("path")
         try:
-            connector_instance: ConnectorInstance = ConnectorInstance.objects.get(pk=id)
+            connector_instance = self._get_connector_or_404(id)
             file_system = FileManagerHelper.get_file_system(connector_instance)
             files = FileManagerHelper.list_files(file_system, path)
             serializer = FileInfoSerializer(files, many=True)
             return Response(serializer.data)
-        except ConnectorInstance.DoesNotExist:
-            raise ConnectorInstanceNotFound()
         except HttpAccessTokenRefreshError as error:
             logger.error(
                 f"HttpAccessTokenRefreshError thrown from file list, error {error}"
@@ -72,7 +85,7 @@ def download(self, request: HttpRequest) -> Response:
         serializer.is_valid(raise_exception=True)
         id: str = serializer.validated_data.get("connector_id")
         path: str = serializer.validated_data.get("path")
-        connector_instance: ConnectorInstance = ConnectorInstance.objects.get(pk=id)
+        connector_instance = self._get_connector_or_404(id)
         file_system = FileManagerHelper.get_file_system(connector_instance)
         return FileManagerHelper.download_file(file_system, path)
 
@@ -84,7 +97,7 @@ def upload(self, request: HttpRequest) -> Response:
 
         path: str = serializer.validated_data.get("path")
         uploaded_files: Any = serializer.validated_data.get("file")
-        connector_instance: ConnectorInstance = ConnectorInstance.objects.get(pk=id)
+        connector_instance = self._get_connector_or_404(id)
         file_system = FileManagerHelper.get_file_system(connector_instance)
 
         for uploaded_file in uploaded_files:

frontend/src/components/helpers/custom-markdown/CustomMarkdown.jsx

@@ -37,7 +37,15 @@ const CustomMarkdown = ({
       case "tripleCode":
         return (
           <Paragraph style={{ margin: 0 }}>
-            <pre style={{ margin: 0 }}>{content}</pre>
+            <pre
+              style={{
+                margin: 0,
+                whiteSpace: "pre-wrap",
+                overflowWrap: "anywhere",
+              }}
+            >
+              {content}
+            </pre>
           </Paragraph>
         );
       case "inlineCode":

tests/e2e/etl/conftest.py

@@ -133,6 +133,7 @@ def create_connector(connector_type: str) -> str:
                     "secret": minio_store.secret_key,
                     "endpoint_url": minio_store.internal_url,
                     "region_name": "us-east-1",
+                    "bucket": minio_store.bucket,
                 },
             },
         )
@@ -164,7 +165,9 @@ def create_connector(connector_type: str) -> str:
             "connection_type": "FILESYSTEM",
             "connector_instance_id": source_id,
             "configuration": {
-                "folders": [f"/{minio_store.bucket}/{input_prefix}"],
+                # UN-3487: the connector is now scoped to its own bucket, so
+                # paths resolve relative to it — no bucket prefix here.
+                "folders": [f"/{input_prefix}"],
                 "processSubDirectories": False,
                 "maxFiles": 1,
                 "fileProcessingOrder": "unordered",
@@ -179,7 +182,8 @@ def create_connector(connector_type: str) -> str:
         json={
             "connection_type": "FILESYSTEM",
             "connector_instance_id": destination_id,
-            "configuration": {"outputFolder": f"{minio_store.bucket}/{output_prefix}"},
+            # UN-3487: same as the source — no bucket prefix, it's implicit.
+            "configuration": {"outputFolder": output_prefix},
         },
     )
     assert resp.status_code == 200, f"destination endpoint: {resp.text}"