feat(auth): implement regional access boundary support for standalone JWT and async service accounts

[nbayati]

This PR implements the following changes:

• Add RAB support to async service account credential type, by providing async manager and fetching methods.
• Update unit tests to accept both mtls and standard lookup endpoint urls.
• Refactor before_request to use a _after_refresh hook so we don't have to override the method.
• Add RAb support for self signed jwt flow through jwt.py
• some small enhancements for test coverage and backward compatibility

[gemini-code-assist]

Code Review

This pull request implements asynchronous support for Regional Access Boundary (RAB) management, including background refresh tasks and mTLS endpoint support. Key changes include the addition of _AsyncRegionalAccessBoundaryRefreshManager, updates to JWT and service account credentials to handle RAB state during cloning and serialization, and comprehensive test coverage for these new flows. However, a critical issue was identified where the refresh method in google/auth/jwt.py was renamed to _perform_refresh_token, which will break token updates as the base class expects a refresh method. Additionally, a typo was found in a test assertion URL.

[lsirac]

In google.auth.aio.transport.Response, the HTTP status code is exposed via the status_code property, not status. Passing a compliant google.auth.aio transport callable raises AttributeError: 'Response' object has no attribute 'status'. Please update the async lookup and grant methods to check .status_code.

[nbayati]

Done!

[nbayati]

Actually decided to only apply the fix to the RAB lookup, and instead open a bug to bring the grant methods up to date separately. This was can keep the blast radius of this PR smaller and limit it to only RAB changes without touching any token fetching flows.

[nbayati]

Created #17139 to track the necessary changes for the token endpoints.

[lsirac]

Unlike the synchronous refresh manager which safely deepcopies the transport (copied_request = copy.deepcopy(request)), the async manager passes the exact same request instance directly into the background coroutine task. Because start_refresh is invoked inside before_request, the main application coroutine immediately proceeds to make its actual service API HTTP call using the exact same request transport while the background task is concurrently using it, risking HTTP state corruption or interleaved headers.

Additionally, spawning asyncio.create_task(_worker()) without tracking cancellation hooks upon client session closure can potentially cause dangling tasks that raise RuntimeError: Session is closed when executing against closed client sessions.

[nbayati]

Both concerns are valid, but safe under the hood:

1. Deepcopying the Transport is impossible and not needed in async
   The async transport object (e.g., aiohttp_requests.Request) contains an active aiohttp.ClientSession with open TCP sockets. Attempting to copy.deepcopy(request) will instantly raise a TypeError: cannot pickle 'ClientSession' object and crash the application at runtime. Unlike synchronous transports running on separate OS threads, async HTTP clients (like aiohttp or httpx) are natively designed to handle concurrent requests sharing the same connection session. All request-specific states (headers, payloads) are stored in localized coroutine call stack frames, preventing any HTTP state corruption or interleaving.

2. Session closure and dangling tasks are handled safely
   The background worker is a single-shot asyncio task that executes exactly one lookup request and then immediately terminates and gets garbage-collected.
   If the user's application closes the underlying client session while a background task is still running, the resulting RuntimeError: Session is closed is safely caught by the worker's generic except Exception as e: block. It logs a warning, fails open cleanly, and does not raise an unhandled exception or crash the application.

I think no code changes are required here, as the current design is fully protected.

[lsirac]

Can you remind me what prompted the mTLS addition?

We should have test(s) that set that signal and makes sure the right endpoint is called on refresh etc.

[nbayati]

I realized that we only had hardcoded the non-mtls endpoint, while the other existing endpoints switch between mtls and non-mtls dynalically at runtime. I added the same treatment to the RAB URL, however, based on my recent talk with Yao, it seems like it's safe to always just send the request to the mtls one. If that ends up being the case, we can simplify the code and always call the mtls RAB endpoint.

[nbayati]

Leaving this comment thread open, so I either come back and add more unit tests as you requested, or simplify the logic and update the existing unit tests.

[nbayati]

After further investigation, it turned out that we couldn't just always call mtls endpoint, and the library had to switch dynamically between the two. I did a refactoring to move the RAB endpoints to the RAB file and out of iam. I'm filing an issue to add unit tests to iam, and have them dynamically switch too, because the current static decision that happens in the beginning of the run won't cover the GCE's CUJ for dynamically changing identity to MWLID. I've opened a separate issue (#17282) to migrate the rest of the endpoints.