Restrict S3 Management to users with write permission

CHANGELOG.md

@@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.31.0] - 2026-05-26
+
+### Changed
+- **S3 Management is now restricted to users with write permission.** The S3 Management navigation entry is only shown to writers and admins (it was previously visible to everyone), and the underlying bucket/object endpoints (`/s3/buckets`, `/s3/objects`) now require the writer role via `get_user_for_write_operation`. Previously any authenticated user — including viewers and users with no role — could list, create and delete buckets and objects. This brings the administrative S3 storage tool in line with the viewer/writer/admin model already used across the Endpoint.
+
+### Security
+- Read-only users can no longer reach the S3 Management endpoints. Requests without write permission receive `403 Forbidden`.
+
+### Backwards compatibility
+- API request/response shapes and route paths are unchanged. The only change is the required permission level on the `/s3/buckets` and `/s3/objects` endpoints: callers that previously relied on viewer-level (or no-role) access to these endpoints will now receive `403` and must hold the writer role. Writers and admins are unaffected.
+
 ## [0.30.0] - 2026-05-20
 
 ### Changed

api/config/swagger_settings.py

@@ -12,7 +12,7 @@ class Settings(BaseSettings):
 
     swagger_title: str = "API Documentation"
     swagger_description: str = "This is the API documentation."
-    swagger_version: str = "0.30.0"
+    swagger_version: str = "0.31.0"
     root_path: str = ""  # API root path prefix (e.g., "/test" or "")
     is_public: bool = True
     metrics_endpoint: str = "https://federation.ndp.utah.edu/metrics/"

api/routes/minio_routes/bucket_routes.py

@@ -7,15 +7,15 @@
     ErrorResponse,
 )
 from api.services.minio_services import bucket_service
-from api.services.auth_services.get_current_user import get_current_user
+from api.services.auth_services import get_user_for_write_operation
 from api.config.minio_settings import s3_settings
 
 router = APIRouter(prefix="/s3/buckets", tags=["S3"])
 
 
 @router.post("/", response_model=dict, status_code=status.HTTP_201_CREATED)
 async def create_bucket(
-    request: BucketCreateRequest, current_user=Depends(get_current_user)
+    request: BucketCreateRequest, current_user=Depends(get_user_for_write_operation)
 ):
     """Create a new bucket."""
     if not s3_settings.is_configured:
@@ -45,7 +45,7 @@ async def create_bucket(
 
 
 @router.get("/", response_model=BucketListResponse)
-async def list_buckets(current_user=Depends(get_current_user)):
+async def list_buckets(current_user=Depends(get_user_for_write_operation)):
     """List all buckets."""
     if not s3_settings.is_configured:
         raise HTTPException(
@@ -65,7 +65,9 @@ async def list_buckets(current_user=Depends(get_current_user)):
 
 
 @router.get("/{bucket_name}", response_model=BucketInfo)
-async def get_bucket_info(bucket_name: str, current_user=Depends(get_current_user)):
+async def get_bucket_info(
+    bucket_name: str, current_user=Depends(get_user_for_write_operation)
+):
     """Get information about a specific bucket."""
     if not s3_settings.is_configured:
         raise HTTPException(
@@ -90,7 +92,9 @@ async def get_bucket_info(bucket_name: str, current_user=Depends(get_current_use
 
 
 @router.delete("/{bucket_name}", response_model=dict)
-async def delete_bucket(bucket_name: str, current_user=Depends(get_current_user)):
+async def delete_bucket(
+    bucket_name: str, current_user=Depends(get_user_for_write_operation)
+):
     """Delete a bucket (must be empty)."""
     if not s3_settings.is_configured:
         raise HTTPException(

api/routes/minio_routes/object_routes.py

@@ -21,7 +21,7 @@
 )
 from api.services.minio_services import object_service
 from api.services.minio_services.minio_client import minio_client
-from api.services.auth_services.get_current_user import get_current_user
+from api.services.auth_services import get_user_for_write_operation
 from api.config.minio_settings import s3_settings
 import io
 
@@ -33,7 +33,7 @@ async def upload_object(
     bucket_name: str,
     file: UploadFile = File(...),
     object_key: Optional[str] = Form(None),
-    current_user=Depends(get_current_user),
+    current_user=Depends(get_user_for_write_operation),
 ):
     """Upload an object to a bucket."""
     if not s3_settings.is_configured:
@@ -79,7 +79,7 @@ async def upload_object(
 async def list_objects(
     bucket_name: str,
     prefix: Optional[str] = Query(None, description="Object prefix filter"),
-    current_user=Depends(get_current_user),
+    current_user=Depends(get_user_for_write_operation),
 ):
     """List objects in a bucket."""
     if not s3_settings.is_configured:
@@ -108,7 +108,9 @@ async def list_objects(
     "/{bucket_name}/{object_key:path}/metadata", response_model=ObjectMetadataResponse
 )
 async def get_object_metadata(
-    bucket_name: str, object_key: str, current_user=Depends(get_current_user)
+    bucket_name: str,
+    object_key: str,
+    current_user=Depends(get_user_for_write_operation),
 ):
     """Get object metadata."""
     if not s3_settings.is_configured:
@@ -146,7 +148,7 @@ async def generate_presigned_upload_url(
     bucket_name: str,
     object_key: str,
     request: PresignedUrlRequest,
-    current_user=Depends(get_current_user),
+    current_user=Depends(get_user_for_write_operation),
 ):
     """Generate a presigned URL for uploading an object."""
     if not s3_settings.is_configured:
@@ -181,7 +183,7 @@ async def generate_presigned_download_url(
     bucket_name: str,
     object_key: str,
     request: PresignedUrlRequest,
-    current_user=Depends(get_current_user),
+    current_user=Depends(get_user_for_write_operation),
 ):
     """Generate a presigned URL for downloading an object."""
     if not s3_settings.is_configured:
@@ -215,7 +217,9 @@ async def generate_presigned_download_url(
 
 @router.get("/{bucket_name}/{object_key:path}")
 async def download_object(
-    bucket_name: str, object_key: str, current_user=Depends(get_current_user)
+    bucket_name: str,
+    object_key: str,
+    current_user=Depends(get_user_for_write_operation),
 ):
     """Download an object from a bucket."""
     if not s3_settings.is_configured:
@@ -269,7 +273,9 @@ async def download_object(
 
 @router.delete("/{bucket_name}/{object_key:path}", response_model=dict)
 async def delete_object(
-    bucket_name: str, object_key: str, current_user=Depends(get_current_user)
+    bucket_name: str,
+    object_key: str,
+    current_user=Depends(get_user_for_write_operation),
 ):
     """Delete an object from a bucket."""
     if not s3_settings.is_configured:

tests/test_minio_routes_auth.py

@@ -0,0 +1,85 @@
+"""
+Authorization tests for the S3 Management routes (``/s3/buckets`` and
+``/s3/objects``).
+
+These endpoints back the S3 Management tool, which is an administrative storage
+feature. They must be guarded by :func:`get_user_for_write_operation` so that
+viewers and users with no role cannot reach them — mirroring the UI, which only
+shows the S3 Management entry to writers.
+
+The minio routers are only mounted on the main app when S3 is enabled, so these
+tests mount them on a minimal standalone app and drive the guard through
+``app.dependency_overrides``. That isolates exactly what we want to verify (the
+routers require write permission) without needing a configured MinIO backend:
+when the guard denies, the request is rejected with 403 before the handler runs;
+when it allows, the request gets past auth and then fails for an unrelated
+reason (503, because S3 is not configured in the test environment).
+"""
+
+from fastapi import FastAPI, HTTPException
+from fastapi.testclient import TestClient
+
+from api.routes.minio_routes.bucket_routes import router as bucket_router
+from api.routes.minio_routes.object_routes import router as object_router
+from api.services.auth_services import get_user_for_write_operation
+
+app = FastAPI()
+app.include_router(bucket_router)
+app.include_router(object_router)
+client = TestClient(app)
+
+# (method, path, kwargs) for every S3 Management endpoint.
+S3_MANAGEMENT_ENDPOINTS = [
+    ("post", "/s3/buckets/", {"json": {"name": "demo"}}),
+    ("get", "/s3/buckets/", {}),
+    ("get", "/s3/buckets/demo", {}),
+    ("delete", "/s3/buckets/demo", {}),
+    ("post", "/s3/objects/demo", {"files": {"file": ("a.txt", b"x")}}),
+    ("get", "/s3/objects/demo", {}),
+    ("get", "/s3/objects/demo/key/metadata", {}),
+    ("post", "/s3/objects/demo/key/presigned-upload", {"json": {"expires_in": 3600}}),
+    ("post", "/s3/objects/demo/key/presigned-download", {"json": {"expires_in": 3600}}),
+    ("get", "/s3/objects/demo/key", {}),
+    ("delete", "/s3/objects/demo/key", {}),
+]
+
+
+def teardown_function():
+    app.dependency_overrides.clear()
+
+
+def test_s3_management_denies_users_without_write_permission():
+    """Every S3 Management endpoint returns 403 when the write guard denies."""
+
+    def _deny():
+        raise HTTPException(
+            status_code=403,
+            detail="You do not have permission to modify resources.",
+        )
+
+    app.dependency_overrides[get_user_for_write_operation] = _deny
+
+    for method, path, kwargs in S3_MANAGEMENT_ENDPOINTS:
+        response = getattr(client, method)(path, **kwargs)
+        assert response.status_code == 403, (
+            f"{method.upper()} {path} should be writer-only "
+            f"but returned {response.status_code}"
+        )
+
+
+def test_s3_management_allows_writers_past_the_guard():
+    """A writer passes the guard (the call then fails for an unrelated reason,
+    not 401/403)."""
+
+    app.dependency_overrides[get_user_for_write_operation] = lambda: {
+        "username": "writer-user",
+        "sub": "writer-sub",
+        "effective_role": "writer",
+    }
+
+    for method, path, kwargs in S3_MANAGEMENT_ENDPOINTS:
+        response = getattr(client, method)(path, **kwargs)
+        assert response.status_code not in (401, 403), (
+            f"{method.upper()} {path} should let a writer past auth "
+            f"but returned {response.status_code}"
+        )

ui/src/components/Navigation.js

@@ -177,7 +177,10 @@ const Navigation = () => {
                 <span>Search</span>
               </Link>
 
-              {/* S3 Management (bucket/object storage tool) */}
+              {/* S3 Management (bucket/object storage tool) — writers only.
+                  It is an administrative storage tool, so viewers and users
+                  with no role do not see it (mirrors the backend guard). */}
+              {canWrite && (
               <Link
                 to="/s3-management"
                 onMouseEnter={handleOtherNavEnter}
@@ -208,6 +211,7 @@ const Navigation = () => {
                 <HardDrive size={18} />
                 <span>S3 Management</span>
               </Link>
+              )}
 
               {/* + New menu — only visible to users that can write */}
               {canWrite && (