Add support for anonymous access to s3 buckets for objectstorage (apa…

…che#35273)

Open data buckets on S3 can require that authentication is turned off. If no credentials 
are obtained we now set the anon flag to make sure that credentials headers (like aws_access_key) 
are dropped and unsigned is set.

airflow/providers/amazon/aws/fs/s3.py

@@ -16,6 +16,7 @@
 # under the License.
 from __future__ import annotations
 
+import asyncio
 import logging
 from functools import partial
 from typing import TYPE_CHECKING, Any, Callable, Dict
@@ -85,7 +86,12 @@ def get_fs(conn_id: str | None) -> AbstractFileSystem:
     if proxy_uri := s3_service_config.get(S3_PROXY_URI, None):
         config_kwargs["proxies"] = {"http": proxy_uri, "https": proxy_uri}
 
-    fs = S3FileSystem(session=session, config_kwargs=config_kwargs, endpoint_url=endpoint_url)
+    anon = False
+    if asyncio.run(session.get_credentials()) is None:
+        log.info("No credentials found, using anonymous access")
+        anon = True
+
+    fs = S3FileSystem(session=session, config_kwargs=config_kwargs, endpoint_url=endpoint_url, anon=anon)
 
     for event_name, event_function in register_events.items():
         fs.s3.meta.events.register_last(event_name, event_function, unique_id=1925)

tests/providers/amazon/aws/fs/test_fs.py

@@ -16,6 +16,9 @@
 # under the License.
 from __future__ import annotations
 
+import os
+from unittest.mock import patch
+
 import pytest
 import responses
 from botocore.awsrequest import AWSRequest
@@ -39,6 +42,19 @@ def test_get_s3fs(self):
 
         assert "s3" in fs.protocol
 
+    @patch("s3fs.S3FileSystem", autospec=True)
+    def test_get_s3fs_anonymous(self, s3fs, monkeypatch):
+        from airflow.providers.amazon.aws.fs.s3 import get_fs
+
+        # remove all AWS_* env vars
+        for env_name in os.environ:
+            if env_name.startswith("AWS"):
+                monkeypatch.delenv(env_name, raising=False)
+
+        get_fs(conn_id=None)
+
+        assert s3fs.call_args.kwargs["anon"] is True
+
     @responses.activate
     def test_signer(self):
         from airflow.providers.amazon.aws.fs.s3 import s3v4_rest_signer