Initial commit for device auth

Author: erwindouna

README.md

@@ -64,6 +64,25 @@ based on the following:
 - `MINOR`: Backwards-compatible new features and enhancements.
 - `PATCH`: Backwards-compatible bugfixes and package updates.
 
+## Usage
+
+As of the 15th of March 2025, Tado has updated their OAuth2 authentication flow. It will now use the device flow, instead of a username/password flow. This means that the user will have to authenticate the device using a browser, and then enter the code that is displayed on the browser into the terminal.
+
+PyTado handles this as following:
+
+1. The `_login_device_flow()` will be invoked at the initialization of a PyTado object. This will start the device flow and will return a URL and a code that the user will have to enter in the browser. The URL can be obtained via the method `device_verification_url()`. Or, when in debug mode, the URL will be printed. Alternatively, you can use the `device_activation_status()` method to check if the device has been activated. It returns three statuses: `NOT_STARTED`, `PENDING`, and `COMPLETED`. Wait to invoke the `device_activation()` method until the status is `PENDING`.
+
+2. Once the URL is obtained, the user will have to enter the code that is displayed on the browser into the terminal. By default, the URL has the `user_code` attached, for the ease of going trough the flow. At this point, run the method `device_activation()`. It will poll every five seconds to see if the flow has been completed. If the flow has been completed, the method will return a token that will be used for all further requests. It will timeout after five minutes.
+
+3. Once the token has been obtained, the user can use the PyTado object to interact with the Tado API. The token will be stored in the `Tado` object, and will be used for all further requests. The token will be refreshed automatically when it expires.
+The `device_verification_url()` will be reset to `None` and the `device_activation_status()` will return `COMPLETED`.
+
+### Screenshots of the device flow
+
+![Tado device flow: invoking](/screenshots/tado-device-flow-0.png)
+![Tado device flow: browser](/screenshots/tado-device-flow-1.png)
+![Tado device flow: complete](/screenshots/tado-device-flow-2.png)
+
 ## Contributing
 
 This is an active open-source project. We are always open to people who want to

examples/example.py

@@ -1,14 +1,27 @@
 """Asynchronous Python client for the Tado API. This is an example file."""
 
 import asyncio
+import logging
 
 from tadoasync import Tado
 
+logging.basicConfig(level=logging.DEBUG)
+
 
 async def main() -> None:
     """Show example on how to use aiohttp.ClientSession."""
-    async with Tado("username", "password") as tado:
-        await tado.get_devices()
+    async with Tado(debug=True) as tado:
+        print("Device activation status: ", tado.device_activation_status)  # noqa: T201
+        print("Device verification URL: ", tado.device_verification_url)  # noqa: T201
+
+        print("Starting device activation")  # noqa: T201
+        await tado.device_activation()
+
+        print("Device activation status: ", tado.device_activation_status)  # noqa: T201
+
+        devices = await tado.get_devices()
+
+        print("Devices: ", devices)  # noqa: T201
 
 
 if __name__ == "__main__":

screenshots/tado-device-flow-0.png

@@ -3,11 +3,14 @@
 from __future__ import annotations
 
 import asyncio
+import enum
+import logging
 import time
 from dataclasses import dataclass
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from importlib import metadata
 from typing import Self
+from urllib.parse import urlencode
 
 import orjson
 from aiohttp import ClientResponseError
@@ -54,10 +57,9 @@
     ZoneState,
 )
 
-CLIENT_ID = "tado-web-app"
-CLIENT_SECRET = "wZaRN7rpjn3FoNyF5IFuxg9uMzYJcvOoQ8QWiIqS3hfk6gLhVlG57j5YNoZL2Rtc"  # noqa: S105
-AUTHORIZATION_BASE_URL = "https://auth.tado.com/oauth/authorize"
-TOKEN_URL = "https://auth.tado.com/oauth/token"  # noqa: S105
+CLIENT_ID = "1bb50063-6b0c-4d11-bd99-387f4a91cc46"
+TOKEN_URL = "https://login.tado.com/oauth2/token"  # noqa: S105
+DEVICE_AUTH_URL = "https://login.tado.com/oauth2/device_authorize"
 API_URL = "my.tado.com/api/v2"
 TADO_HOST_URL = "my.tado.com"
 TADO_API_PATH = "/api/v2"
@@ -66,29 +68,28 @@
 EIQ_API_PATH = "/api"
 VERSION = metadata.version(__package__)
 
+_LOGGER = logging.getLogger(__name__)
+
+
+class DeviceActivationStatus(enum.StrEnum):
+    """Device Activation Status Enum."""
+
+    NOT_STARTED = "NOT_STARTED"
+    PENDING = "PENDING"
+    COMPLETED = "COMPLETED"
+
 
 @dataclass
 class Tado:  # pylint: disable=too-many-instance-attributes
     """Base class for Tado."""
 
     def __init__(
         self,
-        username: str,
-        password: str,
         debug: bool | None = None,
         session: ClientSession | None = None,
         request_timeout: int = 10,
     ) -> None:
-        """Initialize the Tado object.
-
-        :param username: Tado account username.
-        :param password: Tado account password.
-        :param debug: Enable debug logging.
-        :param session: HTTP client session.
-        :param request_timeout: Timeout for HTTP requests.
-        """
-        self._username: str = username
-        self._password: str = password
+        """Initialize the Tado object."""
         self._debug: bool = debug or False
         self._session = session
         self._request_timeout = request_timeout
@@ -107,15 +108,151 @@ def __init__(
         self._me: GetMe | None = None
         self._auto_geofencing_supported: bool | None = None
 
+        self._user_code: str | None = None
+        self._device_verification_url: str | None = None
+        self._device_flow_data: dict[str, str] = {}
+        self._device_activation_status = DeviceActivationStatus.NOT_STARTED
+        self._expires_at: datetime | None = None
+
+        _LOGGER.setLevel(logging.DEBUG if debug else logging.INFO)
+
+    async def async_init(self) -> None:
+        """Asynchronous initialization for the Tado object."""
+        if self._refresh_token is None:
+            self._device_activation_status = await self.login_device_flow()
+        else:
+            self._device_ready()
+
+    @property
+    def device_activation_status(self) -> DeviceActivationStatus:
+        """Return the device activation status."""
+        return self._device_activation_status
+
+    @property
+    def device_verification_url(self) -> str | None:
+        """Return the device verification URL."""
+        return self._device_verification_url
+
+    async def login_device_flow(self) -> DeviceActivationStatus:
+        """Login using device flow."""
+        if self._device_activation_status != DeviceActivationStatus.NOT_STARTED:
+            raise TadoError("Device activation already in progress or completed")
+
+        data = {
+            "client_id": CLIENT_ID,
+            "scope": "offline_access",
+        }
+
+        if self._session is None:
+            self._session = ClientSession()
+            self._close_session = True
+
+        try:
+            async with asyncio.timeout(self._request_timeout):
+                request = await self._session.post(url=DEVICE_AUTH_URL, data=data)
+                request.raise_for_status()
+        except asyncio.TimeoutError as err:
+            raise TadoConnectionError(
+                "Timeout occurred while connecting to Tado."
+            ) from err
+        except ClientResponseError as err:
+            await self.check_request_status(err, login=True)
+
+        if request.status != 200:
+            raise TadoError(f"Failed to start device activation flow: {request.status}")
+
+        self._device_flow_data = await request.json()
+
+        user_code = urlencode({"user_code": self._device_flow_data["user_code"]})
+        visit_url = f"{self._device_flow_data['verification_uri']}?{user_code}"
+        self._user_code = self._device_flow_data["user_code"]
+        self._device_verification_url = visit_url
+
+        _LOGGER.info("Please visit the following URL: %s", visit_url)
+
+        expires_in_seconds = float(self._device_flow_data["expires_in"])
+        self._expires_at = datetime.now(timezone.utc) + timedelta(
+            seconds=expires_in_seconds
+        )
+
+        _LOGGER.info(
+            "Waiting for user to authorize the device. Expires at %s",
+            self._expires_at.strftime("%Y-%m-%d %H:%M:%S"),
+        )
+
+        return DeviceActivationStatus.PENDING
+
+    async def _check_device_activation(self) -> bool:
+        if self._expires_at is not None and datetime.timestamp(
+            datetime.now(timezone.utc)
+        ) > datetime.timestamp(self._expires_at):
+            raise TadoError("User took too long to enter key")
+
+        # Await the desired interval, before polling the API again
+        await asyncio.sleep(float(self._device_flow_data["interval"]))
+
+        data = {
+            "client_id": CLIENT_ID,
+            "device_code": self._device_flow_data["device_code"],
+            "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
+        }
+
+        if self._session is None:
+            self._session = ClientSession()
+            self._close_session = True
+
+        try:
+            async with asyncio.timeout(self._request_timeout):
+                request = await self._session.post(url=TOKEN_URL, data=data)
+                if request.status == 400:
+                    response = await request.json()
+                    if response.get("error") == "authorization_pending":
+                        _LOGGER.info("Authorization pending. Continuing polling...")
+                        return False
+                request.raise_for_status()
+        except asyncio.TimeoutError as err:
+            raise TadoConnectionError(
+                "Timeout occurred while connecting to Tado."
+            ) from err
+
+        if request.status == 200:
+            response = await request.json()
+            self._access_token = response["access_token"]
+            self._token_expiry = time.time() + float(response["expires_in"])
+            self._refresh_token = response["refresh_token"]
+
+            get_me = await self.get_me()
+            self._home_id = get_me.homes[0].id
+
+            return True
+
+        raise TadoError(f"Login failed. Reason: {request.reason}")
+
+    async def device_activation(self) -> None:
+        """Start the device activation process and get the refresh token."""
+        if self._device_activation_status == DeviceActivationStatus.NOT_STARTED:
+            raise TadoError(
+                "Device activation has not yet started or has already completed"
+            )
+
+        while True:
+            if await self._check_device_activation():
+                break
+
+        self._device_ready()
+
+    def _device_ready(self) -> None:
+        """Clear up after device activation."""
+        self._user_code = None
+        self._device_verification_url = None
+        self._device_activation_status = DeviceActivationStatus.COMPLETED
+
     async def login(self) -> None:
         """Perform login to Tado."""
         data = {
             "client_id": CLIENT_ID,
-            "client_secret": CLIENT_SECRET,
             "grant_type": "password",
             "scope": "home.user",
-            "username": self._username,
-            "password": self._password,
         }
 
         if self._session is None:
@@ -189,7 +326,6 @@ async def _refresh_auth(self) -> None:
 
         data = {
             "client_id": CLIENT_ID,
-            "client_secret": CLIENT_SECRET,
             "grant_type": "refresh_token",
             "scope": "home.user",
             "refresh_token": self._refresh_token,
@@ -610,7 +746,7 @@ async def close(self) -> None:
 
     async def __aenter__(self) -> Self:
         """Async enter."""
-        await self.login()
+        await self.async_init()
         return self
 
     async def __aexit__(self, *_exc_info: object) -> None:

screenshots/tado-device-flow-1.png

@@ -10,7 +10,7 @@
 from syrupy import SnapshotAssertion
 from tests import load_fixture
 
-from .const import TADO_API_URL, TADO_TOKEN_URL
+from .const import TADO_API_URL, TADO_DEVICE_AUTH_URL, TADO_TOKEN_URL
 from .syrupy import TadoSnapshotExtension
 
 
@@ -24,8 +24,6 @@ def snapshot_assertion(snapshot: SnapshotAssertion) -> SnapshotAssertion:
 async def client() -> AsyncGenerator[Tado, None]:
     """Return a Tado client."""
     async with aiohttp.ClientSession() as session, Tado(
-        username="username",
-        password="password",
         session=session,
         request_timeout=10,
     ) as tado:
@@ -35,6 +33,18 @@ async def client() -> AsyncGenerator[Tado, None]:
 @pytest.fixture(autouse=True)
 def _tado_oauth(responses: aioresponses) -> None:
     """Mock the Tado token URL."""
+    responses.post(
+        TADO_DEVICE_AUTH_URL,
+        status=200,
+        payload={
+            "device_code": "XXX_code_XXX",
+            "expires_in": 300,
+            "interval": 5,
+            "user_code": "7BQ5ZQ",
+            "verification_uri": "https://login.tado.com/oauth2/device",
+            "verification_uri_complete": "https://login.tado.com/oauth2/device?user_code=7BQ5ZQ",
+        },
+    )
     responses.post(
         TADO_TOKEN_URL,
         status=200,

screenshots/tado-device-flow-2.png

@@ -1,6 +1,7 @@
 """Constants for tests of Python Tado."""
 
 TADO_API_URL = "https://my.tado.com/api/v2"
-TADO_TOKEN_URL = "https://auth.tado.com/oauth/token"
+TADO_TOKEN_URL = "https://login.tado.com/oauth2/token"
+TADO_DEVICE_AUTH_URL = "https://login.tado.com/oauth2/device_authorize"
 
 TADO_EIQ_URL = "https://energy-insights.tado.com/api"

src/tadoasync/tadoasync.py

@@ -38,7 +38,7 @@ async def test_create_session(
         body=load_fixture("me.json"),
     )
     async with aiohttp.ClientSession():
-        tado = Tado(username="username", password="password")
+        tado = Tado()
         await tado.get_me()
         assert tado._session is not None
         assert not tado._session.closed
@@ -48,7 +48,7 @@ async def test_create_session(
 
 async def test_close_session() -> None:
     """Test not closing the session when the session does not exist."""
-    tado = Tado(username="username", password="password")
+    tado = Tado()
     tado._close_session = True
     await tado.close()
 
@@ -61,7 +61,7 @@ async def test_login_success(responses: aioresponses) -> None:
         body=load_fixture("me.json"),
     )
     async with aiohttp.ClientSession() as session:
-        tado = Tado(username="username", password="password", session=session)
+        tado = Tado(session=session)
         await tado.login()
         assert tado._access_token == "test_access_token"
         assert tado._token_expiry is not None
@@ -77,7 +77,7 @@ async def test_login_success_no_session(responses: aioresponses) -> None:
         body=load_fixture("me.json"),
     )
     async with aiohttp.ClientSession():
-        tado = Tado(username="username", password="password")
+        tado = Tado()
         await tado.login()
         assert tado._access_token == "test_access_token"
         assert tado._token_expiry is not None
@@ -142,7 +142,7 @@ async def test_refresh_auth_success(responses: aioresponses) -> None:
         headers={"content-type": "application/json"},
     )
     async with aiohttp.ClientSession() as session:
-        tado = Tado(username="username", password="password", session=session)
+        tado = Tado(session=session)
         tado._access_token = "old_test_access_token"
         tado._token_expiry = time.time() - 10  # make sure the token is expired
         tado._refresh_token = "old_test_refresh_token"
@@ -603,7 +603,7 @@ async def response_handler(_: str, **_kwargs: Any) -> CallbackResult:  # pylint:
     )
 
     async with aiohttp.ClientSession() as session, Tado(
-        username="username", password="password", request_timeout=0, session=session
+        request_timeout=0, session=session
     ) as tado:
         with pytest.raises(TadoConnectionError):
             assert await tado.get_devices()