diff --git a/airflow-core/newsfragments/72042.bugfix.rst b/airflow-core/newsfragments/72042.bugfix.rst new file mode 100644 index 0000000000000..49651487d555b --- /dev/null +++ b/airflow-core/newsfragments/72042.bugfix.rst @@ -0,0 +1 @@ +The write-conflict check performed when setting or updating a team-scoped Variable now resolves against that team rather than the global scope, so a secrets backend shadowing the key within the team is detected and a global-only definition no longer warns about a conflict that would not shadow the read. diff --git a/airflow-core/src/airflow/models/variable.py b/airflow-core/src/airflow/models/variable.py index 9493cab60aa38..bcbc07a9703e3 100644 --- a/airflow-core/src/airflow/models/variable.py +++ b/airflow-core/src/airflow/models/variable.py @@ -258,7 +258,7 @@ def set( # check if the secret exists in the custom secrets' backend. from airflow.sdk import SecretCache - Variable.check_for_write_conflict(key=key) + Variable.check_for_write_conflict(key=key, team_name=team_name) if serialize_json: stored_value = json.dumps(value, indent=2) else: @@ -342,7 +342,7 @@ def update( "Multi-team mode is not configured in the Airflow environment. To assign a team to a variable, multi-mode must be enabled." ) - Variable.check_for_write_conflict(key=key) + Variable.check_for_write_conflict(key=key, team_name=team_name) if Variable.get_variable_from_secrets(key=key, team_name=team_name) is None: raise KeyError(f"Variable {key} does not exist") @@ -430,7 +430,7 @@ def rotate_fernet_key(self): self._val = fernet.rotate(self._val.encode("utf-8")).decode() @staticmethod - def check_for_write_conflict(key: str) -> None: + def check_for_write_conflict(key: str, team_name: str | None = None) -> None: """ Log a warning if a variable exists outside the metastore. @@ -439,11 +439,15 @@ def check_for_write_conflict(key: str) -> None: subsequent reads will not read the set value. :param key: Variable Key + :param team_name: Team name the variable is being written for, so the check resolves + against the same scope the write will use """ for secrets_backend in ensure_secrets_loaded(): if not isinstance(secrets_backend, MetastoreBackend): try: - var_val = secrets_backend.get_variable(key=key) + var_val = call_secrets_backend_method( + secrets_backend.get_variable, team_name=team_name, key=key + ) if var_val is not None: _backend_name = type(secrets_backend).__name__ log.warning( diff --git a/airflow-core/tests/unit/models/test_variable.py b/airflow-core/tests/unit/models/test_variable.py index 0bc32373a305d..cc2f9f94b9367 100644 --- a/airflow-core/tests/unit/models/test_variable.py +++ b/airflow-core/tests/unit/models/test_variable.py @@ -28,6 +28,7 @@ from airflow.models import Variable, crypto, variable from airflow.sdk import SecretCache +from airflow.secrets import BaseSecretsBackend from airflow.secrets.metastore import MetastoreBackend from tests_common.test_utils import db @@ -41,6 +42,30 @@ pytestmark = pytest.mark.db_test +class _TeamUnawareVariableBackend(BaseSecretsBackend): + """A custom backend whose ``get_variable`` override predates the ``team_name`` keyword.""" + + def __init__(self): + self.was_called = False + + # The signature mismatch with the base class is the point of this fixture, so mypy's + # override check has to be waived here rather than fixed. + def get_variable(self, key: str) -> str | None: # type: ignore[override] + self.was_called = True + return "secret_val" + + +class _TeamAwareVariableBackend(BaseSecretsBackend): + """A custom backend whose ``get_variable`` override accepts ``team_name``.""" + + def __init__(self): + self.received_team_name: str | None = None + + def get_variable(self, key: str, team_name: str | None = None) -> str | None: + self.received_team_name = team_name + return "secret_val" + + class TestVariable: @pytest.fixture(autouse=True) def setup_test_cases(self): @@ -192,6 +217,43 @@ def test_variable_set_with_extra_secret_backend(self, mock_ensure_secrets, caplo ) Variable.delete(key="key", session=session) + @mock.patch("airflow.models.variable.ensure_secrets_loaded") + def test_write_conflict_check_forwards_team_name(self, mock_ensure_secrets): + """The check must resolve against the scope being written, not the global one.""" + backend = _TeamAwareVariableBackend() + mock_ensure_secrets.return_value = [backend, MetastoreBackend()] + + Variable.check_for_write_conflict(key="key", team_name="team_a") + + assert backend.received_team_name == "team_a" + + @mock.patch("airflow.models.variable.ensure_secrets_loaded") + def test_write_conflict_check_tolerates_team_unaware_backend(self, mock_ensure_secrets): + """A backend whose override predates ``team_name`` must still be consulted, not error out.""" + backend = _TeamUnawareVariableBackend() + mock_ensure_secrets.return_value = [backend, MetastoreBackend()] + + Variable.check_for_write_conflict(key="key", team_name="team_a") + + assert backend.was_called + + @conf_vars({("core", "multi_team"): "True"}) + @mock.patch.object(Variable, "check_for_write_conflict") + def test_set_forwards_team_name_to_write_conflict_check(self, mock_check, testing_team, session): + Variable.set(key="key", value="db-value", team_name=testing_team.name, session=session) + + assert mock_check.call_args.kwargs["team_name"] == testing_team.name + + @conf_vars({("core", "multi_team"): "True"}) + @mock.patch.object(Variable, "check_for_write_conflict") + def test_update_forwards_team_name_to_write_conflict_check(self, mock_check, testing_team, session): + Variable.set(key="key", value="db-value", team_name=testing_team.name, session=session) + SecretCache.invalidate_variable("key") + + Variable.update(key="key", value="new-value", team_name=testing_team.name, session=session) + + assert mock_check.call_args.kwargs["team_name"] == testing_team.name + def test_variable_set_get_round_trip_json(self): value = {"a": 17, "b": 47} Variable.set(key="tested_var_set_id", value=value, serialize_json=True)