--- a/airflow/providers/google/cloud/transfers/gcs_to_sftp.py	2026-07-06 09:51:08.905429443 +0000
+++ b/airflow/providers/google/cloud/transfers/gcs_to_sftp.py	2026-07-06 09:51:36.944333030 +0000
@@ -20,6 +20,7 @@
 from __future__ import annotations
 
 import os
+import stat
 from collections.abc import Sequence
 from functools import cached_property
 from tempfile import NamedTemporaryFile
@@ -86,6 +87,11 @@
     :param move_object: When move object is True, the object is moved instead
         of copied to the new location. This is the equivalent of a mv command
         as opposed to a cp command.
+    :param allow_destination_symlinks: (Optional) When set to False (default), any existing
+        symlink component found in the resolved destination path under ``destination_path``
+        is rejected before the SFTP write. This prevents attacker-controlled GCS object
+        names from traversing symlinks that escape the configured destination directory.
+        Set to True only when the destination tree is known to contain trusted symlinks.
     :param gcp_conn_id: (Optional) The connection ID used to connect to Google Cloud.
     :param sftp_conn_id: The sftp connection id. The name or identifier for
         establishing a connection to the SFTP server.
@@ -116,6 +122,7 @@
         keep_directory_structure: bool = True,
         create_intermediate_dirs: bool = True,
         move_object: bool = False,
+        allow_destination_symlinks: bool = False,
         gcp_conn_id: str = "google_cloud_default",
         sftp_conn_id: str = "ssh_default",
         impersonation_chain: str | Sequence[str] | None = None,
@@ -129,6 +136,7 @@
         self.keep_directory_structure = keep_directory_structure
         self.create_intermediate_dirs = create_intermediate_dirs
         self.move_object = move_object
+        self.allow_destination_symlinks = allow_destination_symlinks
         self.gcp_conn_id = gcp_conn_id
         self.sftp_conn_id = sftp_conn_id
         self.impersonation_chain = impersonation_chain
@@ -176,7 +184,28 @@
                 source_object = os.path.relpath(source_object, start=prefix)
             else:
                 source_object = os.path.basename(source_object)
-        return os.path.join(self.destination_path, source_object)
+        # GCS object names are arbitrary UTF-8 strings controlled by whoever can
+        # write to the source bucket, so ``..`` segments or an absolute prefix
+        # could canonicalise outside ``destination_path`` on the SFTP server.
+        # Resolve the join and require it to stay within the configured base.
+        resolved = os.path.normpath(os.path.join(self.destination_path, source_object))
+        base = os.path.normpath(self.destination_path)
+        escapes = (
+            resolved == ".."
+            or resolved.startswith(".." + os.sep)
+            # An absolute source_object absorbed a relative base entirely.
+            or (os.path.isabs(resolved) and not os.path.isabs(base))
+            # A configured base directory must remain the prefix of the result.
+            # ``base == "."`` is the SFTP login directory, where any non-escaping
+            # relative path is already in-bounds.
+            or (base != "." and resolved != base and not resolved.startswith(base.rstrip(os.sep) + os.sep))
+        )
+        if escapes:
+            raise ValueError(
+                f"Refusing to copy GCS object {source_object!r}: resolved destination "
+                f"{resolved!r} escapes configured destination_path {self.destination_path!r}."
+            )
+        return resolved
 
     def _copy_single_object(
         self,
@@ -198,6 +227,9 @@
         if self.create_intermediate_dirs:
             sftp_hook.create_directory(dir_path)
 
+        if not self.allow_destination_symlinks:
+            self._reject_destination_symlinks(sftp_hook, destination_path, source_object)
+
         with NamedTemporaryFile("w") as tmp:
             gcs_hook.download(
                 bucket_name=self.source_bucket,
@@ -210,6 +242,49 @@
             self.log.info("Executing delete of gs://%s/%s", self.source_bucket, source_object)
             gcs_hook.delete(self.source_bucket, source_object)
 
+    def _reject_destination_symlinks(
+        self,
+        sftp_hook: SFTPHook,
+        destination_path: str,
+        source_object: str,
+    ) -> None:
+        """
+        Best-effort check that no existing path component under ``destination_path``
+        is a symlink.
+
+        The lexical check in ``_resolve_destination_path`` blocks ``..`` and absolute
+        path escapes, but it cannot see server-side symlinks. An attacker who can
+        write GCS object names and who knows that the destination tree contains a
+        symlink can choose an object name that resolves outside ``destination_path``
+        through that symlink. This method walks the resolved path and rejects any
+        existing symlink component before the file is written.
+        """
+        norm_base = os.path.normpath(self.destination_path)
+        norm_dest = os.path.normpath(destination_path)
+
+        rel = os.path.relpath(norm_dest, norm_base)
+        if rel == "." or rel == ".." or rel.startswith(".." + os.sep):
+            return
+
+        with sftp_hook.get_managed_conn() as conn:
+            current = norm_base
+            for part in rel.split(os.sep):
+                if not part or part == ".":
+                    continue
+                current = os.path.join(current, part)
+                try:
+                    attr = conn.lstat(current)
+                except OSError:
+                    # Path component does not exist yet; the create_directory step
+                    # will create a real directory here, so no symlink to follow.
+                    continue
+                if stat.S_ISLNK(attr.st_mode):
+                    raise ValueError(
+                        f"Refusing to copy GCS object {source_object!r}: destination path component "
+                        f"{current!r} is a symlink, which could escape configured destination_path "
+                        f"{self.destination_path!r}. Set allow_destination_symlinks=True to skip this check."
+                    )
+
     def get_openlineage_facets_on_start(self):
         from airflow.providers.common.compat.openlineage.facet import Dataset
         from airflow.providers.google.cloud.openlineage.utils import extract_ds_name_from_gcs_path