Implement bulk authorization methods in KeycloakAuthManager

[abhishekmauryaKsolves]

Closes #70582

Implements the missing batch/filter authorization methods in
KeycloakAuthManager, following the existing pattern used by
filter_authorized_dag_ids (parallel HTTP calls via ThreadPoolExecutor,
with caching via single_flight).

• batch_is_authorized_connection
• batch_is_authorized_dag
• batch_is_authorized_pool
• batch_is_authorized_variable
• filter_authorized_connections
• filter_authorized_pools
• filter_authorized_variables

Tests added covering normal filtering, empty input, all-denied, and
cache-hit scenarios for each new filter method.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[stephen-bracken]

This implementation matches the implementation in filter_authorized_dag_ids, which is good for consistency, but I think we should utilise the _is_batch_authorized method to streamline the checks here, and avoid calling keycloak N times for N resources

[abhishekmauryaKsolves]

Thanks for the suggestion! Looking at _is_batch_authorized/_get_batch_payload, it doesn't currently support per-permission context attributes (claim_token), which is_authorized_dag/is_authorized_connection/etc rely on for resource_id-level checks. Would extending _get_batch_payload to support per-item attributes be in scope here, or should this refactor apply only where resource_id isn't needed?

[stephen-bracken]

looking at the code, I'm not clear on how the resource_id is being used in policy enforcement, given that the standard keycloak client uses resource type to make decisions based on role / group membership. I think its therefore fine to drop it from the batch authorization requests. Happy to be corrected if I've missed something.

[abhishekmauryaKsolves]

Thanks for the pointer to _is_batch_authorized — I've refactored batch_is_authorized_connection, batch_is_authorized_dag, batch_is_authorized_pool, batch_is_authorized_variable, filter_authorized_connections, filter_authorized_pools, and filter_authorized_variables to use it. This reduces each of these to a single Keycloak call instead of N parallel calls.

A few things worth flagging explicitly since they change behavior:

1. filter_authorized_* is now all-or-nothing. Since resource_id isn't used for policy enforcement (per your earlier comment), the batch permission check happens at the resource-type level, not per-item. That means filter_authorized_connections, filter_authorized_pools, and filter_authorized_variables now return either the full input set or an empty set — they can no longer do true per-item filtering (e.g. "user has access to conn_0 and conn_2 but not conn_1" is no longer expressible). I've updated the unit tests to reflect this, but wanted to confirm this is the intended/acceptable tradeoff before this merges, since it's a real functional change vs. the previous per-item behavior.

2. dag_entity attribute dropped from batch_is_authorized_dag. The old is_authorized_dag-based implementation passed attributes={"dag_entity": ...} per-request. The new batch call doesn't currently support per-item attributes, so this is dropped. I've marked the corresponding test as xfail for now — let me know if this attribute needs to be preserved (would mean extending _get_batch_payload/_is_batch_authorized to accept per-item attributes) or if it's safe to drop, same as resource_id.

3. Caching removed. The single_flight cache-based dedup that existed in the old filter_authorized_connections/pools/variables is gone in the batch versions, since there's now one call per invocation instead of N. I dropped the 3 cache-hit tests that assumed this caching existed — happy to add it back around the batch call if it's still wanted.

Let me know your thoughts on 1 and 2 in particular — I can adjust the implementation depending on which tradeoffs are acceptable.

[vincbeck]

Given the last comment, I am requesting some changes. The goal of Keycloak auth manager is to have an auth manager that support fine grained access. I have not looked at the code but given this comment , the last developments actually REMOVED the fine grained access feature. Big no no

[abhishekmauryaKsolves]

Digging into this further to understand the actual conflict between the two suggestions:

I confirmed that Keycloak resources are registered generically per-type, not per-instance — see resources.py's KeycloakResource enum and commands.py:350 (standard_resources = [(resource.value, scopes) for resource in KeycloakResource]). There's a single "Connection" resource in Keycloak, not one resource per connection ID. Per-instance authorization for connections/pools/variables/dags works entirely through the resource_id claim, pushed via claim_token in _get_payload (see _is_authorized, line ~422).

The problem: Keycloak's batch UMA-ticket endpoint (_get_batch_payload) takes a single permission array and does not support a claim_token at all — there's one shared claim context for the whole request, not one per permission item. That means there is no way to batch N different resource_id values into a single Keycloak call while still discriminating between them.

This is different from filter_authorized_menu_items, where each menu category (Assets, Connections, etc.) actually is its own registered Keycloak resource — no resource_id claim is needed there, since the resource name itself carries the identity, so batching works cleanly.

So @vincbeck is right that the current batch refactor silently drops fine-grained access — and @stephen-bracken's suggestion, while a great optimization for resources that are individually registered in Keycloak (like menu items), doesn't hold for connections/pools/variables/dags, which rely on the resource_id claim mechanism instead.

Proposal: revert batch_is_authorized_connection, batch_is_authorized_dag, batch_is_authorized_pool, batch_is_authorized_variable, filter_authorized_connections, filter_authorized_pools, and filter_authorized_variables back to the original N-parallel-calls-via-is_authorized_* approach (with ThreadPoolExecutor + single_flight caching, as before) — this is the only way to preserve per-instance checks given how these resources are registered. I'll keep _is_batch_authorized reserved for cases where the resource itself is individually registered in Keycloak.

Let me know if this matches your understanding, or if there's a way to register per-instance resources in Keycloak that I'm missing.

[vincbeck]

Digging into this further to understand the actual conflict between the two suggestions:

I confirmed that Keycloak resources are registered generically per-type, not per-instance — see resources.py's KeycloakResource enum and commands.py:350 (standard_resources = [(resource.value, scopes) for resource in KeycloakResource]). There's a single "Connection" resource in Keycloak, not one resource per connection ID. Per-instance authorization for connections/pools/variables/dags works entirely through the resource_id claim, pushed via claim_token in _get_payload (see _is_authorized, line ~422).

The problem: Keycloak's batch UMA-ticket endpoint (_get_batch_payload) takes a single permission array and does not support a claim_token at all — there's one shared claim context for the whole request, not one per permission item. That means there is no way to batch N different resource_id values into a single Keycloak call while still discriminating between them.

This is different from filter_authorized_menu_items, where each menu category (Assets, Connections, etc.) actually is its own registered Keycloak resource — no resource_id claim is needed there, since the resource name itself carries the identity, so batching works cleanly.

So @vincbeck is right that the current batch refactor silently drops fine-grained access — and @stephen-bracken's suggestion, while a great optimization for resources that are individually registered in Keycloak (like menu items), doesn't hold for connections/pools/variables/dags, which rely on the resource_id claim mechanism instead.

Proposal: revert batch_is_authorized_connection, batch_is_authorized_dag, batch_is_authorized_pool, batch_is_authorized_variable, filter_authorized_connections, filter_authorized_pools, and filter_authorized_variables back to the original N-parallel-calls-via-is_authorized_* approach (with ThreadPoolExecutor + single_flight caching, as before) — this is the only way to preserve per-instance checks given how these resources are registered. I'll keep _is_batch_authorized reserved for cases where the resource itself is individually registered in Keycloak.

Let me know if this matches your understanding, or if there's a way to register per-instance resources in Keycloak that I'm missing.

Yep please do that and we'll go from there. But then what is the change you are proposing? Please keep your explanations concise

[abhishekmauryaKsolves]

Reverted batch_is_authorized_connection, batch_is_authorized_dag, batch_is_authorized_pool, batch_is_authorized_variable, filter_authorized_connections, filter_authorized_pools, and filter_authorized_variables back to the original N-parallel-calls approach (ThreadPoolExecutor + single_flight caching) — fine-grained per-resource authorization is fully restored, no behavior change from before this PR touched them.

_is_batch_authorized stays unused by this PR now (it's only exercised by the pre-existing filter_authorized_menu_items, which is safe to batch since menu items are individually registered Keycloak resources).

Net result: this PR now only adds the new batch_is_authorized_* / filter_authorized_* public API surface (per #70582), with the same call pattern and semantics as the existing is_authorized_* methods underneath. All 248 tests pass.

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.