Auth manager cannot push dag authorization into SQL, so list views enumerate every dag

[1fanwang]

Description

BaseAuthManager.filter_authorized_dag_ids returns set[str], and core's only consumer of that set turns it straight into a SQL predicate:

class PermittedDagFilter(OrmClause[set[str]]):
    def to_orm(self, select: Select) -> Select:
        return select.where(DagModel.dag_id.in_(self.value or set()))

core_api/security.py#L246-L251

So an auth manager is asked to materialise every authorized dag id in the deployment, purely so one IN (...) clause can be built. The set is never used for anything else.

Two properties make this expensive:

1. It is deployment-wide, not page-wide. get_authorized_dag_ids selects every dag row with no limit (base_auth_manager.py#L643-L666).
2. It resolves before pagination exists. It is wired as a FastAPI dependency (security.py#L332-L343), so limit/offset are applied to the query after the authorized set is built.

A manager whose policy lives in a database therefore cannot say "just join against my table" — the return type cannot express it. It has to enumerate.

Reported impact

This is already being hit well below large-deployment scale:

• Very slow Keycloak auth manager responses #61686 — Keycloak, 600 dags, ~25s responses. Closed after per-manager caching and connection pooling; the reporter noted a real fix "is quite complicated to implement".
• Multi Team: /dags screen is very slow to load with multiple teams #69041 — multi-team, hundreds of dags across 10 teams, ~10s /dags load.
• Implement bulk authorization methods in KeycloakAuthManager #70582 — bulk authorization methods added to the Keycloak manager.

Each fix so far has been inside one manager. The interface is unchanged, so every new manager rediscovers the problem.

Proposal

Let an auth manager contribute a SQLAlchemy predicate instead of a materialised set, with the current behaviour as the default so nothing breaks:

def authorized_dag_ids_clause(self, *, user, method="GET") -> ColumnElement[bool] | None:
    """A predicate restricting DagModel to the dags this user may access.

    Returning None (the default) keeps the existing behaviour: core calls
    get_authorized_dag_ids and builds an IN clause from the result.
    """
    return None

A manager backed by a table returns something like DagModel.dag_id.in_(select(...)), and the whole filter becomes one join the database plans, with LIMIT/OFFSET applied in the same statement. Managers that do not override it are unaffected.

What this does and does not fix

Fixes: any manager whose policy is in the metadata database. FAB is the obvious one — it currently overrides get_authorized_dag_ids and enumerates, when its permission tables are joinable.

Does not fix: managers backed by an external PDP (Amazon Verified Permissions, Keycloak Authorization Services). Their policy cannot be expressed as SQL, so they still enumerate. #61686 would not be solved by this, and I do not want to overstate it. Those need either a reverse-lookup API on the PDP side ("which resources may this principal access?") or a local projection of the policy — both outside Airflow.

So this is the DB-backed half of the problem. It is the half Airflow can fix on its own.

Evidence

I have a manager running on a live cluster that reads per-dag grants out of the serialized dag. With a bulk override it resolves 5,000 dags in 6 SQL queries, 20.8ms; per-dag lookups against an external service on the same data are ~6.7ms per dag, which extrapolates to minutes at 41,606 dags. The difference is entirely whether the filter can be one query or must be N decisions.

Happy to prototype this if the shape seems reasonable. Wanted to check the interface direction before writing an API, since it touches a public extension point.