From 71f5f4bcddf8a98636f09eabb3219913fb9ddc45 Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Sat, 7 Dec 2024 05:43:37 +1000 Subject: [PATCH] Remove default OAuth implementation from Tesla Fleet (#132431) --- .../components/tesla_fleet/__init__.py | 7 -- .../components/tesla_fleet/config_flow.py | 6 -- homeassistant/components/tesla_fleet/oauth.py | 56 +------------ .../tesla_fleet/test_config_flow.py | 84 ++----------------- 4 files changed, 10 insertions(+), 143 deletions(-) diff --git a/homeassistant/components/tesla_fleet/__init__.py b/homeassistant/components/tesla_fleet/__init__.py index e7030b568b3..bc837aa4cac 100644 --- a/homeassistant/components/tesla_fleet/__init__.py +++ b/homeassistant/components/tesla_fleet/__init__.py @@ -34,7 +34,6 @@ from homeassistant.helpers.config_entry_oauth2_flow import ( import homeassistant.helpers.config_validation as cv from homeassistant.helpers.device_registry import DeviceInfo -from .config_flow import OAuth2FlowHandler from .const import DOMAIN, LOGGER, MODELS from .coordinator import ( TeslaFleetEnergySiteInfoCoordinator, @@ -42,7 +41,6 @@ from .coordinator import ( TeslaFleetVehicleDataCoordinator, ) from .models import TeslaFleetData, TeslaFleetEnergyData, TeslaFleetVehicleData -from .oauth import TeslaSystemImplementation PLATFORMS: Final = [ Platform.BINARY_SENSOR, @@ -73,11 +71,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslaFleetConfigEntry) - scopes: list[Scope] = [Scope(s) for s in token["scp"]] region: str = token["ou_code"].lower() - OAuth2FlowHandler.async_register_implementation( - hass, - TeslaSystemImplementation(hass), - ) - implementation = await async_get_config_entry_implementation(hass, entry) oauth_session = OAuth2Session(hass, entry, implementation) refresh_lock = asyncio.Lock() diff --git a/homeassistant/components/tesla_fleet/config_flow.py b/homeassistant/components/tesla_fleet/config_flow.py index ca36c6f511b..feeb5e74ca6 100644 --- a/homeassistant/components/tesla_fleet/config_flow.py +++ b/homeassistant/components/tesla_fleet/config_flow.py @@ -12,7 +12,6 @@ from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlowResult from homeassistant.helpers import config_entry_oauth2_flow from .const import DOMAIN, LOGGER -from .oauth import TeslaSystemImplementation class OAuth2FlowHandler( @@ -31,11 +30,6 @@ class OAuth2FlowHandler( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Handle a flow start.""" - self.async_register_implementation( - self.hass, - TeslaSystemImplementation(self.hass), - ) - return await super().async_step_user() async def async_oauth_create_entry( diff --git a/homeassistant/components/tesla_fleet/oauth.py b/homeassistant/components/tesla_fleet/oauth.py index 8b43460436b..b25c5216009 100644 --- a/homeassistant/components/tesla_fleet/oauth.py +++ b/homeassistant/components/tesla_fleet/oauth.py @@ -1,8 +1,5 @@ """Provide oauth implementations for the Tesla Fleet integration.""" -import base64 -import hashlib -import secrets from typing import Any from homeassistant.components.application_credentials import ( @@ -11,59 +8,8 @@ from homeassistant.components.application_credentials import ( ClientCredential, ) from homeassistant.core import HomeAssistant -from homeassistant.helpers import config_entry_oauth2_flow -from .const import AUTHORIZE_URL, CLIENT_ID, DOMAIN, SCOPES, TOKEN_URL - - -class TeslaSystemImplementation(config_entry_oauth2_flow.LocalOAuth2Implementation): - """Tesla Fleet API open source Oauth2 implementation.""" - - code_verifier: str - code_challenge: str - - def __init__(self, hass: HomeAssistant) -> None: - """Initialize open source Oauth2 implementation.""" - - # Setup PKCE - self.code_verifier = secrets.token_urlsafe(32) - hashed_verifier = hashlib.sha256(self.code_verifier.encode()).digest() - self.code_challenge = ( - base64.urlsafe_b64encode(hashed_verifier).decode().replace("=", "") - ) - super().__init__( - hass, - DOMAIN, - CLIENT_ID, - "", - AUTHORIZE_URL, - TOKEN_URL, - ) - - @property - def name(self) -> str: - """Name of the implementation.""" - return "Built-in open source client ID" - - @property - def extra_authorize_data(self) -> dict[str, Any]: - """Extra data that needs to be appended to the authorize url.""" - return { - "prompt": "login", - "scope": " ".join(SCOPES), - "code_challenge": self.code_challenge, # PKCE - } - - async def async_resolve_external_data(self, external_data: Any) -> dict: - """Resolve the authorization code to tokens.""" - return await self._token_request( - { - "grant_type": "authorization_code", - "code": external_data["code"], - "redirect_uri": external_data["state"]["redirect_uri"], - "code_verifier": self.code_verifier, # PKCE - } - ) +from .const import AUTHORIZE_URL, SCOPES, TOKEN_URL class TeslaUserImplementation(AuthImplementation): diff --git a/tests/components/tesla_fleet/test_config_flow.py b/tests/components/tesla_fleet/test_config_flow.py index b49e090cd5d..6cb8c60ac0c 100644 --- a/tests/components/tesla_fleet/test_config_flow.py +++ b/tests/components/tesla_fleet/test_config_flow.py @@ -11,7 +11,6 @@ from homeassistant.components.application_credentials import ( ) from homeassistant.components.tesla_fleet.const import ( AUTHORIZE_URL, - CLIENT_ID, DOMAIN, SCOPES, TOKEN_URL, @@ -52,69 +51,18 @@ async def access_token(hass: HomeAssistant) -> str: ) -@pytest.mark.usefixtures("current_request_with_host") -async def test_full_flow( - hass: HomeAssistant, - hass_client_no_auth: ClientSessionGenerator, - aioclient_mock: AiohttpClientMocker, - access_token: str, -) -> None: - """Check full flow.""" - result = await hass.config_entries.flow.async_init( - DOMAIN, context={"source": SOURCE_USER} - ) - - state = config_entry_oauth2_flow._encode_jwt( +@pytest.fixture(autouse=True) +async def create_credential(hass: HomeAssistant) -> None: + """Create a user credential.""" + # Create user application credential + assert await async_setup_component(hass, "application_credentials", {}) + await async_import_client_credential( hass, - { - "flow_id": result["flow_id"], - "redirect_uri": REDIRECT, - }, + DOMAIN, + ClientCredential("user_client_id", "user_client_secret"), + "user_cred", ) - assert result["type"] is FlowResultType.EXTERNAL_STEP - - assert result["url"].startswith(AUTHORIZE_URL) - parsed_url = urlparse(result["url"]) - parsed_query = parse_qs(parsed_url.query) - assert parsed_query["response_type"][0] == "code" - assert parsed_query["client_id"][0] == CLIENT_ID - assert parsed_query["redirect_uri"][0] == REDIRECT - assert parsed_query["state"][0] == state - assert parsed_query["scope"][0] == " ".join(SCOPES) - assert parsed_query["code_challenge"][0] is not None - - client = await hass_client_no_auth() - resp = await client.get(f"/auth/external/callback?code=abcd&state={state}") - assert resp.status == 200 - assert resp.headers["content-type"] == "text/html; charset=utf-8" - - aioclient_mock.clear_requests() - aioclient_mock.post( - TOKEN_URL, - json={ - "refresh_token": "mock-refresh-token", - "access_token": access_token, - "type": "Bearer", - "expires_in": 60, - }, - ) - with patch( - "homeassistant.components.tesla_fleet.async_setup_entry", return_value=True - ) as mock_setup: - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - - assert len(hass.config_entries.async_entries(DOMAIN)) == 1 - assert len(mock_setup.mock_calls) == 1 - - assert result["type"] is FlowResultType.CREATE_ENTRY - assert result["title"] == UNIQUE_ID - assert "result" in result - assert result["result"].unique_id == UNIQUE_ID - assert "token" in result["result"].data - assert result["result"].data["token"]["access_token"] == access_token - assert result["result"].data["token"]["refresh_token"] == "mock-refresh-token" - @pytest.mark.usefixtures("current_request_with_host") async def test_full_flow_user_cred( @@ -125,24 +73,10 @@ async def test_full_flow_user_cred( ) -> None: """Check full flow.""" - # Create user application credential - assert await async_setup_component(hass, "application_credentials", {}) - await async_import_client_credential( - hass, - DOMAIN, - ClientCredential("user_client_id", "user_client_secret"), - "user_cred", - ) - result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": SOURCE_USER} ) - assert result["type"] is FlowResultType.FORM - - result = await hass.config_entries.flow.async_configure( - result["flow_id"], {"implementation": "user_cred"} - ) assert result["type"] is FlowResultType.EXTERNAL_STEP state = config_entry_oauth2_flow._encode_jwt(