diff --git a/airflow-core/docs/migrations-ref.rst b/airflow-core/docs/migrations-ref.rst index b6f7fbcf87933..cc18c48f49323 100644 --- a/airflow-core/docs/migrations-ref.rst +++ b/airflow-core/docs/migrations-ref.rst @@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are executed via when you ru +-------------------------+------------------+-------------------+--------------------------------------------------------------+ | Revision ID | Revises ID | Airflow Version | Description | +=========================+==================+===================+==============================================================+ -| ``76c46545c91e`` (head) | ``3c525f44bea8`` | ``3.4.0`` | Add new index for trigger. | +| ``c7f0a5d2e9b4`` (head) | ``76c46545c91e`` | ``3.4.0`` | Lower case team names. | ++-------------------------+------------------+-------------------+--------------------------------------------------------------+ +| ``76c46545c91e`` | ``3c525f44bea8`` | ``3.4.0`` | Add new index for trigger. | +-------------------------+------------------+-------------------+--------------------------------------------------------------+ | ``3c525f44bea8`` | ``b2f1a9c7d4e0`` | ``3.4.0`` | Add indexes on serialized_dag and dag_code. | +-------------------------+------------------+-------------------+--------------------------------------------------------------+ diff --git a/airflow-core/newsfragments/70886.significant.rst b/airflow-core/newsfragments/70886.significant.rst new file mode 100644 index 0000000000000..d772b43d169f5 --- /dev/null +++ b/airflow-core/newsfragments/70886.significant.rst @@ -0,0 +1,19 @@ +Team names must now be lower case, and cannot contain two consecutive underscores + +``airflow teams create`` and ``airflow teams sync`` now accept only names matching +``(?!.*__)[a-z0-9_-]{3,50}`` -- lower case letters, digits, hyphens and single +underscores. Upper case letters and ``__`` were previously accepted. + +A team name is embedded in the ``AIRFLOW_CONN_____`` environment variable +namespace that holds team scoped secrets, and the backend upper-cases the name when it +builds that variable. ``data_eng`` and ``Data_Eng`` therefore resolved a single +namespace between them, and a name containing ``___`` could not be told apart from the +separator. + +``airflow db migrate`` lower-cases every stored team name, moving that team's Dag +bundles, Connections, Variables, Pools and Triggers with it. Two stored names that +differ only in case would collapse onto one row and silently merge two teams, so +migration refuses to start while any such pair exists -- rename all but one first. Names +that stay invalid for another reason, such as containing ``__``, are migrated as they +are; they keep working, but ``airflow teams sync`` exits non-zero until they are +renamed. diff --git a/airflow-core/src/airflow/cli/commands/team_command.py b/airflow-core/src/airflow/cli/commands/team_command.py index c108ac203f8bf..4d669408dc0de 100644 --- a/airflow-core/src/airflow/cli/commands/team_command.py +++ b/airflow-core/src/airflow/cli/commands/team_command.py @@ -30,7 +30,12 @@ from airflow.dag_processing.bundles.manager import DagBundlesManager from airflow.models.connection import Connection from airflow.models.pool import Pool -from airflow.models.team import Team, dag_bundle_team_association_table +from airflow.models.team import ( + TEAM_NAME_PATTERN, + Team, + dag_bundle_team_association_table, + find_invalid_team_names, +) from airflow.models.variable import Variable from airflow.utils import cli as cli_utils from airflow.utils.providers_configuration_loader import providers_configuration_loaded @@ -54,12 +59,23 @@ def _show_teams(teams, output): def _extract_team_name(args): - """Extract and validate team name from args.""" + """ + Extract a team name from args, without holding it to the name rule. + + ``teams delete`` has to accept any name a row can hold, including one stored before the + current rule applied -- otherwise a team that violates it could never be removed. + """ team_name = args.name.strip() if not team_name: raise SystemExit("Team name cannot be empty") - if not re.match(r"^[a-zA-Z0-9_-]{3,50}$", team_name): - raise SystemExit("Invalid team name: must match regex ^[a-zA-Z0-9_-]{3,50}$") + return team_name + + +def _extract_new_team_name(args): + """Extract a team name for a creation path, holding it to the name rule.""" + team_name = _extract_team_name(args) + if not re.fullmatch(TEAM_NAME_PATTERN, team_name): + raise SystemExit(f"Invalid team name: must match regex {TEAM_NAME_PATTERN}") return team_name @@ -78,8 +94,8 @@ def _create_default_team_pool(team_name: str, *, session: Session) -> None: @providers_configuration_loaded @provide_session def team_create(args, *, session=NEW_SESSION): - """Create a new team. Team names must be 3-50 characters long and contain only alphanumeric characters, hyphens, and underscores.""" - team_name = _extract_team_name(args) + """Create a new team. Team names must be 3-50 characters long and contain only lower case alphanumeric characters, hyphens, and single underscores.""" + team_name = _extract_new_team_name(args) # Check if team with this name already exists if session.scalar(select(Team).where(Team.name == team_name)): @@ -210,11 +226,22 @@ def team_sync(args, *, session=NEW_SESSION): for bundle in DagBundlesManager()._bundle_config.values() if bundle.team_name is not None } + existing_teams = Team.get_all_team_names(session=session) + + # The bundle config is a second creation path for teams, so it enforces the same rule as + # `teams create`. Stored names are checked too: `teams sync` shipped without any validation, + # so a deployment can hold a name the rule rejects, and checking only the incoming config + # would miss a team created by an earlier sync and since dropped from it. + if invalid := find_invalid_team_names(dag_bundle_teams | existing_teams): + raise SystemExit( + f"Invalid team name(s): {', '.join(invalid)}. " + f"Team names must match regex {TEAM_NAME_PATTERN}. " + "Names already stored must be corrected before syncing." + ) teams_added = 0 try: - existing_teams = Team.get_all_team_names(session=session) for team_name in dag_bundle_teams: if team_name not in existing_teams: session.add(Team(name=team_name)) diff --git a/airflow-core/src/airflow/migrations/versions/0131_3_4_0_lower_case_team_names.py b/airflow-core/src/airflow/migrations/versions/0131_3_4_0_lower_case_team_names.py new file mode 100644 index 0000000000000..c62866b935b17 --- /dev/null +++ b/airflow-core/src/airflow/migrations/versions/0131_3_4_0_lower_case_team_names.py @@ -0,0 +1,97 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +""" +Lower case team names. + +Revision ID: c7f0a5d2e9b4 +Revises: 76c46545c91e +Create Date: 2026-08-12 09:14:22.518233 + +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import sqlalchemy as sa +from alembic import op + +if TYPE_CHECKING: + from sqlalchemy.sql import ClauseElement + +# revision identifiers, used by Alembic. +revision = "c7f0a5d2e9b4" +down_revision = "76c46545c91e" +branch_labels = None +depends_on = None +airflow_version = "3.4.0" + +# Tables carrying a foreign key to ``team.name``. +_REFERRING_TABLES = ("dag_bundle_team", "connection", "variable", "slot_pool", "trigger") + +_TEAM = sa.table("team", sa.column("name", sa.String)) + + +def build_lower_casing_statements() -> list[ClauseElement]: + """ + Build the statements that fold every stored team name to lower case. + + Kept free of anything read back from a row so that ``db migrate --show-sql-only`` can emit + it: an offline migration has no rows to read. + + Their order carries both collations a database may have, without having to tell them apart: + + * Case sensitive, where ``name <> lower(name)`` selects the names to fold. ``team.name`` is + the primary key and the foreign keys do not cascade on update, so the lower case row has to + exist before anything can point at it, and the old one can only go once nothing does. Two + stored names folding onto one collide on the primary key at the insert, deliberately -- + merging two teams is not a migration's decision to make, and ``db migrate`` refuses earlier + with an explanation. + * Case insensitive, where the two spellings are a single value: the insert and the delete + match nothing, the referring rows keep matching their team while they are rewritten, and + the closing unconditional update renames the row in place. + """ + lowered = sa.func.lower(_TEAM.c.name) + statements: list[ClauseElement] = [ + _TEAM.insert().from_select(["name"], sa.select(lowered).where(_TEAM.c.name != lowered)) + ] + for table_name in _REFERRING_TABLES: + referring = sa.table(table_name, sa.column("team_name", sa.String)) + # ``connection`` and ``trigger`` carry the rows of team-less deployments as well; only + # the team scoped ones are rewritten. + statements.append( + referring.update() + .where(referring.c.team_name.is_not(None)) + .values(team_name=sa.func.lower(referring.c.team_name)) + ) + statements.append(_TEAM.delete().where(_TEAM.c.name != lowered)) + statements.append(_TEAM.update().values(name=lowered)) + return statements + + +def upgrade(): + """Lower case team names.""" + for statement in build_lower_casing_statements(): + op.execute(statement) + + +def downgrade(): + """Unapply Lower case team names.""" + # The original casing is recorded nowhere, so it cannot be restored. Lower case names are + # valid under every rule that preceded this one, so leaving them as they are is safe. diff --git a/airflow-core/src/airflow/models/team.py b/airflow-core/src/airflow/models/team.py index 64877489c66a8..fab0e27240976 100644 --- a/airflow-core/src/airflow/models/team.py +++ b/airflow-core/src/airflow/models/team.py @@ -17,6 +17,7 @@ # under the License. from __future__ import annotations +import re from typing import TYPE_CHECKING from sqlalchemy import Column, ForeignKey, Index, String, Table, select @@ -26,8 +27,28 @@ from airflow.utils.session import NEW_SESSION, provide_session if TYPE_CHECKING: + from collections.abc import Iterable + from sqlalchemy.orm import Session +# What ``airflow teams create`` and ``airflow teams sync`` accept as a team name. +# +# Lower case only: the environment secrets backend upper-cases the team name to build +# ``AIRFLOW_CONN_____``, so ``data_eng`` and ``Data_Eng`` resolve one namespace between +# them and each team would read the other's Connections and Variables. Two consecutive +# underscores are excluded so that a name can never contain the ``___`` separator itself. +# +# Deliberately unanchored and always used with ``re.fullmatch``: ``re.match`` against a +# ``$``-anchored pattern also accepts a trailing newline, and unlike ``teams create`` the +# ``teams sync`` path does not strip what the bundle config gives it. +TEAM_NAME_PATTERN = r"(?!.*__)[a-z0-9_-]{3,50}" + + +def find_invalid_team_names(names: Iterable[str]) -> list[str]: + """Return the names that do not satisfy :data:`TEAM_NAME_PATTERN`, sorted.""" + return sorted({name for name in names if not re.fullmatch(TEAM_NAME_PATTERN, name)}) + + dag_bundle_team_association_table = Table( "dag_bundle_team", Base.metadata, diff --git a/airflow-core/src/airflow/utils/db.py b/airflow-core/src/airflow/utils/db.py index 4b873b4214eba..19f96a2ea6e0c 100644 --- a/airflow-core/src/airflow/utils/db.py +++ b/airflow-core/src/airflow/utils/db.py @@ -117,7 +117,7 @@ class MappedClassProtocol(Protocol): "3.1.8": "509b94a1042d", "3.2.0": "1d6611b6ab7c", "3.3.0": "d2f4e1b3c5a7", - "3.4.0": "76c46545c91e", + "3.4.0": "c7f0a5d2e9b4", } # Prefix used to identify tables holding data moved during migration. @@ -1117,10 +1117,40 @@ def reflect_tables(tables: list[MappedClassProtocol | str] | None, session, sche return metadata +def check_team_names_can_be_lower_cased(*, session: Session) -> Iterable[str]: + """ + Yield an error for each group of stored team names that cannot be folded to lower case. + + Team names became lower case only, and a migration rewrites the stored ones. Names that + differ only in case collapse onto a single row, which would silently merge two teams -- one + would inherit the other's Dag bundles, Connections, Variables, Pools and Triggers. Nothing + here can choose which name survives, so migration stops and the operator renames one first. + """ + from collections import defaultdict + + from airflow.models.team import Team + + if not inspect(session.get_bind()).has_table(Team.__tablename__): + return + + by_lower: dict[str, list[str]] = defaultdict(list) + for name in Team.get_all_team_names(session=session): + by_lower[name.lower()].append(name) + + for lowered, names in sorted(by_lower.items()): + if len(names) > 1: + yield ( + f"Teams {', '.join(repr(name) for name in sorted(names))} differ only in case and " + f"would both become '{lowered}'. Rename all but one with `airflow teams create` / " + "`airflow teams delete`, moving its Dag bundles, Connections, Variables and Pools " + "across, before migrating." + ) + + @provide_session def _check_migration_errors(*, session: Session = NEW_SESSION) -> Iterable[str]: """:session: session of the sqlalchemy.""" - check_functions: Iterable[Callable[..., Iterable[str]]] = () + check_functions: Iterable[Callable[..., Iterable[str]]] = (check_team_names_can_be_lower_cased,) for check_fn in check_functions: log.debug("running check function %s", check_fn.__name__) yield from check_fn(session=session) diff --git a/airflow-core/tests/unit/cli/commands/test_team_command.py b/airflow-core/tests/unit/cli/commands/test_team_command.py index 02d8a44516f0e..ff15a9167ec76 100644 --- a/airflow-core/tests/unit/cli/commands/test_team_command.py +++ b/airflow-core/tests/unit/cli/commands/test_team_command.py @@ -107,6 +107,34 @@ def test_team_create_invalid_name(self): with pytest.raises(SystemExit, match="Invalid team name"): team_command.team_create(self.parser.parse_args(["teams", "create", "test with space"])) + @pytest.mark.parametrize( + "bad_name", + [ + pytest.param("Data-Eng", id="upper-case"), + pytest.param("team__x", id="consecutive-underscores"), + pytest.param("team_a___prod", id="spans-the-separator"), + ], + ) + def test_team_create_rejects_a_name_the_secrets_namespace_cannot_hold(self, bad_name): + with pytest.raises(SystemExit, match="Invalid team name"): + team_command.team_create(self.parser.parse_args(["teams", "create", bad_name])) + + assert self.session.scalar(select(Team).where(Team.name == bad_name)) is None + + def test_team_create_allows_a_single_underscore(self): + team_command.team_create(self.parser.parse_args(["teams", "create", "team_a"])) + + assert self.session.scalar(select(Team).where(Team.name == "team_a")) is not None + + def test_team_delete_accepts_a_name_the_current_rule_rejects(self): + """A team stored before the rule tightened still has to be removable.""" + self.session.add(Team(name="Data-Eng")) + self.session.commit() + + team_command.team_delete(self.parser.parse_args(["teams", "delete", "Data-Eng", "--yes"])) + + assert self.session.scalar(select(Team).where(Team.name == "Data-Eng")) is None + def test_team_create_whitespace_name(self): """Test team creation with whitespace-only name.""" with pytest.raises(SystemExit, match="Team name cannot be empty"): @@ -393,6 +421,67 @@ def test_team_operations_integration(self): assert "integration-2" not in team_names assert "integration-3" in team_names + @pytest.mark.parametrize( + "bad_name", + [ + pytest.param("a", id="too-short"), + pytest.param("Team1", id="upper-case"), + pytest.param("team__x", id="consecutive-underscores"), + pytest.param("team1\n", id="trailing-newline"), + ], + ) + def test_team_sync_rejects_an_invalid_bundle_team_name(self, bad_name): + """``teams sync`` is a second creation path, so it holds names to the same rule. + + Unlike ``teams create`` this path does not strip its input, so a trailing newline + reaches the pattern -- which is why the check is ``re.fullmatch`` against an unanchored + pattern rather than ``re.match`` against a ``$``-anchored one. + """ + bundle_config = [ + { + "name": "bundleone", + "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle", + "kwargs": {"path": "/dev/null", "refresh_interval": 0}, + "team_name": bad_name, + }, + ] + + with conf_vars( + { + ("core", "multi_team"): "True", + ("dag_processor", "dag_bundle_config_list"): json.dumps(bundle_config), + } + ): + with pytest.raises(SystemExit, match="Invalid team name"): + team_command.team_sync(self.parser.parse_args(["teams", "sync"])) + + assert self.session.scalars(select(Team)).all() == [] + + def test_team_sync_rejects_an_invalid_name_already_stored(self): + """``teams sync`` shipped with no validation, so stored names can already break the rule.""" + self.session.add(Team(name="Data-Eng")) + self.session.commit() + + bundle_config = [ + { + "name": "bundleone", + "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle", + "kwargs": {"path": "/dev/null", "refresh_interval": 0}, + "team_name": "team1", + }, + ] + + with conf_vars( + { + ("core", "multi_team"): "True", + ("dag_processor", "dag_bundle_config_list"): json.dumps(bundle_config), + } + ): + with pytest.raises(SystemExit, match="Data-Eng"): + team_command.team_sync(self.parser.parse_args(["teams", "sync"])) + + assert self.session.scalar(select(Team).where(Team.name == "team1")) is None + def test_team_sync(self): bundle_config = [ { diff --git a/airflow-core/tests/unit/migrations/test_0131_lower_case_team_names.py b/airflow-core/tests/unit/migrations/test_0131_lower_case_team_names.py new file mode 100644 index 0000000000000..f536ebc094897 --- /dev/null +++ b/airflow-core/tests/unit/migrations/test_0131_lower_case_team_names.py @@ -0,0 +1,109 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +""" +Tests for migration 0131 (c7f0a5d2e9b4), which lower-cases stored team names. + +``team.name`` is a primary key whose foreign keys do not cascade on update, so the rename has +to carry every referring row with it. A row left behind would either break the constraint or +strand that team's Connections, Variables and Pools under a name nothing resolves. +""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path + +import pytest +import sqlalchemy as sa +from alembic import command + +from airflow import settings +from airflow.models import Connection, Pool, Variable +from airflow.models.team import Team +from airflow.utils.db import _get_alembic_config +from airflow.utils.session import create_session + +from tests_common.test_utils.paths import AIRFLOW_CORE_SOURCES_PATH + +pytestmark = pytest.mark.db_test + +# Migration filenames start with a digit so they cannot be imported via the normal import +# system; load the module by file path instead. +_MIGRATION_PATH = ( + Path(AIRFLOW_CORE_SOURCES_PATH) / "airflow/migrations/versions/0131_3_4_0_lower_case_team_names.py" +) +_spec = importlib.util.spec_from_file_location("migration_0131", _MIGRATION_PATH) +_migration = importlib.util.module_from_spec(_spec) # type: ignore[arg-type] +_spec.loader.exec_module(_migration) # type: ignore[union-attr] + +_OLD = "Data-Eng" +_NEW = "data-eng" + + +def _team_names(conn) -> set[str]: + return {row.name for row in conn.execute(sa.text("SELECT name FROM team"))} + + +class TestMigration0131: + def test_rename_carries_referring_rows_with_the_team(self): + with create_session() as session: + session.add(Team(name=_OLD)) + session.flush() + session.add_all( + [ + Connection(conn_id="c_0131", conn_type="http", team_name=_OLD), + Variable(key="v_0131", val="x", team_name=_OLD), + Pool(pool="p_0131", slots=1, description="", include_deferred=False, team_name=_OLD), + ] + ) + session.commit() + + try: + with settings.engine.begin() as conn: + for statement in _migration.build_lower_casing_statements(): + conn.execute(statement) + + with settings.engine.connect() as conn: + assert _OLD not in _team_names(conn) + assert _NEW in _team_names(conn) + for table, key_column, key in ( + ("connection", "conn_id", "c_0131"), + ("variable", "key", "v_0131"), + ("slot_pool", "pool", "p_0131"), + ): + team_name = conn.execute( + sa.text(f"SELECT team_name FROM {table} WHERE {key_column} = :k"), {"k": key} + ).scalar() + assert team_name == _NEW, table + finally: + with settings.engine.begin() as conn: + conn.execute(sa.text("DELETE FROM connection WHERE conn_id = 'c_0131'")) + conn.execute(sa.text("DELETE FROM variable WHERE key = 'v_0131'")) + conn.execute(sa.text("DELETE FROM slot_pool WHERE pool = 'p_0131'")) + conn.execute( + sa.text("DELETE FROM team WHERE name IN (:old, :new)"), {"old": _OLD, "new": _NEW} + ) + + def test_upgrade_emits_sql_offline(self, capsys): + """``db migrate --show-sql-only`` has no rows to read, so the migration must not read any.""" + command.upgrade(_get_alembic_config(), f"{_migration.down_revision}:{_migration.revision}", sql=True) + + emitted = capsys.readouterr().out + assert "INSERT INTO team (name) SELECT lower(team.name)" in emitted + assert "UPDATE team SET name=lower(team.name)" in emitted diff --git a/airflow-core/tests/unit/utils/test_db.py b/airflow-core/tests/unit/utils/test_db.py index c81859edbdcfa..a33f7b667f43b 100644 --- a/airflow-core/tests/unit/utils/test_db.py +++ b/airflow-core/tests/unit/utils/test_db.py @@ -31,16 +31,18 @@ from alembic.migration import MigrationContext from alembic.runtime.environment import EnvironmentContext from alembic.script import ScriptDirectory -from sqlalchemy import Column, Integer, MetaData, Table, select +from sqlalchemy import Column, Integer, MetaData, Table, delete, select from airflow import settings from airflow.models import Base as airflow_base +from airflow.models.team import Team from airflow.utils.db import ( AutocommitEngineForMySQL, LazySelectSequence, _get_alembic_config, _get_current_revision, check_migrations, + check_team_names_can_be_lower_cased, compare_server_default, compare_type, create_default_connections, @@ -50,6 +52,7 @@ upgradedb, ) from airflow.utils.db_manager import RunDBManager +from airflow.utils.session import create_session from tests_common.test_utils.config import conf_vars @@ -237,6 +240,36 @@ def test_check_migrations(self): check_migrations(0) check_migrations(1) + @pytest.mark.usefixtures("initialized_db") + def test_check_team_names_can_be_lower_cased_passes_when_no_names_collide(self): + with create_session() as session: + session.add_all([Team(name="data-eng"), Team(name="Platform")]) + session.commit() + try: + assert list(check_team_names_can_be_lower_cased(session=session)) == [] + finally: + session.execute(delete(Team)) + session.commit() + + # MySQL's default collation is case insensitive, so the two rows this needs cannot coexist + # there -- which is also why the collision it guards against cannot arise on MySQL. + @pytest.mark.backend("postgres", "sqlite") + @pytest.mark.usefixtures("initialized_db") + def test_check_team_names_can_be_lower_cased_reports_a_collision(self): + """Two names folding onto one row would merge two teams, so migration has to stop.""" + with create_session() as session: + session.add_all([Team(name="data-eng"), Team(name="Data-Eng")]) + session.commit() + try: + errors = list(check_team_names_can_be_lower_cased(session=session)) + finally: + session.execute(delete(Team)) + session.commit() + + assert len(errors) == 1 + assert "'Data-Eng', 'data-eng'" in errors[0] + assert "would both become 'data-eng'" in errors[0] + @pytest.mark.parametrize( ("auth", "expected"), [