Contain remote log upload paths within base_log_folder

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

@@ -84,6 +84,18 @@ def upload(self, path: os.PathLike | str, ti: RuntimeTI | None = None) -> None:
             local_loc = self.base_log_folder.joinpath(path)
             remote_loc = os.path.join(self.remote_base, path)
 
+        # The log path is supplied by the caller and is not guaranteed to stay within
+        # ``base_log_folder``: ``joinpath`` and ``PurePath.relative_to`` are purely lexical and
+        # do not normalise ``..``. Without this check a traversing path would have its contents
+        # uploaded to the remote log store and, with ``delete_local_copy``, its parent directory
+        # removed. Mirrors the containment check in ``CloudWatchRemoteLogIO.upload``.
+        base = self.base_log_folder.resolve()
+        try:
+            local_loc.resolve().relative_to(base)
+        except ValueError:
+            self.log.warning("Skipping upload: path %s is outside base_log_folder %s", local_loc, base)
+            return
+
         if local_loc.is_file():
             # read log and remove old logs to get just the latest additions
             log = local_loc.read_text()

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

@@ -538,3 +538,26 @@ def test_close_with_delete_local_logs_conf(self, delete_local_copy, expected_exi
     def test_filename_template_for_backward_compatibility(self):
         # filename_template arg support for running the latest provider on airflow 2
         S3TaskHandler(self.local_log_location, self.remote_log_base, filename_template=None)
+
+
+def test_upload_skips_path_outside_base_log_folder(tmp_path, caplog):
+    """A traversing log path is refused before the file is read or its parent removed.
+
+    ``base_log_folder.joinpath(path)`` is purely lexical, so a ``..``-bearing relative path
+    escapes the log folder. Without the containment check the file would be uploaded to the
+    remote log store and, with ``delete_local_copy``, its parent directory deleted.
+    """
+    base = tmp_path / "logs"
+    base.mkdir()
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    secret = outside / "secret.log"
+    secret.write_text("sensitive")
+
+    subject = S3RemoteLogIO(remote_base="s3://bucket/remote", base_log_folder=base, delete_local_copy=True)
+    with caplog.at_level(logging.WARNING):
+        subject.upload(os.path.join("..", "outside", "secret.log"))
+
+    assert secret.exists()
+    assert outside.exists()
+    assert "outside base_log_folder" in caplog.text

providers/google/src/airflow/providers/google/cloud/log/gcs_task_handler.py

@@ -107,6 +107,18 @@ def upload(self, path: os.PathLike | str, ti: RuntimeTI | None = None) -> None:
             local_loc = self.base_log_folder.joinpath(path)
             remote_loc = os.path.join(self.remote_base, path)
 
+        # The log path is supplied by the caller and is not guaranteed to stay within
+        # ``base_log_folder``: ``joinpath`` and ``PurePath.relative_to`` are purely lexical and
+        # do not normalise ``..``. Without this check a traversing path would have its contents
+        # uploaded to the remote log store and, with ``delete_local_copy``, its parent directory
+        # removed. Mirrors the containment check in ``CloudWatchRemoteLogIO.upload``.
+        base = self.base_log_folder.resolve()
+        try:
+            local_loc.resolve().relative_to(base)
+        except ValueError:
+            self.log.warning("Skipping upload: path %s is outside base_log_folder %s", local_loc, base)
+            return
+
         if local_loc.is_file():
             # read log and remove old logs to get just the latest additions
             log = local_loc.read_text()

providers/google/tests/unit/google/cloud/log/test_gcs_task_handler.py

@@ -727,3 +727,26 @@ def test_hook_silent_when_no_remote_log_conn_id_configured(self, mock_hook, capl
             "remote_log_conn_id" in r.getMessage() for r in caplog.records if r.levelno == logging.WARNING
         )
         mock_hook.assert_not_called()
+
+
+def test_upload_skips_path_outside_base_log_folder(tmp_path, caplog):
+    """A traversing log path is refused before the file is read or its parent removed.
+
+    ``base_log_folder.joinpath(path)`` is purely lexical, so a ``..``-bearing relative path
+    escapes the log folder. Without the containment check the file would be uploaded to the
+    remote log store and, with ``delete_local_copy``, its parent directory deleted.
+    """
+    base = tmp_path / "logs"
+    base.mkdir()
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    secret = outside / "secret.log"
+    secret.write_text("sensitive")
+
+    subject = GCSRemoteLogIO(remote_base="gs://bucket/remote", base_log_folder=base, delete_local_copy=True)
+    with caplog.at_level(logging.WARNING):
+        subject.upload(os.path.join("..", "outside", "secret.log"))
+
+    assert secret.exists()
+    assert outside.exists()
+    assert "outside base_log_folder" in caplog.text

providers/microsoft/azure/src/airflow/providers/microsoft/azure/log/wasb_task_handler.py

@@ -89,6 +89,18 @@ def upload(self, path: str | os.PathLike, ti: RuntimeTI | None = None) -> None:
             local_loc = self.base_log_folder.joinpath(path)
             remote_loc = os.path.join(self.remote_base, path)
 
+        # The log path is supplied by the caller and is not guaranteed to stay within
+        # ``base_log_folder``: ``joinpath`` and ``PurePath.relative_to`` are purely lexical and
+        # do not normalise ``..``. Without this check a traversing path would have its contents
+        # uploaded to the remote log store and, with ``delete_local_copy``, its parent directory
+        # removed. Mirrors the containment check in ``CloudWatchRemoteLogIO.upload``.
+        base = self.base_log_folder.resolve()
+        try:
+            local_loc.resolve().relative_to(base)
+        except ValueError:
+            self.log.warning("Skipping upload: path %s is outside base_log_folder %s", local_loc, base)
+            return
+
         if local_loc.is_file():
             # read log and remove old logs to get just the latest additions
             log = local_loc.read_text()

providers/microsoft/azure/tests/unit/microsoft/azure/log/test_wasb_task_handler.py

@@ -465,3 +465,31 @@ def test_filename_template_for_backward_compatibility(self):
             delete_local_copy=True,
             filename_template=None,
         )
+
+
+def test_upload_skips_path_outside_base_log_folder(tmp_path, caplog):
+    """A traversing log path is refused before the file is read or its parent removed.
+
+    ``base_log_folder.joinpath(path)`` is purely lexical, so a ``..``-bearing relative path
+    escapes the log folder. Without the containment check the file would be uploaded to the
+    remote log store and, with ``delete_local_copy``, its parent directory deleted.
+    """
+    base = tmp_path / "logs"
+    base.mkdir()
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    secret = outside / "secret.log"
+    secret.write_text("sensitive")
+
+    subject = WasbRemoteLogIO(
+        remote_base="remote/log/location",
+        base_log_folder=base,
+        delete_local_copy=True,
+        wasb_container="container",
+    )
+    with caplog.at_level(logging.WARNING):
+        subject.upload(os.path.join("..", "outside", "secret.log"))
+
+    assert secret.exists()
+    assert outside.exists()
+    assert "outside base_log_folder" in caplog.text