diff --git a/doc/source/reference/job_def.rst b/doc/source/reference/job_def.rst
index 791d4fadad..3438e67c6e 100644
--- a/doc/source/reference/job_def.rst
+++ b/doc/source/reference/job_def.rst
@@ -670,6 +670,15 @@ Here is an example of two job definitions:
       will not change.  Untrusted playbooks dynamically evaluate
       variables and are not limited by this restriction.
 
+      Un-frozen versions of all the original job variables are
+      available tagged with the ``!unsafe`` YAML tag under the
+      ``unsafe_vars`` variable hierarchy.  This tag prevents Ansible
+      from evaluating them as Jinja templates.  For example, the job
+      variable `myvar` would be available under `unsafe_vars.myvar`.
+      Advanced users may force Ansible to evaluate these values, but
+      it is not recommended to do so except in the most controlled of
+      circumstances.  They are almost impossible to render safely.
+
    .. attr:: extra-vars
 
       A dictionary of variables to supply to Ansible with higher
diff --git a/releasenotes/notes/unsafe-vars-a997ccabd745176d.yaml b/releasenotes/notes/unsafe-vars-a997ccabd745176d.yaml
index d7414ae7ae..c6bcea5f05 100644
--- a/releasenotes/notes/unsafe-vars-a997ccabd745176d.yaml
+++ b/releasenotes/notes/unsafe-vars-a997ccabd745176d.yaml
@@ -41,4 +41,15 @@ security:
        - set_fact:
            unsafe_var_eval: "{{ hostvars['localhost'].secret.var }}"
 
-    This will force an explicit evaluation of the variable.
+    This will force an explicit evaluation of the variable.  This is
+    generally safe to do in a situation where a playbook is accessing
+    a single secret by name, with no other secrets in scope.  Do not
+    use this capability with more than one secret that is not under
+    the control of the project where the playbook is defined.
+
+    Similarly, versions of all the original job variables tagged with
+    ``!unsafe`` are available under the ``unsafe_vars`` variable
+    hierarchy.  For example, the job variable `myvar` would be
+    available under `unsafe_vars.myvar`.  It is not recommended to
+    evaluate ``unsafe_vars`` expressions except in the most controlled
+    of circumstances.  They are almost impossible to render safely.
diff --git a/tests/fixtures/config/unsafe-vars/git/common-config/playbooks/base-pre.yaml b/tests/fixtures/config/unsafe-vars/git/common-config/playbooks/base-pre.yaml
index 96b755c8f4..d4ccd93715 100644
--- a/tests/fixtures/config/unsafe-vars/git/common-config/playbooks/base-pre.yaml
+++ b/tests/fixtures/config/unsafe-vars/git/common-config/playbooks/base-pre.yaml
@@ -9,3 +9,5 @@
         msg: "BASE SECRETSUB: {{ base_secret.secretsub }}"
     - debug:
         msg: "BASE LATESUB: {{ latesub }}"
+    - debug:
+        msg: "BASE LATESUB UNSAFE: {{ unsafe_vars.latesub }}"
diff --git a/tests/fixtures/config/unsafe-vars/git/org_project/playbooks/testjob-run.yaml b/tests/fixtures/config/unsafe-vars/git/org_project/playbooks/testjob-run.yaml
index 1b9fedd4fc..bd505776a9 100644
--- a/tests/fixtures/config/unsafe-vars/git/org_project/playbooks/testjob-run.yaml
+++ b/tests/fixtures/config/unsafe-vars/git/org_project/playbooks/testjob-run.yaml
@@ -4,6 +4,8 @@
         msg: "TESTJOB SUB: {{ sub }}"
     - debug:
         msg: "TESTJOB LATESUB: {{ latesub }}"
+    - debug:
+        msg: "TESTJOB LATESUB UNSAFE: {{ unsafe_vars.latesub }}"
     - debug:
         msg: "TESTJOB SECRET: {{ project_secret.secretsub }}"
       when: project_secret is defined
diff --git a/tests/unit/test_v3.py b/tests/unit/test_v3.py
index d11863577d..5d0ea503b4 100644
--- a/tests/unit/test_v3.py
+++ b/tests/unit/test_v3.py
@@ -7283,11 +7283,18 @@ class TestUnsafeVars(AnsibleZuulTestCase):
         self.assertIn("BASE SECRETSUB: {{ subtext }}", job_output)
         # latefact wasn't present when frozen
         self.assertIn("BASE LATESUB: undefined", job_output)
+        # check the !unsafe tagged version
+        self.assertIn("BASE LATESUB UNSAFE: "
+                      "{{ latefact | default('undefined') }}", job_output)
 
         # Both of these are dynamically evaluated
         self.assertIn("TESTJOB SUB: text", job_output)
         self.assertIn("TESTJOB LATESUB: late", job_output)
 
+        # check the !unsafe tagged version
+        self.assertIn("TESTJOB LATESUB UNSAFE: "
+                      "{{ latefact | default('undefined') }}", job_output)
+
         # The project secret is not defined
         self.assertNotIn("TESTJOB SECRET:", job_output)
 
@@ -7300,10 +7307,17 @@ class TestUnsafeVars(AnsibleZuulTestCase):
         self.assertIn("BASE SECRETSUB: {{ subtext }}", job_output)
         # latefact wasn't present when frozen
         self.assertIn("BASE LATESUB: undefined", job_output)
+        # check the !unsafe tagged version
+        self.assertIn("BASE LATESUB UNSAFE: "
+                      "{{ latefact | default('undefined') }}", job_output)
 
         # These are frozen
         self.assertIn("TESTJOB SUB: text", job_output)
         self.assertIn("TESTJOB LATESUB: undefined", job_output)
 
+        # check the !unsafe tagged version
+        self.assertIn("TESTJOB LATESUB UNSAFE: "
+                      "{{ latefact | default('undefined') }}", job_output)
+
         # This is marked unsafe
         self.assertIn("TESTJOB SECRET: {{ subtext }}", job_output)
diff --git a/tests/unit/test_yamlutil.py b/tests/unit/test_yamlutil.py
index 706436afec..9c3fb3f66a 100644
--- a/tests/unit/test_yamlutil.py
+++ b/tests/unit/test_yamlutil.py
@@ -63,11 +63,22 @@ class TestYamlDumper(BaseTestCase):
 
     def test_ansible_dumper(self):
         data = {'foo': 'bar'}
-        expected = "!unsafe 'foo': !unsafe 'bar'\n"
+        data = yamlutil.mark_strings_unsafe(data)
+        expected = "foo: !unsafe 'bar'\n"
         yaml_out = yamlutil.ansible_unsafe_dump(data, default_flow_style=False)
         self.assertEqual(yaml_out, expected)
 
-        data = {'foo': {'bar': 'baz'}}
-        expected = "!unsafe 'foo':\n  !unsafe 'bar': !unsafe 'baz'\n"
+        data = {'foo': {'bar': 'baz'}, 'list': ['bar', 1, 3.0, True, None]}
+        data = yamlutil.mark_strings_unsafe(data)
+        expected = """\
+foo:
+  bar: !unsafe 'baz'
+list:
+- !unsafe 'bar'
+- 1
+- 3.0
+- true
+- null
+"""
         yaml_out = yamlutil.ansible_unsafe_dump(data, default_flow_style=False)
         self.assertEqual(yaml_out, expected)
diff --git a/zuul/configloader.py b/zuul/configloader.py
index 605c96207c..c32467f06d 100644
--- a/zuul/configloader.py
+++ b/zuul/configloader.py
@@ -873,30 +873,31 @@ class JobParser(object):
                 setattr(job, k, v)
 
         variables = conf.get('vars', None)
+        forbidden = {'zuul', 'nodepool', 'unsafe_vars'}
         if variables:
-            if 'zuul' in variables or 'nodepool' in variables:
-                raise Exception("Variables named 'zuul' or 'nodepool' "
-                                "are not allowed.")
+            if set(variables.keys()).intersection(forbidden):
+                raise Exception("Variables named 'zuul', 'nodepool', "
+                                "or 'unsafe_vars' are not allowed.")
             job.variables = variables
         extra_variables = conf.get('extra-vars', None)
         if extra_variables:
-            if 'zuul' in extra_variables or 'nodepool' in extra_variables:
-                raise Exception("Variables named 'zuul' or 'nodepool' "
-                                "are not allowed.")
+            if set(extra_variables.keys()).intersection(forbidden):
+                raise Exception("Variables named 'zuul', 'nodepool', "
+                                "or 'unsafe_vars' are not allowed.")
             job.extra_variables = extra_variables
         host_variables = conf.get('host-vars', None)
         if host_variables:
             for host, hvars in host_variables.items():
-                if 'zuul' in hvars or 'nodepool' in hvars:
-                    raise Exception("Variables named 'zuul' or 'nodepool' "
-                                    "are not allowed.")
+                if set(hvars.keys()).intersection(forbidden):
+                    raise Exception("Variables named 'zuul', 'nodepool', "
+                                    "or 'unsafe_vars'are not allowed.")
             job.host_variables = host_variables
         group_variables = conf.get('group-vars', None)
         if group_variables:
             for group, gvars in group_variables.items():
-                if 'zuul' in group_variables or 'nodepool' in gvars:
-                    raise Exception("Variables named 'zuul' or 'nodepool' "
-                                    "are not allowed.")
+                if set(gvars.keys()).intersection(forbidden):
+                    raise Exception("Variables named 'zuul', 'nodepool', "
+                                    "or 'unsafe_vars'are not allowed.")
             job.group_variables = group_variables
 
         allowed_projects = conf.get('allowed-projects', None)
@@ -1008,10 +1009,11 @@ class ProjectTemplateParser(object):
             project_template.setImpliedBranchMatchers(branches)
 
         variables = conf.get('vars', {})
+        forbidden = {'zuul', 'nodepool', 'unsafe_vars'}
         if variables:
-            if 'zuul' in variables or 'nodepool' in variables:
-                raise Exception("Variables named 'zuul' or 'nodepool' "
-                                "are not allowed.")
+            if set(variables.keys()).intersection(forbidden):
+                raise Exception("Variables named 'zuul', 'nodepool', "
+                                "or 'unsafe_vars' are not allowed.")
             project_template.variables = variables
 
         if freeze:
@@ -1131,10 +1133,11 @@ class ProjectParser(object):
         project_config.queue_name = conf.get('queue', None)
 
         variables = conf.get('vars', {})
+        forbidden = {'zuul', 'nodepool', 'unsafe_vars'}
         if variables:
-            if 'zuul' in variables or 'nodepool' in variables:
-                raise Exception("Variables named 'zuul' or 'nodepool' "
-                                "are not allowed.")
+            if set(variables.keys()).intersection(forbidden):
+                raise Exception("Variables named 'zuul', 'nodepool', "
+                                "or 'unsafe_vars' are not allowed.")
             project_config.variables = variables
 
         project_config.freeze()
diff --git a/zuul/executor/server.py b/zuul/executor/server.py
index dd1ddc9aea..7a7da54f2a 100644
--- a/zuul/executor/server.py
+++ b/zuul/executor/server.py
@@ -752,6 +752,9 @@ def check_varnames(var):
         raise Exception("Defining variables named 'zuul' is not allowed")
     if 'nodepool' in var:
         raise Exception("Defining variables named 'nodepool' is not allowed")
+    if 'unsafe_vars' in var:
+        raise Exception("Defining variables named 'unsafe_vars' "
+                        "is not allowed")
     for varname in var.keys():
         if not VARNAME_RE.match(varname):
             raise Exception("Variable names may only contain letters, "
@@ -1892,6 +1895,7 @@ class AnsibleJob(object):
         secrets = self.mergeSecretVars(secrets, args)
         if secrets:
             check_varnames(secrets)
+            secrets = yaml.mark_strings_unsafe(secrets)
             jobdir_playbook.secrets_content = yaml.ansible_unsafe_dump(
                 secrets, default_flow_style=False)
             jobdir_playbook.secrets_keys = set(secrets.keys())
@@ -2325,6 +2329,16 @@ class AnsibleJob(object):
             with open(path, 'w') as f:
                 f.write(json.dumps(facts))
 
+            # While we're here, update both hostvars dicts with
+            # an !unsafe copy of the original input as well.
+            unsafe = yaml.mark_strings_unsafe(
+                self.original_hostvars[host['name']])
+            self.frozen_hostvars[host['name']]['unsafe_vars'] = unsafe
+
+            unsafe = yaml.mark_strings_unsafe(
+                self.original_hostvars[host['name']])
+            self.original_hostvars[host['name']]['unsafe_vars'] = unsafe
+
     def writeDebugInventory(self):
         # This file is unused by Zuul, but the base jobs copy it to logs
         # for debugging, so let's continue to put something there.
@@ -2334,12 +2348,13 @@ class AnsibleJob(object):
 
         with open(self.jobdir.inventory, 'w') as inventory_yaml:
             inventory_yaml.write(
-                yaml.safe_dump(inventory, default_flow_style=False))
+                yaml.ansible_unsafe_dump(inventory, default_flow_style=False))
 
     def writeSetupInventory(self):
         jobdir_playbook = self.jobdir.setup_playbook
         setup_inventory = make_setup_inventory_dict(
             self.host_list, self.original_hostvars)
+        setup_inventory = yaml.mark_strings_unsafe(setup_inventory)
 
         with open(jobdir_playbook.inventory, 'w') as inventory_yaml:
             # Write this inventory with !unsafe tags to avoid mischief
@@ -2356,7 +2371,7 @@ class AnsibleJob(object):
 
         with open(jobdir_playbook.inventory, 'w') as inventory_yaml:
             inventory_yaml.write(
-                yaml.safe_dump(inventory, default_flow_style=False))
+                yaml.ansible_unsafe_dump(inventory, default_flow_style=False))
 
     def writeLoggingConfig(self):
         self.log.debug("Writing logging config for job %s %s",
diff --git a/zuul/lib/yamlutil.py b/zuul/lib/yamlutil.py
index 4b7d28a919..a040c426c5 100644
--- a/zuul/lib/yamlutil.py
+++ b/zuul/lib/yamlutil.py
@@ -114,19 +114,42 @@ def encrypted_load(stream, *args, **kwargs):
 
 # Add support for the Ansible !unsafe tag
 # Note that "unsafe" here is used differently than "safe" from PyYAML
+
+class AnsibleUnsafeStr:
+    yaml_tag = u'!unsafe'
+
+    def __init__(self, value):
+        self.value = value
+
+    def __ne__(self, other):
+        return not self.__eq__(other)
+
+    def __eq__(self, other):
+        if isinstance(other, AnsibleUnsafeStr):
+            return self.value == other.value
+        return self.value == other
+
+    @classmethod
+    def from_yaml(cls, loader, node):
+        return cls(node.value)
+
+    @classmethod
+    def to_yaml(cls, dumper, data):
+        return yaml.ScalarNode(tag=cls.yaml_tag, value=data.value)
+
+
 class AnsibleUnsafeDumper(yaml.SafeDumper):
-    def represent_str(self, data):
-        return self.represent_scalar('!unsafe', data)
+    pass
 
 
 class AnsibleUnsafeLoader(yaml.SafeLoader):
     pass
 
 
-AnsibleUnsafeDumper.add_representer(
-    str, AnsibleUnsafeDumper.represent_str)
-AnsibleUnsafeLoader.add_constructor(
-    '!unsafe', AnsibleUnsafeLoader.construct_yaml_str)
+AnsibleUnsafeDumper.add_representer(AnsibleUnsafeStr,
+                                    AnsibleUnsafeStr.to_yaml)
+AnsibleUnsafeLoader.add_constructor(AnsibleUnsafeStr.yaml_tag,
+                                    AnsibleUnsafeStr.from_yaml)
 
 
 def ansible_unsafe_dump(data, *args, **kwargs):
@@ -135,3 +158,35 @@ def ansible_unsafe_dump(data, *args, **kwargs):
 
 def ansible_unsafe_load(stream, *args, **kwargs):
     return yaml.load(stream, *args, Loader=AnsibleUnsafeLoader, **kwargs)
+
+
+def mark_strings_unsafe(d):
+    """Traverse a json-style data structure and replace every string value
+    with an AnsibleUnsafeStr
+
+    Returns the new structure.
+    """
+    if isinstance(d, tuple):
+        d = list(d)
+
+    if isinstance(d, dict):
+        newdict = {}
+        for key, value in d.items():
+            newdict[key] = mark_strings_unsafe(value)
+        return newdict
+    elif isinstance(d, list):
+        return [mark_strings_unsafe(v) for v in d]
+    elif isinstance(d, int):
+        return d
+    elif isinstance(d, float):
+        return d
+    elif isinstance(d, type(None)):
+        return d
+    elif isinstance(d, bool):
+        return d
+    elif isinstance(d, AnsibleUnsafeStr):
+        return d
+    elif isinstance(d, str):
+        return AnsibleUnsafeStr(d)
+    else:
+        raise Exception("Unhandled type: %s", type(d))
diff --git a/zuul/model.py b/zuul/model.py
index e9c1813288..e1a7eb8881 100644
--- a/zuul/model.py
+++ b/zuul/model.py
@@ -1106,9 +1106,9 @@ class PlaybookContext(ConfigObject):
                 raise Exception(
                     'The secret "{name}" was not found.'.format(
                         name=secret_use.name))
-            if secret_use.alias == 'zuul' or secret_use.alias == 'nodepool':
-                raise Exception('Secrets named "zuul" or "nodepool" '
-                                'are not allowed.')
+            if secret_use.alias in ('zuul', 'nodepool', 'unsafe_vars'):
+                raise Exception("Secrets named 'zuul', 'nodepool', "
+                                "or 'unsafe_vars' are not allowed.")
             if not VARNAME_RE.match(secret_use.alias):
                 raise Exception("Variable names may only contain letters, "
                                 "numbers, and underscores")
@@ -1570,8 +1570,10 @@ class Job(ConfigObject):
         v = Job._deepUpdate(v, other_vars)
         # To avoid running afoul of checks that jobs don't set zuul
         # variables, remove them from parent data here.
-        if 'zuul' in v:
-            del v['zuul']
+        v.pop('zuul', None)
+        # For safety, also drop nodepool and unsafe_vars
+        v.pop('nodepool', None)
+        v.pop('unsafe_vars', None)
         self.parent_data = v
 
         secret_other_vars = other_build.secret_result_data