Refuse only ids whose leading segment could name a team

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

@@ -48,13 +48,16 @@ def _show_teams(teams, output):
     )
 
 
+TEAM_NAME_PATTERN = r"^[a-zA-Z0-9_-]{3,50}$"
+
+
 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
 
 
@@ -169,6 +172,18 @@ 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}."
+        )
+
     teams_added = 0
 
     try:

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.