diff --git a/changelogs/fragments/fetch_no_slurp.yml b/changelogs/fragments/fetch_no_slurp.yml
new file mode 100644
index 00000000000..c742d40c3ba
--- /dev/null
+++ b/changelogs/fragments/fetch_no_slurp.yml
@@ -0,0 +1,2 @@
+bugfixes:
+    - In fetch action, avoid using slurp return to set up dest, also ensure no dir traversal CVE-2019-3828.
diff --git a/lib/ansible/plugins/action/fetch.py b/lib/ansible/plugins/action/fetch.py
index 475c9df9f6f..eee9cca8184 100644
--- a/lib/ansible/plugins/action/fetch.py
+++ b/lib/ansible/plugins/action/fetch.py
@@ -20,13 +20,13 @@ __metaclass__ = type
 import os
 import base64
 
-from ansible.errors import AnsibleError
+from ansible.errors import AnsibleActionFail, AnsibleActionSkip
 from ansible.module_utils._text import to_bytes
 from ansible.module_utils.six import string_types
 from ansible.module_utils.parsing.convert_bool import boolean
 from ansible.plugins.action import ActionBase
 from ansible.utils.hashing import checksum, checksum_s, md5, secure_hash
-from ansible.utils.path import makedirs_safe
+from ansible.utils.path import makedirs_safe, is_subpath
 
 try:
     from __main__ import display
@@ -47,24 +47,23 @@ class ActionModule(ActionBase):
 
         try:
             if self._play_context.check_mode:
-                result['skipped'] = True
-                result['msg'] = 'check mode not (yet) supported for this module'
-                return result
+                raise AnsibleActionSkip('check mode not (yet) supported for this module')
 
             source = self._task.args.get('src', None)
-            dest = self._task.args.get('dest', None)
+            original_dest = dest = self._task.args.get('dest', None)
             flat = boolean(self._task.args.get('flat'), strict=False)
             fail_on_missing = boolean(self._task.args.get('fail_on_missing', True), strict=False)
             validate_checksum = boolean(self._task.args.get('validate_checksum',
                                                             self._task.args.get('validate_md5', True)),
                                         strict=False)
 
+            msg = ''
             # validate source and dest are strings FIXME: use basic.py and module specs
             if not isinstance(source, string_types):
-                result['msg'] = "Invalid type supplied for source option, it must be a string"
+                msg = "Invalid type supplied for source option, it must be a string"
 
             if not isinstance(dest, string_types):
-                result['msg'] = "Invalid type supplied for dest option, it must be a string"
+                msg = "Invalid type supplied for dest option, it must be a string"
 
             # validate_md5 is the deprecated way to specify validate_checksum
             if 'validate_md5' in self._task.args and 'validate_checksum' in self._task.args:
@@ -74,11 +73,10 @@ class ActionModule(ActionBase):
                 display.deprecated('Use validate_checksum instead of validate_md5', version='2.8')
 
             if source is None or dest is None:
-                result['msg'] = "src and dest are required"
+                msg = "src and dest are required"
 
-            if result.get('msg'):
-                result['failed'] = True
-                return result
+            if msg:
+                raise AnsibleActionFail(msg)
 
             source = self._connection._shell.join_path(source)
             source = self._remote_expand_user(source)
@@ -106,12 +104,6 @@ class ActionModule(ActionBase):
                         remote_data = base64.b64decode(slurpres['content'])
                     if remote_data is not None:
                         remote_checksum = checksum_s(remote_data)
-                    # the source path may have been expanded on the
-                    # target system, so we compare it here and use the
-                    # expanded version if it's different
-                    remote_source = slurpres.get('source')
-                    if remote_source and remote_source != source:
-                        source = remote_source
 
             # calculate the destination name
             if os.path.sep not in self._connection._shell.join_path('a', ''):
@@ -120,13 +112,14 @@ class ActionModule(ActionBase):
             else:
                 source_local = source
 
-            dest = os.path.expanduser(dest)
+            # ensure we only use file name, avoid relative paths
+            if not is_subpath(dest, original_dest):
+                # TODO: ? dest = os.path.expanduser(dest.replace(('../','')))
+                raise AnsibleActionFail("Detected directory traversal, expected to be contained in '%s' but got '%s'" % (original_dest, dest))
+
             if flat:
                 if os.path.isdir(to_bytes(dest, errors='surrogate_or_strict')) and not dest.endswith(os.sep):
-                    result['msg'] = "dest is an existing directory, use a trailing slash if you want to fetch src into that directory"
-                    result['file'] = dest
-                    result['failed'] = True
-                    return result
+                    raise AnsibleActionFail("dest is an existing directory, use a trailing slash if you want to fetch src into that directory")
                 if dest.endswith(os.sep):
                     # if the path ends with "/", we'll use the source filename as the
                     # destination filename
@@ -143,8 +136,6 @@ class ActionModule(ActionBase):
                     target_name = self._play_context.remote_addr
                 dest = "%s/%s/%s" % (self._loader.path_dwim(dest), target_name, source_local)
 
-            dest = dest.replace("//", "/")
-
             if remote_checksum in ('0', '1', '2', '3', '4', '5'):
                 result['changed'] = False
                 result['file'] = source
@@ -172,6 +163,8 @@ class ActionModule(ActionBase):
                     result['msg'] += ", not transferring, ignored"
                 return result
 
+            dest = os.path.normpath(dest)
+
             # calculate checksum for the local file
             local_checksum = checksum(dest)
 
@@ -188,7 +181,7 @@ class ActionModule(ActionBase):
                         f.write(remote_data)
                         f.close()
                     except (IOError, OSError) as e:
-                        raise AnsibleError("Failed to fetch the file: %s" % e)
+                        raise AnsibleActionFail("Failed to fetch the file: %s" % e)
                 new_checksum = secure_hash(dest)
                 # For backwards compatibility. We'll return None on FIPS enabled systems
                 try:
diff --git a/lib/ansible/utils/path.py b/lib/ansible/utils/path.py
index 717cef6cf01..3b50102f2ec 100644
--- a/lib/ansible/utils/path.py
+++ b/lib/ansible/utils/path.py
@@ -97,3 +97,26 @@ def basedir(source):
         dname = os.path.abspath(dname)
 
     return to_text(dname, errors='surrogate_or_strict')
+
+
+def is_subpath(child, parent):
+    """
+    Compares paths to check if one is contained in the other
+    :arg: child: Path to test
+    :arg parent; Path to test against
+     """
+    test = False
+
+    abs_child = unfrackpath(child, follow=False)
+    abs_parent = unfrackpath(parent, follow=False)
+
+    c = abs_child.split(os.path.sep)
+    p = abs_parent.split(os.path.sep)
+
+    try:
+        test = c[:len(p)] == p
+    except IndexError:
+        # child is shorter than parent so cannot be subpath
+        pass
+
+    return test
diff --git a/test/integration/targets/fetch/aliases b/test/integration/targets/fetch/aliases
index 765b70da796..fb5d6faa350 100644
--- a/test/integration/targets/fetch/aliases
+++ b/test/integration/targets/fetch/aliases
@@ -1 +1,2 @@
 shippable/posix/group2
+needs/target/setup_remote_tmp_dir
diff --git a/test/integration/targets/fetch/injection/avoid_slurp_return.yml b/test/integration/targets/fetch/injection/avoid_slurp_return.yml
new file mode 100644
index 00000000000..af62dcf4b8b
--- /dev/null
+++ b/test/integration/targets/fetch/injection/avoid_slurp_return.yml
@@ -0,0 +1,26 @@
+- name: ensure that 'fake slurp' does not poison fetch source
+  hosts: localhost
+  gather_facts: False
+  tasks:
+    - name: fetch with relative source path
+      fetch: src=../injection/here.txt dest={{output_dir}}
+      become: true
+      register: islurp
+
+    - name: fetch with normal source path
+      fetch: src=here.txt dest={{output_dir}}
+      become: true
+      register: islurp2
+
+    - name: ensure all is good in hollywood
+      assert:
+        that:
+            - "'..' not in islurp['dest']"
+            - "'..' not in islurp2['dest']"
+            - "'foo' not in islurp['dest']"
+            - "'foo' not in islurp2['dest']"
+
+    - name: try to trip dest anyways
+      fetch: src=../injection/here.txt dest={{output_dir}}
+      become: true
+      register: islurp2
diff --git a/test/integration/targets/fetch/injection/here.txt b/test/integration/targets/fetch/injection/here.txt
new file mode 100644
index 00000000000..493021b1c9e
--- /dev/null
+++ b/test/integration/targets/fetch/injection/here.txt
@@ -0,0 +1 @@
+this is a test file
diff --git a/test/integration/targets/fetch/injection/library/slurp.py b/test/integration/targets/fetch/injection/library/slurp.py
new file mode 100644
index 00000000000..7b78ba18056
--- /dev/null
+++ b/test/integration/targets/fetch/injection/library/slurp.py
@@ -0,0 +1,29 @@
+#!/usr/bin/python
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+
+DOCUMENTATION = """
+    module: fakeslurp
+    short_desciptoin: fake slurp module
+    description:
+        - this is a fake slurp module
+    options:
+        _notreal:
+            description: really not a real slurp
+    author:
+        - me
+"""
+
+import json
+import random
+
+bad_responses = ['../foo', '../../foo', '../../../foo', '/../../../foo', '/../foo', '//..//foo', '..//..//foo']
+
+
+def main():
+    print(json.dumps(dict(changed=False, content='', encoding='base64', source=random.choice(bad_responses))))
+
+
+if __name__ == '__main__':
+    main()
diff --git a/test/integration/targets/fetch/meta/main.yml b/test/integration/targets/fetch/roles/fetch_tests/meta/main.yml
similarity index 100%
rename from test/integration/targets/fetch/meta/main.yml
rename to test/integration/targets/fetch/roles/fetch_tests/meta/main.yml
diff --git a/test/integration/targets/fetch/tasks/main.yml b/test/integration/targets/fetch/roles/fetch_tests/tasks/main.yml
similarity index 100%
rename from test/integration/targets/fetch/tasks/main.yml
rename to test/integration/targets/fetch/roles/fetch_tests/tasks/main.yml
diff --git a/test/integration/targets/fetch/run_fetch_tests.yml b/test/integration/targets/fetch/run_fetch_tests.yml
new file mode 100644
index 00000000000..f2ff1df3ddd
--- /dev/null
+++ b/test/integration/targets/fetch/run_fetch_tests.yml
@@ -0,0 +1,5 @@
+- name: call fetch_tests role
+  hosts: testhost
+  gather_facts: false
+  roles:
+    - fetch_tests
diff --git a/test/integration/targets/fetch/runme.sh b/test/integration/targets/fetch/runme.sh
new file mode 100755
index 00000000000..7e909dde095
--- /dev/null
+++ b/test/integration/targets/fetch/runme.sh
@@ -0,0 +1,12 @@
+#!/usr/bin/env bash
+
+set -eux
+
+# setup required roles
+ln -s ../../setup_remote_tmp_dir roles/setup_remote_tmp_dir
+
+# run old type role tests
+ansible-playbook -i ../../inventory run_fetch_tests.yml -e "output_dir=${OUTPUT_DIR}" -v "$@"
+
+# run tests to avoid path injection from slurp when fetch uses become
+ansible-playbook -i ../../inventory injection/avoid_slurp_return.yml -e "output_dir=${OUTPUT_DIR}" -v "$@"