Fix Internal Server Error in API server due to KeycloakPostError when refreshing user in Keycloak provider

providers/keycloak/src/airflow/providers/keycloak/auth_manager/keycloak_auth_manager.py

@@ -26,7 +26,7 @@
 
 import requests
 from fastapi import FastAPI
-from keycloak import KeycloakOpenID
+from keycloak import KeycloakOpenID, KeycloakPostError
 from requests.adapters import HTTPAdapter
 from urllib3.util import Retry
 
@@ -149,12 +149,19 @@ def get_url_logout(self) -> str | None:
 
     def refresh_user(self, *, user: KeycloakAuthManagerUser) -> KeycloakAuthManagerUser | None:
         if self._token_expired(user.access_token):
-            log.debug("Refreshing the token")
-            client = self.get_keycloak_client()
-            tokens = client.refresh_token(user.refresh_token)
-            user.refresh_token = tokens["refresh_token"]
-            user.access_token = tokens["access_token"]
-            return user
+            try:
+                log.debug("Refreshing the token")
+                client = self.get_keycloak_client()
+                tokens = client.refresh_token(user.refresh_token)
+                user.refresh_token = tokens["refresh_token"]
+                user.access_token = tokens["access_token"]
+                return user
+            except KeycloakPostError as exc:
+                log.warning(
+                    "KeycloakPostError encountered during token refresh. "
+                    "Suppressing the exception and returning None.",
+                    exc_info=exc,
+                )
 
         return None
 

providers/keycloak/tests/unit/keycloak/auth_manager/test_keycloak_auth_manager.py

@@ -20,6 +20,7 @@
 from unittest.mock import Mock, patch
 
 import pytest
+from keycloak import KeycloakPostError
 
 from airflow.api_fastapi.app import AUTH_MANAGER_FASTAPI_APP_PREFIX
 from airflow.api_fastapi.auth.managers.models.resource_details import (
@@ -133,6 +134,24 @@ def test_refresh_user_expired(self, mock_token_expired, mock_get_keycloak_client
         assert result.access_token == "new_access_token"
         assert result.refresh_token == "new_refresh_token"
 
+    @patch.object(KeycloakAuthManager, "get_keycloak_client")
+    @patch.object(KeycloakAuthManager, "_token_expired")
+    def test_refresh_user_expired_with_invalid_token(
+        self, mock_token_expired, mock_get_keycloak_client, auth_manager, user
+    ):
+        mock_token_expired.return_value = True
+        keycloak_client = Mock()
+        keycloak_client.refresh_token.side_effect = KeycloakPostError(
+            response_code=400,
+            response_body=b'{"error":"invalid_grant","error_description":"Token is not active"}',
+        )
+
+        mock_get_keycloak_client.return_value = keycloak_client
+
+        assert auth_manager.refresh_user(user=user) is None
+
+        keycloak_client.refresh_token.assert_called_with("refresh_token")
+
     @pytest.mark.parametrize(
         ("function", "method", "details", "permission", "attributes"),
         [