From ab99529ab2c9cea6777d6bbc233856cd6017a4ba Mon Sep 17 00:00:00 2001
From: Jarek Potiuk <jarek@potiuk.com>
Date: Thu, 16 Apr 2026 04:44:47 +0200
Subject: [PATCH] [v3-2-test] Refuse to follow log symlinks that resolve
 outside the base log folder (#65325)

* Refuse to follow log symlinks that resolve outside the base log folder

FileTaskHandler._read_from_local used to open every file that matched
the task's log glob pattern, including symlinks whose real path was
outside the configured base_log_folder. On deployments where worker
logs are accessible from the api-server, that meant the log viewer
could end up streaming content from files outside the configured log
tree whenever a symlink in the task log directory happened to match
the glob pattern.

Canonicalise self.local_base once via os.path.realpath and, for every
glob hit, resolve the path with os.path.realpath and skip it if the
resolved form is not contained in the canonicalised base log folder
(using os.path.commonpath, with a ValueError fallback for the
different-drive case on Windows). Open the resolved path rather than
the original glob hit so the file we open is the one we just
validated. Append to sources only after a successful open so sources
and log_streams stay aligned.

Drop the @staticmethod decorator so the method can read
self.local_base; existing call sites already invoke it via self.

Add a test class covering: regular-file-inside-base is still streamed;
a symlink whose real path is outside base_log_folder is skipped; a
symlink that stays inside base_log_folder is followed (legitimate
rotation case); and base_log_folder itself being a symlink works.

Generated-by: Claude Opus 4.6 (1M context) following the guidelines at
https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions

* Fix test__read_from_local to use valid base_log_folder

The existing test passed an empty string as base_log_folder, which
after the containment check resolves to CWD via os.path.realpath(""),
causing all files under tmp_path to be rejected. Use tmp_path instead.
(cherry picked from commit 3eda84547e743397ec1027733f97d37ab2e628b4)

Co-authored-by: Jarek Potiuk <jarek@potiuk.com>
---
 .../airflow/utils/log/file_task_handler.py    | 28 +++++-
 .../unit/utils/log/test_file_task_handler.py  | 99 +++++++++++++++++++
 .../tests/unit/utils/test_log_handlers.py     |  2 +-
 3 files changed, 125 insertions(+), 4 deletions(-)

diff --git a/airflow-core/src/airflow/utils/log/file_task_handler.py b/airflow-core/src/airflow/utils/log/file_task_handler.py
index 1481d0a315e9b..13138d1cff575 100644
--- a/airflow-core/src/airflow/utils/log/file_task_handler.py
+++ b/airflow-core/src/airflow/utils/log/file_task_handler.py
@@ -857,20 +857,42 @@ def _init_file(self, ti, *, identifier: str | None = None):
 
         return full_path
 
-    @staticmethod
     def _read_from_local(
+        self,
         worker_log_path: Path,
     ) -> StreamingLogResponse:
         sources: LogSourceInfo = []
         log_streams: list[RawLogStream] = []
+        # The glob below can match symlinks as well as regular files, so
+        # resolve each hit and only open the ones that stay inside the base
+        # log folder. Canonicalising ``self.local_base`` once up front makes
+        # the containment check compare two already-resolved paths.
+        base_log_folder = os.path.realpath(self.local_base)
         paths = sorted(worker_log_path.parent.glob(worker_log_path.name + "*"))
         if not paths:
             return sources, log_streams
 
         for path in paths:
+            resolved_path = os.path.realpath(path)
+            try:
+                if os.path.commonpath([base_log_folder, resolved_path]) != base_log_folder:
+                    continue
+            except ValueError:
+                # ``os.path.commonpath`` raises ``ValueError`` when the two
+                # paths have nothing in common (e.g. different drives on
+                # Windows); treat that as "not contained" and skip the file.
+                continue
+
+            # Open the resolved path so the file we read is the same one we
+            # just validated above. Append to ``sources`` only after a
+            # successful ``open`` so ``sources`` and ``log_streams`` stay
+            # aligned.
+            try:
+                log_stream = _stream_lines_by_chunk(open(resolved_path, encoding="utf-8"))
+            except OSError:
+                continue
             sources.append(os.fspath(path))
-            # Read the log file and yield lines
-            log_streams.append(_stream_lines_by_chunk(open(path, encoding="utf-8")))
+            log_streams.append(log_stream)
         return sources, log_streams
 
     def _read_from_logs_server(
diff --git a/airflow-core/tests/unit/utils/log/test_file_task_handler.py b/airflow-core/tests/unit/utils/log/test_file_task_handler.py
index cdda0e66de84c..93606cffe1dd1 100644
--- a/airflow-core/tests/unit/utils/log/test_file_task_handler.py
+++ b/airflow-core/tests/unit/utils/log/test_file_task_handler.py
@@ -90,3 +90,102 @@ def test_403_shows_secret_key_message(self, mock_get_url, mock_fetch):
         assert len(sources) == 1
         assert "secret_key" in sources[0]
         assert streams == []
+
+
+class TestFileTaskHandlerReadFromLocal:
+    """Tests for ``FileTaskHandler._read_from_local`` path containment."""
+
+    @staticmethod
+    def _drain(stream) -> str:
+        return "".join(list(stream))
+
+    def test_reads_regular_log_file_inside_base(self, tmp_path):
+        """A regular file under ``base_log_folder`` is streamed as before."""
+        log_dir = tmp_path / "dag" / "run" / "task"
+        log_dir.mkdir(parents=True)
+        log_file = log_dir / "1.log"
+        log_file.write_text("legitimate log line\n")
+
+        handler = FileTaskHandler(base_log_folder=str(tmp_path))
+        sources, streams = handler._read_from_local(log_file)
+
+        assert sources == [str(log_file)]
+        assert len(streams) == 1
+        assert "legitimate log line" in self._drain(streams[0])
+
+    def test_skips_symlink_resolving_outside_base_log_folder(self, tmp_path):
+        """A glob hit that resolves outside ``base_log_folder`` is not streamed.
+
+        This documents the intended containment behaviour: a file under the
+        task's log directory that is actually a symlink whose real path is
+        outside the configured base log folder must be skipped, even though
+        it matches the glob pattern used to discover the task's log files.
+        """
+        base_log_folder = tmp_path / "logs"
+        log_dir = base_log_folder / "dag" / "run" / "task"
+        log_dir.mkdir(parents=True)
+
+        # A regular log file inside the base log folder.
+        legit = log_dir / "1.log"
+        legit.write_text("legitimate log line\n")
+
+        # A file that lives outside the base log folder.
+        external_dir = tmp_path / "external"
+        external_dir.mkdir()
+        external_file = external_dir / "other.txt"
+        external_file.write_text("external content\n")
+
+        # A glob hit that matches ``1.log*`` but resolves outside the base.
+        escape_link = log_dir / "1.log.external"
+        escape_link.symlink_to(external_file)
+
+        handler = FileTaskHandler(base_log_folder=str(base_log_folder))
+        sources, streams = handler._read_from_local(legit)
+
+        assert str(legit) in sources
+        assert str(escape_link) not in sources
+        content = "".join(self._drain(s) for s in streams)
+        assert "legitimate log line" in content
+        assert "external content" not in content
+
+    def test_follows_symlink_within_base_log_folder(self, tmp_path):
+        """A symlink that resolves back into the base log folder is allowed.
+
+        The containment check compares the realpath of the glob hit to the
+        realpath of the base log folder, so a symlink that stays entirely
+        inside the log tree (for example from log rotation) still works.
+        """
+        base_log_folder = tmp_path / "logs"
+        log_dir = base_log_folder / "dag" / "run" / "task"
+        log_dir.mkdir(parents=True)
+
+        real_file = log_dir / "real.log"
+        real_file.write_text("inner content\n")
+
+        link = log_dir / "1.log.link"
+        link.symlink_to(real_file)
+
+        handler = FileTaskHandler(base_log_folder=str(base_log_folder))
+        sources, streams = handler._read_from_local(log_dir / "1.log")
+
+        assert str(link) in sources
+        assert "inner content" in "".join(self._drain(s) for s in streams)
+
+    def test_handles_base_log_folder_that_is_itself_a_symlink(self, tmp_path):
+        """``base_log_folder`` itself is realpath'd so a base that is a
+        symlink to the actual log directory is treated as contained."""
+        real_base = tmp_path / "real_logs"
+        real_base.mkdir()
+        base_link = tmp_path / "logs"
+        base_link.symlink_to(real_base)
+
+        log_dir = base_link / "dag" / "run" / "task"
+        log_dir.mkdir(parents=True)
+        log_file = log_dir / "1.log"
+        log_file.write_text("through-symlink content\n")
+
+        handler = FileTaskHandler(base_log_folder=str(base_link))
+        sources, streams = handler._read_from_local(log_file)
+
+        assert len(sources) == 1
+        assert "through-symlink content" in self._drain(streams[0])
diff --git a/airflow-core/tests/unit/utils/test_log_handlers.py b/airflow-core/tests/unit/utils/test_log_handlers.py
index 6a93e940e7254..4c95327c9fb71 100644
--- a/airflow-core/tests/unit/utils/test_log_handlers.py
+++ b/airflow-core/tests/unit/utils/test_log_handlers.py
@@ -510,7 +510,7 @@ def test__read_from_local(self, tmp_path):
         path2 = tmp_path / "hello1.log.suffix.log"
         path1.write_text("file1 content\nfile1 content2")
         path2.write_text("file2 content\nfile2 content2")
-        fth = FileTaskHandler("")
+        fth = FileTaskHandler(str(tmp_path))
         log_source_info, log_streams = fth._read_from_local(path1)
         assert log_source_info == [str(path1), str(path2)]
         assert len(log_streams) == 2
