Support setting an S3 object ACL for remote task logs

providers/amazon/docs/logging/s3-task-handler.rst

@@ -43,6 +43,26 @@ To enable this feature, ``airflow.cfg`` must be configured as follows:
 
 In the above example, Airflow will try to use ``S3Hook(aws_conn_id='my_s3_conn')``.
 
+Setting an object ACL for cross-account logging
+'''''''''''''''''''''''''''''''''''''''''''''''
+
+When Airflow runs under one AWS account but writes logs to a bucket owned by a different
+account, S3 makes the writing account the owner of each uploaded log object. As a result the
+bucket owner cannot read or manage the logs. Applying the ``bucket-owner-full-control`` ACL
+on upload grants the bucket owner full control over the log objects.
+
+Set the ACL with the ``[aws] s3_task_handler_acl_policy`` option:
+
+.. code-block:: ini
+
+    [aws]
+    # ACL applied to every task log object uploaded to S3, e.g. for cross-account buckets.
+    s3_task_handler_acl_policy = bucket-owner-full-control
+
+When unset, no ACL is sent and the bucket's default object ownership applies. The same value can
+also be supplied through ``[logging] remote_task_handler_kwargs``, for example
+``{"acl_policy": "bucket-owner-full-control"}``.
+
 You can also use `LocalStack <https://localstack.cloud/>`_ to emulate Amazon S3 locally.
 To configure it, you must additionally set the endpoint url to point to your local stack.
 You can do this via the Connection Extra ``endpoint_url`` field.

providers/amazon/provider.yaml

@@ -1166,6 +1166,21 @@ config:
         version_added: 8.7.2
         example: airflow.providers.amazon.aws.log.cloudwatch_task_handler.json_serialize
         default: airflow.providers.amazon.aws.log.cloudwatch_task_handler.json_serialize_legacy
+      s3_task_handler_acl_policy:
+        description: |
+          The ACL applied to task log objects uploaded to S3 by the S3 remote log handler,
+          for example ``bucket-owner-full-control``.
+
+          This is primarily useful for cross-account remote logging: when Airflow runs under one AWS
+          account but writes logs to a bucket owned by another account, S3 makes the writing account
+          the object owner, so the bucket owner cannot read or manage the log objects. Setting
+          ``bucket-owner-full-control`` grants the bucket owner full control over the uploaded logs.
+
+          When unset, no ACL is sent and the bucket's default object ownership applies.
+        type: string
+        version_added: 9.33.0
+        example: bucket-owner-full-control
+        default: ~
   aws_batch_executor:
     description: |
       This section only applies if you are using the AwsBatchExecutor in

providers/amazon/src/airflow/providers/amazon/aws/log/s3_task_handler.py

@@ -44,6 +44,7 @@ class S3RemoteLogIO(LoggingMixin):  # noqa: D101
     remote_base: str
     base_log_folder: pathlib.Path = attrs.field(converter=pathlib.Path)
     delete_local_copy: bool
+    acl_policy: str | None = None
 
     processors = ()
 
@@ -68,6 +69,7 @@ def from_config(cls) -> S3RemoteLogIO:
                 "base_log_folder": os.path.expanduser(conf.get_mandatory_value("logging", "base_log_folder")),
                 "remote_base": conf.get_mandatory_value("logging", "remote_base_log_folder"),
                 "delete_local_copy": conf.getboolean("logging", "delete_local_logs"),
+                "acl_policy": conf.get("aws", "s3_task_handler_acl_policy", fallback=None) or None,
             }
             | io_kwargs,
         )
@@ -161,6 +163,8 @@ def write(
         extra_args = {}
         if conf.getboolean("logging", "ENCRYPT_S3_LOGS"):
             extra_args["ServerSideEncryption"] = "AES256"
+        if self.acl_policy:
+            extra_args["ACL"] = self.acl_policy
 
         # Default to a single retry attempt because s3 upload failures are
         # rare but occasionally occur.  Multiple retry attempts are unlikely
@@ -235,6 +239,9 @@ def __init__(
             delete_local_copy=kwargs.get(
                 "delete_local_copy", conf.getboolean("logging", "delete_local_logs")
             ),
+            acl_policy=kwargs.get(
+                "acl_policy", conf.get("aws", "s3_task_handler_acl_policy", fallback=None) or None
+            ),
         )
 
     def set_context(self, ti: TaskInstance, *, identifier: str | None = None) -> None:

providers/amazon/src/airflow/providers/amazon/get_provider_info.py

@@ -1279,6 +1279,13 @@ def get_provider_info():
                         "example": "airflow.providers.amazon.aws.log.cloudwatch_task_handler.json_serialize",
                         "default": "airflow.providers.amazon.aws.log.cloudwatch_task_handler.json_serialize_legacy",
                     },
+                    "s3_task_handler_acl_policy": {
+                        "description": "The ACL applied to task log objects uploaded to S3 by the S3 remote log handler,\nfor example ``bucket-owner-full-control``.\n\nThis is primarily useful for cross-account remote logging: when Airflow runs under one AWS\naccount but writes logs to a bucket owned by another account, S3 makes the writing account\nthe object owner, so the bucket owner cannot read or manage the log objects. Setting\n``bucket-owner-full-control`` grants the bucket owner full control over the uploaded logs.\n\nWhen unset, no ACL is sent and the bucket's default object ownership applies.\n",
+                        "type": "string",
+                        "version_added": "9.33.0",
+                        "example": "bucket-owner-full-control",
+                        "default": None,
+                    },
                 },
             },
             "aws_batch_executor": {

providers/amazon/tests/unit/amazon/aws/log/test_s3_task_handler.py

@@ -132,6 +132,50 @@ def test_resolve_remote_task_log_uses_provider_dispatch_not_local_settings(self,
         assert conn_id == "aws_default"
         legacy_discover.assert_not_called()
 
+    @conf_vars(
+        {
+            ("logging", "remote_base_log_folder"): "s3://bucket/remote/log/location",
+            ("aws", "s3_task_handler_acl_policy"): "bucket-owner-full-control",
+        }
+    )
+    def test_from_config_reads_acl_policy(self):
+        subject = S3RemoteLogIO.from_config()
+
+        assert subject.acl_policy == "bucket-owner-full-control"
+
+    @conf_vars({("logging", "remote_base_log_folder"): "s3://bucket/remote/log/location"})
+    def test_from_config_acl_policy_defaults_to_none(self):
+        subject = S3RemoteLogIO.from_config()
+
+        assert subject.acl_policy is None
+
+    @conf_vars(
+        {
+            ("logging", "remote_base_log_folder"): "s3://bucket/remote/log/location",
+            ("logging", "remote_task_handler_kwargs"): '{"acl_policy": "bucket-owner-full-control"}',
+        }
+    )
+    def test_from_config_acl_policy_via_remote_task_handler_kwargs(self):
+        subject = S3RemoteLogIO.from_config()
+
+        assert subject.acl_policy == "bucket-owner-full-control"
+
+
+class TestS3TaskHandlerInit:
+    @conf_vars({("aws", "s3_task_handler_acl_policy"): "bucket-owner-full-control"})
+    def test_init_reads_acl_policy_from_conf(self):
+        handler = S3TaskHandler("/tmp/local", "s3://bucket/remote/log/location")
+
+        assert handler.io.acl_policy == "bucket-owner-full-control"
+
+    @conf_vars({("aws", "s3_task_handler_acl_policy"): "bucket-owner-full-control"})
+    def test_init_acl_policy_kwarg_overrides_conf(self):
+        handler = S3TaskHandler(
+            "/tmp/local", "s3://bucket/remote/log/location", acl_policy="bucket-owner-read"
+        )
+
+        assert handler.io.acl_policy == "bucket-owner-read"
+
 
 @pytest.mark.db_test
 class TestS3RemoteLogIO:
@@ -280,6 +324,15 @@ def test_write_with_encryption(self):
         body = boto3.resource("s3").Object("bucket", self.remote_log_key).get()["Body"].read()
         assert body == b"text"
 
+    def test_write_with_acl_policy(self):
+        self.subject.acl_policy = "bucket-owner-full-control"
+        conn = self.subject.hook.get_conn()
+        with mock.patch.object(conn, "put_object", wraps=conn.put_object) as mock_put_object:
+            self.subject.write("text", self.remote_log_location)
+        assert mock_put_object.call_args.kwargs["ACL"] == "bucket-owner-full-control"
+        body = boto3.resource("s3").Object("bucket", self.remote_log_key).get()["Body"].read()
+        assert body == b"text"
+
     def test_upload_repeated_appends_no_duplication(self):
         """Simulate reschedule-mode sensor: each cycle appends to the local log, then uploads.