fix: support ECS task role for S3 storage

packages/shared-python/shared/core/config/storage.py

@@ -6,7 +6,7 @@
 import boto3
 from botocore.client import BaseClient
 from botocore.config import Config
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, model_validator
 
 from shared.core.exceptions.domain_exceptions import (
     DependencyMissingException,
@@ -30,8 +30,8 @@ class StorageConfig(BaseModel):
 
     # Shared S3-style configuration used by S3, OSS, and MinIO.
     S3_BUCKET_NAME: str = Field(..., description="Bucket name")
-    S3_ACCESS_KEY_ID: str = Field(..., description="Access key ID")
-    S3_SECRET_ACCESS_KEY: str = Field(..., description="Secret access key")
+    S3_ACCESS_KEY_ID: str = Field(default="", description="Access key ID")
+    S3_SECRET_ACCESS_KEY: str = Field(default="", description="Secret access key")
     S3_ENDPOINT_URL: str = Field(
         default="", description="Endpoint URL for S3-compatible services such as MinIO"
     )
@@ -113,6 +113,27 @@ class StorageConfig(BaseModel):
         default=True, description="Verify OSS event signatures"
     )
 
+    @model_validator(mode="after")
+    def validate_storage_credentials(self) -> "StorageConfig":
+        """Validate credentials according to the selected storage backend."""
+        storage_type = self.S3_TYPE.lower()
+        has_access_key = bool(self.S3_ACCESS_KEY_ID)
+        has_secret_key = bool(self.S3_SECRET_ACCESS_KEY)
+
+        if has_access_key != has_secret_key:
+            raise ValueError(
+                "S3_ACCESS_KEY_ID and S3_SECRET_ACCESS_KEY must be configured together"
+            )
+
+        if storage_type in {"oss", "minio"} and not (
+            has_access_key and has_secret_key
+        ):
+            raise ValueError(
+                f"Explicit storage credentials are required when S3_TYPE={storage_type}"
+            )
+
+        return self
+
     def get_s3_client(self) -> BaseClient:
         """Return an S3 client for S3-compatible backends."""
         # Build the client config.
@@ -128,11 +149,14 @@ def get_s3_client(self) -> BaseClient:
         config = Config(**config_kwargs) if config_kwargs else None
 
         # Build client kwargs.
-        client_kwargs: dict[str, object] = {
-            "service_name": "s3",
-            "aws_access_key_id": self.S3_ACCESS_KEY_ID,
-            "aws_secret_access_key": self.S3_SECRET_ACCESS_KEY,
-        }
+        client_kwargs: dict[str, object] = {"service_name": "s3"}
+
+        # When explicit keys are omitted, boto3 automatically retrieves temporary
+        # authenticated credentials from the ECS task role. Explicit credentials
+        # remain supported for local and legacy deployments.
+        if self.S3_ACCESS_KEY_ID and self.S3_SECRET_ACCESS_KEY:
+            client_kwargs["aws_access_key_id"] = self.S3_ACCESS_KEY_ID
+            client_kwargs["aws_secret_access_key"] = self.S3_SECRET_ACCESS_KEY
 
         # Add endpoint_url for MinIO or custom S3-compatible services.
         if self.S3_ENDPOINT_URL:

packages/shared-python/shared/tests/test_storage_config_contract.py

@@ -0,0 +1,96 @@
+"""Contracts for storage credentials at the boto3 boundary."""
+
+import os
+from unittest.mock import Mock
+
+import pytest
+from pydantic import ValidationError
+
+os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test")
+os.environ.setdefault("TMP_PATH", "/tmp/knowhere-test")
+os.environ.setdefault("S3_BUCKET_NAME", "test-bucket")
+os.environ.setdefault("S3_ACCESS_KEY_ID", "")
+os.environ.setdefault("S3_SECRET_ACCESS_KEY", "")
+os.environ.setdefault("S3_TEMP_PATH", "/tmp/knowhere-storage-contract")
+
+from shared.core.config.storage import StorageConfig
+
+
+def create_storage_config(**overrides: str) -> StorageConfig:
+    """Create the smallest valid storage configuration for a contract test."""
+    values: dict[str, str] = {
+        "S3_BUCKET_NAME": "test-bucket",
+        "S3_TEMP_PATH": "/tmp/knowhere-storage-contract",
+    }
+    values.update(overrides)
+    return StorageConfig(**values)
+
+
+def test_aws_s3_uses_default_credential_chain_when_keys_are_empty(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """AWS S3 must allow ECS task-role credentials through boto3's chain."""
+    boto3_client: Mock = Mock()
+    monkeypatch.setattr("shared.core.config.storage.boto3.client", boto3_client)
+
+    config: StorageConfig = create_storage_config(S3_TYPE="s3", S3_REGION="us-east-1")
+
+    config.get_s3_client()
+
+    boto3_client.assert_called_once()
+    client_arguments: dict[str, object] = dict(boto3_client.call_args.kwargs)
+    assert client_arguments["service_name"] == "s3"
+    assert client_arguments["region_name"] == "us-east-1"
+    assert "aws_access_key_id" not in client_arguments
+    assert "aws_secret_access_key" not in client_arguments
+
+
+def test_aws_s3_passes_complete_explicit_credentials(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Existing deployments may continue to provide an explicit key pair."""
+    boto3_client: Mock = Mock()
+    monkeypatch.setattr("shared.core.config.storage.boto3.client", boto3_client)
+    config: StorageConfig = create_storage_config(
+        S3_TYPE="s3",
+        S3_ACCESS_KEY_ID="access-key",
+        S3_SECRET_ACCESS_KEY="secret-key",
+    )
+
+    config.get_s3_client()
+
+    client_arguments: dict[str, object] = dict(boto3_client.call_args.kwargs)
+    assert client_arguments["aws_access_key_id"] == "access-key"
+    assert client_arguments["aws_secret_access_key"] == "secret-key"
+
+
+@pytest.mark.parametrize("storage_type", ["oss", "minio"])
+def test_s3_compatible_backends_require_explicit_credentials(
+    storage_type: str,
+) -> None:
+    """OSS and MinIO must not silently fall back to an AWS identity chain."""
+    with pytest.raises(
+        ValidationError,
+        match=f"Explicit storage credentials are required when S3_TYPE={storage_type}",
+    ):
+        create_storage_config(S3_TYPE=storage_type)
+
+
+@pytest.mark.parametrize(
+    ("access_key_id", "secret_access_key"),
+    [("access-key", ""), ("", "secret-key")],
+)
+def test_storage_rejects_partial_explicit_credentials(
+    access_key_id: str,
+    secret_access_key: str,
+) -> None:
+    """A partial key pair must fail before an unusable client is created."""
+    with pytest.raises(
+        ValidationError,
+        match="S3_ACCESS_KEY_ID and S3_SECRET_ACCESS_KEY must be configured together",
+    ):
+        create_storage_config(
+            S3_TYPE="s3",
+            S3_ACCESS_KEY_ID=access_key_id,
+            S3_SECRET_ACCESS_KEY=secret_access_key,
+        )