This is an automated email from the ASF dual-hosted git repository.
o-nikolas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 3e20d9054d7 Log the AWS identity used by AwsDuckDBHook (#75997)
3e20d9054d7 is described below
commit 3e20d9054d70cfbc111aa237a07b232358b961be
Author: Niko Oliveira <[email protected]>
AuthorDate: Mon Nov 2 18:34:44 2026 -0500
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.
---
.../airflow/providers/amazon/aws/hooks/duckdb.py | 39 ++++++++++++++-
.../tests/unit/amazon/aws/hooks/test_duckdb.py | 57 ++++++++++++++++++++++
2 files changed, 95 insertions(+), 1 deletion(-)
diff --git a/providers/amazon/src/airflow/providers/amazon/aws/hooks/duckdb.py
b/providers/amazon/src/airflow/providers/amazon/aws/hooks/duckdb.py
index 3f6e800ae7f..bf91a0d345d 100644
--- a/providers/amazon/src/airflow/providers/amazon/aws/hooks/duckdb.py
+++ b/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"]
diff --git a/providers/amazon/tests/unit/amazon/aws/hooks/test_duckdb.py
b/providers/amazon/tests/unit/amazon/aws/hooks/test_duckdb.py
index ce72e4313fe..4ee6d6a8b6e 100644
--- a/providers/amazon/tests/unit/amazon/aws/hooks/test_duckdb.py
+++ b/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()
