Fix leak sensitive field via V1EnvVar on exception

[ecerulm]

Currently the KubernetesPodOperator env_vars will be printed on the task logs if there is any templating error (like an UndefinedError, TemplateSyntaxError or KeyError)

[2023-01-16, 23:03:17 UTC] {abstractoperator.py:592} ERROR - Exception rendering Jinja template for task 'dry_run_demo', field 'env_vars'. Template: [{'name': 'password', 'value': 'secretpassword', 'value_from': None}, {'name': 'VAR2', 'value': '{{ var.value.nonexisting}}', 'value_from': None}]
Traceback (most recent call last):
  File "/Users/rubelagu/.pyenv/versions/3.10.7/envs/venv-airflow-250/lib/python3.10/site-packages/airflow/models/abstractoperator.py", line 585, in _do_render_template_fields
    rendered_content = self.render_template(
  File "/Users/rubelagu/.pyenv/versions/3.10.7/envs/venv-airflow-250/lib/python3.10/site-packages/airflow/models/abstractoperator.py", line 657, in render_template
    return [self.render_template(element, context, jinja_env, oids) for element in value]
  File "/Users/rubelagu/.pyenv/versions/3.10.7/envs/venv-airflow-250/lib/python3.10/site-packages/airflow/models/abstractoperator.py", line 657, in <listcomp>
    return [self.render_template(element, context, jinja_env, oids) for element in value]
  File "/Users/rubelagu/.pyenv/versions/3.10.7/envs/venv-airflow-250/lib/python3.10/site-packages/airflow/models/abstractoperator.py", line 664, in render_template
    self._render_nested_template_fields(value, context, jinja_env, oids)
  File "/Users/rubelagu/.pyenv/versions/3.10.7/envs/venv-airflow-250/lib/python3.10/site-packages/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py", line 321, in _render_nested_template_fields
    self._do_render_template_fields(content, ("value", "name"), context, jinja_env, seen_oids)
  ...
  ...
  File "/Users/rubelagu/.pyenv/versions/3.10.7/envs/venv-airflow-250/lib/python3.10/site-packages/airflow/models/variable.py", line 141, in get
    raise KeyError(f"Variable {key} does not exist")
KeyError: 'Variable nonexisting does not exist'

this happens when there is any error on the templates. For example a KeyError raised when using var.value.somemistypedvalue:

        env_vars={
            "password": "{{ conn.test_connection.password }}",
            "VAR2": "{{ var.value.nonexisting}}",
        },

This PR uses the airflow.utils.log.secrets_maker.redact to remove any field contained in DEFAULT_SENSITIVE_FIELDS or
sensitive_var_conn_names.

---

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[ecerulm]

I had to introduce a ConvertableToDict type , in order to keep the mypy happy since k8s.V1EnvVar is not a dict. Initially I used hasattr(item, "to_dict") but mypy will complain about the typing since Redactable is not guaranteed to have to_dict, and it seems that the mypy does not understand the hasattr() the same way it recognizes the isinstance() .

[uranusjr]

What are other instances this would be desired? Personally I’d prefer to keep this more conservative and only check for V1EnvVar.

[ecerulm]

If I check explicitly for V1EnvVar then I guess we introduce a direct dependency to the kubernetes provider (which is the one that provides the kubernetes client, if I'm not mistaken ). That was my only motivation to use the to_dict/ConvertableToDict , I didn't want to introduce that dependency because I thought that would be not allowed.

[ecerulm]

The other alternative to avoid introducing a dependency to V1EnvVar would be to mask everything that is not explicitly Redactable. Which one shall I go for? the ConvertableToDict, introducing a dependency to V1EnvVar or mask everything not redactable?

[ecerulm]

@uranusjr , what do you think?

[uranusjr]

We could try-import kubernetes (if we can’t import it the object would obviously not be a V1EnvVar), but this is OK I think. We can always make the check stricter later if something breaks.

Maybe add a note on ConvertableToDict describing the rationale? (e.g. we don’t want to depend directly on kubernetes so this checks for the “shape” of a V1EnvVar instead)

[ecerulm]

Nevermind, I realized now that the ConvertableToDict approach does not work on Python 3.7 (as typing.Protocol was introduced in 3.8). So I guess I'll try your try-import approach

[uranusjr]

We actually have Protocol available on 3.7 via airflow.typing_compat, but either way is fine as I mentioned above, so since you already made the change let’s just keep this.

[ecerulm]

I had to fix some formatting issues and some stuff from the rebase from main but now "all checks have passed".

[ecerulm]

Is there anything else that I can do? I just wonder if I need to do anything else before you merge it?

[Taragolis]

Just in case run full test