diff --git a/README.md b/README.md index 1eb8a635..9a4c30e0 100644 --- a/README.md +++ b/README.md @@ -120,6 +120,43 @@ async def main(): print(f"- {device.name} at {device.ip}") +if __name__ == "__main__": + asyncio.run(main()) +``` + +### API key from the device + +Devices from 2022 onward can hand out their API key locally, after someone +presses the button on top of the device. No LaMetric account needed. An +LM 37X8 TIME does not support this. This uses the web interface of the +device, which LaMetric does not document. + +```python +"""Asynchronous Python client for LaMetric TIME devices.""" + +import asyncio + +from demetriek import LaMetricLocalAuth + + +async def main(): + """Get the API key of a LaMetric device with a press on its button.""" + async with LaMetricLocalAuth("192.168.1.2") as auth: + challenge = await auth.request_challenge() + print(f"Press the button on your LaMetric within {challenge.duration}s") + + while challenge.state == "in-progress": + await asyncio.sleep(1) + challenge = await auth.challenge(challenge_id=challenge.challenge_id) + + if not challenge.resolved: + print("The button was not pressed in time") + return + + api_key = await auth.api_key(challenge_id=challenge.challenge_id) + print(f"The API key is {api_key}") + + if __name__ == "__main__": asyncio.run(main()) ``` diff --git a/src/demetriek/__init__.py b/src/demetriek/__init__.py index d04fe676..af6a2ebe 100644 --- a/src/demetriek/__init__.py +++ b/src/demetriek/__init__.py @@ -26,11 +26,13 @@ LaMetricConnectionTimeoutError, LaMetricError, ) +from .local_auth import LaMetricLocalAuth from .models import ( API, App, AppParameter, Audio, + AuthChallenge, Bluetooth, BluetoothLowEnergy, Chart, @@ -69,6 +71,7 @@ "App", "AppParameter", "Audio", + "AuthChallenge", "Bluetooth", "BluetoothLowEnergy", "BrightnessMode", @@ -92,6 +95,7 @@ "LaMetricConnectionTimeoutError", "LaMetricDevice", "LaMetricError", + "LaMetricLocalAuth", "LaMetricStream", "Model", "Notification", diff --git a/src/demetriek/exceptions.py b/src/demetriek/exceptions.py index 39001872..05afa41d 100644 --- a/src/demetriek/exceptions.py +++ b/src/demetriek/exceptions.py @@ -25,7 +25,8 @@ def error_message(body: str) -> str | None: """Extract the error messages from an error response of the LaMetric API. Both the device and the cloud answer errors with a body like - `{"errors": [{"message": "..."}]}`. + `{"errors": [{"message": "..."}]}`. The web interface of the device + answers with a single `{"error": {"message": "..."}}` instead. Args: ---- @@ -37,7 +38,10 @@ def error_message(body: str) -> str | None: """ try: - errors = orjson.loads(body).get("errors") + data = orjson.loads(body) + errors = data.get("errors") + if isinstance(error := data.get("error"), dict): + errors = [error] except (ValueError, AttributeError): return None diff --git a/src/demetriek/local_auth.py b/src/demetriek/local_auth.py new file mode 100644 index 00000000..184c6b69 --- /dev/null +++ b/src/demetriek/local_auth.py @@ -0,0 +1,280 @@ +"""Asynchronous Python client for LaMetric TIME devices.""" + +from __future__ import annotations + +import asyncio +import socket +from dataclasses import dataclass +from http import HTTPStatus +from typing import Any, Self + +import aiohttp +from aiohttp import hdrs +from mashumaro.exceptions import MissingField +from yarl import URL + +from .exceptions import ( + LaMetricAuthenticationError, + LaMetricConnectionError, + LaMetricConnectionTimeoutError, + LaMetricError, + error_message, +) +from .models import AuthChallenge + + +@dataclass +class LaMetricLocalAuth: + """Get the API key of a LaMetric device, with a press on its button. + + This talks to the web interface of the device, which LaMetric does not + document. Devices from 2022 onward have it, an LM 37X8 TIME does not. + + Request a challenge, have someone press the button on top of the device + while it shows it, and exchange the resolved challenge for the API key. + No LaMetric account or cloud is involved. + """ + + host: str + request_timeout: float = 8.0 + session: aiohttp.client.ClientSession | None = None + + _close_session: bool = False + + async def _request( + self, + uri: str, + *, + method: str = hdrs.METH_GET, + query: dict[str, str] | None = None, + data: dict[str, Any] | None = None, + admin_key: str | None = None, + ) -> Any: + """Handle a request to the web interface of a LaMetric device. + + Args: + ---- + uri: Request URI, for example `/api/v1/user`. + method: HTTP method to use for the request. + query: Query parameters to add to the URI. + data: Dictionary of data to send to the device. + admin_key: The web admin key, for requests that need it. + + Returns: + ------- + A Python dictionary (JSON decoded) with the response. + + Raises: + ------ + LaMetricAuthenticationError: The web admin key was not accepted. + LaMetricConnectionError: An error occurred while communicating with + the LaMetric device. + LaMetricConnectionTimeoutError: A timeout occurred while communicating + with the LaMetric device. + LaMetricError: Received an unexpected response from the device. + + """ + url = URL.build(scheme="https", host=self.host, path=uri, query=query or {}) + + headers = {"Accept": "application/json"} + if admin_key is not None: + # The web interface takes the key from a cookie, not from basic auth. + headers["Cookie"] = f"no-auth-challenge=1; authorization={admin_key}:" + + if self.session is None: + self.session = aiohttp.ClientSession() + self._close_session = True + + try: + async with asyncio.timeout(self.request_timeout): + response = await self.session.request( + method, + url, + headers=headers, + json=data, + ssl=False, + ) + body = await response.text() + + if response.status >= HTTPStatus.BAD_REQUEST: + reason = error_message(body) or response.reason + if response.status == HTTPStatus.UNAUTHORIZED: + msg = ( + f"Authentication to the LaMetric device at {self.host}" + f" failed: {reason}" + ) + raise LaMetricAuthenticationError(msg) + msg = ( + f"The LaMetric device at {self.host} returned an error" + f" ({response.status}): {reason}" + ) + raise LaMetricError(msg) + + try: + return await response.json(content_type=None) + except ValueError as exception: + msg = f"The LaMetric device at {self.host} answered with invalid JSON" + raise LaMetricError(msg) from exception + + except TimeoutError as exception: + msg = ( + "Timeout occurred while connecting to the LaMetric device" + f" at {self.host}" + ) + raise LaMetricConnectionTimeoutError(msg) from exception + except (aiohttp.ClientError, socket.gaierror) as exception: + msg = ( + "Error occurred while communicating with the LaMetric device" + f" at {self.host}" + ) + raise LaMetricConnectionError(msg) from exception + + def _parse_challenge(self, data: Any) -> AuthChallenge: + """Parse a challenge answered by the device. + + Args: + ---- + data: The JSON decoded challenge. + + Returns: + ------- + The challenge. + + Raises: + ------ + LaMetricError: The challenge does not look like one. + + """ + try: + return AuthChallenge.from_dict(data) + except (MissingField, ValueError) as exception: + msg = ( + f"The LaMetric device at {self.host} answered with data this" + f" library does not understand: {exception}" + ) + raise LaMetricError(msg) from exception + + async def request_challenge(self) -> AuthChallenge: + """Ask the device to have its button pressed. + + The device shows the challenge on its screen, someone then has + `duration` seconds to press the button on top of the device. + + Returns + ------- + The challenge, to poll with `challenge()`. + + Raises + ------ + LaMetricError: The device does not support this, for example an + LM 37X8 TIME. + + """ + try: + response = await self._request( + "/api/v1/user/request", + method=hdrs.METH_POST, + query={"group": "web_admin"}, + ) + except LaMetricAuthenticationError as exception: + # A device without this flow asks for credentials instead. + msg = ( + f"The LaMetric device at {self.host} does not support getting" + " its API key with a press on its button" + ) + raise LaMetricError(msg) from exception + + challenge = response.get("challenge") if isinstance(response, dict) else None + return self._parse_challenge(challenge) + + async def challenge(self, *, challenge_id: str) -> AuthChallenge: + """Get the current state of a challenge. + + Args: + ---- + challenge_id: ID of the challenge, from `request_challenge()`. + + Returns: + ------- + The challenge, which is resolved once the button was pressed. + + """ + response = await self._request(f"/api/v1/user/challenge/{challenge_id}") + return self._parse_challenge(response) + + async def api_key(self, *, challenge_id: str) -> str: + """Exchange a resolved challenge for the API key of the device. + + Args: + ---- + challenge_id: ID of the resolved challenge. + + Returns: + ------- + The API key, to use with `LaMetricDevice`. + + Raises: + ------ + LaMetricError: The challenge is not resolved, or the device has + no API key. + + """ + response = await self._request( + "/api/v1/user/request/exchange", + method=hdrs.METH_POST, + data={"challenge_id": challenge_id}, + ) + + try: + admin_key = response["user"]["key"] + except (KeyError, TypeError) as exception: + msg = f"The LaMetric device at {self.host} did not hand out a key" + raise LaMetricError(msg) from exception + + # The exchange hands out a web admin key, which stays the same when + # the API key is regenerated. The API key itself is in the users. + users = await self._request("/api/v1/user", admin_key=admin_key) + keys = [ + user + for user in (users if isinstance(users, list) else []) + if isinstance(user, dict) + and user.get("group") == "integration" + and user.get("status") == "active" + and isinstance(user.get("key"), str) + ] + if not keys: + msg = ( + f"The LaMetric device at {self.host} has no API key yet," + " generate one in the LaMetric app first" + ) + raise LaMetricError(msg) + + # A key generated in the LaMetric app stays on the device, the one + # for the LaMetric account comes from the cloud. Prefer the local one. + keys.sort(key=lambda user: user.get("origin") != "local") + return keys[0]["key"] + + async def close(self) -> None: + """Close open client session.""" + if self.session and self._close_session: + await self.session.close() + + async def __aenter__(self) -> Self: + """Async enter. + + Returns + ------- + The LaMetricLocalAuth object. + + """ + return self + + async def __aexit__(self, *_exc_info: object) -> None: + """Async exit. + + Args: + ---- + _exc_info: Exec type. + + """ + await self.close() diff --git a/src/demetriek/models.py b/src/demetriek/models.py index 50b52ae4..642f5b2f 100644 --- a/src/demetriek/models.py +++ b/src/demetriek/models.py @@ -71,6 +71,28 @@ class BluetoothLowEnergy(DataClassORJSONMixin): connectable: bool | None = None +@dataclass(kw_only=True) +class AuthChallenge(DataClassORJSONMixin): + """Object holding a challenge to get the API key of an LaMetric device.""" + + challenge_id: str = field(metadata=field_options(alias="uuid")) + duration: int + + # Seen so far: "in-progress", "resolved", and "expired" once the time + # runs out. Kept as text, so a state not seen yet does not break parsing. + state: str + + @property + def resolved(self) -> bool: + """Return whether the button on the device has been pressed.""" + return self.state == "resolved" + + class Config(BaseConfig): + """AuthChallenge model configuration.""" + + allow_deserialization_not_by_alias = True + + @dataclass(kw_only=True) class Bluetooth(DataClassORJSONMixin): """Object holding the Bluetooth state of an LaMetric device.""" diff --git a/tests/conftest.py b/tests/conftest.py index 399b4501..90698429 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -13,7 +13,7 @@ from aioresponses import core as aioresponses_core from yarl import URL -from demetriek import LaMetricCloud, LaMetricDevice +from demetriek import LaMetricCloud, LaMetricDevice, LaMetricLocalAuth if TYPE_CHECKING: from collections.abc import AsyncGenerator, Generator @@ -103,3 +103,10 @@ async def cloud() -> AsyncGenerator[LaMetricCloud, None]: """Yield a LaMetric cloud client on a shared session.""" async with aiohttp.ClientSession() as session: yield LaMetricCloud(token="abc", session=session) # noqa: S106 + + +@pytest.fixture +async def auth() -> AsyncGenerator[LaMetricLocalAuth, None]: + """Yield a local auth client on a shared session.""" + async with aiohttp.ClientSession() as session: + yield LaMetricLocalAuth(host="127.0.0.2", session=session) diff --git a/tests/fixtures/auth_challenge_resolved_sa8.json b/tests/fixtures/auth_challenge_resolved_sa8.json new file mode 100644 index 00000000..f5a101ce --- /dev/null +++ b/tests/fixtures/auth_challenge_resolved_sa8.json @@ -0,0 +1,7 @@ +{ + "date": "Sat Oct 3 13:50:12 2026", + "duration": 60, + "state": "resolved", + "type": "press_key", + "uuid": "7b3636393566323030" +} diff --git a/tests/fixtures/auth_challenge_sa8.json b/tests/fixtures/auth_challenge_sa8.json new file mode 100644 index 00000000..8b6a0ea3 --- /dev/null +++ b/tests/fixtures/auth_challenge_sa8.json @@ -0,0 +1,9 @@ +{ + "challenge": { + "date": "Sat Oct 3 13:50:08 2026", + "duration": 60, + "state": "in-progress", + "type": "press_key", + "uuid": "7b3636393566323030" + } +} diff --git a/tests/fixtures/auth_exchange_sa8.json b/tests/fixtures/auth_exchange_sa8.json new file mode 100644 index 00000000..6aaaefe1 --- /dev/null +++ b/tests/fixtures/auth_exchange_sa8.json @@ -0,0 +1,10 @@ +{ + "user": { + "group": "web_admin", + "key": "webadminkey", + "name": "", + "origin": "local", + "status": "active", + "uuid": "7b3636393566323030" + } +} diff --git a/tests/fixtures/auth_users_sa8.json b/tests/fixtures/auth_users_sa8.json new file mode 100644 index 00000000..e69d45f6 --- /dev/null +++ b/tests/fixtures/auth_users_sa8.json @@ -0,0 +1,26 @@ +[ + { + "group": "guest", + "key": "guest", + "name": "", + "origin": "local", + "status": "inactive", + "uuid": "guestuuid" + }, + { + "group": "integration", + "key": "accountintegrationkey", + "name": "Account Integration Key", + "origin": "remote", + "status": "active", + "uuid": "integrationuuid" + }, + { + "group": "web_admin", + "key": "webadminkey", + "name": "", + "origin": "local", + "status": "active", + "uuid": "7b3636393566323030" + } +] diff --git a/tests/test_lametric.py b/tests/test_lametric.py index 6053bfd4..3e238359 100644 --- a/tests/test_lametric.py +++ b/tests/test_lametric.py @@ -311,6 +311,8 @@ async def test_http_error401_message( [ ('{"errors": [{"message": "Forbidden"}]}', "Forbidden"), ('{"errors": [{"message": "One"}, {"message": "Two"}]}', "One; Two"), + ('{"error": {"message": "Invalid auth"}}', "Invalid auth"), + ('{"error": "nonsense"}', None), ('{"errors": [{"code": 1}, "nonsense"]}', None), ('{"errors": "nonsense"}', None), ('{"errors": []}', None), diff --git a/tests/test_local_auth.py b/tests/test_local_auth.py new file mode 100644 index 00000000..fca70a58 --- /dev/null +++ b/tests/test_local_auth.py @@ -0,0 +1,255 @@ +"""Asynchronous Python client for LaMetric TIME devices. + +The fixtures are answers of the web interface of a real sa8 TIME on +firmware 3.2.6, with the keys and IDs replaced. +""" + +# pylint: disable=protected-access +import json + +import aiohttp +import pytest +from aioresponses import aioresponses +from yarl import URL + +from demetriek import ( + LaMetricAuthenticationError, + LaMetricConnectionError, + LaMetricConnectionTimeoutError, + LaMetricError, + LaMetricLocalAuth, +) + +from .conftest import load_fixture + +WEB_URL = "https://127.0.0.2" +CHALLENGE_ID = "7b3636393566323030" +REQUEST_URL = f"{WEB_URL}/api/v1/user/request?group=web_admin" +CHALLENGE_URL = f"{WEB_URL}/api/v1/user/challenge/{CHALLENGE_ID}" +EXCHANGE_URL = f"{WEB_URL}/api/v1/user/request/exchange" +USERS_URL = f"{WEB_URL}/api/v1/user" + + +async def test_request_challenge( + responses: aioresponses, auth: LaMetricLocalAuth +) -> None: + """Test requesting a challenge, which the device shows on its screen.""" + responses.post( + REQUEST_URL, status=200, body=load_fixture("auth_challenge_sa8.json") + ) + + challenge = await auth.request_challenge() + + assert challenge.challenge_id == CHALLENGE_ID + assert challenge.duration == 60 + assert challenge.state == "in-progress" + assert challenge.resolved is False + assert ("POST", URL(REQUEST_URL)) in responses.requests + + +async def test_request_challenge_not_supported( + responses: aioresponses, auth: LaMetricLocalAuth +) -> None: + """Test a device without this flow, like an LM 37X8 TIME, says so. + + It asks for credentials instead, which is no reason to ask for new ones. + """ + responses.post( + REQUEST_URL, + status=401, + body='{"errors":[{"message":"Authorization is required"}]}', + ) + + with pytest.raises(LaMetricError, match="does not support") as error: + await auth.request_challenge() + + assert error.type is LaMetricError + + +@pytest.mark.parametrize("body", ['{"nope": true}', "[]"]) +async def test_request_challenge_unexpected_data( + responses: aioresponses, auth: LaMetricLocalAuth, body: str +) -> None: + """Test an answer that is no challenge raises a LaMetricError.""" + responses.post(REQUEST_URL, status=200, body=body) + + with pytest.raises(LaMetricError, match="does not understand"): + await auth.request_challenge() + + +@pytest.mark.parametrize( + ("state", "resolved"), + [("in-progress", False), ("resolved", True), ("expired", False)], +) +async def test_challenge( + responses: aioresponses, + auth: LaMetricLocalAuth, + state: str, + resolved: bool, # noqa: FBT001 +) -> None: + """Test polling a challenge, including the states seen on a real device.""" + data = json.loads(load_fixture("auth_challenge_resolved_sa8.json")) + data["state"] = state + responses.get(CHALLENGE_URL, status=200, body=json.dumps(data)) + + challenge = await auth.challenge(challenge_id=CHALLENGE_ID) + + assert challenge.state == state + assert challenge.resolved is resolved + + +async def test_challenge_unknown( + responses: aioresponses, auth: LaMetricLocalAuth +) -> None: + """Test the reason of the device is passed on, in its own error shape.""" + responses.get( + CHALLENGE_URL, + status=500, + body='{"error":{ "message":"Failed to get info about challenge"}}', + ) + + with pytest.raises(LaMetricError, match="Failed to get info about challenge"): + await auth.challenge(challenge_id=CHALLENGE_ID) + + +async def test_api_key(responses: aioresponses, auth: LaMetricLocalAuth) -> None: + """Test exchanging a resolved challenge for the API key.""" + responses.post( + EXCHANGE_URL, status=200, body=load_fixture("auth_exchange_sa8.json") + ) + responses.get(USERS_URL, status=200, body=load_fixture("auth_users_sa8.json")) + + assert await auth.api_key(challenge_id=CHALLENGE_ID) == "accountintegrationkey" + + exchange = responses.requests[("POST", URL(EXCHANGE_URL))][0] + assert exchange.kwargs["json"] == {"challenge_id": CHALLENGE_ID} + + # The web interface takes the web admin key from a cookie. + users = responses.requests[("GET", URL(USERS_URL))][0] + assert users.kwargs["headers"]["Cookie"] == ( + "no-auth-challenge=1; authorization=webadminkey:" + ) + + +async def test_api_key_prefers_local( + responses: aioresponses, auth: LaMetricLocalAuth +) -> None: + """Test a key generated in the LaMetric app wins over the account one.""" + users = json.loads(load_fixture("auth_users_sa8.json")) + users.append( + { + "group": "integration", + "key": "localkey", + "name": "", + "origin": "local", + "status": "active", + "uuid": "localuuid", + } + ) + responses.post( + EXCHANGE_URL, status=200, body=load_fixture("auth_exchange_sa8.json") + ) + responses.get(USERS_URL, status=200, body=json.dumps(users)) + + assert await auth.api_key(challenge_id=CHALLENGE_ID) == "localkey" + + +async def test_api_key_none(responses: aioresponses, auth: LaMetricLocalAuth) -> None: + """Test a device without any API key yet says what to do.""" + users = [ + user + for user in json.loads(load_fixture("auth_users_sa8.json")) + if user["group"] != "integration" + ] + responses.post( + EXCHANGE_URL, status=200, body=load_fixture("auth_exchange_sa8.json") + ) + responses.get(USERS_URL, status=200, body=json.dumps(users)) + + with pytest.raises(LaMetricError, match="generate one in the LaMetric app"): + await auth.api_key(challenge_id=CHALLENGE_ID) + + +async def test_api_key_not_resolved( + responses: aioresponses, auth: LaMetricLocalAuth +) -> None: + """Test exchanging before the button was pressed passes on the reason.""" + responses.post( + EXCHANGE_URL, + status=400, + body='{"error":{ "message":"Invalid challenge status in-progress"}}', + ) + + with pytest.raises(LaMetricError, match="Invalid challenge status in-progress"): + await auth.api_key(challenge_id=CHALLENGE_ID) + + +async def test_api_key_no_key_handed_out( + responses: aioresponses, auth: LaMetricLocalAuth +) -> None: + """Test an exchange without a key raises a LaMetricError.""" + responses.post(EXCHANGE_URL, status=200, body='{"user": {}}') + + with pytest.raises(LaMetricError, match="did not hand out a key"): + await auth.api_key(challenge_id=CHALLENGE_ID) + + +async def test_api_key_admin_key_refused( + responses: aioresponses, auth: LaMetricLocalAuth +) -> None: + """Test a refused web admin key raises an authentication error.""" + responses.post( + EXCHANGE_URL, status=200, body=load_fixture("auth_exchange_sa8.json") + ) + responses.get(USERS_URL, status=401, body='{"error":{ "message":"Invalid auth"}}') + + with pytest.raises(LaMetricAuthenticationError, match="Invalid auth"): + await auth.api_key(challenge_id=CHALLENGE_ID) + + +async def test_invalid_json(responses: aioresponses, auth: LaMetricLocalAuth) -> None: + """Test a broken JSON answer raises a LaMetricError.""" + responses.get(CHALLENGE_URL, status=200, body="{") + + with pytest.raises(LaMetricError, match="invalid JSON"): + await auth.challenge(challenge_id=CHALLENGE_ID) + + +@pytest.mark.parametrize( + ("exception", "expected"), + [ + (TimeoutError(), LaMetricConnectionTimeoutError), + (aiohttp.ClientError(), LaMetricConnectionError), + ], +) +async def test_connection_errors( + responses: aioresponses, + auth: LaMetricLocalAuth, + exception: Exception, + expected: type[Exception], +) -> None: + """Test connection problems raise connection errors, without retrying. + + Requesting a challenge twice would show it twice on the device. + """ + responses.post(REQUEST_URL, exception=exception, repeat=True) + + with pytest.raises(expected): + await auth.request_challenge() + + assert len(responses.requests[("POST", URL(REQUEST_URL))]) == 1 + + +async def test_internal_session(responses: aioresponses) -> None: + """Test the client creates and closes its own session.""" + responses.get( + CHALLENGE_URL, status=200, body=load_fixture("auth_challenge_resolved_sa8.json") + ) + + async with LaMetricLocalAuth(host="127.0.0.2") as auth: + challenge = await auth.challenge(challenge_id=CHALLENGE_ID) + session = auth.session + + assert challenge.resolved is True + assert session is not None + assert session.closed