Fix remote_tmp when become with non admin user (#42396)

* Fix tmpdir on non root become

 - also avoid exception if tmpdir and remote_tmp are None
 - give 'None' on deescalation so tempfile will fallback to it's default behaviour
   and use system dirs
 - fix issue with bad tempdir (not existing/not createable/not writeable)
   i.e nobody and ~/.ansible/tmp
 - added tests for blockfile case

* Revert "Temporarily revert c119d54"

This reverts commit 5c614a59a6.

* changes based on PR feedback and changelog fragment

* changes based on the review

* Fix tmpdir when makedirs failed so we just use the system tmp

* Let missing remote_tmp fail

If remote_tmp is missing then there's something more basic wrong in the
communication from the controller to the module-side.  It's better to
be alerted in this case than to silently ignore it.

jborean and I have independently checked what happens if the user sets
ansible_remote_tmp to empty string and !!null and both cases work fine.
(null is turned into a default value controller-side.  empty string
triggers the warning because it is probably not a directory that the
become user is able to use).

test/integration/targets/remote_tmp/aliases

@@ -0,0 +1 @@
+posix/ci/group3

test/integration/targets/remote_tmp/playbook.yml

@@ -0,0 +1,26 @@
+- name: Test temp dir on de escalation
+  hosts: testhost
+  gather_facts: false
+  become: yes
+  tasks:
+    - name: create test user
+      user: name=tmptest state=present
+
+    - name: execute test case
+      become_user: tmptest
+      block:
+        - name: Test case from issue 41340
+          blockinfile:
+              create: yes
+              block: |
+                export foo=bar
+              marker: "# {mark} Here there be a marker"
+              dest: /tmp/testing.txt
+              mode: 0644
+      always:
+        - name: clean up file
+          file: path=/tmp/testing.txt state=absent
+
+        - name: clean up test user
+          user: name=tmptest state=absent
+          become_user: root

test/integration/targets/remote_tmp/runme.sh

@@ -0,0 +1,5 @@
+#!/usr/bin/env bash
+
+set -ux
+
+ansible-playbook -i ../../inventory playbook.yml -v "$@"

test/units/module_utils/basic/test_tmpdir.py

@@ -12,7 +12,7 @@ import tempfile
 
 import pytest
 
-from ansible.compat.tests.mock import patch
+from ansible.compat.tests.mock import patch, MagicMock
 
 
 class TestAnsibleModuleTmpDir:
@@ -87,3 +87,25 @@ class TestAnsibleModuleTmpDir:
 
         if not stat_exists:
             assert makedirs['called']
+
+    @pytest.mark.parametrize('stdin', ({"_ansible_tmpdir": None,
+                                        "_ansible_remote_tmp": "$HOME/.test",
+                                        "_ansible_keep_remote_files": True},),
+                             indirect=['stdin'])
+    def test_tmpdir_makedirs_failure(self, am, monkeypatch):
+
+        mock_mkdtemp = MagicMock(return_value="/tmp/path")
+        mock_makedirs = MagicMock(side_effect=OSError("Some OS Error here"))
+
+        monkeypatch.setattr(tempfile, 'mkdtemp', mock_mkdtemp)
+        monkeypatch.setattr(os.path, 'exists', lambda x: False)
+        monkeypatch.setattr(os, 'makedirs', mock_makedirs)
+
+        actual = am.tmpdir
+        assert actual == "/tmp/path"
+        assert mock_makedirs.call_args[0] == (os.path.expanduser(os.path.expandvars("$HOME/.test")),)
+        assert mock_makedirs.call_args[1] == {"mode": 0o700}
+
+        # because makedirs failed the dir should be None so it uses the System tmp
+        assert mock_mkdtemp.call_args[1]['dir'] is None
+        assert mock_mkdtemp.call_args[1]['prefix'].startswith("ansible-moduletmp-")

test/units/plugins/action/test_action.py

@@ -415,7 +415,7 @@ class TestActionBase(unittest.TestCase):
             return " ".join(to_run)
 
         def get_option(option):
-            return {}.get(option)
+            return {'admin_users': ['root', 'toor']}.get(option)
 
         mock_connection = MagicMock()
         mock_connection.build_module_command.side_effect = build_module_command

test/units/plugins/action/test_raw.py

@@ -18,11 +18,19 @@
 from __future__ import (absolute_import, division, print_function)
 __metaclass__ = type
 
+import os
+
 from ansible.errors import AnsibleActionFail
 from ansible.compat.tests import unittest
 from ansible.compat.tests.mock import patch, MagicMock, Mock
 from ansible.plugins.action.raw import ActionModule
 from ansible.playbook.task import Task
+from ansible.plugins.loader import connection_loader
+
+
+play_context = Mock()
+play_context.shell = 'sh'
+connection = connection_loader.get('local', play_context, os.devnull)
 
 
 class TestCopyResultExclude(unittest.TestCase):
@@ -41,10 +49,8 @@ class TestCopyResultExclude(unittest.TestCase):
 
     def test_raw_executable_is_not_empty_string(self):
 
-        play_context = Mock()
         task = MagicMock(Task)
         task.async_val = False
-        connection = Mock()
 
         task.args = {'_raw_params': 'Args1'}
         play_context.check_mode = False
@@ -52,16 +58,15 @@ class TestCopyResultExclude(unittest.TestCase):
         self.mock_am = ActionModule(task, connection, play_context, loader=None, templar=None, shared_loader_obj=None)
         self.mock_am._low_level_execute_command = Mock(return_value={})
         self.mock_am.display = Mock()
+        self.mock_am._admin_users = ['root', 'toor']
 
         self.mock_am.run()
         self.mock_am._low_level_execute_command.assert_called_with('Args1', executable=False)
 
     def test_raw_check_mode_is_True(self):
 
-        play_context = Mock()
         task = MagicMock(Task)
         task.async_val = False
-        connection = Mock()
 
         task.args = {'_raw_params': 'Args1'}
         play_context.check_mode = True
@@ -73,10 +78,8 @@ class TestCopyResultExclude(unittest.TestCase):
 
     def test_raw_test_environment_is_None(self):
 
-        play_context = Mock()
         task = MagicMock(Task)
         task.async_val = False
-        connection = Mock()
 
         task.args = {'_raw_params': 'Args1'}
         task.environment = None
@@ -90,10 +93,8 @@ class TestCopyResultExclude(unittest.TestCase):
 
     def test_raw_task_vars_is_not_None(self):
 
-        play_context = Mock()
         task = MagicMock(Task)
         task.async_val = False
-        connection = Mock()
 
         task.args = {'_raw_params': 'Args1'}
         task.environment = None

test/units/plugins/action/test_synchronize.py

@@ -67,10 +67,13 @@ class ConnectionMock(object):
     transport = None
     _new_stdin = StdinMock()
 
+    get_option = MagicMock(return_value='root')
+
     # my shell
     _shell = MagicMock()
     _shell.mkdtemp.return_value = 'mkdir command'
     _shell.join_path.side_effect = os.path.join
+    _shell.get_option = MagicMock(return_value=['root', 'toor'])
 
 
 class PlayContextMock(object):