feat(auth): [aiohttp] Add mTLS reconfiguration logic when certificate mismatch for existing credentials & Agent Identity workloads

[agrawalradhika-cell]

feat: [aiohttp] Add mTLS reconfiguration logic when certificate mismatch for existing credentials & Agent Identity workloads

Changes included:

• Adds mTLS reconfiguration logic when a certificate mismatch occurs.
• Aligns async session behavior with the synchronous session by triggering a credential refresh and request retry on all 401 Unauthorized responses (not just mTLS).

• Make sure to open an issue as a bug/issue before writing your code! That way we can discuss the change, evaluate designs, and agree on the general idea
• Ensure the tests and linter pass
• Code coverage does not decrease (if any source code was changed)
• Appropriate docs were updated (if necessary)

Fixes #18227 #18227 🦕

[gemini-code-assist]

Code Review

This pull request introduces client certificate rotation handling for asynchronous authorized sessions when encountering an unauthorized response under mTLS. The review feedback highlights a violation of the repository style guide regarding exception contract compliance, suggesting that the certificate parameter check should be wrapped in a try-except block to gracefully fall back to the original response rather than crashing. Additionally, the feedback recommends updating the corresponding unit tests to assert this resilient fallback behavior.

[lsirac]

Every rotation appends the old AiohttpRequest to self._old_auth_requests. These requests stay open until session.close(). For long running services with periodic certificate rotation, this list grows without bounds, leaking connection pools and open sockets. Prune or close old sessions in the background after in flight requests finish.

[agrawalradhika-cell]

Thanks for the feedback!

It was an intentional decision to tie the cleanup of old sessions to AsyncAuthorizedSession.close() for a few practical reasons:

• Low Frequency of Rotation: Certificate rotations for mTLS and Agent Identity typically occur on the scale of hours or days (e.g., every 12 to 24 hours). Even for a long-running service operating continuously for weeks or months, the list will only accumulate a very small number of stale session objects. The memory overhead is negligible in practice.
• Natural Connection Timeouts (No Socket Leaks): Retaining an aiohttp.ClientSession object in a list does not mean its underlying TCP sockets stay open forever. HTTP connection pools rely on keep-alive timeouts. Once the final in-flight request finishes, the connection sits idle in the pool. Shortly after, the client-side timeout or the server-side idle timeout will trigger and cleanly close the TCP socket. The lightweight session object is retained in Python memory, but the underlying system sockets and connection pool resources are properly freed.
• Complexity of Background Pruning: As noted in previous comments, closing the active ClientSession immediately aborts concurrent in-flight requests. Working around this by implementing a mechanism to "prune after in-flight requests finish" introduces significant architectural complexity. We would have to implement manual request reference-counting or orchestrate a background asyncio.Task to monitor the session. Background tasks in library code are notoriously tricky and carry a high risk of dangling task warnings on shutdown or unhandled exceptions.

Given the infrequency of rotations and the fact that idle sockets naturally time out and close on their own, accepting a slow-growing list of empty session objects seemed like the safest tradeoff compared to the complexity and risk of implementing background garbage collection.

Let me know if you still have concerns here

[agrawalradhika-cell]

The Kokoro System Tests failure is unrelated to the code change - https://btx.cloud.google.com/invocations/61f25d3a-d504-4b68-a0b8-38871927bd46/targets/cloud-devrel%2Fclient-libraries%2Fpython%2Fgoogleapis%2Fgoogle-cloud-python%2Fpresubmit%2Fsystem;config=default/log

[nolanleastin]

LGTM

[nbayati]

Passing headers=headers into the recursive self.request call forwards a dictionary that was already modified in place by self._credentials.before_request during the initial attempt.

To address this, consider creating a shallow copy on entry via request_headers = dict(headers) if headers is not None else {} for before_request and _auth_request, while passing the untouched original headers to the recursive retry (matching synchronous requests.py:L611, L706).

[nbayati]

Calling await self.configure_mtls_channel(self._client_cert_callback) discards the (call_cert_bytes, call_key_bytes) already retrieved and validated at lines 374–375.

Re-running discovery causes two problems:

1. It reads certificates from disk a second time unnecessarily, creating a race condition if files on disk change between check and use.
2. In Enterprise Certificate Provider (ECP) setups where self._client_cert_callback is None, discovery via aio.transport.mtls fails because it does not support ECP. This returns False and silently disables mTLS instead of rotating the certificate.

Note that passing a temporary callback directly (e.g. returning the validated bytes) will overwrite self._client_cert_callback at line 191, which breaks future rotations.

To fix this, consider either:

• Extracting transport recreation into a helper (e.g. _apply_client_cert(cert, key)) so the 401 handler can apply the validated bytes directly without touching self._client_cert_callback.
• Preserving self._client_cert_callback around configure_mtls_channel with a try...finally block if reusing the existing method.