Reject team names that differ only in case

airflow-core/src/airflow/cli/commands/team_command.py

@@ -48,13 +48,40 @@ def _show_teams(teams, output):
     )
 
 
+TEAM_NAME_PATTERN = r"^[a-zA-Z0-9_-]{3,50}$"
+
+
+def _case_collision(team_name: str, existing_names) -> str | None:
+    """
+    Return an existing team name that shares ``team_name``'s secrets namespace, if any.
+
+    Team names are case sensitive -- ``data_eng`` and ``Data_Eng`` are two distinct teams --
+    but the environment secrets backend upper-cases the team name when it builds the variable
+    name, so both resolve ``AIRFLOW_CONN__DATA_ENG___<ID>``. Two such teams therefore share a
+    single namespace and each reads the other's Connections and Variables.
+
+    Whether the two rows can coexist at all is decided by the database's collation on a
+    ``String`` primary key rather than by Airflow, so the same configuration behaves
+    differently across backends. Rejecting the collision here makes the behaviour uniform and
+    fails at the moment the name is chosen, rather than silently merging two teams' secrets.
+
+    ``upper()`` rather than ``casefold()``: it is what the backend applies, and team names are
+    restricted to ASCII by ``TEAM_NAME_PATTERN``.
+    """
+    folded = team_name.upper()
+    for existing in existing_names:
+        if existing != team_name and existing.upper() == folded:
+            return existing
+    return None
+
+
 def _extract_team_name(args):
     """Extract and validate team name from args."""
     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}$")
+    if not re.match(TEAM_NAME_PATTERN, team_name):
+        raise SystemExit(f"Invalid team name: must match regex {TEAM_NAME_PATTERN}")
     return team_name
 
 
@@ -69,6 +96,15 @@ def team_create(args, *, session=NEW_SESSION):
     if session.scalar(select(Team).where(Team.name == team_name)):
         raise SystemExit(f"Team with name '{team_name}' already exists")
 
+    collision = _case_collision(team_name, Team.get_all_team_names(session=session))
+    if collision:
+        raise SystemExit(
+            f"Team name '{team_name}' differs only in case from the existing team "
+            f"'{collision}'. They would share one secrets namespace, so each team could read "
+            f"the other's Connections and Variables. Choose a name that differs by more than "
+            f"case."
+        )
+
     # Create new team (UUID will be auto-generated by the database)
     new_team = Team(name=team_name)
 
@@ -169,10 +205,36 @@ def team_sync(args, *, session=NEW_SESSION):
         if bundle.team_name is not None
     }
 
+    # The bundle config is a second creation path for teams, so it has to enforce the same
+    # name rule as `teams create`. Other code relies on a stored team name being a valid one
+    # -- the environment secrets backend decides whether a supplied secret id could name a
+    # team namespace by testing the leading segment against this pattern, and an unvalidated
+    # short name such as "a" would make that test miss.
+    invalid = sorted(name for name in dag_bundle_teams if not re.match(TEAM_NAME_PATTERN, name))
+    if invalid:
+        raise SystemExit(
+            f"Invalid team name(s) in the dag bundle config: {', '.join(invalid)}. "
+            f"Team names must match regex {TEAM_NAME_PATTERN}."
+        )
+
+    existing_names = Team.get_all_team_names(session=session)
+    # Check the incoming names against each other as well as against the stored ones: a bundle
+    # config can introduce both halves of a collision in the same sync.
+    seen: set[str] = set(existing_names)
+    for team_name in sorted(dag_bundle_teams):
+        collision = _case_collision(team_name, seen)
+        if collision:
+            raise SystemExit(
+                f"Team name '{team_name}' in the dag bundle config differs only in case from "
+                f"'{collision}'. They would share one secrets namespace, so each team could "
+                f"read the other's Connections and Variables."
+            )
+        seen.add(team_name)
+
     teams_added = 0
 
     try:
-        for team_name in dag_bundle_teams - Team.get_all_team_names(session=session):
+        for team_name in dag_bundle_teams - existing_names:
             team = Team(name=team_name)
             session.add(team)
             teams_added += 1

airflow-core/src/airflow/secrets/environment_variables.py

@@ -27,6 +27,15 @@
 CONN_ENV_PREFIX = "AIRFLOW_CONN_"
 VAR_ENV_PREFIX = "AIRFLOW_VAR_"
 
+# Separator between the team name and the secret id in a team namespaced
+# environment variable name: AIRFLOW_CONN__<TEAM>___<ID>.
+TEAM_SEP = "___"
+
+# What ``airflow teams create`` accepts as a team name, so what a team
+# namespace can actually be spelled with.
+# Kept in sync with ``airflow.cli.commands.team_command``.
+_TEAM_NAME = re.compile(r"[a-zA-Z0-9_-]{3,50}")
+
 
 class EnvironmentVariablesBackend(BaseSecretsBackend):
     """Retrieves Connection object and Variable from environment variable."""
@@ -65,25 +74,36 @@ def get_variable(self, key: str, team_name: str | None = None) -> str | None:
     @staticmethod
     def _names_a_team_namespace(secret_id: str) -> bool:
         """
-        Whether ``secret_id`` spells out a team namespaced environment variable name.
+        Whether ``secret_id`` could spell out a team namespaced environment variable name.
 
         A team specific secret lives in the ``_<TEAM_NAME>___<SECRET_ID>`` namespace of the
-        environment. An id of that shape therefore makes the team agnostic lookup -- which
-        prepends only ``AIRFLOW_CONN_`` / ``AIRFLOW_VAR_`` -- land inside some team's namespace,
-        so that lookup is refused for such an id.
+        environment. An id of that shape makes the team agnostic lookup -- which prepends only
+        ``AIRFLOW_CONN_`` / ``AIRFLOW_VAR_`` -- land inside some team's namespace, so that
+        lookup is refused for such an id.
+
+        The test is whether the leading segment **could be a team name**, not merely whether
+        the id contains the separator. Team names are validated on creation, so an id like
+        ``_a___b`` cannot name a team namespace -- ``a`` is too short to be a team -- and
+        refusing it would block a legitimate team agnostic secret for no benefit. Every id
+        that does spell a real team's namespace has a valid team name in that position by
+        construction, so nothing reachable is let through.
 
         **The id is never attributed to a particular team**, because it cannot be: a team name
-        may itself contain the ``___`` separator, so ``_a___b___c`` is both team ``a`` with id
-        ``b___c`` and team ``a___b`` with id ``c``, and nothing in the string distinguishes them.
-        Only the caller's own namespace is ever constructed, never parsed.
-
-        This is why the check is not "does the id belong to some team other than the caller's".
-        Comparing the id against the prefix the caller's team builds looks equivalent and is not:
-        for a caller in team ``a`` the id ``_a___b___c`` starts with ``_A___``, yet the variable
-        it resolves, ``AIRFLOW_CONN__A___B___C``, is team ``a___b``'s. Treating a prefix match as
-        ownership hands one team the secrets of every team whose name extends it. Callers reach
-        their own team's secrets with the bare id plus their team scope -- handled above, and safe
-        because that path can only ever build the caller's own namespace -- so nothing legitimate
-        needs a namespaced id here.
+        may itself contain the separator, so ``_a___b___c`` is both team ``a`` with id
+        ``b___c`` and team ``a___b`` with id ``c``. Every split is therefore considered, and
+        one plausible team name is enough to refuse. Comparing the id against the prefix the
+        caller's own team builds looks equivalent and is not: for a caller in team ``a`` the
+        id ``_a___b___c`` starts with ``_A___``, yet the variable it resolves,
+        ``AIRFLOW_CONN__A___B___C``, is team ``a___b``'s. Treating a prefix match as ownership
+        hands one team the secrets of every team whose name extends it.
         """
-        return re.fullmatch(r"_.+___.+", secret_id) is not None
+        if not secret_id.startswith("_"):
+            return False
+        # Overlapping positions matter: ``_abc____x`` splits at both index 4 and 5.
+        for i in range(1, len(secret_id) - len(TEAM_SEP) + 1):
+            if secret_id[i : i + len(TEAM_SEP)] != TEAM_SEP:
+                continue
+            team, rest = secret_id[1:i], secret_id[i + len(TEAM_SEP) :]
+            if rest and _TEAM_NAME.fullmatch(team):
+                return True
+        return False

airflow-core/tests/unit/always/test_secrets_environment_variables.py

@@ -98,6 +98,22 @@ def test_id_spelling_out_a_team_namespace_is_never_resolved(
 
         assert lookup(env_prefix, method, team_scoped_id(team_name), team_name) is None
 
+    @pytest.mark.parametrize(("env_prefix", "method"), LOOKUPS)
+    @pytest.mark.parametrize("secret_id", ["_a___b", "_ab___x"])
+    @pytest.mark.parametrize("team_name", [None, *TEAM_NAMES])
+    def test_id_that_cannot_name_a_team_is_still_resolved(
+        self, monkeypatch, env_prefix, method, secret_id, team_name
+    ):
+        """Refusing every ``_x___y`` id would block legitimate team agnostic secrets.
+
+        Team names are validated as ``^[a-zA-Z0-9_-]{3,50}$``, so a one- or two-character
+        leading segment cannot be a team. Such an id names no team's namespace and must stay
+        resolvable through the team agnostic lookup, for any caller.
+        """
+        monkeypatch.setenv(env_prefix + secret_id.upper(), GLOBAL_VALUE)
+
+        assert lookup(env_prefix, method, secret_id, team_name) == GLOBAL_VALUE
+
     @pytest.mark.parametrize(("env_prefix", "method"), LOOKUPS)
     def test_team_whose_name_extends_the_callers_is_not_readable(self, monkeypatch, env_prefix, method):
         """A team name may contain the ``___`` separator, so one team's namespace can start with another's.

airflow-core/tests/unit/cli/commands/test_team_command.py

@@ -88,6 +88,26 @@ 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"]))
 
+    def test_team_create_rejects_a_name_differing_only_in_case(self):
+        """Two teams differing only in case would share one secrets namespace.
+
+        The environment secrets backend upper-cases the team name when building the variable
+        name, so ``data_eng`` and ``Data_Eng`` both resolve ``AIRFLOW_CONN__DATA_ENG___<ID>``
+        and each team would read the other's Connections and Variables.
+        """
+        team_command.team_create(self.parser.parse_args(["teams", "create", "data_eng"]))
+
+        with pytest.raises(SystemExit, match="differs only in case from the existing team"):
+            team_command.team_create(self.parser.parse_args(["teams", "create", "Data_Eng"]))
+
+        assert self.session.scalar(select(Team).where(Team.name == "Data_Eng")) is None
+
+    def test_team_create_allows_a_name_differing_by_more_than_case(self):
+        team_command.team_create(self.parser.parse_args(["teams", "create", "data_eng"]))
+        team_command.team_create(self.parser.parse_args(["teams", "create", "data_eng2"]))
+
+        assert self.session.scalar(select(Team).where(Team.name == "data_eng2")) is not 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"):