diff --git a/changelogs/fragments/vault_tmp_file.yml b/changelogs/fragments/vault_tmp_file.yml
new file mode 100644
index 00000000000..1eaea6f3e7f
--- /dev/null
+++ b/changelogs/fragments/vault_tmp_file.yml
@@ -0,0 +1,2 @@
+bugfixes:
+    - Ensure DataLoader temp files are removed at appropriate times and that we observe the LOCAL_TMP setting. 
diff --git a/lib/ansible/executor/process/worker.py b/lib/ansible/executor/process/worker.py
index d01ff7f5cc5..0b18fc351f3 100644
--- a/lib/ansible/executor/process/worker.py
+++ b/lib/ansible/executor/process/worker.py
@@ -67,6 +67,10 @@ class WorkerProcess(multiprocessing_context.Process):
         self._variable_manager = variable_manager
         self._shared_loader_obj = shared_loader_obj
 
+        # NOTE: this works due to fork, if switching to threads this should change to per thread storage of temp files
+        # clear var to ensure we only delete files for this child
+        self._loader._tempfiles = set()
+
     def _save_stdin(self):
         self._new_stdin = os.devnull
         try:
@@ -200,6 +204,8 @@ class WorkerProcess(multiprocessing_context.Process):
                 except Exception:
                     display.debug(u"WORKER EXCEPTION: %s" % to_text(e))
                     display.debug(u"WORKER TRACEBACK: %s" % to_text(traceback.format_exc()))
+                finally:
+                    self._clean_up()
 
         display.debug("WORKER PROCESS EXITING")
 
@@ -210,3 +216,8 @@ class WorkerProcess(multiprocessing_context.Process):
         # ps.print_stats()
         # with open('worker_%06d.stats' % os.getpid(), 'w') as f:
         #     f.write(s.getvalue())
+
+    def _clean_up(self):
+        # NOTE: see note in init about forks
+        # ensure we cleanup all temp files for this worker
+        self._loader.cleanup_all_tmp_files()
diff --git a/lib/ansible/parsing/dataloader.py b/lib/ansible/parsing/dataloader.py
index 3e8bce70968..41095147346 100644
--- a/lib/ansible/parsing/dataloader.py
+++ b/lib/ansible/parsing/dataloader.py
@@ -51,8 +51,16 @@ class DataLoader:
     '''
 
     def __init__(self):
+
         self._basedir = '.'
+
+        # NOTE: not effective with forks as the main copy does not get updated.
+        # avoids rereading files
         self._FILE_CACHE = dict()
+
+        # NOTE: not thread safe, also issues with forks not returning data to main proc
+        #       so they need to be cleaned independantly. See WorkerProcess for example.
+        # used to keep track of temp files for cleaning
         self._tempfiles = set()
 
         # initialize the vault stuff with an empty password
@@ -322,7 +330,7 @@ class DataLoader:
 
     def _create_content_tempfile(self, content):
         ''' Create a tempfile containing defined content '''
-        fd, content_tempfile = tempfile.mkstemp()
+        fd, content_tempfile = tempfile.mkstemp(dir=C.DEFAULT_LOCAL_TMP)
         f = os.fdopen(fd, 'wb')
         content = to_bytes(content)
         try:
@@ -385,6 +393,10 @@ class DataLoader:
             self._tempfiles.remove(file_path)
 
     def cleanup_all_tmp_files(self):
+        """
+        Removes all temporary files that DataLoader has created
+        NOTE: not thread safe, forks also need special handling see __init__ for details.
+        """
         for f in self._tempfiles:
             try:
                 self.cleanup_tmp_file(f)
diff --git a/lib/ansible/parsing/vault/__init__.py b/lib/ansible/parsing/vault/__init__.py
index 566d6999758..dde1fecb236 100644
--- a/lib/ansible/parsing/vault/__init__.py
+++ b/lib/ansible/parsing/vault/__init__.py
@@ -850,7 +850,7 @@ class VaultEditor:
 
         # Create a tempfile
         root, ext = os.path.splitext(os.path.realpath(filename))
-        fd, tmp_path = tempfile.mkstemp(suffix=ext)
+        fd, tmp_path = tempfile.mkstemp(suffix=ext, dir=C.DEFAULT_LOCAL_TMP)
         os.close(fd)
 
         cmd = self._editor_shell_command(tmp_path)
diff --git a/test/integration/targets/vault/files/test_assemble/nonsecret.txt b/test/integration/targets/vault/files/test_assemble/nonsecret.txt
new file mode 100644
index 00000000000..320b6b4ca0e
--- /dev/null
+++ b/test/integration/targets/vault/files/test_assemble/nonsecret.txt
@@ -0,0 +1 @@
+THIS IS OK
diff --git a/test/integration/targets/vault/files/test_assemble/secret.vault b/test/integration/targets/vault/files/test_assemble/secret.vault
new file mode 100644
index 00000000000..fd278564916
--- /dev/null
+++ b/test/integration/targets/vault/files/test_assemble/secret.vault
@@ -0,0 +1,7 @@
+$ANSIBLE_VAULT;1.1;AES256
+37626439373465656332623633333336353334326531333666363766303339336134313136616165
+6561333963343739386334653636393363396366396338660a663537666561643862343233393265
+33336436633864323935356337623861663631316530336532633932623635346364363338363437
+3365313831366365350a613934313862313538626130653539303834656634353132343065633162
+34316135313837623735653932663139353164643834303534346238386435373832366564646236
+3461333465343434666639373432366139363566303564643066
diff --git a/test/integration/targets/vault/runme.sh b/test/integration/targets/vault/runme.sh
index c4d17dbd26e..c8ecde9ba03 100755
--- a/test/integration/targets/vault/runme.sh
+++ b/test/integration/targets/vault/runme.sh
@@ -517,3 +517,7 @@ ansible-playbook "$@" -i invalid_format/inventory --vault-id invalid_format/vaul
 
 # Run playbook with vault file with unicode in filename (https://github.com/ansible/ansible/issues/50316)
 ansible-playbook -i ../../inventory -v "$@" --vault-password-file vault-password test_utf8_value_in_filename.yml
+
+# Ensure we don't leave unencrypted temp files dangling
+ansible-playbook -v "$@" --vault-password-file vault-password test_dangling_temp.yml
+
diff --git a/test/integration/targets/vault/test_dangling_temp.yml b/test/integration/targets/vault/test_dangling_temp.yml
new file mode 100644
index 00000000000..71a9d73aaf6
--- /dev/null
+++ b/test/integration/targets/vault/test_dangling_temp.yml
@@ -0,0 +1,34 @@
+- hosts: localhost
+  gather_facts: False
+  vars:
+    od: "{{output_dir|default('/tmp')}}/test_vault_assemble"
+  tasks:
+    - name: create target directory
+      file:
+        path: "{{od}}"
+        state: directory
+
+    - name: assemble_file file with secret
+      assemble:
+        src: files/test_assemble
+        dest: "{{od}}/dest_file"
+        remote_src: no
+        mode: 0600
+
+    - name: remove assembled file with secret (so nothing should have unencrypted secret)
+      file: path="{{od}}/dest_file" state=absent
+
+    - name: find temp files with secrets
+      find:
+        paths: '{{temp_paths}}'
+        contains: 'VAULT TEST IN WHICH BAD THING HAPPENED'
+        recurse: yes
+      register: badthings
+      vars:
+        temp_paths: "{{[lookup('env', 'TMP'), lookup('env', 'TEMP'), hardcoded]|flatten(1)|unique|list}}"
+        hardcoded: ['/tmp', '/var/tmp']
+
+    - name: ensure we failed to find any
+      assert:
+        that:
+            - badthings['matched'] == 0