Merge branch 'main' into sgarg/add-ybdb-connector-to-unstructured

Author: gargsans-yb

CHANGELOG.md

@@ -1,3 +1,11 @@
+## 1.2.14
+
+* **Fix: IBM watsonx.data S3 bucket authentication fix**
+
+## 1.2.13
+
+* **Feat: Make Bedrock embedding credentials optional and add IAM support**
+
 ## 1.2.12
 
 * **Fix: retry with wait when throttling error happens in Sharepoint connector**

test/integration/connectors/test_s3.py

@@ -48,6 +48,18 @@ def anon_connection_config() -> S3ConnectionConfig:
     return S3ConnectionConfig(access_config=S3AccessConfig(), anonymous=True)
 
 
+@pytest.fixture
+def ambient_credentials_config() -> S3ConnectionConfig:
+    """Test fixture for ambient credentials with mock values."""
+    access_config = S3AccessConfig(
+        use_ambient_credentials=True,
+        presigned_url="https://example.com/mock-presigned-url",
+        role_arn="arn:aws:iam::123456789012:role/test-role"
+    )
+    return S3ConnectionConfig(access_config=access_config)
+
+
+
 @pytest.mark.asyncio
 @pytest.mark.tags(CONNECTOR_TYPE, SOURCE_TAG, BLOB_STORAGE_TAG)
 async def test_s3_source(anon_connection_config: S3ConnectionConfig):

test/unit/connectors/ibm_watsonx/test_ibm_watsonx_s3.py

@@ -243,6 +243,7 @@ def test_ibm_watsonx_connection_config_get_catalog_success(
             "s3.access-key-id": "test_access_key_id",
             "s3.secret-access-key": "test_secret_access_key",
             "s3.region": "test_region",
+            "header.X-Iceberg-Access-Delegation": None,
         }
     )
 

unstructured_ingest/__version__.py

@@ -1 +1 @@
-__version__ = "1.2.12"  # pragma: no cover
+__version__ = "1.2.14"  # pragma: no cover

unstructured_ingest/embed/bedrock.py

@@ -58,9 +58,22 @@ def conform_query(query: str, provider: str) -> dict:
 
 
 class BedrockEmbeddingConfig(EmbeddingConfig):
-    aws_access_key_id: SecretStr = Field(description="aws access key id")
-    aws_secret_access_key: SecretStr = Field(description="aws secret access key")
-    region_name: str = Field(description="aws region name", default="us-west-2")
+    aws_access_key_id: SecretStr | None = Field(description="aws access key id", default=None)
+    aws_secret_access_key: SecretStr | None = Field(
+        description="aws secret access key", default=None
+    )
+    region_name: str = Field(
+        description="aws region name", 
+        default_factory=lambda: (
+            os.getenv("BEDROCK_REGION_NAME") or 
+            os.getenv("AWS_DEFAULT_REGION") or 
+            "us-west-2"
+        )
+    )
+    endpoint_url: str | None = Field(description="custom bedrock endpoint url", default=None)
+    access_method: str = Field(
+        description="authentication method", default="credentials"
+    )  # "credentials" or "iam"
     embedder_model_name: str = Field(
         default="amazon.titan-embed-text-v1",
         alias="model_name",
@@ -96,6 +109,20 @@ def wrap_error(self, e: Exception) -> Exception:
         return e
 
     def run_precheck(self) -> None:
+        # Validate access method and credentials configuration
+        if self.access_method == "credentials":
+            if not (self.aws_access_key_id and self.aws_secret_access_key):
+                raise ValueError(
+                    "Credentials access method requires aws_access_key_id and aws_secret_access_key"
+                )
+        elif self.access_method == "iam":
+            # For IAM, credentials are handled by AWS SDK
+            pass
+        else:
+            raise ValueError(
+                f"Invalid access_method: {self.access_method}. Must be 'credentials' or 'iam'"
+            )
+            
         client = self.get_bedrock_client()
         try:
             model_info = client.list_foundation_models(byOutputModality="EMBEDDING")
@@ -113,11 +140,30 @@ def run_precheck(self) -> None:
             raise self.wrap_error(e=e)
 
     def get_client_kwargs(self) -> dict:
-        return {
-            "aws_access_key_id": self.aws_access_key_id.get_secret_value(),
-            "aws_secret_access_key": self.aws_secret_access_key.get_secret_value(),
+        kwargs = {
             "region_name": self.region_name,
         }
+        
+        if self.endpoint_url:
+            kwargs["endpoint_url"] = self.endpoint_url
+            
+        if self.access_method == "credentials":
+            if self.aws_access_key_id and self.aws_secret_access_key:
+                kwargs["aws_access_key_id"] = self.aws_access_key_id.get_secret_value()
+                kwargs["aws_secret_access_key"] = self.aws_secret_access_key.get_secret_value()
+            else:
+                raise ValueError(
+                    "Credentials access method requires aws_access_key_id and aws_secret_access_key"
+                )
+        elif self.access_method == "iam":
+            # For IAM, boto3 will use default credential chain (IAM roles, environment, etc.)
+            pass
+        else:
+            raise ValueError(
+                f"Invalid access_method: {self.access_method}. Must be 'credentials' or 'iam'"
+            )
+            
+        return kwargs
 
     @requires_dependencies(
         ["boto3"],

unstructured_ingest/processes/connectors/ibm_watsonx/ibm_watsonx_s3.py

@@ -147,6 +147,10 @@ def get_catalog_config(self) -> dict[str, Any]:
             "s3.access-key-id": self.access_config.get_secret_value().access_key_id,
             "s3.secret-access-key": self.access_config.get_secret_value().secret_access_key,
             "s3.region": self.object_storage_region,
+            # By default this header is set to `vended-credentials`, and default bucket
+            # configuration doesn't allow vending credentials. We need to set it to `None`
+            # in order to use user-provided S3 credentials.
+            "header.X-Iceberg-Access-Delegation": None,
         }
 
     @requires_dependencies(["pyiceberg"], extras="ibm-watsonx-s3")