Regression with use of bearer token for keycloak

### Under which category would you file this issue?

Providers

### Apache Airflow version

3.3.0

### What happened and how to reproduce it?

This code worked with airflow 3.3.0 and apache-airflow-providers-keycloak==0.8.2 to initialize connections and pools:

```python
"""Create team-owned pilot Connections and pools through Airflow's public API."""

from __future__ import annotations

import json
import os
import time
import urllib.error
import urllib.parse
import urllib.request

BASE_URL = "http://airflow-apiserver:8080"

def request(method: str, path: str, token: str | None = None, body: dict | None = None):
data = json.dumps(body).encode() if body is not None else None
headers = {"Content-Type": "application/json"}
if token:
headers["Authorization"] = f"Bearer {token}"
req = urllib.request.Request(BASE_URL + path, data=data, headers=headers, method=method)
with urllib.request.urlopen(req, timeout=15) as response:
payload = response.read()
return json.loads(payload) if payload else None

def upsert(path: str, item_path: str, token: str, body: dict) -> None:
try:
# The auth manager cannot infer a team for a GET on an item that does
# not exist yet. Create first so authorization can use body.team_name;
# a conflict means this is an idempotent bootstrap rerun.
request("POST", path, token, body)
except urllib.error.HTTPError as exc:
if exc.code != 409:
raise
request("PATCH", item_path, token, body)

def main() -> None:
token_body = {
"grant_type": "client_credentials",
"client_id": "airflow-bootstrap",
"client_secret": os.environ["AIRFLOW_BOOTSTRAP_CLIENT_SECRET"],
}
for attempt in range(20):
try:
token = request("POST", "/auth/token", body=token_body)["access_token"]
break
except (urllib.error.URLError, KeyError):
if attempt == 19:
raise
time.sleep(2)

resources = [
(
"team-analytics",
"team-analytics__demo-service",
os.environ["TEAM_ANALYTICS_DEMO_PASSWORD"],
"team-analytics__pilot-pool",
),
]
for team, connection_id, password, pool_name in resources:
connection = {
"connection_id": connection_id,
"conn_type": "http",
"host": f"https://{team}.example.invalid",
"login": "pilot-service",
"password": password,
"extra": json.dumps({"pilot": True}),
"team_name": team,
}
upsert(
"/api/v2/connections",
f"/api/v2/connections/{urllib.parse.quote(connection_id, safe='')}",
token,
connection,
)
pool = {
"name": pool_name,
"slots": 4,
"description": f"Local pilot pool for {team}",
"include_deferred": False,
"team_name": team,
}
upsert(
"/api/v2/pools",
f"/api/v2/pools/{urllib.parse.quote(pool_name, safe='')}",
token,
pool,
)
print("Team-owned Connections and pools are ready.")

if __name__ == "__main__":
main()
```

However with both airflow 3.3.0 and 3.3.1 I get errors like:
```
File ".../airflow/api_fastapi/core_api/security.py", line 605, in inner
_requires_access(is_authorized_callback=_callback)
File ".../airflow/api_fastapi/core_api/security.py", line 1001, in _requires_access
if not is_authorized_callback():
File ".../airflow/api_fastapi/core_api/security.py", line 599, in _callback
return get_auth_manager().is_authorized_connection(
method=method,
details=ConnectionDetails(conn_id=connection_id, team_name=tn),
user=user,
)
File ".../airflow/providers/keycloak/auth_manager/keycloak_auth_manager.py", line 249, in is_authorized_connection
return self._is_authorized(
method=method,
...
team_name=team_name,
)
File ".../airflow/providers/keycloak/auth_manager/keycloak_auth_manager.py", line 471, in _is_authorized
headers=self._get_headers(user.access_token),
^^^^^^^^^^^^^^^^^
AttributeError: 'NoneType' object has no attribute 'access_token'
```

The initial error is:
```
airflow-apiserver-1 | 2026-08-31T19:49:19.922649Z [info ] request finished [http.access] client_addr=172.22.0.7:36610 duration_us=49458 loc=http_access_log.py:130 method=POST path=/auth/token query= status_code=201
airflow-apiserver-1 | 2026-08-31T19:49:19.926767Z [info ] request finished [http.access] client_addr=172.22.0.7:36626 duration_us=3211 loc=http_access_log.py:130 method=POST path=/api/v2/connections query= status_code=500
airflow-apiserver-1 | 2026-08-31T19:49:19.926926Z [error ] Exception in ASGI application
airflow-apiserver-1 | [uvicorn.error] loc=httptools_impl.py:427
airflow-apiserver-1 | + Exception Group Traceback (most recent call last):
airflow-apiserver-1 | | File "/home/airflow/.local/lib/python3.13/site-packages/starlette/_utils.py", line 85, in create_collapsing_task_group
airflow-apiserver-1 | | async with anyio.create_task_group() as tg:
airflow-apiserver-1 | | ~~~~~~~~~~~~~~~~~~~~~~~^^
airflow-apiserver-1 | | File "/home/airflow/.local/lib/python3.13/site-packages/anyio/_backends/_asyncio.py", line 815, in __aexit__
airflow-apiserver-1 | | raise BaseExceptionGroup(
airflow-apiserver-1 | | "unhandled errors in a TaskGroup", self._exceptions
airflow-apiserver-1 | | ) from None
```

Steps to reproduce:
1. Configure `KeycloakAuthManager` on Airflow ≥3.3.0 with `apache-airflow-providers-keycloak==0.9.0`.
2. Create a confidential Keycloak client (e.g. `airflow-bootstrap`) with a service account enabled (client-credentials grant), with a role/permission grant sufficient to create Connections and Pools for a team.
3. Run the script above against a running Airflow API server with `AIRFLOW_BOOTSTRAP_CLIENT_SECRET` and `TEAM_ANALYTICS_DEMO_PASSWORD` set appropriately.
4. The `POST /auth/token` call succeeds (HTTP 201) and returns a token.
5. The subsequent `POST /api/v2/connections` call fails with HTTP 500 / `AttributeError: 'NoneType' object has no attribute 'access_token'`.

Downgrading only `apache-airflow-providers-keycloak` to `0.8.2` (same Airflow core version, same script, same Keycloak client config) resolves the issue, isolating the regression to the provider package.

### What you think should happen instead?

Ideally, it would continue to work like the 0.8.2 version.

The rest is Claude's analysis of what went wrong (Claude was concerned "there's context I'm missing about why the `None` return is intentional for this case."):

## What do you think went wrong?

This is a conflict between two independent authentication code paths that both call `KeycloakAuthManager.get_user_from_token`, with incompatible expectations introduced in #70800 and refined in #70981.

**Path 1 — generic bearer/OAuth2 token auth** (`airflow/api_fastapi/core_api/security.py`, core Airflow, not keycloak-specific):

```python
async def resolve_user_from_token(token_str: str | None) -> BaseUser:
if not token_str:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
try:
return await get_auth_manager().get_user_from_token(token_str)
...
```

This is called from `get_user()`, which is what handles any `Authorization: Bearer ` or OAuth2 client-credentials caller — including the `/auth/token` client-credentials flow the keycloak provider's own docs describe for service-to-service API access. It calls `get_user_from_token` with a single positional argument.

**Path 2 — browser-cookie session refresh** (`KeycloakJWTMiddleware`, added in #70800):

```python
@staticmethod
async def _refresh_user(request: Request) -> tuple[...]:
jwt_token = request.cookies.get(COOKIE_NAME_JWT_TOKEN)
access_token = request.cookies.get(COOKIE_NAME_ACCESS_TOKEN)
refresh_token = request.cookies.get(COOKIE_NAME_REFRESH_TOKEN)
...
user = await auth_manager.get_user_from_token(jwt_token, access_token, refresh_token)
```

This path always supplies `access_token`/`refresh_token` sourced from dedicated cookies (`_access_token`, `_refresh_token`), per the cookie-splitting scheme introduced in #70800 and scoped to Airflow ≥3.3.0 in #70981.

**The override in `KeycloakAuthManager.get_user_from_token` only accounts for Path 2 once `AIRFLOW_V_3_3_PLUS` is true:**

```python
async def get_user_from_token(
self, token: str, access_token: str | None = None, refresh_token: str | None = None
):
user = cast("KeycloakAuthManagerUser", await super().get_user_from_token(token))
if not AIRFLOW_V_3_3_PLUS:
return user
if access_token:
user.access_token = access_token
user.refresh_token = refresh_token
return user
# Skip refreshing JWT if Keycloak JWTs are not included.
return None
```

When called via Path 1 (bearer/OAuth2 token, no separate cookies exist), `access_token` is `None` by default. `super().get_user_from_token(token)` successfully resolves a valid `user` object on the line above — but the method then discards it and returns `None` anyway, because it assumes the *absence* of a separate Keycloak-cookie access token means the caller isn't authenticated, rather than considering that the caller may simply be using a different, equally valid auth path that was never designed to carry that cookie in the first place.

`None` then propagates through `get_user` → `_requires_access` → `is_authorized_connection` → `_is_authorized`, which unconditionally does `user.access_token`, causing the `AttributeError`.

## What you think should happen instead?

`get_user_from_token` should not discard a validly-resolved `user` solely because the request didn't include a separate Keycloak-cookie `access_token`. A bearer-token/client-credentials caller that authenticates successfully via `super().get_user_from_token(token)` should remain authenticated; the cookie-splitting logic should only apply refresh-token bookkeeping when those cookies are present, not gate authentication success on their presence.

A minimal fix might look like:

```python
async def get_user_from_token(
self, token: str, access_token: str | None = None, refresh_token: str | None = None
):
user = cast("KeycloakAuthManagerUser", await super().get_user_from_token(token))
if not AIRFLOW_V_3_3_PLUS:
return user
if access_token:
user.access_token = access_token
user.refresh_token = refresh_token
return user
# No separate Keycloak cookies (e.g. bearer/client-credentials caller rather than
# a browser session) — the user is still authenticated, just not refreshable here.
return user
```

### Operating System

Debian GNU/Linux 12 (bookworm)

### Deployment

Docker-Compose

### Apache Airflow Provider(s)

keycloak

### Versions of Apache Airflow Providers

apache-airflow-providers-keycloak==0.9.0

### Official Helm Chart version

Not Applicable

### Kubernetes Version

_No response_

### Helm Chart configuration

_No response_

### Docker Image customizations

Switched to KeycloakAuthManager and using multi-teams.

### Anything else?

_No response_

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

Start in airflow/providers/keycloak/auth_manager/keycloak_auth_manager.py at get_user_from_token and trace the bearer-token path through airflow/api_fastapi/core_api/security.py. Compare calls with and without the cookie-derived access token, then inspect existing Keycloak auth-manager tests for the appropriate regression-test location. Done means client-credentials bearer requests remain authenticated and the Connections or Pools API no longer raises AttributeError.

Written by the indexing model from the issue text.