diff --git a/airflow/providers/amazon/aws/fs/s3.py b/airflow/providers/amazon/aws/fs/s3.py
index afe13be1b43b7..c2eefcc379ba6 100644
--- a/airflow/providers/amazon/aws/fs/s3.py
+++ b/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)
diff --git a/tests/providers/amazon/aws/fs/test_fs.py b/tests/providers/amazon/aws/fs/test_fs.py
index 7a392a28329af..babddf078a3e2 100644
--- a/tests/providers/amazon/aws/fs/test_fs.py
+++ b/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