Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions airflow-core/newsfragments/72042.bugfix.rst
Original file line number Diff line number Diff line change
@@ -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.
12 changes: 8 additions & 4 deletions airflow-core/src/airflow/models/variable.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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.

Expand All @@ -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(
Expand Down
62 changes: 62 additions & 0 deletions airflow-core/tests/unit/models/test_variable.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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):
Expand Down Expand Up @@ -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)
Expand Down