Require team names to be lower case with no doubled underscores

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.                  |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+

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__<TEAM>___<ID>`` 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.

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))

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.

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__<TEAM>___<ID>``, 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,

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)

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 = [
             {