fix: address PR review — head_bucket and partial-creds validation

- Replace list_buckets() with head_bucket() in config_validator and
  health check; least-privilege IAM roles need only s3:ListBucket on
  the configured bucket, not s3:ListAllMyBuckets
- Raise ValueError in make_client() when exactly one of S3_ACCESS_KEY /
  S3_SECRET_KEY is set — partial static credentials are a misconfiguration
- Remove total_buckets from health check details (no longer available)

Author: ynotoast

src/config/s3.py

@@ -37,8 +37,19 @@ def make_client(self) -> Any:
         ``secret_key`` are set. When they are ``None``, boto3 falls through to
         its default credential chain (env vars, ``~/.aws/credentials``, EC2/ECS
         instance metadata).
+
+        Raises ``ValueError`` when exactly one of the pair is set — partial
+        static config is always a misconfiguration.
         """
-        kwargs: dict[str, Any] = {"endpoint_url": self.endpoint_url, "region_name": self.region}
+        if bool(self.access_key) != bool(self.secret_key):
+            raise ValueError(
+                "S3_ACCESS_KEY and S3_SECRET_KEY must both be set or both be unset. "
+                "Partial static credentials are not supported."
+            )
+        kwargs: dict[str, Any] = {
+            "endpoint_url": self.endpoint_url,
+            "region_name": self.region,
+        }
         if self.access_key and self.secret_key:
             kwargs["aws_access_key_id"] = self.access_key
             kwargs["aws_secret_access_key"] = self.secret_key

src/services/health.py

@@ -226,12 +226,8 @@ async def check_s3(self) -> HealthCheckResult:
                 self._s3_client = settings.s3.make_client()
 
             loop = asyncio.get_event_loop()
-            buckets_resp = await loop.run_in_executor(
-                None, self._s3_client.list_buckets
-            )
-            buckets = buckets_resp.get("Buckets", [])
 
-            # Check if our bucket exists
+            # Check if our bucket exists; create it if not
             try:
                 await loop.run_in_executor(
                     None,
@@ -293,7 +289,6 @@ async def check_s3(self) -> HealthCheckResult:
                 "endpoint": settings.s3_endpoint,
                 "bucket": settings.s3_bucket,
                 "bucket_exists": bucket_exists,
-                "total_buckets": len(buckets),
                 "secure": settings.s3_secure,
             }
 

src/utils/config_validator.py

@@ -125,18 +125,16 @@ def _validate_s3_connection(self):
         try:
             client = settings.s3.make_client()
 
-            # Test connection by listing buckets
-            response = client.list_buckets()
-            buckets = response.get("Buckets", [])
-
-            # Check if our bucket exists
-            bucket_exists = any(
-                bucket["Name"] == settings.s3_bucket for bucket in buckets
-            )
-            if not bucket_exists:
-                self.warnings.append(
-                    f"S3 bucket '{settings.s3_bucket}' does not exist - will be created"
-                )
+            try:
+                client.head_bucket(Bucket=settings.s3_bucket)
+            except ClientError as e:
+                code = e.response["Error"]["Code"]
+                if code in ("404", "NoSuchBucket"):
+                    self.warnings.append(
+                        f"S3 bucket '{settings.s3_bucket}' does not exist"
+                    )
+                else:
+                    raise
 
         except ClientError as e:
             if settings.api_debug: