Apache Airflow: Potentially harmful permission changing by log task handler

@@ -159,6 +159,31 @@ def _ensure_ti(ti: TaskInstanceKey | TaskInstance, session) -> TaskInstance:
         raise AirflowException(f"Could not find TaskInstance for {ti}")
 
 
+def _change_directory_permissions_up(directory: Path, folder_permissions: int):
+    """
+    Change permissions of the given directory and its parents.
+
+    Only attempt to change permissions for directories owned by the current user.
+
+    :param directory: directory to change permissions of (including parents)
+    :param folder_permissions: permissions to set
+    """
+    if directory.stat().st_uid == os.getuid():
+        if directory.stat().st_mode % 0o1000 != folder_permissions % 0o1000:
+            print(f"Changing {directory} permission to {folder_permissions}")
+            try:
+                directory.chmod(folder_permissions)
+            except PermissionError as e:
+                # In some circumstances (depends on user and filesystem) we might not be able to
+                # change the permission for the folder (when the folder was created by another user
+                # before or when the filesystem does not allow to change permission). We should not
+                # fail in this case but rather ignore it.
+                print(f"Failed to change {directory} permission to {folder_permissions}: {e}")
+                return
+        if directory.parent != directory:
+            _change_directory_permissions_up(directory.parent, folder_permissions)
+
+
 class FileTaskHandler(logging.Handler):
     """
     FileTaskHandler is a python log handler that handles and reads task instance logs.
@@ -484,17 +509,7 @@ def _prepare_log_folder(self, directory: Path):
             conf.get("logging", "file_task_handler_new_folder_permissions", fallback="0o775"), 8
         )
         directory.mkdir(mode=new_folder_permissions, parents=True, exist_ok=True)
-        if directory.stat().st_mode % 0o1000 != new_folder_permissions % 0o1000:
-            print(f"Changing {directory} permission to {new_folder_permissions}")
-            try:
-                directory.chmod(new_folder_permissions)
-            except PermissionError as e:
-                # In some circumstances (depends on user and filesystem) we might not be able to
-                # change the permission for the folder (when the folder was created by another user
-                # before or when the filesystem does not allow to change permission). We should not
-                # fail in this case but rather ignore it.
-                print(f"Failed to change {directory} permission to {new_folder_permissions}: {e}")
-                pass
+        _change_directory_permissions_up(directory, new_folder_permissions)
 
     def _init_file(self, ti, *, identifier: str | None = None):
         """

@@ -21,6 +21,8 @@
 import logging.config
 import os
 import re
+import shutil
+import tempfile
 from pathlib import Path
 from unittest import mock
 from unittest.mock import patch
@@ -40,6 +42,7 @@
 from airflow.utils.log.file_task_handler import (
     FileTaskHandler,
     LogType,
+    _change_directory_permissions_up,
     _interleave_logs,
     _parse_timestamps_in_log_file,
 )
@@ -719,3 +722,30 @@ def test_interleave_logs_correct_ordering():
     """
 
     assert sample_with_dupe == "\n".join(_interleave_logs(sample_with_dupe, "", sample_with_dupe))
+
+
+def test_permissions_for_new_directories():
+    tmp_path = Path(tempfile.mkdtemp())
+    try:
+        # Set umask to 0o027: owner rwx, group rx-w, other -rwx
+        old_umask = os.umask(0o027)
+        try:
+            subdir = tmp_path / "subdir1" / "subdir2"
+            # force permissions for the new folder to be owner rwx, group -rxw, other -rwx
+            new_folder_permissions = 0o700
+            # default permissions are owner rwx, group rx-w, other -rwx (umask bit negative)
+            default_permissions = 0o750
+            subdir.mkdir(mode=new_folder_permissions, parents=True, exist_ok=True)
+            assert subdir.exists()
+            assert subdir.is_dir()
+            assert subdir.stat().st_mode % 0o1000 == new_folder_permissions
+            # initially parent permissions are as per umask
+            assert subdir.parent.stat().st_mode % 0o1000 == default_permissions
+            _change_directory_permissions_up(subdir, new_folder_permissions)
+            assert subdir.stat().st_mode % 0o1000 == new_folder_permissions
+            # now parent permissions are as per new_folder_permissions
+            assert subdir.parent.stat().st_mode % 0o1000 == new_folder_permissions
+        finally:
+            os.umask(old_umask)
+    finally:
+        shutil.rmtree(tmp_path)