Keep S3 Dag bundle downloads within the configured directory

providers/amazon/docs/bundles/index.rst

@@ -27,6 +27,11 @@ S3DagBundle
 Use the :class:`~airflow.providers.amazon.aws.bundles.s3.S3DagBundle` to configure an S3 bundle in your Airflow's
 ``[dag_processor] dag_bundle_config_list``.
 
+The ``prefix`` selects a subdirectory, with or without a trailing slash. For example,
+``dags`` and ``dags/`` both download objects under ``dags/``. An empty prefix downloads
+the whole bucket. IAM policies that restrict listing with the ``s3:prefix`` condition
+must allow the directory prefix including its trailing slash.
+
 Example of using the S3DagBundle:
 
 **JSON format example**:

providers/amazon/docs/changelog.rst

@@ -26,6 +26,12 @@
 Changelog
 ---------
 
+.. warning::
+  ``S3DagBundle`` now appends ``/`` to non-empty directory prefixes when downloading Dags.
+  For a configured prefix of ``dags``, listing requests now use ``dags/``. IAM policies
+  that restrict ``s3:prefix`` by exact value must permit the directory prefix with its
+  trailing slash. Empty prefixes and prefixes that already end in ``/`` are unchanged.
+
 9.37.0
 ......
 

providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py

@@ -36,6 +36,7 @@ class S3DagBundle(BaseDagBundle):
     :param aws_conn_id: Airflow connection ID for AWS.  Defaults to AwsBaseHook.default_conn_name.
     :param bucket_name: The name of the S3 bucket containing the Dag files.
     :param prefix:  Optional subdirectory within the S3 bucket where the Dags are stored.
+                    A trailing slash is optional.
                     If None, Dags are assumed to be at the root of the bucket (Optional).
     """
 
@@ -129,9 +130,12 @@ def refresh(self) -> None:
             self._log.debug(
                 "Downloading Dags from s3://%s/%s to %s", self.bucket_name, self.prefix, self.s3_dags_dir
             )
+            sync_prefix = self.prefix
+            if sync_prefix and not sync_prefix.endswith("/"):
+                sync_prefix += "/"
             self.s3_hook.sync_to_local_dir(
                 bucket_name=self.bucket_name,
-                s3_prefix=self.prefix,
+                s3_prefix=sync_prefix,
                 local_dir=self.s3_dags_dir,
                 delete_stale=True,
             )

providers/amazon/tests/unit/amazon/aws/bundles/test_s3.py

@@ -221,3 +221,36 @@ def test_refresh_without_prefix(self, s3_bucket, s3_client):
         bundle.refresh()
         assert bundle._log.debug.call_count == 2
         assert bundle._log.debug.call_args_list == [download_log_call, download_log_call]
+
+    @pytest.mark.parametrize("prefix", ["dags", "dags/", "project/dags", "project/dags/"])
+    @pytest.mark.parametrize("extra_suffix", ["_archive/other.py", pytest.param("", id="key_equals_prefix")])
+    def test_refresh_uses_directory_prefix(self, s3_client, prefix, extra_suffix):
+        s3_client.create_bucket(Bucket=S3_BUCKET_NAME)
+        directory = prefix.rstrip("/")
+        old_key = f"{directory}/nested/old.py"
+        extra_key = directory + extra_suffix
+        s3_client.put_object(Bucket=S3_BUCKET_NAME, Key=old_key, Body=b"old")
+        s3_client.put_object(Bucket=S3_BUCKET_NAME, Key=extra_key, Body=b"outside")
+
+        bundle = S3DagBundle(name="test", bucket_name=S3_BUCKET_NAME, prefix=prefix)
+        original_url = bundle.view_url_template()
+        original_repr = repr(bundle)
+        bundle.initialize()
+        assert bundle.is_initialized
+        assert (bundle.path / "nested/old.py").read_bytes() == b"old"
+        assert {p.relative_to(bundle.path).as_posix() for p in bundle.path.rglob("*") if p.is_file()} == {
+            "nested/old.py"
+        }
+
+        s3_client.delete_object(Bucket=S3_BUCKET_NAME, Key=old_key)
+        s3_client.put_object(Bucket=S3_BUCKET_NAME, Key=f"{directory}/new.py", Body=b"new-content")
+        bundle.refresh()
+        assert not (bundle.path / "nested").exists()
+        assert (bundle.path / "new.py").read_bytes() == b"new-content"
+        assert {p.relative_to(bundle.path).as_posix() for p in bundle.path.rglob("*") if p.is_file()} == {
+            "new.py"
+        }
+        assert bundle.prefix == prefix
+        assert repr(bundle) == original_repr
+        assert bundle.view_url_template() == original_url
+        assert s3_client.get_object(Bucket=S3_BUCKET_NAME, Key=extra_key)["Body"].read() == b"outside"