Secure tempfiles

ansible_runner/config/_base.py

@@ -20,8 +20,8 @@
 import os
 import pexpect
 import re
+import tempfile
 
-from tempfile import gettempdir
 from uuid import uuid4
 try:
     from collections.abc import Mapping
@@ -88,8 +88,12 @@ def __init__(self,
         # setup initial environment
         if private_data_dir:
             self.private_data_dir = os.path.abspath(private_data_dir)
+            # Note that os.makedirs, exist_ok=True is dangerous.  If there's a directory writable
+            # by someone other than the user anywhere in the path to be created, an attacker can
+            # attempt to compromise the directories via a race.
+            os.makedirs(self.private_data_dir, exist_ok=True, mode=0o700)
         else:
-            self.private_data_dir = os.path.join(gettempdir(), ".ansible-runner")
+            self.private_data_dir = tempfile.mkdtemp(prefix=".ansible-runner-")
 
         if artifact_dir is None:
             artifact_dir = os.path.join(self.private_data_dir, 'artifacts')
@@ -120,7 +124,6 @@ def __init__(self,
         else:
             self.cwd = os.getcwd()
 
-        os.makedirs(self.private_data_dir, exist_ok=True, mode=0o700)
         os.makedirs(self.artifact_dir, exist_ok=True, mode=0o700)
 
     _CONTAINER_ENGINES = ('docker', 'podman')

ansible_runner/streaming.py

@@ -73,7 +73,7 @@ def __init__(self, _input=None, _output=None, **kwargs):
 
         private_data_dir = kwargs.get('private_data_dir')
         if private_data_dir is None:
-            private_data_dir = tempfile.TemporaryDirectory().name
+            private_data_dir = tempfile.mkdtemp()
         self.private_data_dir = private_data_dir
 
         self.status = "unstarted"
@@ -164,7 +164,7 @@ def __init__(self, _input=None, status_handler=None, event_handler=None,
 
         private_data_dir = kwargs.get('private_data_dir')
         if private_data_dir is None:
-            private_data_dir = tempfile.TemporaryDirectory().name
+            private_data_dir = tempfile.mkdtemp()
         self.private_data_dir = private_data_dir
         self._loader = ArtifactLoader(self.private_data_dir)
 

test/unit/config/test__base.py

@@ -3,13 +3,13 @@
 from functools import partial
 import os
 import re
+from tempfile import gettempdir
 
 import six
 from pexpect import TIMEOUT, EOF
 
 import pytest
 
-from tempfile import gettempdir
 from unittest.mock import (Mock, patch, PropertyMock)
 
 from ansible_runner.config._base import BaseConfig, BaseExecutionMode
@@ -47,7 +47,12 @@ def test_base_config_init_defaults():
 def test_base_config_with_artifact_dir():
     rc = BaseConfig(artifact_dir='/tmp/this-is-some-dir')
     assert rc.artifact_dir == os.path.join('/tmp/this-is-some-dir', rc.ident)
-    assert rc.private_data_dir == os.path.join(gettempdir(), ".ansible-runner")
+
+    # Check that the private data dir is placed in our default location with our default prefix
+    # and has some extra uniqueness on the end.
+    base_private_data_dir = os.path.join(gettempdir(), '.ansible-runner-')
+    assert rc.private_data_dir.startswith(base_private_data_dir)
+    assert len(rc.private_data_dir) > len(base_private_data_dir)
 
 
 def test_base_config_init_with_ident():

test/unit/config/test_ansible_cfg.py

@@ -13,7 +13,13 @@
 
 def test_ansible_cfg_init_defaults():
     rc = AnsibleCfgConfig()
-    assert rc.private_data_dir == os.path.join(gettempdir(), ".ansible-runner")
+
+    # Check that the private data dir is placed in our default location with our default prefix
+    # and has some extra uniqueness on the end.
+    base_private_data_dir = os.path.join(gettempdir(), '.ansible-runner-')
+    assert rc.private_data_dir.startswith(base_private_data_dir)
+    assert len(rc.private_data_dir) > len(base_private_data_dir)
+
     assert rc.execution_mode == BaseExecutionMode.ANSIBLE_COMMANDS
 
 

test/unit/config/test_command.py

@@ -12,7 +12,13 @@
 
 def test_ansible_config_defaults():
     rc = CommandConfig()
-    assert rc.private_data_dir == os.path.join(gettempdir(), ".ansible-runner")
+
+    # Check that the private data dir is placed in our default location with our default prefix
+    # and has some extra uniqueness on the end.
+    base_private_data_dir = os.path.join(gettempdir(), '.ansible-runner-')
+    assert rc.private_data_dir.startswith(base_private_data_dir)
+    assert len(rc.private_data_dir) > len(base_private_data_dir)
+
     assert rc.execution_mode == BaseExecutionMode.NONE
     assert rc.runner_mode is None
 

test/unit/config/test_container_volmount_generation.py

@@ -70,25 +70,27 @@ def generate_volmount_args(src_str, dst_str, labels):
     return ["-v", vol_mount_str]
 
 
-@patch("os.path.exists", return_value=True)
 @pytest.mark.parametrize("not_safe", not_safe)
-def test_check_not_safe_to_mount_dir(_mock_ope, not_safe):
+def test_check_not_safe_to_mount_dir(not_safe):
     """Ensure unsafe directories are not mounted"""
     with pytest.raises(ConfigurationError):
-        BaseConfig()._update_volume_mount_paths(
-            args_list=[], src_mount_path=not_safe, dst_mount_path=None
-        )
+        bc = BaseConfig()
+        with patch("os.path.exists", return_value=True):
+            bc._update_volume_mount_paths(
+                args_list=[], src_mount_path=not_safe, dst_mount_path=None
+            )
 
 
-@patch("os.path.exists", return_value=True)
 @pytest.mark.parametrize("not_safe", not_safe)
-def test_check_not_safe_to_mount_file(_mock_ope, not_safe):
+def test_check_not_safe_to_mount_file(not_safe):
     """Ensure unsafe directories for a given file are not mounted"""
     file_path = os.path.join(not_safe, "file.txt")
     with pytest.raises(ConfigurationError):
-        BaseConfig()._update_volume_mount_paths(
-            args_list=[], src_mount_path=file_path, dst_mount_path=None
-        )
+        bc = BaseConfig()
+        with patch("os.path.exists", return_value=True):
+            bc._update_volume_mount_paths(
+                args_list=[], src_mount_path=file_path, dst_mount_path=None
+            )
 
 
 @patch("os.path.exists", return_value=True)
@@ -166,10 +168,9 @@ def test_src_dst_all_dirs(_mock_ope, _mock_isdir, src, dst, labels):
 
 
 
-@patch("os.path.exists", return_value=True)
 @pytest.mark.parametrize("labels", labels, ids=id_for_label)
 @pytest.mark.parametrize("path", dir_variations, ids=id_for_src)
-def test_src_dst_all_files(_mock_ope, path, labels):
+def test_src_dst_all_files(path, labels):
     """Ensure file paths are tranformed correctly into dir paths"""
     src_str = os.path.join(resolve_path(path.path), "")
     dst_str = src_str
@@ -180,10 +181,11 @@ def test_src_dst_all_files(_mock_ope, path, labels):
     dest_file = src_file
 
     base_config = BaseConfig()
-    with patch("os.path.isdir", return_value=False):
-        base_config._update_volume_mount_paths(
-            args_list=result, src_mount_path=src_file, dst_mount_path=dest_file, labels=labels
-        )
+    with patch("os.path.exists", return_value=True):
+        with patch("os.path.isdir", return_value=False):
+            base_config._update_volume_mount_paths(
+                args_list=result, src_mount_path=src_file, dst_mount_path=dest_file, labels=labels
+            )
 
     explanation = (
         f"provided: {src_file}:{dest_file}",

test/unit/config/test_doc.py

@@ -13,7 +13,13 @@
 
 def test_ansible_doc_defaults():
     rc = DocConfig()
-    assert rc.private_data_dir == os.path.join(gettempdir(), ".ansible-runner")
+
+    # Check that the private data dir is placed in our default location with our default prefix
+    # and has some extra uniqueness on the end.
+    base_private_data_dir = os.path.join(gettempdir(), '.ansible-runner-')
+    assert rc.private_data_dir.startswith(base_private_data_dir)
+    assert len(rc.private_data_dir) > len(base_private_data_dir)
+
     assert rc.execution_mode == BaseExecutionMode.ANSIBLE_COMMANDS
     assert rc.runner_mode == 'subprocess'
 

test/unit/config/test_inventory.py

@@ -13,7 +13,13 @@
 
 def test_ansible_inventory_init_defaults():
     rc = InventoryConfig()
-    assert rc.private_data_dir == os.path.join(gettempdir(), ".ansible-runner")
+
+    # Check that the private data dir is placed in our default location with our default prefix
+    # and has some extra uniqueness on the end.
+    base_private_data_dir = os.path.join(gettempdir(), '.ansible-runner-')
+    assert rc.private_data_dir.startswith(base_private_data_dir)
+    assert len(rc.private_data_dir) > len(base_private_data_dir)
+
     assert rc.execution_mode == BaseExecutionMode.ANSIBLE_COMMANDS