diff --git a/providers/keycloak/src/airflow/providers/keycloak/auth_manager/keycloak_auth_manager.py b/providers/keycloak/src/airflow/providers/keycloak/auth_manager/keycloak_auth_manager.py index 955b431742f5e..dc8de7c25954e 100644 --- a/providers/keycloak/src/airflow/providers/keycloak/auth_manager/keycloak_auth_manager.py +++ b/providers/keycloak/src/airflow/providers/keycloak/auth_manager/keycloak_auth_manager.py @@ -412,7 +412,8 @@ def _is_batch_authorized( @staticmethod def _get_token_url(server_url, realm): - return f"{server_url}/realms/{realm}/protocol/openid-connect/token" + # Normalize server_url to avoid double slashes (required for Keycloak 26.4+ strict path validation). + return f"{server_url.rstrip('/')}/realms/{realm}/protocol/openid-connect/token" @staticmethod def _get_payload(client_id: str, permission: str, attributes: dict[str, str] | None = None): diff --git a/providers/keycloak/tests/unit/keycloak/auth_manager/test_keycloak_auth_manager.py b/providers/keycloak/tests/unit/keycloak/auth_manager/test_keycloak_auth_manager.py index 10732a636a782..27747f506961b 100644 --- a/providers/keycloak/tests/unit/keycloak/auth_manager/test_keycloak_auth_manager.py +++ b/providers/keycloak/tests/unit/keycloak/auth_manager/test_keycloak_auth_manager.py @@ -559,3 +559,29 @@ def test_get_keycloak_client_with_no_credentials(self, mock_keycloak_openid, aut client_secret_key="client_secret", ) assert client == mock_keycloak_openid.return_value + + @pytest.mark.parametrize( + ("server_url", "expected_url"), + [ + ( + "https://keycloak.example.com/auth", + "https://keycloak.example.com/auth/realms/myrealm/protocol/openid-connect/token", + ), + ( + "https://keycloak.example.com/auth/", + "https://keycloak.example.com/auth/realms/myrealm/protocol/openid-connect/token", + ), + ( + "https://keycloak.example.com/auth///", + "https://keycloak.example.com/auth/realms/myrealm/protocol/openid-connect/token", + ), + ( + "https://keycloak.example.com/", + "https://keycloak.example.com/realms/myrealm/protocol/openid-connect/token", + ), + ], + ) + def test_get_token_url_normalization(self, auth_manager, server_url, expected_url): + """Test that _get_token_url normalizes server_url by stripping trailing slashes.""" + token_url = auth_manager._get_token_url(server_url, "myrealm") + assert token_url == expected_url