prevent path traversal in s3 to sftp/ftp transfer destinations

providers/amazon/src/airflow/providers/amazon/aws/transfers/s3_to_ftp.py

@@ -22,6 +22,7 @@
 from typing import TYPE_CHECKING
 
 from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+from airflow.providers.amazon.aws.utils import validate_destination_path
 from airflow.providers.common.compat.sdk import BaseOperator
 from airflow.providers.ftp.hooks.ftp import FTPHook
 
@@ -85,6 +86,7 @@ def __init__(
         self.fail_on_file_not_exist = fail_on_file_not_exist
 
     def _download_from_s3(self, s3_hook: S3Hook, ftp_hook: FTPHook, s3_key: str, ftp_path: str) -> None:
+        validate_destination_path(ftp_path, self.ftp_path, base_name="ftp_path")
         if not s3_hook.check_for_key(s3_key, self.s3_bucket):
             if self.fail_on_file_not_exist:
                 raise FileNotFoundError(f"Key {s3_key!r} not found in S3 bucket {self.s3_bucket!r}")

providers/amazon/src/airflow/providers/amazon/aws/transfers/s3_to_sftp.py

@@ -23,6 +23,7 @@
 from urllib.parse import urlsplit
 
 from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+from airflow.providers.amazon.aws.utils import validate_destination_path
 from airflow.providers.common.compat.sdk import BaseOperator
 from airflow.providers.ssh.hooks.ssh import SSHHook
 
@@ -113,6 +114,7 @@ def _download_from_s3(
         s3_key: str,
         sftp_path: str,
     ) -> None:
+        validate_destination_path(sftp_path, self.sftp_path, base_name="sftp_path")
         if not s3_hook.check_for_key(s3_key, self.s3_bucket):
             if self.fail_on_file_not_exist:
                 raise FileNotFoundError(f"Key {s3_key!r} not found in S3 bucket {self.s3_bucket!r}")

providers/amazon/src/airflow/providers/amazon/aws/utils/__init__.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import logging
+import posixpath
 import re
 from datetime import datetime, timezone
 from enum import Enum
@@ -74,6 +75,31 @@ def validate_execute_complete_event(event: dict[str, Any] | None = None) -> dict
     return event
 
 
+def validate_destination_path(destination: str, base_path: str, *, base_name: str) -> None:
+    """
+    Ensure ``destination`` stays within ``base_path`` once resolved.
+
+    In multi-file transfers the destination is ``base_path`` concatenated with a key suffix
+    returned by ``list_keys``. S3 object names are arbitrary strings controlled by whoever can
+    write to the source bucket, so ``..`` segments or an absolute name could place the upload
+    outside ``base_path`` once the remote server resolves it on its host. ``base_name`` is the
+    operator argument name used in the error message (e.g. ``sftp_path`` or ``ftp_path``).
+    """
+    base = posixpath.normpath(base_path)
+    resolved = posixpath.normpath(destination)
+    escapes = (
+        resolved == ".."
+        or resolved.startswith("../")
+        or (posixpath.isabs(resolved) and not posixpath.isabs(base))
+        or (base != "." and resolved != base and not resolved.startswith(base.rstrip("/") + "/"))
+    )
+    if escapes:
+        raise ValueError(
+            f"Refusing to upload S3 object to {destination!r}: resolved path "
+            f"escapes configured {base_name} {base_path!r}."
+        )
+
+
 class _StringCompareEnum(Enum):
     """
     An Enum class which can be compared with regular `str` and subclasses.

providers/amazon/tests/unit/amazon/aws/transfers/test_s3_to_ftp.py

@@ -22,6 +22,7 @@
 import pytest
 
 from airflow.providers.amazon.aws.transfers.s3_to_ftp import S3ToFTPOperator
+from airflow.providers.amazon.aws.utils import validate_destination_path
 
 TASK_ID = "test_s3_to_ftp"
 BUCKET = "test-s3-bucket"
@@ -101,3 +102,22 @@ def test_fail_on_file_not_exist_skip(self, fail_on_file_not_exist):
             with patch.object(op.log, "info") as mock_log:
                 op._download_from_s3(mock_s3_hook, MagicMock(), S3_KEY, FTP_PATH)
             mock_log.assert_called_once()
+
+    @pytest.mark.parametrize(
+        "object_suffix",
+        [
+            pytest.param("../../../../etc/cron.d/evil", id="dotdot-segments"),
+            pytest.param("subdir/../../../escape", id="dotdot-cancels-base"),
+        ],
+    )
+    def test_validate_destination_path_rejects_escape(self, object_suffix):
+        # In multi-file mode the destination is ``ftp_path`` plus a key suffix
+        # returned by ``list_keys``; a crafted object name must not place the
+        # upload outside the configured ``ftp_path`` on the FTP server.
+        ftp_path = "/srv/ftp/incoming/"
+        with pytest.raises(ValueError, match="escapes configured ftp_path"):
+            validate_destination_path(ftp_path + object_suffix, ftp_path, base_name="ftp_path")
+
+    def test_validate_destination_path_allows_contained(self):
+        ftp_path = "/srv/ftp/incoming/"
+        validate_destination_path(ftp_path + "sub/dir/report.csv", ftp_path, base_name="ftp_path")

providers/amazon/tests/unit/amazon/aws/transfers/test_s3_to_sftp.py

@@ -24,6 +24,7 @@
 from airflow.models import DAG
 from airflow.providers.amazon.aws.hooks.s3 import S3Hook
 from airflow.providers.amazon.aws.transfers.s3_to_sftp import S3ToSFTPOperator
+from airflow.providers.amazon.aws.utils import validate_destination_path
 from airflow.providers.ssh.hooks.ssh import SSHHook
 from airflow.providers.ssh.operators.ssh import SSHOperator
 from airflow.utils.timezone import datetime
@@ -375,3 +376,22 @@ def test_fail_on_file_not_exist_skip(self, fail_on_file_not_exist):
             with patch.object(op.log, "info") as mock_log:
                 op._download_from_s3(MagicMock(), mock_s3_hook, S3_KEY, SFTP_PATH)
             mock_log.assert_called_once()
+
+    @pytest.mark.parametrize(
+        "object_suffix",
+        [
+            pytest.param("../../../../etc/cron.d/evil", id="dotdot-segments"),
+            pytest.param("subdir/../../../escape", id="dotdot-cancels-base"),
+        ],
+    )
+    def test_validate_destination_path_rejects_escape(self, object_suffix):
+        # In multi-file mode the destination is ``sftp_path`` plus a key suffix
+        # returned by ``list_keys``; a crafted object name must not place the
+        # upload outside the configured ``sftp_path`` on the SFTP server.
+        sftp_path = "/srv/sftp/incoming/"
+        with pytest.raises(ValueError, match="escapes configured sftp_path"):
+            validate_destination_path(sftp_path + object_suffix, sftp_path, base_name="sftp_path")
+
+    def test_validate_destination_path_allows_contained(self):
+        sftp_path = "/srv/sftp/incoming/"
+        validate_destination_path(sftp_path + "sub/dir/report.csv", sftp_path, base_name="sftp_path")