Do not silently allow the use of undefined variables in jinja2 templates (#11016)

This can have *extremely* bad consequences. After this change, a jinja2
template like the one below will cause the task instance to fail, if the
DAG being executed is not a sub-DAG. This may also display an error on
the Rendered tab of the Task Instance page.

task_instance.xcom_pull('z', key='return_value', dag_id=dag.parent_dag.dag_id)

Prior to the change in this commit, the above template would pull the
latest value for task_id 'z', for the given execution_date, from *any DAG*.
If your task_ids between DAGs are all unique, or if DAGs using the same
task_id always have different execution_date values, this will appear to
act like dag_id=None.

Our current theory is SQLAlchemy/Python doesn't behave as expected when
comparing `jinja2.Undefined` to `None`.

UPDATING.md

@@ -138,6 +138,20 @@ with third party services to the ``airflow.providers`` package.
 All changes made are backward compatible, but if you use the old import paths you will
 see a deprecation warning. The old import paths can be abandoned in the future.
 
+#### Change to undefined variable handling in templates
+
+Prior to Airflow 2.0 Jinja Templates would permit the use of undefined variables. They would render as an
+empty string, with no indication to the user an undefined variable was used. With this release, any template
+rendering involving undefined variables will fail the task, as well as displaying an error in the UI when
+rendering.
+
+The behavior can be reverted when instantiating a DAG.
+```python
+import jinja2
+
+dag = DAG('simple_dag', template_undefined=jinja2.Undefined)
+```
+
 ### Breaking Change in OAuth
 
 The flask-ouathlib has been replaced with authlib because flask-outhlib has

airflow/models/dag.py

@@ -123,7 +123,7 @@ class DAG(BaseDag, LoggingMixin):
         default
     :type template_searchpath: str or list[str]
     :param template_undefined: Template undefined type.
-    :type template_undefined: jinja2.Undefined
+    :type template_undefined: jinja2.StrictUndefined
     :param user_defined_macros: a dictionary of macros that will be exposed
         in your jinja templates. For example, passing ``dict(foo='bar')``
         to this argument allows you to ``{{ foo }}`` in all jinja
@@ -224,7 +224,7 @@ class DAG(BaseDag, LoggingMixin):
         end_date: Optional[datetime] = None,
         full_filepath: Optional[str] = None,
         template_searchpath: Optional[Union[str, Iterable[str]]] = None,
-        template_undefined: Type[jinja2.Undefined] = jinja2.Undefined,
+        template_undefined: Type[jinja2.StrictUndefined] = jinja2.StrictUndefined,
         user_defined_macros: Optional[Dict] = None,
         user_defined_filters: Optional[Dict] = None,
         default_args: Optional[Dict] = None,

tests/models/test_baseoperator.py

@@ -71,7 +71,6 @@ class TestBaseOperator(unittest.TestCase):
     @parameterized.expand(
         [
             ("{{ foo }}", {"foo": "bar"}, "bar"),
-            ("{{ foo }}", {}, ""),
             (["{{ foo }}_1", "{{ foo }}_2"], {"foo": "bar"}, ["bar_1", "bar_2"]),
             (("{{ foo }}_1", "{{ foo }}_2"), {"foo": "bar"}, ("bar_1", "bar_2")),
             (
@@ -184,6 +183,14 @@ class TestBaseOperator(unittest.TestCase):
         result = task.render_template(content, {"foo": "bar"})
         self.assertEqual(content, result)
 
+    def test_render_template_field_undefined_default(self):
+        """Test render_template with template_undefined unchanged."""
+        with DAG("test-dag", start_date=DEFAULT_DATE):
+            task = DummyOperator(task_id="op1")
+
+        with self.assertRaises(jinja2.UndefinedError):
+            task.render_template("{{ foo }}", {})
+
     def test_render_template_field_undefined_strict(self):
         """Test render_template with template_undefined configured."""
         with DAG("test-dag", start_date=DEFAULT_DATE, template_undefined=jinja2.StrictUndefined):
@@ -192,6 +199,13 @@ class TestBaseOperator(unittest.TestCase):
         with self.assertRaises(jinja2.UndefinedError):
             task.render_template("{{ foo }}", {})
 
+    def test_render_template_field_undefined_not_strict(self):
+        """Test render_template with template_undefined configured to silently error."""
+        with DAG("test-dag", start_date=DEFAULT_DATE, template_undefined=jinja2.Undefined):
+            task = DummyOperator(task_id="op1")
+
+        self.assertEqual(task.render_template("{{ foo }}", {}), "")
+
     def test_nested_template_fields_declared_must_exist(self):
         """Test render_template when a nested template field is missing."""
         with DAG("test-dag", start_date=DEFAULT_DATE):