Repository: apache/airflow
File: providers/keycloak/src/airflow/providers/keycloak/auth_manager/keycloak_auth_manager.py (Python)

Code before:
        if self._token_expired(user.access_token):
            tokens = self.refresh_tokens(user=user)

            if tokens:
                user.refresh_token = tokens["refresh_token"]
                user.access_token = tokens["access_token"]
                return user

        return None

    def refresh_tokens(self, *, user: KeycloakAuthManagerUser) -> dict[str, str]:
        if not user.refresh_token:
            # It is a service account. It used the client credentials flow and no refresh token is issued.
            return {}

        try:
            log.debug("Refreshing the token")
            client = self.get_keycloak_client()
            return client.refresh_token(user.refresh_token)
        except KeycloakPostError as exc:
            if AIRFLOW_V_3_2_0_PLUS:
                from airflow.api_fastapi.auth.managers.exceptions import (
                    AuthManagerRefreshTokenExpiredException,
                )

                raise AuthManagerRefreshTokenExpiredException(exc)
            log.warning(
                "KeycloakPostError encountered during token refresh. "
                "Suppressing the exception and returning None.",
                exc_info=exc,
            )

            return {}

    def is_authorized_configuration(
        self,
        *,
        method: ResourceMethod,
        user: KeycloakAuthManagerUser,
        details: ConfigurationDetails | None = None,
    ) -> bool:
        config_section = details.section if details else None
        return self._is_authorized(
            method=method,
            resource_type=KeycloakResource.CONFIGURATION,
            user=user,
            resource_id=config_section,
        )

    def is_authorized_connection(
        self,

Review comment (security):
Generally in cases like this I'd say "try to import, catch the error" instead of a specific version check (in case we need to backport this etc)

```suggestion
            try:
                from airflow.api_fastapi.auth.managers.exceptions import (
                    AuthManagerRefreshTokenExpiredException,
                )
            except ImportError:
                pass
            else:
                raise AuthManagerRefreshTokenExpiredException(exc)
```

Code after:
            tokens = self.refresh_tokens(user=user)

            if tokens:
                user.refresh_token = tokens["refresh_token"]
                user.access_token = tokens["access_token"]
                return user

        return None

    def refresh_tokens(self, *, user: KeycloakAuthManagerUser) -> dict[str, str]:
        if not user.refresh_token:
            # It is a service account. It used the client credentials flow and no refresh token is issued.
            return {}

        try:
            log.debug("Refreshing the token")
            client = self.get_keycloak_client()
            return client.refresh_token(user.refresh_token)
        except KeycloakPostError as exc:
            try:
                from airflow.api_fastapi.auth.managers.exceptions import (
                    AuthManagerRefreshTokenExpiredException,
                )
            except ImportError:
                return {}
            else:
                raise AuthManagerRefreshTokenExpiredException(exc)

    def is_authorized_configuration(
        self,
        *,
        method: ResourceMethod,
        user: KeycloakAuthManagerUser,
        details: ConfigurationDetails | None = None,
    ) -> bool:
        config_section = details.section if details else None
        return self._is_authorized(
            method=method,
            resource_type=KeycloakResource.CONFIGURATION,
            user=user,
            resource_id=config_section,
        )

    def is_authorized_connection(
        self,
        *,
        method: ResourceMethod,
        user: KeycloakAuthManagerUser,
        details: ConnectionDetails | None = None,
    ) -> bool:
        connection_id = details.conn_id if details else None