security: restrict MinIO app to one bucket

apps/api/app/readiness.py

@@ -6,6 +6,7 @@ from typing import Any
 from sqlalchemy import func, select, text
 from sqlalchemy.orm import Session
 
+from .config import settings
 from .models import CommunityImportRun, DataSource, ImportStatus, OfficialRecordImport
 
 
@@ -32,7 +33,7 @@ def readiness_report(
         ready = False
 
     try:
-        s3.list_buckets()
+        s3.head_bucket(Bucket=settings.s3_bucket)
         components["minio"] = {"status": "ready"}
     except Exception:
         components["minio"] = {"status": "unavailable"}

apps/api/app/storage.py

@@ -6,7 +6,6 @@ from functools import lru_cache
 
 import boto3
 from botocore.client import BaseClient
-from botocore.exceptions import ClientError
 from PIL import Image, UnidentifiedImageError
 
 from .config import settings
@@ -63,10 +62,7 @@ def upload_screenshot(raw: bytes, *, filename: str | None = None, content_type:
     body, extension, mime = prepare_image(raw)
     key = f"reports/{uuid.uuid4()}.{extension}"
     s3 = client()
-    try:
-        s3.head_bucket(Bucket=settings.s3_bucket)
-    except ClientError:
-        s3.create_bucket(Bucket=settings.s3_bucket)
+    s3.head_bucket(Bucket=settings.s3_bucket)
     s3.put_object(Bucket=settings.s3_bucket, Key=key, Body=body, ContentType=mime)
     return key
 

apps/api/tests/test_readiness.py

@@ -11,12 +11,13 @@ from app.readiness import readiness_report
 
 
 class AvailableStorage:
-    def list_buckets(self) -> dict[str, list[object]]:
-        return {"Buckets": []}
+    def head_bucket(self, *, Bucket: str) -> dict[str, object]:
+        assert Bucket
+        return {}
 
 
 class UnavailableStorage:
-    def list_buckets(self) -> None:
+    def head_bucket(self, *, Bucket: str) -> None:
         raise ConnectionError("fixture unavailable")