Use PKCE for Weheat OAuth2 authorization (#181881)

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

homeassistant/components/weheat/application_credentials.py

@@ -1,11 +1,23 @@
 """application_credentials platform the Weheat integration."""
 
-from homeassistant.components.application_credentials import AuthorizationServer
+from homeassistant.components.application_credentials import ClientCredential
 from homeassistant.core import HomeAssistant
+from homeassistant.helpers.config_entry_oauth2_flow import (
+    LocalOAuth2ImplementationWithPkce,
+)
 
 from .const import OAUTH2_AUTHORIZE, OAUTH2_TOKEN
 
 
-async def async_get_authorization_server(hass: HomeAssistant) -> AuthorizationServer:
-    """Return authorization server."""
-    return AuthorizationServer(authorize_url=OAUTH2_AUTHORIZE, token_url=OAUTH2_TOKEN)
+async def async_get_auth_implementation(
+    hass: HomeAssistant, auth_domain: str, credential: ClientCredential
+) -> LocalOAuth2ImplementationWithPkce:
+    """Return auth implementation with PKCE support."""
+    return LocalOAuth2ImplementationWithPkce(
+        hass,
+        auth_domain,
+        credential.client_id,
+        OAUTH2_AUTHORIZE,
+        OAUTH2_TOKEN,
+        credential.client_secret,
+    )

tests/components/weheat/test_config_flow.py

@@ -4,11 +4,13 @@ from unittest.mock import AsyncMock, patch
 
 import pytest
 from weheat.exceptions import ApiException
+from yarl import URL
 
 from homeassistant.components.weheat.const import (
     DOMAIN,
     ENTRY_TITLE,
     OAUTH2_AUTHORIZE,
+    OAUTH2_SCOPES,
     OAUTH2_TOKEN,
 )
 from homeassistant.config_entries import SOURCE_USER, ConfigFlowResult
@@ -58,6 +60,10 @@ async def test_full_flow(
     assert len(mock_setup_entry.mock_calls) == 1
     assert len(mock_weheat.mock_calls) == 1
 
+    token_request_data = aioclient_mock.mock_calls[-1][2]
+    assert token_request_data["grant_type"] == "authorization_code"
+    assert token_request_data["code_verifier"]
+
     assert result["type"] is FlowResultType.CREATE_ENTRY
     assert result["result"].unique_id == USER_UUID_1
     assert result["result"].title == ENTRY_TITLE
@@ -189,12 +195,17 @@ async def handle_oauth(
         },
     )
 
-    assert result["url"] == (
-        f"{OAUTH2_AUTHORIZE}?response_type=code&client_id={CLIENT_ID}"
-        "&redirect_uri=https://example.com/auth/external/callback"
-        f"&state={state}"
-        "&scope=openid+offline_access"
+    result_url = URL(result["url"])
+    assert f"{result_url.origin()}{result_url.path}" == OAUTH2_AUTHORIZE
+    assert result_url.query["response_type"] == "code"
+    assert result_url.query["client_id"] == CLIENT_ID
+    assert (
+        result_url.query["redirect_uri"] == "https://example.com/auth/external/callback"
     )
+    assert result_url.query["state"] == state
+    assert result_url.query["scope"] == " ".join(OAUTH2_SCOPES)
+    assert result_url.query["code_challenge"]
+    assert result_url.query["code_challenge_method"] == "S256"
 
     client = await hass_client_no_auth()
     resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")