Repository: apache/airflow
File: airflow-core/src/airflow/api_fastapi/core_api/security.py (Python)

Code before:
ReadableTagsFilterDep = Annotated[
    PermittedTagFilter, Depends(permitted_dag_filter_factory("GET", PermittedTagFilter))
]


def requires_access_backfill(method: ResourceMethod) -> Callable[[Request, BaseUser, Session], None]:
    """Wrap ``requires_access_dag`` and extract the dag_id from the backfill_id."""

    async def inner(
        request: Request,
        user: GetUserDep,
        session: SessionDep,
    ) -> None:
        dag_id = None

        # Try to retrieve the dag_id from the backfill_id path param
        backfill_id = request.path_params.get("backfill_id")
        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)
        if dag_id is None:
            try:
                dag_id = (await request.json()).get("dag_id")
            except JSONDecodeError:
                # Not a json body, ignore
                pass

        requires_access_dag(method, DagAccessEntity.RUN, dag_id)(
            request=request,
            user=user,
        )

    return inner


class PermittedPoolFilter(OrmClause[set[str]]):
    """A parameter that filters the permitted pools for the user."""

    def to_orm(self, select: Select) -> Select:
        return select.where(Pool.pool.in_(self.value or set()))


def permitted_pool_filter_factory(
    method: ResourceMethod,
) -> Callable[[BaseUser, BaseAuthManager], PermittedPoolFilter]:
    """
    Create a callable for Depends in FastAPI that returns a filter of the permitted pools for the user.

    :param method: whether filter readable or writable.

Review comment (bug):
Fixes a bug, we should check access control based on the 'dag_id' in the payload when we try to create a backfill.

Code after:
]

ReadableTagsFilterDep = Annotated[
    PermittedTagFilter, Depends(permitted_dag_filter_factory("GET", PermittedTagFilter))
]


def requires_access_backfill(
    method: ResourceMethod,
) -> Callable[[Request, BaseUser, Session], Coroutine[Any, Any, None]]:
    """Wrap ``requires_access_dag`` and extract the dag_id from the backfill_id."""

    async def inner(
        request: Request,
        user: GetUserDep,
        session: SessionDep,
    ) -> None:
        dag_id = None

        # Try to retrieve the dag_id from the backfill_id path param
        backfill_id = request.path_params.get("backfill_id")
        if backfill_id is not None and isinstance(backfill_id, int):
            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)
        if dag_id is None:
            try:
                dag_id = (await request.json()).get("dag_id")
            except JSONDecodeError:
                # Not a json body, ignore
                pass

        requires_access_dag(method, DagAccessEntity.RUN, dag_id)(
            request,
            user,
        )

    return inner


class PermittedPoolFilter(OrmClause[set[str]]):
    """A parameter that filters the permitted pools for the user."""

    def to_orm(self, select: Select) -> Select:
        return select.where(Pool.pool.in_(self.value or set()))


def permitted_pool_filter_factory(
    method: ResourceMethod,
) -> Callable[[BaseUser, BaseAuthManager], PermittedPoolFilter]: