Resolve backfill_id in the access dependency with the type the routes declare

airflow-core/src/airflow/api_fastapi/core_api/security.py

@@ -27,6 +27,7 @@
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer, OAuth2PasswordBearer
 from itsdangerous import BadSignature, URLSafeSerializer
 from jwt import ExpiredSignatureError, InvalidTokenError
+from pydantic import NonNegativeInt, TypeAdapter, ValidationError
 from sqlalchemy import or_, select
 from sqlalchemy.orm import Session
 
@@ -390,6 +391,12 @@ def depends_readable_event_logs_filter(
 ]
 
 
+# The type the backfill routes declare for the `backfill_id` path parameter. Shared with
+# `requires_access_backfill` so the authorization decision parses the id exactly as the handler
+# does; see the comment there for why any divergence is a cross-Dag authorization bypass.
+_BACKFILL_ID_ADAPTER: TypeAdapter[NonNegativeInt] = TypeAdapter(NonNegativeInt)
+
+
 def requires_access_backfill(
     method: ResourceMethod,
 ) -> Callable[[Request, BaseUser, Session], Coroutine[Any, Any, None]]:
@@ -405,15 +412,25 @@ async def inner(
         # Try to retrieve the dag_id from the backfill_id path param
         backfill_id_raw = request.path_params.get("backfill_id")
         try:
-            backfill_id = int(backfill_id_raw) if backfill_id_raw is not None else None
-        except ValueError:
+            # Must parse exactly as the handler does (e.g. pydantic's lax mode coerces "1.0" to 1
+            # where int() raises), or the two can authorize and act on different backfills.
+            backfill_id = (
+                _BACKFILL_ID_ADAPTER.validate_python(backfill_id_raw) if backfill_id_raw is not None else None
+            )
+        except ValidationError:
+            # Rejected by the endpoint's parser too, so the handler cannot run: FastAPI answers
+            # 422 before it is reached. Left as None, preserving that response.
             backfill_id = None
 
         if backfill_id is not None:
             backfill = session.scalars(select(Backfill).where(Backfill.id == backfill_id)).one_or_none()
             dag_id = backfill.dag_id if backfill else None
 
         # Try to retrieve the dag_id from the request body (POST backfill)
+        # TODO: a backfill_id that parses but matches no row also lands here, so an unknown
+        # backfill is authorized against the body's dag_id and answers 404 where an unauthorized
+        # one answers 403 - disclosing which ids exist. Not exploitable for a cross-Dag action;
+        # tracked at https://github.com/apache/airflow/issues/71080
         if dag_id is None:
             # Not a json body, ignore
             with suppress(JSONDecodeError):

airflow-core/tests/unit/api_fastapi/core_api/test_security.py

@@ -20,12 +20,14 @@
 from unittest.mock import AsyncMock, Mock, patch
 
 import pytest
-from fastapi import HTTPException
+from fastapi import HTTPException, Request
 from jwt import ExpiredSignatureError, InvalidTokenError
+from sqlalchemy.orm import Session
 
 from airflow import settings
 from airflow.api_fastapi.app import create_app
-from airflow.api_fastapi.auth.managers.base_auth_manager import COOKIE_NAME_JWT_TOKEN
+from airflow.api_fastapi.auth.managers.base_auth_manager import COOKIE_NAME_JWT_TOKEN, BaseAuthManager
+from airflow.api_fastapi.auth.managers.models.base_user import BaseUser
 from airflow.api_fastapi.auth.managers.models.resource_details import (
     AccessView,
     ConnectionDetails,
@@ -55,6 +57,7 @@
     resolve_user_from_token,
 )
 from airflow.models import Connection, Pool, Variable
+from airflow.models.backfill import Backfill
 from airflow.models.dag import DagModel
 from airflow.models.dagbundle import DagBundleModel
 from airflow.models.team import Team
@@ -367,7 +370,7 @@ async def test_requires_access_backfill_authorized_from_path(
     async def test_requires_access_backfill_authorized_from_body(
         self, mock_get_auth_manager, mock_get_team_name
     ):
-        """When backfill_id is missing or not int, dag_id can come from request body (POST backfill)."""
+        """With no backfill_id in the path, dag_id comes from the request body (POST backfill)."""
         auth_manager = Mock()
         auth_manager.is_authorized_dag.return_value = True
         mock_get_auth_manager.return_value = auth_manager
@@ -455,6 +458,48 @@ async def test_requires_access_backfill_backfill_not_found_falls_back_to_body(
             user=user,
         )
 
+    @pytest.mark.db_test
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("backfill_id", ["42", "42.0", "42.00"])
+    @patch.object(DagModel, "get_team_name")
+    @patch("airflow.api_fastapi.core_api.security.get_auth_manager")
+    async def test_requires_access_backfill_authorizes_the_backfill_the_handler_will_act_on(
+        self, mock_get_auth_manager, mock_get_team_name, backfill_id
+    ):
+        """The dependency must resolve the same backfill the handler does, for every spelling.
+
+        The endpoints declare ``backfill_id: NonNegativeInt``, and pydantic's lax mode coerces
+        ``"42.0"`` and ``"42.00"`` to ``42`` -- both are spellings the handler accepts and serves
+        against backfill 42. Parsing with ``int()`` here rejected them and left ``dag_id``
+        unresolved, so the two disagreed about which Dag the request concerned.
+        """
+        auth_manager = Mock(spec=BaseAuthManager)
+        auth_manager.is_authorized_dag.return_value = True
+        mock_get_auth_manager.return_value = auth_manager
+        mock_get_team_name.return_value = "team1"
+
+        backfill = Mock(spec=Backfill)
+        backfill.dag_id = "backfill_dag"
+        session = Mock(spec=Session)
+        session.scalars.return_value.one_or_none.return_value = backfill
+
+        request = Mock(spec=Request)
+        request.path_params = {"backfill_id": backfill_id}
+        request.query_params = {"dag_id": "some_other_dag"}
+        request.json = AsyncMock(return_value={"dag_id": "some_other_dag"})
+
+        user = Mock(spec=BaseUser)
+
+        await requires_access_backfill("PUT")(request, user, session)
+
+        # the backfill's own Dag, not the one supplied on the request
+        auth_manager.is_authorized_dag.assert_called_once_with(
+            method="PUT",
+            access_entity=DagAccessEntity.RUN,
+            details=DagDetails(id="backfill_dag", team_name="team1"),
+            user=user,
+        )
+
     @pytest.mark.db_test
     @pytest.mark.asyncio
     @patch.object(DagModel, "get_team_name")