Log the AWS identity used by AwsDuckDBHook (#75997)

DuckDB reports a denied S3 request only as "HTTP 403", without saying
which credentials it used, so the AWS identity is the first thing to
check. sts:GetCallerIdentity returns it and needs no IAM permission.

The lookup is only a debugging aid and must never affect the task. It
has short timeouts and no retries, and a failure is logged at debug
level and ignored.

providers/amazon/src/airflow/providers/amazon/aws/hooks/duckdb.py

@@ -20,7 +20,10 @@ from functools import cached_property
 from typing import TYPE_CHECKING, Any, Literal
 from urllib.parse import urlsplit
 
+from botocore.config import Config
+
 from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+from airflow.providers.amazon.aws.hooks.sts import StsHook
 from airflow.providers.amazon.version_compat import NOTSET, ArgNotSet
 from airflow.providers.common.compat.sdk import AirflowOptionalProviderFeatureException
 
@@ -138,6 +141,7 @@ class AwsDuckDBHook(DuckDBHook):
         self._credential_chain = credential_chain
         self._s3_endpoint_url = s3_endpoint_url
         self._secret_name = secret_name
+        self._aws_identity_looked_up = False
 
     @property
     def aws_conn_id(self) -> str | None:
@@ -197,7 +201,7 @@ class AwsDuckDBHook(DuckDBHook):
         return self.connection_extra.get("endpoint_url") or self.aws_hook.conn_config.endpoint_url
 
     def configure_secrets(self, conn: DuckDBPyConnection) -> None:
-        """Create the DuckDB S3 secret for this connection."""
+        """Create the DuckDB S3 secret for this connection, and log the AWS identity it uses."""
         strategy = self.credential_strategy
         if strategy == "none":
             self.log.debug("credential_strategy is 'none'; not creating a DuckDB S3 secret.")
@@ -232,6 +236,39 @@ class AwsDuckDBHook(DuckDBHook):
                 error,
             )
 
+        self._log_aws_identity(strategy, region, endpoint_url)
+
+    def _log_aws_identity(
+        self, strategy: CredentialStrategy, region: str | None, endpoint_url: str | None
+    ) -> None:
+        """Log the AWS identity once per hook, to help debug S3 ``HTTP 403`` errors."""
+        if self._aws_identity_looked_up:
+            return
+        self._aws_identity_looked_up = True
+        try:
+            if endpoint_url:
+                host = urlsplit(endpoint_url if "//" in endpoint_url else f"//{endpoint_url}").hostname or ""
+                if not host.endswith((".amazonaws.com", ".amazonaws.com.cn")):
+                    return  # Not AWS (e.g. MinIO), so no AWS identity applies.
+            aws_config = self.aws_hook.config
+            bounded = Config(
+                # Keeps an unreachable STS endpoint from delaying the task by more than a few seconds.
+                connect_timeout=2,
+                read_timeout=2,
+                retries={"total_max_attempts": 1},
+                # Caller names only the operator (_PythonDecoratedOperator for @task), so name the hook too.
+                user_agent_extra=f"AwsDuckDBHook {aws_config.user_agent_extra or ''}".strip(),
+            )
+            sts_hook = StsHook(
+                # DuckDB's credential chain ignores the connection's credentials, so boto3 has to as well.
+                aws_conn_id=self.aws_conn_id if strategy == "config" else None,
+                region_name=region,
+                config=aws_config.merge(bounded),
+            )
+            self.log.info("AWS identity: %s", sts_hook.get_conn().get_caller_identity()["Arn"])
+        except Exception as error:  # Not BaseException: a task timeout must still stop the task.
+            self.log.debug("Could not look up the AWS identity: %s", error)
+
     def _provider_clauses(self, strategy: CredentialStrategy) -> list[str]:
         if strategy == "credential_chain":
             clauses = ["PROVIDER credential_chain"]

providers/amazon/tests/unit/amazon/aws/hooks/test_duckdb.py

@@ -21,6 +21,8 @@ from unittest import mock
 
 import duckdb
 import pytest
+from botocore.config import Config
+from botocore.exceptions import EndpointConnectionError
 
 from airflow.models.connection import Connection
 from airflow.providers.amazon.aws.hooks.duckdb import AwsDuckDBHook
@@ -29,6 +31,9 @@ from airflow.providers.duckdb.operators.duckdb import DuckDBExecuteQueryOperator
 
 GET_CONNECTION = "airflow.providers.amazon.aws.hooks.duckdb.AwsDuckDBHook.get_connection"
 AWS_BASE_HOOK = "airflow.providers.amazon.aws.hooks.duckdb.AwsBaseHook"
+STS_HOOK = "airflow.providers.amazon.aws.hooks.duckdb.StsHook"
+ARN = "arn:aws:sts::123456789012:assumed-role/airflow-worker/session"
+S3_VPC_ENDPOINT = "https://bucket.vpce-1a2b3c4d-5e6f.s3.us-east-1.vpce.amazonaws.com"
 
 try:
     from airflow.sdk.exceptions import AirflowNotFoundException
@@ -305,3 +310,55 @@ class TestAwsDuckDBHookConnectionOverrides:
         with mock.patch(GET_CONNECTION, return_value=self.duckdb_connection(secret_name="from_extra")):
             statement = secret_statement(AwsDuckDBHook(secret_name="explicit"))
         assert statement.startswith("CREATE OR REPLACE SECRET explicit (")
+
+
+class TestAwsDuckDBHookIdentityLookup:
+    @pytest.mark.usefixtures("no_duckdb_connection")
+    @pytest.mark.parametrize(
+        ("hook_kwargs", "sts_conn_id"),
+        [
+            pytest.param({}, None, id="credential-chain"),
+            pytest.param({"credential_strategy": "config"}, "aws_default", id="config"),
+            pytest.param({"s3_endpoint_url": S3_VPC_ENDPOINT}, None, id="s3-vpc-endpoint"),
+        ],
+    )
+    @mock.patch(STS_HOOK, autospec=True)
+    def test_identity_is_logged(self, mock_sts_hook, aws_base_hook, hook_kwargs, sts_conn_id):
+        aws_base_hook.config = Config(user_agent_extra="from-connection")
+        mock_sts_hook.return_value.get_conn.return_value.get_caller_identity.return_value = {"Arn": ARN}
+
+        with mock.patch.object(AwsDuckDBHook, "log") as mock_log:
+            AwsDuckDBHook(**hook_kwargs).configure_secrets(mock.MagicMock(spec=duckdb.DuckDBPyConnection))
+
+        mock_sts_hook.assert_called_once_with(
+            aws_conn_id=sts_conn_id, region_name="us-west-2", config=mock.ANY
+        )
+        assert mock_sts_hook.call_args.kwargs["config"].user_agent_extra == "AwsDuckDBHook from-connection"
+        mock_log.info.assert_any_call("AWS identity: %s", ARN)
+
+    @pytest.mark.usefixtures("aws_base_hook", "no_duckdb_connection")
+    @pytest.mark.parametrize(
+        "hook_kwargs",
+        [
+            pytest.param({"credential_strategy": "none"}, id="no-s3-secret"),
+            pytest.param({"s3_endpoint_url": "http://minio:9000"}, id="s3-compatible-endpoint"),
+        ],
+    )
+    @mock.patch(STS_HOOK, autospec=True)
+    def test_lookup_is_skipped_without_aws_s3(self, mock_sts_hook, hook_kwargs):
+        AwsDuckDBHook(**hook_kwargs).configure_secrets(mock.MagicMock(spec=duckdb.DuckDBPyConnection))
+
+        mock_sts_hook.assert_not_called()
+
+    @pytest.mark.usefixtures("aws_base_hook", "no_duckdb_connection")
+    @mock.patch(STS_HOOK, autospec=True)
+    def test_lookup_is_tried_once_and_its_failure_is_ignored(self, mock_sts_hook):
+        get_caller_identity = mock_sts_hook.return_value.get_conn.return_value.get_caller_identity
+        get_caller_identity.side_effect = EndpointConnectionError(endpoint_url="https://sts.amazonaws.com")
+        hook = AwsDuckDBHook()
+        conn = mock.MagicMock(spec=duckdb.DuckDBPyConnection)
+
+        hook.configure_secrets(conn)
+        hook.configure_secrets(conn)
+
+        get_caller_identity.assert_called_once_with()