diff --git a/changelogs/fragments/remote_tmp_de-escalated_user.yaml b/changelogs/fragments/remote_tmp_de-escalated_user.yaml
new file mode 100644
index 00000000000..32376373d31
--- /dev/null
+++ b/changelogs/fragments/remote_tmp_de-escalated_user.yaml
@@ -0,0 +1,2 @@
+bugfixes:
+- fix the remote tmp folder permissions issue when becoming a non admin user - https://github.com/ansible/ansible/issues/41340, https://github.com/ansible/ansible/issues/42117
diff --git a/lib/ansible/module_utils/basic.py b/lib/ansible/module_utils/basic.py
index 8e04f134db2..65adce93194 100644
--- a/lib/ansible/module_utils/basic.py
+++ b/lib/ansible/module_utils/basic.py
@@ -934,20 +934,35 @@ class AnsibleModule(object):
 
     @property
     def tmpdir(self):
-        # if _ansible_tmpdir was not set, the module needs to create it and
-        # clean it up once finished.
+        # if _ansible_tmpdir was not set and we have a remote_tmp,
+        # the module needs to create it and clean it up once finished.
+        # otherwise we create our own module tmp dir from the system defaults
         if self._tmpdir is None:
+            basedir = None
+
             basedir = os.path.expanduser(os.path.expandvars(self._remote_tmp))
             if not os.path.exists(basedir):
-                self.warn("Module remote_tmp %s did not exist and was created "
-                          "with a mode of 0700, this may cause issues when "
-                          "running as another user. To avoid this, create the "
-                          "remote_tmp dir with the correct permissions "
-                          "manually" % basedir)
-                os.makedirs(basedir, mode=0o700)
+                try:
+                    os.makedirs(basedir, mode=0o700)
+                except (OSError, IOError) as e:
+                    self.warn("Unable to use %s as temporary directory, "
+                              "failing back to system: %s" % (basedir, to_native(e)))
+                    basedir = None
+                else:
+                    self.warn("Module remote_tmp %s did not exist and was "
+                              "created with a mode of 0700, this may cause"
+                              " issues when running as another user. To "
+                              "avoid this, create the remote_tmp dir with "
+                              "the correct permissions manually" % basedir)
 
             basefile = "ansible-moduletmp-%s-" % time.time()
-            tmpdir = tempfile.mkdtemp(prefix=basefile, dir=basedir)
+            try:
+                tmpdir = tempfile.mkdtemp(prefix=basefile, dir=basedir)
+            except (OSError, IOError) as e:
+                self.fail_json(
+                    msg="Failed to create remote module tmp path at dir %s "
+                        "with prefix %s: %s" % (basedir, basefile, to_native(e))
+                )
             if not self._keep_remote_files:
                 atexit.register(shutil.rmtree, tmpdir)
             self._tmpdir = tmpdir
diff --git a/lib/ansible/modules/files/assemble.py b/lib/ansible/modules/files/assemble.py
index 49696753812..b4a74f104dc 100644
--- a/lib/ansible/modules/files/assemble.py
+++ b/lib/ansible/modules/files/assemble.py
@@ -107,9 +107,9 @@ from ansible.module_utils.six import b
 from ansible.module_utils._text import to_native
 
 
-def assemble_from_fragments(src_path, delimiter=None, compiled_regexp=None, ignore_hidden=False):
+def assemble_from_fragments(src_path, delimiter=None, compiled_regexp=None, ignore_hidden=False, tmpdir=None):
     ''' assemble a file from a directory of fragments '''
-    tmpfd, temp_path = tempfile.mkstemp()
+    tmpfd, temp_path = tempfile.mkstemp(dir=tmpdir)
     tmp = os.fdopen(tmpfd, 'wb')
     delimit_me = False
     add_newline = False
@@ -204,7 +204,7 @@ def main():
     if validate and "%s" not in validate:
         module.fail_json(msg="validate must contain %%s: %s" % validate)
 
-    path = assemble_from_fragments(src, delimiter, compiled_regexp, ignore_hidden)
+    path = assemble_from_fragments(src, delimiter, compiled_regexp, ignore_hidden, module.tmpdir)
     path_hash = module.sha1(path)
     result['checksum'] = path_hash
 
diff --git a/lib/ansible/modules/files/blockinfile.py b/lib/ansible/modules/files/blockinfile.py
index 27dfa2ef2b3..26beb497d5e 100644
--- a/lib/ansible/modules/files/blockinfile.py
+++ b/lib/ansible/modules/files/blockinfile.py
@@ -160,7 +160,7 @@ from ansible.module_utils._text import to_bytes
 
 def write_changes(module, contents, path):
 
-    tmpfd, tmpfile = tempfile.mkstemp()
+    tmpfd, tmpfile = tempfile.mkstemp(dir=module.tmpdir)
     f = os.fdopen(tmpfd, 'wb')
     f.write(contents)
     f.close()
diff --git a/lib/ansible/modules/files/replace.py b/lib/ansible/modules/files/replace.py
index 1a652b6b45e..30615b70087 100644
--- a/lib/ansible/modules/files/replace.py
+++ b/lib/ansible/modules/files/replace.py
@@ -154,7 +154,7 @@ from ansible.module_utils.basic import AnsibleModule
 
 def write_changes(module, contents, path):
 
-    tmpfd, tmpfile = tempfile.mkstemp()
+    tmpfd, tmpfile = tempfile.mkstemp(dir=module.tmpdir)
     f = os.fdopen(tmpfd, 'wb')
     f.write(contents)
     f.close()
diff --git a/lib/ansible/modules/net_tools/basics/get_url.py b/lib/ansible/modules/net_tools/basics/get_url.py
index 89c69994ed1..65d23548ba6 100644
--- a/lib/ansible/modules/net_tools/basics/get_url.py
+++ b/lib/ansible/modules/net_tools/basics/get_url.py
@@ -49,7 +49,8 @@ options:
   tmp_dest:
     description:
       - Absolute path of where temporary file is downloaded to.
-      - Defaults to C(TMPDIR), C(TEMP) or C(TMP) env variables or a platform specific value.
+      - When run on Ansible 2.5 or greater, path defaults to ansible's remote_tmp setting
+      - When run on Ansible prior to 2.5, it defaults to C(TMPDIR), C(TEMP) or C(TMP) env variables or a platform specific value.
       - U(https://docs.python.org/2/library/tempfile.html#tempfile.tempdir)
     version_added: '2.1'
   force:
@@ -340,18 +341,17 @@ def url_get(module, url, dest, use_proxy, last_mod_time, force, timeout=10, head
                 module.fail_json(msg="%s is a file but should be a directory." % tmp_dest)
             else:
                 module.fail_json(msg="%s directory does not exist." % tmp_dest)
-
-        fd, tempname = tempfile.mkstemp(dir=tmp_dest)
     else:
-        fd, tempname = tempfile.mkstemp()
+        tmp_dest = module.tmpdir
+
+    fd, tempname = tempfile.mkstemp(dir=tmp_dest)
 
     f = os.fdopen(fd, 'wb')
     try:
         shutil.copyfileobj(rsp, f)
     except Exception as e:
         os.remove(tempname)
-        module.fail_json(msg="failed to create temporary content file: %s" % to_native(e),
-                         exception=traceback.format_exc())
+        module.fail_json(msg="failed to create temporary content file: %s" % to_native(e), exception=traceback.format_exc())
     f.close()
     rsp.close()
     return tempname, info
diff --git a/lib/ansible/modules/net_tools/basics/uri.py b/lib/ansible/modules/net_tools/basics/uri.py
index a7d87607576..727ea80a2d2 100644
--- a/lib/ansible/modules/net_tools/basics/uri.py
+++ b/lib/ansible/modules/net_tools/basics/uri.py
@@ -252,7 +252,7 @@ JSON_CANDIDATES = ('text', 'json', 'javascript')
 
 def write_file(module, url, dest, content):
     # create a tempfile with some test content
-    fd, tmpsrc = tempfile.mkstemp()
+    fd, tmpsrc = tempfile.mkstemp(dir=module.tmpdir)
     f = open(tmpsrc, 'wb')
     try:
         f.write(content)
diff --git a/lib/ansible/modules/packaging/os/yum.py b/lib/ansible/modules/packaging/os/yum.py
index e0577ee8fa9..da6aa8b65e4 100644
--- a/lib/ansible/modules/packaging/os/yum.py
+++ b/lib/ansible/modules/packaging/os/yum.py
@@ -333,7 +333,7 @@ def ensure_yum_utils(module):
 def fetch_rpm_from_url(spec, module=None):
     # download package so that we can query it
     package_name, _ = os.path.splitext(str(spec.rsplit('/', 1)[1]))
-    package_file = tempfile.NamedTemporaryFile(prefix=package_name, suffix='.rpm', delete=False)
+    package_file = tempfile.NamedTemporaryFile(dir=module.tmpdir, prefix=package_name, suffix='.rpm', delete=False)
     module.add_cleanup_file(package_file.name)
     try:
         rsp, info = fetch_url(module, spec)
diff --git a/lib/ansible/plugins/action/__init__.py b/lib/ansible/plugins/action/__init__.py
index 5f8529424b9..9bc27931ebc 100644
--- a/lib/ansible/plugins/action/__init__.py
+++ b/lib/ansible/plugins/action/__init__.py
@@ -226,18 +226,43 @@ class ActionBase(with_metaclass(ABCMeta, object)):
 
         return True
 
-    def _make_tmp_path(self, remote_user=None):
+    def _get_admin_users(self):
         '''
-        Create and return a temporary path on a remote box.
+        Returns a list of admin users that are configured for the current shell
+        plugin
         '''
+        try:
+            admin_users = self._connection._shell.get_option('admin_users')
+        except AnsibleError:
+            # fallback for old custom plugins w/o get_option
+            admin_users = ['root']
+        return admin_users
 
-        if remote_user is None:
-            remote_user = self._play_context.remote_user
-
+    def _is_become_unprivileged(self):
+        '''
+        The user is not the same as the connection user and is not part of the
+        shell configured admin users
+        '''
+        # if we don't use become then we know we aren't switching to a
+        # different unprivileged user
+        if not self._play_context.become:
+            return False
+
+        # if we use become and the user is not an admin (or same user) then
+        # we need to return become_unprivileged as True
+        admin_users = self._get_admin_users()
         try:
-            admin_users = self._connection._shell.get_option('admin_users') + [remote_user]
+            remote_user = self._connection.get_option('remote_user')
         except AnsibleError:
-            admin_users = ['root', remote_user]  # plugin does not support admin_users
+            remote_user = self._play_context.remote_user
+        return bool(self._play_context.become_user not in admin_users + [remote_user])
+
+    def _make_tmp_path(self, remote_user=None):
+        '''
+        Create and return a temporary path on a remote box.
+        '''
+
+        become_unprivileged = self._is_become_unprivileged()
         try:
             remote_tmp = self._connection._shell.get_option('remote_tmp')
         except AnsibleError:
@@ -245,7 +270,6 @@ class ActionBase(with_metaclass(ABCMeta, object)):
 
         # deal with tmpdir creation
         basefile = 'ansible-tmp-%s-%s' % (time.time(), random.randint(0, 2**48))
-        use_system_tmp = bool(self._play_context.become and self._play_context.become_user not in admin_users)
         # Network connection plugins (network_cli, netconf, etc.) execute on the controller, rather than the remote host.
         # As such, we want to avoid using remote_user for paths  as remote_user may not line up with the local user
         # This is a hack and should be solved by more intelligent handling of remote_tmp in 2.7
@@ -253,7 +277,7 @@ class ActionBase(with_metaclass(ABCMeta, object)):
             tmpdir = C.DEFAULT_LOCAL_TMP
         else:
             tmpdir = self._remote_expand_user(remote_tmp, sudoable=False)
-        cmd = self._connection._shell.mkdtemp(basefile=basefile, system=use_system_tmp, tmpdir=tmpdir)
+        cmd = self._connection._shell.mkdtemp(basefile=basefile, system=become_unprivileged, tmpdir=tmpdir)
         result = self._low_level_execute_command(cmd, sudoable=False)
 
         # error handling on this seems a little aggressive?
@@ -297,7 +321,7 @@ class ActionBase(with_metaclass(ABCMeta, object)):
 
         self._connection._shell.tmpdir = rc
 
-        if not use_system_tmp:
+        if not become_unprivileged:
             self._connection._shell.env.update({'ANSIBLE_REMOTE_TMP': self._connection._shell.tmpdir})
         return rc
 
@@ -409,12 +433,7 @@ class ActionBase(with_metaclass(ABCMeta, object)):
             # we have a need for it, at which point we'll have to do something different.
             return remote_paths
 
-        try:
-            admin_users = self._connection._shell.get_option('admin_users')
-        except AnsibleError:
-            admin_users = ['root']  # plugin does not support admin users
-
-        if self._play_context.become and self._play_context.become_user and self._play_context.become_user not in admin_users + [remote_user]:
+        if self._is_become_unprivileged():
             # Unprivileged user that's different than the ssh user.  Let's get
             # to work!
 
@@ -441,8 +460,8 @@ class ActionBase(with_metaclass(ABCMeta, object)):
                         raise AnsibleError('Failed to set file mode on remote temporary files (rc: {0}, err: {1})'.format(res['rc'], to_native(res['stderr'])))
 
                 res = self._remote_chown(remote_paths, self._play_context.become_user)
-                if res['rc'] != 0 and remote_user in admin_users:
-                    # chown failed even if remove_user is root
+                if res['rc'] != 0 and remote_user in self._get_admin_users():
+                    # chown failed even if remote_user is administrator/root
                     raise AnsibleError('Failed to change ownership of the temporary files Ansible needs to create despite connecting as a privileged user. '
                                        'Unprivileged become user would be unable to read the file.')
                 elif res['rc'] != 0:
@@ -665,7 +684,10 @@ class ActionBase(with_metaclass(ABCMeta, object)):
         module_args['_ansible_keep_remote_files'] = C.DEFAULT_KEEP_REMOTE_FILES
 
         # make sure all commands use the designated temporary directory if created
-        module_args['_ansible_tmpdir'] = self._connection._shell.tmpdir
+        if self._is_become_unprivileged():  # force fallback on remote_tmp as user cannot normally write to dir
+            module_args['_ansible_tmpdir'] = None
+        else:
+            module_args['_ansible_tmpdir'] = self._connection._shell.tmpdir
 
         # make sure the remote_tmp value is sent through in case modules needs to create their own
         try:
diff --git a/test/integration/targets/remote_tmp/aliases b/test/integration/targets/remote_tmp/aliases
new file mode 100644
index 00000000000..79d8b9285eb
--- /dev/null
+++ b/test/integration/targets/remote_tmp/aliases
@@ -0,0 +1 @@
+posix/ci/group3
diff --git a/test/integration/targets/remote_tmp/playbook.yml b/test/integration/targets/remote_tmp/playbook.yml
new file mode 100644
index 00000000000..46a2846590c
--- /dev/null
+++ b/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
diff --git a/test/integration/targets/remote_tmp/runme.sh b/test/integration/targets/remote_tmp/runme.sh
new file mode 100755
index 00000000000..69efd6e0164
--- /dev/null
+++ b/test/integration/targets/remote_tmp/runme.sh
@@ -0,0 +1,5 @@
+#!/usr/bin/env bash
+
+set -ux
+
+ansible-playbook -i ../../inventory playbook.yml -v "$@"
diff --git a/test/units/module_utils/basic/test_tmpdir.py b/test/units/module_utils/basic/test_tmpdir.py
index 95c864c2cde..2315313253e 100644
--- a/test/units/module_utils/basic/test_tmpdir.py
+++ b/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-")
diff --git a/test/units/plugins/action/test_action.py b/test/units/plugins/action/test_action.py
index df90c06bf65..04f287561f5 100644
--- a/test/units/plugins/action/test_action.py
+++ b/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
diff --git a/test/units/plugins/action/test_raw.py b/test/units/plugins/action/test_raw.py
index f082025d6b1..5e4be9b03b1 100644
--- a/test/units/plugins/action/test_raw.py
+++ b/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
diff --git a/test/units/plugins/action/test_synchronize.py b/test/units/plugins/action/test_synchronize.py
index 23e64e9149d..c669422d180 100644
--- a/test/units/plugins/action/test_synchronize.py
+++ b/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):