CVE-2026-40861

@@ -856,20 +856,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(

@@ -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])

@@ -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

@@ -24,7 +24,7 @@
 
 .. towncrier release notes start
 
-Airflow 3.2.2 (2026-05-27)
+Airflow 3.2.2 (2026-05-29)
 --------------------------
 
 Significant Changes
@@ -81,7 +81,8 @@ Significant Changes
 
 Bug Fixes
 ^^^^^^^^^
-
+- Fix ``Callback.handle_event`` triggerer crash when OpenTelemetry metrics receive dict typed tag values (#67527) (#67529)
+- UI: Rewrite ``modulepreload hrefs`` to the api-server static path (#67548) (#67556)
 - Correctly pre-allocate ``external_executor_id`` with multiple executors on PostgreSQL (#67388) (#67458)
 - Return raw import-error stacktrace when a Dag file has no registered Dag (#67465) (#67478)
 - UI: Fix Expand/Collapse All on XComs and Audit Log JSON cells (#67316) (#67361)

@@ -1,2 +1,2 @@
-release-notes-hash: 6407b48d1054fe3ce68c09bf4435d91d
-source-date-epoch: 1779745327
+release-notes-hash: 504288db9a9dc13a0db859232fab98d0
+source-date-epoch: 1779811737